This is an automated email from the ASF dual-hosted git repository. szaszm pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit fb9cfe38fcf012f626f605d551560fa90d33e597 Author: Ferenc Gerlits <[email protected]> AuthorDate: Wed Aug 2 18:00:45 2023 +0200 MINIFICPP-2134 Do not swallow parsing exceptions If the value of a typed property (Time Period, Data Size, Port etc) is set, but to an invalid value, then we should throw in getProperty<T>(). In some cases, we were using a typed getProperty<T>(), but the corresponding PropertyDefinition did not have a PropertyType, so the C2 server had no way of validating the property. In these cases, I have kept the current behavior, which is to treat the invalid value as if no value was set -- but I have also added the PropertyType to the PropertyDefinition so we can add validation in a later release. Additionally, the Max Poll Records property of ConsumeKafka used to internally treat the value as uint64, but in the manifest we sent a uint32 validator; I have changed the validator to be uint64 ("unsigned long"), as well. Closes #1610 Signed-off-by: Marton Szasz <[email protected]> --- extensions/http-curl/processors/InvokeHTTP.cpp | 9 +++-- extensions/http-curl/processors/InvokeHTTP.h | 8 ++-- extensions/librdkafka/ConsumeKafka.h | 2 +- .../mqtt/processors/AbstractMQTTProcessor.cpp | 6 +-- extensions/mqtt/processors/AbstractMQTTProcessor.h | 3 ++ extensions/mqtt/processors/ConsumeMQTT.cpp | 13 ++++--- extensions/mqtt/processors/ConsumeMQTT.h | 6 +++ extensions/mqtt/processors/PublishMQTT.cpp | 4 +- extensions/mqtt/processors/PublishMQTT.h | 2 + extensions/sftp/processors/ListSFTP.cpp | 2 +- extensions/sftp/processors/ListSFTP.h | 7 ++-- .../processors/DefragmentText.cpp | 2 +- .../processors/DefragmentText.h | 1 + .../standard-processors/processors/ListFile.cpp | 2 +- .../standard-processors/processors/ListFile.h | 3 +- libminifi/include/core/ProcessContext.h | 16 +++----- libminifi/include/utils/ValueParser.h | 31 ++++++++-------- libminifi/test/unit/ValueParserTests.cpp | 43 ++++++++++++++++++++++ 18 files changed, 108 insertions(+), 52 deletions(-) diff --git a/extensions/http-curl/processors/InvokeHTTP.cpp b/extensions/http-curl/processors/InvokeHTTP.cpp index 75f70b19f..b9f1799c4 100644 --- a/extensions/http-curl/processors/InvokeHTTP.cpp +++ b/extensions/http-curl/processors/InvokeHTTP.cpp @@ -71,8 +71,9 @@ void setupClientProxy(extensions::curl::HTTPClient& client, const core::ProcessC } void setupClientPeerVerification(extensions::curl::HTTPClient& client, const core::ProcessContext& context) { - if (auto disable_peer_verification = context.getProperty<bool>(InvokeHTTP::DisablePeerVerification)) + if (auto disable_peer_verification = context.getProperty(InvokeHTTP::DisablePeerVerification) | utils::flatMap(&utils::StringUtils::toBool)) { client.setPeerVerification(*disable_peer_verification); + } } void setupClientFollowRedirects(extensions::curl::HTTPClient& client, const core::ProcessContext& context) { @@ -104,8 +105,8 @@ void InvokeHTTP::setupMembersFromProperties(const core::ProcessContext& context) | utils::orElse([this] { logger_->log_debug("%s is missing, so the default value will be used", std::string{AttributesToSend.name}); }); - always_output_response_ = context.getProperty<bool>(AlwaysOutputResponse).value_or(false); - penalize_no_retry_ = context.getProperty<bool>(PenalizeOnNoRetry).value_or(false); + always_output_response_ = (context.getProperty(AlwaysOutputResponse) | utils::flatMap(&utils::StringUtils::toBool)).value_or(false); + penalize_no_retry_ = (context.getProperty(PenalizeOnNoRetry) | utils::flatMap(&utils::StringUtils::toBool)).value_or(false); invalid_http_header_field_handling_strategy_ = utils::parseEnumProperty<invoke_http::InvalidHTTPHeaderFieldHandlingOption>(context, InvalidHTTPHeaderFieldHandlingStrategy); @@ -115,7 +116,7 @@ void InvokeHTTP::setupMembersFromProperties(const core::ProcessContext& context) put_response_body_in_attribute_.reset(); } - use_chunked_encoding_ = context.getProperty<bool>(UseChunkedEncoding).value_or(false); + use_chunked_encoding_ = (context.getProperty(UseChunkedEncoding) | utils::flatMap(&utils::StringUtils::toBool)).value_or(false); send_date_header_ = context.getProperty<bool>(DateHeader).value_or(true); } diff --git a/extensions/http-curl/processors/InvokeHTTP.h b/extensions/http-curl/processors/InvokeHTTP.h index 6d5e72667..c9d291a34 100644 --- a/extensions/http-curl/processors/InvokeHTTP.h +++ b/extensions/http-curl/processors/InvokeHTTP.h @@ -104,9 +104,7 @@ class InvokeHTTP : public core::Processor { .withDescription("Regular expression that defines which attributes to send as HTTP headers in the request. If not defined, no attributes are sent as headers.") .build(); EXTENSIONAPI static constexpr auto SSLContext = core::PropertyDefinitionBuilder<0, 1, 0, 1>::createProperty("SSL Context Service") - .withDescription( - "The SSL Context Service used to provide client certificate " - "information for TLS/SSL (https) connections.") + .withDescription("The SSL Context Service used to provide client certificate information for TLS/SSL (https) connections.") .isRequired(false) .withAllowedTypes({core::className<minifi::controllers::SSLContextService>()}) .withExclusiveOfProperties({{{"Remote URL", "^http:.*$"}}}) @@ -147,10 +145,12 @@ class InvokeHTTP : public core::Processor { .withDescription("When POST'ing, PUT'ing or PATCH'ing content set this property to true in order to not pass the 'Content-length' header" " and instead send 'Transfer-Encoding' with a value of 'chunked'." " This will enable the data transfer mechanism which was introduced in HTTP 1.1 to pass data of unknown lengths in chunks.") + .withPropertyType(core::StandardPropertyTypes::BOOLEAN_TYPE) .withDefaultValue("false") .build(); EXTENSIONAPI static constexpr auto DisablePeerVerification = core::PropertyDefinitionBuilder<>::createProperty("Disable Peer Verification") .withDescription("Disables peer verification for the SSL session") + .withPropertyType(core::StandardPropertyTypes::BOOLEAN_TYPE) .withDefaultValue("false") .build(); EXTENSIONAPI static constexpr auto PutResponseBodyInAttribute = core::PropertyDefinitionBuilder<>::createProperty("Put Response Body in Attribute") @@ -160,10 +160,12 @@ class InvokeHTTP : public core::Processor { .build(); EXTENSIONAPI static constexpr auto AlwaysOutputResponse = core::PropertyDefinitionBuilder<>::createProperty("Always Output Response") .withDescription("Will force a response FlowFile to be generated and routed to the 'Response' relationship regardless of what the server status code received is ") + .withPropertyType(core::StandardPropertyTypes::BOOLEAN_TYPE) .withDefaultValue("false") .build(); EXTENSIONAPI static constexpr auto PenalizeOnNoRetry = core::PropertyDefinitionBuilder<>::createProperty("Penalize on \"No Retry\"") .withDescription("Enabling this property will penalize FlowFiles that are routed to the \"No Retry\" relationship.") + .withPropertyType(core::StandardPropertyTypes::BOOLEAN_TYPE) .withDefaultValue("false") .build(); EXTENSIONAPI static constexpr auto InvalidHTTPHeaderFieldHandlingStrategy diff --git a/extensions/librdkafka/ConsumeKafka.h b/extensions/librdkafka/ConsumeKafka.h index 07e64bec2..6095d6d9b 100644 --- a/extensions/librdkafka/ConsumeKafka.h +++ b/extensions/librdkafka/ConsumeKafka.h @@ -172,7 +172,7 @@ class ConsumeKafka : public KafkaProcessorBase { .build(); EXTENSIONAPI static constexpr auto MaxPollRecords = core::PropertyDefinitionBuilder<>::createProperty("Max Poll Records") .withDescription("Specifies the maximum number of records Kafka should return when polling each time the processor is triggered.") - .withPropertyType(core::StandardPropertyTypes::UNSIGNED_INT_TYPE) + .withPropertyType(core::StandardPropertyTypes::UNSIGNED_LONG_TYPE) .withDefaultValue(DEFAULT_MAX_POLL_RECORDS) .build(); EXTENSIONAPI static constexpr auto MaxPollTime = core::PropertyDefinitionBuilder<>::createProperty("Max Poll Time") diff --git a/extensions/mqtt/processors/AbstractMQTTProcessor.cpp b/extensions/mqtt/processors/AbstractMQTTProcessor.cpp index 8d20b41e9..06ecf3d3e 100644 --- a/extensions/mqtt/processors/AbstractMQTTProcessor.cpp +++ b/extensions/mqtt/processors/AbstractMQTTProcessor.cpp @@ -51,12 +51,12 @@ void AbstractMQTTProcessor::onSchedule(const std::shared_ptr<core::ProcessContex } logger_->log_debug("AbstractMQTTProcessor: Password [%s]", password_); - if (const auto keep_alive_interval = context->getProperty<core::TimePeriodValue>(KeepAliveInterval)) { + if (const auto keep_alive_interval = context->getProperty(KeepAliveInterval) | utils::flatMap(&core::TimePeriodValue::fromString)) { keep_alive_interval_ = std::chrono::duration_cast<std::chrono::seconds>(keep_alive_interval->getMilliseconds()); } logger_->log_debug("AbstractMQTTProcessor: KeepAliveInterval [%" PRId64 "] s", int64_t{keep_alive_interval_.count()}); - if (const auto connection_timeout = context->getProperty<core::TimePeriodValue>(ConnectionTimeout)) { + if (const auto connection_timeout = context->getProperty(ConnectionTimeout) | utils::flatMap(&core::TimePeriodValue::fromString)) { connection_timeout_ = std::chrono::duration_cast<std::chrono::seconds>(connection_timeout->getMilliseconds()); } logger_->log_debug("AbstractMQTTProcessor: ConnectionTimeout [%" PRId64 "] s", int64_t{connection_timeout_.count()}); @@ -107,7 +107,7 @@ void AbstractMQTTProcessor::onSchedule(const std::shared_ptr<core::ProcessContex logger_->log_debug("AbstractMQTTProcessor: Last Will QoS [%u]", static_cast<uint8_t>(last_will_qos_)); last_will_->qos = static_cast<int>(last_will_qos_); - if (const auto value = context->getProperty<bool>(LastWillRetain)) { + if (const auto value = context->getProperty(LastWillRetain) | utils::flatMap(&utils::StringUtils::toBool)) { logger_->log_debug("AbstractMQTTProcessor: Last Will Retain [%d]", *value); last_will_retain_ = {*value}; last_will_->retained = last_will_retain_; diff --git a/extensions/mqtt/processors/AbstractMQTTProcessor.h b/extensions/mqtt/processors/AbstractMQTTProcessor.h index 3e8b8ee85..79049ef20 100644 --- a/extensions/mqtt/processors/AbstractMQTTProcessor.h +++ b/extensions/mqtt/processors/AbstractMQTTProcessor.h @@ -115,10 +115,12 @@ class AbstractMQTTProcessor : public core::Processor { .build(); EXTENSIONAPI static constexpr auto ConnectionTimeout = core::PropertyDefinitionBuilder<>::createProperty("Connection Timeout") .withDescription("Maximum time interval the client will wait for the network connection to the MQTT broker") + .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE) .withDefaultValue("10 sec") .build(); EXTENSIONAPI static constexpr auto KeepAliveInterval = core::PropertyDefinitionBuilder<>::createProperty("Keep Alive Interval") .withDescription("Defines the maximum time interval between messages sent or received") + .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE) .withDefaultValue("60 sec") .build(); EXTENSIONAPI static constexpr auto LastWillTopic = core::PropertyDefinitionBuilder<>::createProperty("Last Will Topic") @@ -135,6 +137,7 @@ class AbstractMQTTProcessor : public core::Processor { .build(); EXTENSIONAPI static constexpr auto LastWillRetain = core::PropertyDefinitionBuilder<>::createProperty("Last Will Retain") .withDescription("Whether to retain the client's Last Will") + .withPropertyType(core::StandardPropertyTypes::BOOLEAN_TYPE) .withDefaultValue("false") .build(); EXTENSIONAPI static constexpr auto LastWillContentType = core::PropertyDefinitionBuilder<>::createProperty("Last Will Content Type") diff --git a/extensions/mqtt/processors/ConsumeMQTT.cpp b/extensions/mqtt/processors/ConsumeMQTT.cpp index f8be50268..54d497063 100644 --- a/extensions/mqtt/processors/ConsumeMQTT.cpp +++ b/extensions/mqtt/processors/ConsumeMQTT.cpp @@ -23,6 +23,7 @@ #include <vector> #include "utils/StringUtils.h" +#include "utils/ValueParser.h" #include "core/ProcessContext.h" #include "core/ProcessSession.h" #include "core/Resource.h" @@ -50,22 +51,22 @@ void ConsumeMQTT::readProperties(const std::shared_ptr<core::ProcessContext>& co } logger_->log_debug("ConsumeMQTT: Topic [%s]", topic_); - if (const auto value = context->getProperty<bool>(CleanSession)) { + if (const auto value = context->getProperty(CleanSession) | utils::flatMap(&utils::StringUtils::toBool)) { clean_session_ = *value; } logger_->log_debug("ConsumeMQTT: CleanSession [%d]", clean_session_); - if (const auto value = context->getProperty<bool>(CleanStart)) { + if (const auto value = context->getProperty(CleanStart) | utils::flatMap(&utils::StringUtils::toBool)) { clean_start_ = *value; } logger_->log_debug("ConsumeMQTT: CleanStart [%d]", clean_start_); - if (const auto session_expiry_interval = context->getProperty<core::TimePeriodValue>(SessionExpiryInterval)) { + if (const auto session_expiry_interval = context->getProperty(SessionExpiryInterval) | utils::flatMap(&core::TimePeriodValue::fromString)) { session_expiry_interval_ = std::chrono::duration_cast<std::chrono::seconds>(session_expiry_interval->getMilliseconds()); } logger_->log_debug("ConsumeMQTT: SessionExpiryInterval [%" PRId64 "] s", int64_t{session_expiry_interval_.count()}); - if (const auto value = context->getProperty<uint64_t>(QueueBufferMaxMessage)) { + if (const auto value = context->getProperty(QueueBufferMaxMessage) | utils::flatMap(&utils::toNumber<uint64_t>)) { max_queue_size_ = *value; } logger_->log_debug("ConsumeMQTT: Queue Max Message [%" PRIu64 "]", max_queue_size_); @@ -75,12 +76,12 @@ void ConsumeMQTT::readProperties(const std::shared_ptr<core::ProcessContext>& co } logger_->log_debug("ConsumeMQTT: Attribute From Content Type [%s]", attribute_from_content_type_); - if (const auto topic_alias_maximum = context->getProperty<uint32_t>(TopicAliasMaximum)) { + if (const auto topic_alias_maximum = context->getProperty(TopicAliasMaximum) | utils::flatMap(&utils::toNumber<uint32_t>)) { topic_alias_maximum_ = gsl::narrow<uint16_t>(*topic_alias_maximum); } logger_->log_debug("ConsumeMQTT: Topic Alias Maximum [%" PRIu16 "]", topic_alias_maximum_); - if (const auto receive_maximum = context->getProperty<uint32_t>(ReceiveMaximum)) { + if (const auto receive_maximum = context->getProperty(ReceiveMaximum) | utils::flatMap(&utils::toNumber<uint32_t>)) { receive_maximum_ = gsl::narrow<uint16_t>(*receive_maximum); } logger_->log_debug("ConsumeMQTT: Receive Maximum [%" PRIu16 "]", receive_maximum_); diff --git a/extensions/mqtt/processors/ConsumeMQTT.h b/extensions/mqtt/processors/ConsumeMQTT.h index 02c899ea7..670587073 100644 --- a/extensions/mqtt/processors/ConsumeMQTT.h +++ b/extensions/mqtt/processors/ConsumeMQTT.h @@ -52,18 +52,22 @@ class ConsumeMQTT : public processors::AbstractMQTTProcessor { .build(); EXTENSIONAPI static constexpr auto CleanSession = core::PropertyDefinitionBuilder<>::createProperty("Clean Session") .withDescription("Whether to start afresh rather than remembering previous subscriptions. If true, then make broker forget subscriptions after disconnected. MQTT 3.x only.") + .withPropertyType(core::StandardPropertyTypes::BOOLEAN_TYPE) .withDefaultValue("true") .build(); EXTENSIONAPI static constexpr auto CleanStart = core::PropertyDefinitionBuilder<>::createProperty("Clean Start") .withDescription("Whether to start afresh rather than remembering previous subscriptions. MQTT 5.x only.") + .withPropertyType(core::StandardPropertyTypes::BOOLEAN_TYPE) .withDefaultValue("true") .build(); EXTENSIONAPI static constexpr auto SessionExpiryInterval = core::PropertyDefinitionBuilder<>::createProperty("Session Expiry Interval") .withDescription("Time to delete session on broker after client is disconnected. MQTT 5.x only.") + .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE) .withDefaultValue("0 s") .build(); EXTENSIONAPI static constexpr auto QueueBufferMaxMessage = core::PropertyDefinitionBuilder<>::createProperty("Queue Max Message") .withDescription("Maximum number of messages allowed on the received MQTT queue") + .withPropertyType(core::StandardPropertyTypes::UNSIGNED_LONG_TYPE) .withDefaultValue("1000") .build(); EXTENSIONAPI static constexpr auto AttributeFromContentType = core::PropertyDefinitionBuilder<>::createProperty("Attribute From Content Type") @@ -71,10 +75,12 @@ class ConsumeMQTT : public processors::AbstractMQTTProcessor { .build(); EXTENSIONAPI static constexpr auto TopicAliasMaximum = core::PropertyDefinitionBuilder<>::createProperty("Topic Alias Maximum") .withDescription("Maximum number of topic aliases to use. If set to 0, then topic aliases cannot be used. MQTT 5.x only.") + .withPropertyType(core::StandardPropertyTypes::UNSIGNED_INT_TYPE) .withDefaultValue("0") .build(); EXTENSIONAPI static constexpr auto ReceiveMaximum = core::PropertyDefinitionBuilder<>::createProperty("Receive Maximum") .withDescription("Maximum number of unacknowledged messages allowed. MQTT 5.x only.") + .withPropertyType(core::StandardPropertyTypes::UNSIGNED_INT_TYPE) .withDefaultValue(MQTT_MAX_RECEIVE_MAXIMUM_STR) .build(); EXTENSIONAPI static constexpr auto Properties = utils::array_cat(AbstractMQTTProcessor::BasicProperties, std::array<core::PropertyReference, 8>{ diff --git a/extensions/mqtt/processors/PublishMQTT.cpp b/extensions/mqtt/processors/PublishMQTT.cpp index 8ad317dec..9b2591524 100644 --- a/extensions/mqtt/processors/PublishMQTT.cpp +++ b/extensions/mqtt/processors/PublishMQTT.cpp @@ -47,12 +47,12 @@ void PublishMQTT::readProperties(const std::shared_ptr<core::ProcessContext>& co throw Exception(PROCESS_SCHEDULE_EXCEPTION, "PublishMQTT: Topic is required"); } - if (const auto retain_opt = context->getProperty<bool>(Retain)) { + if (const auto retain_opt = context->getProperty(Retain) | utils::flatMap(&utils::StringUtils::toBool)) { retain_ = *retain_opt; } logger_->log_debug("PublishMQTT: Retain [%d]", retain_); - if (const auto message_expiry_interval = context->getProperty<core::TimePeriodValue>(MessageExpiryInterval)) { + if (const auto message_expiry_interval = context->getProperty(MessageExpiryInterval) | utils::flatMap(&core::TimePeriodValue::fromString)) { message_expiry_interval_ = std::chrono::duration_cast<std::chrono::seconds>(message_expiry_interval->getMilliseconds()); logger_->log_debug("PublishMQTT: MessageExpiryInterval [%" PRId64 "] s", int64_t{message_expiry_interval_->count()}); } diff --git a/extensions/mqtt/processors/PublishMQTT.h b/extensions/mqtt/processors/PublishMQTT.h index 73abdccd4..ed4e8aa90 100644 --- a/extensions/mqtt/processors/PublishMQTT.h +++ b/extensions/mqtt/processors/PublishMQTT.h @@ -53,10 +53,12 @@ class PublishMQTT : public processors::AbstractMQTTProcessor { .build(); EXTENSIONAPI static constexpr auto Retain = core::PropertyDefinitionBuilder<>::createProperty("Retain") .withDescription("Retain published message in broker") + .withPropertyType(core::StandardPropertyTypes::BOOLEAN_TYPE) .withDefaultValue("false") .build(); EXTENSIONAPI static constexpr auto MessageExpiryInterval = core::PropertyDefinitionBuilder<>::createProperty("Message Expiry Interval") .withDescription("Time while message is valid and will be forwarded by broker. MQTT 5.x only.") + .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE) .build(); EXTENSIONAPI static constexpr auto ContentType = core::PropertyDefinitionBuilder<>::createProperty("Content Type") .withDescription("Content type of the message. MQTT 5.x only.") diff --git a/extensions/sftp/processors/ListSFTP.cpp b/extensions/sftp/processors/ListSFTP.cpp index d157c1c97..42490b260 100644 --- a/extensions/sftp/processors/ListSFTP.cpp +++ b/extensions/sftp/processors/ListSFTP.cpp @@ -127,7 +127,7 @@ void ListSFTP::onSchedule(const std::shared_ptr<core::ProcessContext> &context, logger_->log_error("Minimum File Age attribute is missing or invalid"); } - if (auto maximum_file_age = context->getProperty<core::TimePeriodValue>(MaximumFileAge)) { + if (auto maximum_file_age = context->getProperty(MaximumFileAge) | utils::flatMap(&core::TimePeriodValue::fromString)) { maximum_file_age_ = maximum_file_age->getMilliseconds(); } else { logger_->log_error("Maximum File Age attribute is missing or invalid"); diff --git a/extensions/sftp/processors/ListSFTP.h b/extensions/sftp/processors/ListSFTP.h index 128dc0aa3..432c98002 100644 --- a/extensions/sftp/processors/ListSFTP.h +++ b/extensions/sftp/processors/ListSFTP.h @@ -135,16 +135,15 @@ class ListSFTP : public SFTPProcessorBase { .withDefaultValue(ENTITY_TRACKING_INITIAL_LISTING_TARGET_ALL_AVAILABLE) .build(); EXTENSIONAPI static constexpr auto MinimumFileAge = core::PropertyDefinitionBuilder<>::createProperty("Minimum File Age") - .withDescription("The minimum age that a file must be in order to be pulled; " - "any file younger than this amount of time (according to last modification date) will be ignored") + .withDescription("The minimum age that a file must be in order to be pulled; any file younger than this amount of time (according to last modification date) will be ignored") .isRequired(true) .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE) .withDefaultValue("0 sec") .build(); EXTENSIONAPI static constexpr auto MaximumFileAge = core::PropertyDefinitionBuilder<>::createProperty("Maximum File Age") - .withDescription("The maximum age that a file must be in order to be pulled; " - "any file older than this amount of time (according to last modification date) will be ignored") + .withDescription("The maximum age that a file must be in order to be pulled; any file older than this amount of time (according to last modification date) will be ignored") .isRequired(false) + .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE) .build(); EXTENSIONAPI static constexpr auto MinimumFileSize = core::PropertyDefinitionBuilder<>::createProperty("Minimum File Size") .withDescription("The minimum size that a file must be in order to be pulled") diff --git a/extensions/standard-processors/processors/DefragmentText.cpp b/extensions/standard-processors/processors/DefragmentText.cpp index 722de5c66..7fbbe76d3 100644 --- a/extensions/standard-processors/processors/DefragmentText.cpp +++ b/extensions/standard-processors/processors/DefragmentText.cpp @@ -41,7 +41,7 @@ void DefragmentText::initialize() { void DefragmentText::onSchedule(core::ProcessContext* context, core::ProcessSessionFactory*) { gsl_Expects(context); - if (auto max_buffer_age = context->getProperty<core::TimePeriodValue>(MaxBufferAge)) { + if (auto max_buffer_age = context->getProperty(MaxBufferAge) | utils::flatMap(&core::TimePeriodValue::fromString)) { max_age_ = max_buffer_age->getMilliseconds(); setTriggerWhenEmpty(true); logger_->log_trace("The Buffer maximum age is configured to be %" PRId64 " ms", int64_t{max_buffer_age->getMilliseconds().count()}); diff --git a/extensions/standard-processors/processors/DefragmentText.h b/extensions/standard-processors/processors/DefragmentText.h index 586930009..3531f1e82 100644 --- a/extensions/standard-processors/processors/DefragmentText.h +++ b/extensions/standard-processors/processors/DefragmentText.h @@ -79,6 +79,7 @@ class DefragmentText : public core::Processor { EXTENSIONAPI static constexpr auto MaxBufferAge = core::PropertyDefinitionBuilder<>::createProperty("Max Buffer Age") .withDescription("The maximum age of the buffer after which it will be transferred to success when matching Start of Message patterns or to failure when matching End of Message patterns. " "Expected format is <duration> <time unit>") + .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE) .withDefaultValue("10 min") .build(); EXTENSIONAPI static constexpr auto MaxBufferSize = core::PropertyDefinitionBuilder<>::createProperty("Max Buffer Size") diff --git a/extensions/standard-processors/processors/ListFile.cpp b/extensions/standard-processors/processors/ListFile.cpp index 7fbd84890..ee22a0781 100644 --- a/extensions/standard-processors/processors/ListFile.cpp +++ b/extensions/standard-processors/processors/ListFile.cpp @@ -59,7 +59,7 @@ void ListFile::onSchedule(const std::shared_ptr<core::ProcessContext> &context, minimum_file_age_ = minimum_file_age->getMilliseconds(); } - if (auto maximum_file_age = context->getProperty<core::TimePeriodValue>(MaximumFileAge)) { + if (auto maximum_file_age = context->getProperty(MaximumFileAge) | utils::flatMap(&core::TimePeriodValue::fromString)) { maximum_file_age_ = maximum_file_age->getMilliseconds(); } diff --git a/extensions/standard-processors/processors/ListFile.h b/extensions/standard-processors/processors/ListFile.h index a50033d05..8fead27a1 100644 --- a/extensions/standard-processors/processors/ListFile.h +++ b/extensions/standard-processors/processors/ListFile.h @@ -68,7 +68,8 @@ class ListFile : public core::Processor { .build(); EXTENSIONAPI static constexpr auto MaximumFileAge = core::PropertyDefinitionBuilder<>::createProperty("Maximum File Age") .withDescription("The maximum age that a file must be in order to be pulled; any file older than this amount of time (according to last modification date) will be ignored") - .build(); + .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE) + .build(); EXTENSIONAPI static constexpr auto MinimumFileSize = core::PropertyDefinitionBuilder<>::createProperty("Minimum File Size") .withDescription("The minimum size that a file must be in order to be pulled") .isRequired(true) diff --git a/libminifi/include/core/ProcessContext.h b/libminifi/include/core/ProcessContext.h index a9c3212d4..1304b2532 100644 --- a/libminifi/include/core/ProcessContext.h +++ b/libminifi/include/core/ProcessContext.h @@ -100,23 +100,19 @@ class ProcessContext : public controller::ControllerServiceLookup, public core:: template<std::default_initializable T = std::string> std::optional<T> getProperty(const Property& property) const { T value; - try { - if (!getProperty(property.getName(), value)) return std::nullopt; - } catch (const utils::internal::ValueException&) { - return std::nullopt; + if (getProperty(property.getName(), value)) { + return value; } - return value; + return std::nullopt; } template<std::default_initializable T = std::string> std::optional<T> getProperty(const PropertyReference& property) const { T value; - try { - if (!getProperty(property.name, value)) return std::nullopt; - } catch (const utils::internal::ValueException&) { - return std::nullopt; + if (getProperty(property.name, value)) { + return value; } - return value; + return std::nullopt; } bool getProperty(std::string_view name, detail::NotAFlowFile auto& value) const { diff --git a/libminifi/include/utils/ValueParser.h b/libminifi/include/utils/ValueParser.h index 5d6cec9ae..ff04306be 100644 --- a/libminifi/include/utils/ValueParser.h +++ b/libminifi/include/utils/ValueParser.h @@ -16,8 +16,7 @@ * limitations under the License. */ -#ifndef LIBMINIFI_INCLUDE_UTILS_VALUEPARSER_H_ -#define LIBMINIFI_INCLUDE_UTILS_VALUEPARSER_H_ +#pragma once #include <exception> #include <string> @@ -30,11 +29,7 @@ #include "PropertyErrors.h" -namespace org { -namespace apache { -namespace nifi { -namespace minifi { -namespace utils { +namespace org::apache::nifi::minifi::utils { namespace internal { class ValueParser { @@ -191,11 +186,17 @@ class ValueParser { const std::string& str; std::size_t offset; }; -} /* namespace internal */ -} /* namespace utils */ -} /* namespace minifi */ -} /* namespace nifi */ -} /* namespace apache */ -} /* namespace org */ - -#endif // LIBMINIFI_INCLUDE_UTILS_VALUEPARSER_H_ +} // namespace internal + +template<typename T> +std::optional<T> toNumber(const std::string& input) { + try { + T output; + internal::ValueParser(input).parse(output).parseEnd(); + return output; + } catch(const internal::ParseException&) { + return std::nullopt; + } +} + +} // namespace org::apache::nifi::minifi::utils diff --git a/libminifi/test/unit/ValueParserTests.cpp b/libminifi/test/unit/ValueParserTests.cpp new file mode 100644 index 000000000..cade20ba2 --- /dev/null +++ b/libminifi/test/unit/ValueParserTests.cpp @@ -0,0 +1,43 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include <cstdint> + +#include "utils/ValueParser.h" +#include "../Catch.h" + +namespace utils = org::apache::nifi::minifi::utils; + +TEST_CASE("toNumber() can parse valid numbers") { + CHECK(utils::toNumber<bool>("false") == std::optional<bool>(false)); + CHECK(utils::toNumber<bool>("true") == std::optional<bool>(true)); + CHECK(utils::toNumber<int32_t>("-56483") == std::optional<int32_t>(-56483)); + CHECK(utils::toNumber<uint32_t>("8431") == std::optional<uint32_t>(8431)); + CHECK(utils::toNumber<int64_t>("4781362456") == std::optional<int64_t>(4781362456)); + CHECK(utils::toNumber<uint64_t>("56447216560") == std::optional<uint64_t>(56447216560)); + CHECK(utils::toNumber<double>("-4.1248457") == std::optional<double>(-4.1248457)); +} + +TEST_CASE("toNumber() returns nullopt if the argument is not a valid number") { + CHECK(utils::toNumber<bool>("") == std::nullopt); + CHECK(utils::toNumber<bool>("maybe") == std::nullopt); + CHECK(utils::toNumber<int32_t>("999000999000") == std::nullopt); + CHECK(utils::toNumber<uint32_t>("-561") == std::nullopt); + CHECK(utils::toNumber<int64_t>("0.5") == std::nullopt); + CHECK(utils::toNumber<uint64_t>("MAXINT") == std::nullopt); + CHECK(utils::toNumber<double>("sqrt(-1)") == std::nullopt); +}
