martinzink commented on code in PR #1383:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1383#discussion_r954695141
##########
extensions/http-curl/processors/InvokeHTTP.cpp:
##########
@@ -83,165 +84,188 @@ const core::Property InvokeHTTP::ProxyUsername(
core::PropertyBuilder::createProperty("invokehttp-proxy-username", "Proxy
Username")->withDescription("Username to set when authenticating against
proxy")->isRequired(false)->build());
const core::Property InvokeHTTP::ProxyPassword(
core::PropertyBuilder::createProperty("invokehttp-proxy-password", "Proxy
Password")->withDescription("Password to set when authenticating against
proxy")->isRequired(false)->build());
-const core::Property InvokeHTTP::ContentType("Content-type", "The Content-Type
to specify for when content is being transmitted through a PUT, "
- "POST or PATCH. In the case of an empty
value after evaluating an expression language expression, "
- "Content-Type defaults to",
- "application/octet-stream");
+const core::Property InvokeHTTP::ContentType("Content-type",
+ "The Content-Type to specify for when content is being transmitted through
a PUT, "
+ "POST or PATCH. In the case of an empty value after evaluating an
expression language expression, "
+ "Content-Type defaults to",
+ "application/octet-stream");
const core::Property InvokeHTTP::SendBody(
core::PropertyBuilder::createProperty("send-message-body", "Send Body")
- ->withDescription("DEPRECATED. Only kept for backwards compatibility, no
functionality is included.")
- ->withDefaultValue<bool>(true)
- ->build());
+ ->withDescription("DEPRECATED. Only kept for backwards compatibility,
no functionality is included.")
+ ->withDefaultValue<bool>(true)
+ ->build());
const core::Property InvokeHTTP::SendMessageBody(
core::PropertyBuilder::createProperty("Send Message Body")
- ->withDescription("If true, sends the HTTP message body on
POST/PUT/PATCH requests (default). "
- "If false, suppresses the message body and
content-type header for these requests.")
- ->withDefaultValue<bool>(true)
- ->build());
-const core::Property InvokeHTTP::UseChunkedEncoding("Use Chunked Encoding",
"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.",
- "false");
-const core::Property InvokeHTTP::PropPutOutputAttributes("Put Response Body in
Attribute", "If set, the response body received back will be put into an
attribute of the original "
- "FlowFile instead of a
separate FlowFile. The attribute key to put to is determined by evaluating
value of this property. ",
- "");
-const core::Property InvokeHTTP::AlwaysOutputResponse("Always Output
Response", "Will force a response FlowFile to be generated and routed to the
'Response' relationship "
- "regardless of what the server
status code received is ",
- "false");
-const core::Property InvokeHTTP::PenalizeOnNoRetry("Penalize on \"No Retry\"",
"Enabling this property will penalize FlowFiles that are routed to the \"No
Retry\" relationship.", "false");
+ ->withDescription("If true, sends the HTTP message body on
POST/PUT/PATCH requests (default). "
+ "If false, suppresses the message body and
content-type header for these requests.")
+ ->withDefaultValue<bool>(true)
+ ->build());
+const core::Property InvokeHTTP::UseChunkedEncoding("Use Chunked Encoding",
+ "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.",
+ "false");
+const core::Property InvokeHTTP::PutResponseBodyInAttribute("Put Response Body
in Attribute",
+ "If set, the response body received back will be put into an attribute of
the original "
+ "FlowFile instead of a separate FlowFile. "
+ "The attribute key to put to is determined by evaluating value of this
property. ",
+ "");
+const core::Property InvokeHTTP::AlwaysOutputResponse("Always Output Response",
+ "Will force a response FlowFile to be generated and routed to the
'Response' relationship regardless of what the server status code received is ",
+ "false");
+const core::Property InvokeHTTP::PenalizeOnNoRetry("Penalize on \"No Retry\"",
+ "Enabling this property will penalize FlowFiles that are routed to the
\"No Retry\" relationship.",
+ "false");
const core::Property InvokeHTTP::DisablePeerVerification("Disable Peer
Verification", "Disables peer verification for the SSL session", "false");
const core::Property InvokeHTTP::InvalidHTTPHeaderFieldHandlingStrategy(
core::PropertyBuilder::createProperty("Invalid HTTP Header Field Handling
Strategy")
- ->withDescription("Indicates what should happen when an attribute's name
is not a valid HTTP header field name. "
- "Options: transform - invalid characters are replaced, fail - flow
file is transferred to failure, drop - drops invalid attributes from HTTP
message")
- ->isRequired(true)
-
->withDefaultValue<std::string>(toString(InvalidHTTPHeaderFieldHandlingOption::TRANSFORM))
-
->withAllowableValues<std::string>(InvalidHTTPHeaderFieldHandlingOption::values())
- ->build());
+ ->withDescription("Indicates what should happen when an attribute's
name is not a valid HTTP header field name. "
+ "Options: transform - invalid characters are
replaced, fail - flow file is transferred to failure, drop - drops invalid
attributes from HTTP message")
+ ->isRequired(true)
+
->withDefaultValue<std::string>(toString(InvalidHTTPHeaderFieldHandlingOption::TRANSFORM))
+
->withAllowableValues<std::string>(InvalidHTTPHeaderFieldHandlingOption::values())
+ ->build());
-const core::Relationship InvokeHTTP::Success("success", "The original FlowFile
will be routed upon success (2xx status codes). "
- "It will have new attributes detailing
the success of the request.");
+const core::Relationship InvokeHTTP::Success("success",
+ "The original FlowFile will be routed upon success (2xx status codes). It
will have new attributes detailing the success of the request.");
-const core::Relationship InvokeHTTP::RelResponse("response", "A Response
FlowFile will be routed upon success (2xx status codes). "
- "If the 'Always Output Response'
property is true then the response will be sent "
- "to this relationship regardless of
the status code received.");
+const core::Relationship InvokeHTTP::RelResponse("response",
+ "A Response FlowFile will be routed upon success (2xx status codes). "
+ "If the 'Always Output Response' property is true then the response will
be sent "
+ "to this relationship regardless of the status code received.");
-const core::Relationship InvokeHTTP::RelRetry("retry", "The original FlowFile
will be routed on any status code that can be retried "
- "(5xx status codes). It will have new
attributes detailing the request.");
+const core::Relationship InvokeHTTP::RelRetry("retry",
+ "The original FlowFile will be routed on any status code that can be
retried "
+ "(5xx status codes). It will have new attributes detailing the request.");
-const core::Relationship InvokeHTTP::RelNoRetry("no retry", "The original
FlowFile will be routed on any status code that should NOT "
- "be retried (1xx, 3xx, 4xx status
codes). It will have new attributes detailing the request.");
+const core::Relationship InvokeHTTP::RelNoRetry("no retry",
+ "The original FlowFile will be routed on any status code that should NOT "
+ "be retried (1xx, 3xx, 4xx status codes). It will have new attributes
detailing the request.");
-const core::Relationship InvokeHTTP::RelFailure("failure", "The original
FlowFile will be routed on any type of connection failure, "
- "timeout or general exception. It
will have new attributes detailing the request.");
+const core::Relationship InvokeHTTP::RelFailure("failure",
+ "The original FlowFile will be routed on any type of connection failure, "
+ "timeout or general exception. It will have new attributes detailing the
request.");
void InvokeHTTP::initialize() {
logger_->log_trace("Initializing InvokeHTTP");
setSupportedProperties(properties());
setSupportedRelationships(relationships());
}
-void InvokeHTTP::onSchedule(const std::shared_ptr<core::ProcessContext>
&context, const std::shared_ptr<core::ProcessSessionFactory>&
/*sessionFactory*/) {
- if (!context->getProperty(Method.getName(), method_)) {
- logger_->log_debug("%s attribute is missing, so default value of %s will
be used", Method.getName(), Method.getValue());
- return;
- }
+namespace {
+void setupClientTimeouts(extensions::curl::HTTPClient& client, const
core::ProcessContext& context) {
+ if (auto connection_timeout =
context.getProperty<core::TimePeriodValue>(InvokeHTTP::ConnectTimeout))
+ client.setConnectionTimeout(connection_timeout->getMilliseconds());
- if (!context->getProperty(URL.getName(), url_)) {
- logger_->log_debug("%s attribute is missing, so default value of %s will
be used", URL.getName(), URL.getValue());
- return;
- }
+ if (auto read_timeout =
context.getProperty<core::TimePeriodValue>(InvokeHTTP::ReadTimeout))
+ client.setReadTimeout(read_timeout->getMilliseconds());
+}
- if (auto connect_timeout =
context->getProperty<core::TimePeriodValue>(ConnectTimeout)) {
- connect_timeout_ms_ = connect_timeout->getMilliseconds();
- } else {
- logger_->log_debug("%s attribute is missing, so default value of %s will
be used", ConnectTimeout.getName(), ConnectTimeout.getValue());
- return;
+void setupClientProxy(extensions::curl::HTTPClient& client, const
core::ProcessContext& context) {
+ utils::HTTPProxy proxy = {};
+ context.getProperty(InvokeHTTP::ProxyHost.getName(), proxy.host);
+ std::string port_str;
+ if (context.getProperty(InvokeHTTP::ProxyPort.getName(), port_str) &&
!port_str.empty()) {
+ core::Property::StringToInt(port_str, proxy.port);
}
+ context.getProperty(InvokeHTTP::ProxyUsername.getName(), proxy.username);
+ context.getProperty(InvokeHTTP::ProxyPassword.getName(), proxy.password);
- std::string content_type_str;
- if (context->getProperty(ContentType.getName(), content_type_str)) {
- content_type_ = content_type_str;
- }
+ client.setHTTPProxy(proxy);
+}
- if (auto read_timeout =
context->getProperty<core::TimePeriodValue>(ReadTimeout)) {
- read_timeout_ms_ = read_timeout->getMilliseconds();
- } else {
- logger_->log_debug("%s attribute is missing, so default value of %s will
be used", ReadTimeout.getName(), ReadTimeout.getValue());
- }
+void setupClientPeerVerification(extensions::curl::HTTPClient& client, const
core::ProcessContext& context) {
+ if (auto disable_peer_verification =
context.getProperty<bool>(InvokeHTTP::DisablePeerVerification))
+ client.setPeerVerification(*disable_peer_verification);
+}
+
+void setupClientFollowRedirects(extensions::curl::HTTPClient& client, const
core::ProcessContext& context) {
+ if (auto follow_redirects =
context.getProperty<bool>(InvokeHTTP::FollowRedirects))
+ client.setFollowRedirects(*follow_redirects);
+}
- std::string date_header_str;
- if (!context->getProperty(DateHeader.getName(), date_header_str)) {
- logger_->log_debug("%s attribute is missing, so default value of %s will
be used", DateHeader.getName(), DateHeader.getValue());
+void setupClientContentType(extensions::curl::HTTPClient& client, const
core::ProcessContext& context, bool send_body) {
+ if (auto content_type = context.getProperty(InvokeHTTP::ContentType)) {
+ if (send_body)
+ client.setContentType(*content_type);
}
+}
- date_header_include_ =
utils::StringUtils::toBool(date_header_str).value_or(DateHeader.getValue());
+void setupClientTransferEncoding(extensions::curl::HTTPClient& client, bool
use_chunked_encoding) {
+ if (use_chunked_encoding)
+ client.setRequestHeader("Transfer-Encoding", "chunked");
+ else
+ client.setRequestHeader("Transfer-Encoding", std::nullopt);
+}
+} // namespace
- if (!context->getProperty(PropPutOutputAttributes.getName(),
put_attribute_name_)) {
- logger_->log_debug("%s attribute is missing, so default value of %s will
be used", PropPutOutputAttributes.getName(),
PropPutOutputAttributes.getValue());
- }
+void InvokeHTTP::setupMembersFromProperties(const core::ProcessContext&
context) {
+ context.getProperty(SendMessageBody.getName(), send_body_);
- attributes_to_send_ = context->getProperty(AttributesToSend)
- | utils::filter([](const std::string& s) { return !s.empty(); }) //
avoid compiling an empty string to regex
- | utils::map([](const std::string& regex_str) { return
utils::Regex{regex_str}; })
- | utils::orElse([this] { logger_->log_debug("%s is missing, so the
default value will be used", AttributesToSend.getName()); });
+ attributes_to_send_ = context.getProperty(AttributesToSend)
+ | utils::filter([](const std::string& s) { return
!s.empty(); }) // avoid compiling an empty string to regex
+ | utils::map([](const std::string& regex_str) { return
utils::Regex{regex_str}; })
+ | utils::orElse([this] { logger_->log_debug("%s is
missing, so the default value will be used", AttributesToSend.getName()); });
- std::string always_output_response;
- if (!context->getProperty(AlwaysOutputResponse.getName(),
always_output_response)) {
- logger_->log_debug("%s attribute is missing, so default value of %s will
be used", AlwaysOutputResponse.getName(), AlwaysOutputResponse.getValue());
- }
- always_output_response_ =
utils::StringUtils::toBool(always_output_response).value_or(false);
+ always_output_response_ =
context.getProperty<bool>(AlwaysOutputResponse).value_or(false);
+ penalize_no_retry_ =
context.getProperty<bool>(PenalizeOnNoRetry).value_or(false);
- std::string penalize_no_retry = "false";
- if (!context->getProperty(PenalizeOnNoRetry.getName(), penalize_no_retry)) {
- logger_->log_debug("%s attribute is missing, so default value of %s will
be used", PenalizeOnNoRetry.getName(), PenalizeOnNoRetry.getValue());
- }
+ invalid_http_header_field_handling_strategy_ =
utils::parseEnumProperty<InvalidHTTPHeaderFieldHandlingOption>(context,
InvalidHTTPHeaderFieldHandlingStrategy);
- penalize_no_retry_ =
utils::StringUtils::toBool(penalize_no_retry).value_or(false);
+ put_response_body_in_attribute_ =
context.getProperty(PutResponseBodyInAttribute);
+ if (put_response_body_in_attribute_ &&
put_response_body_in_attribute_->empty()) {
+ logger_->log_warn("%s is set to an empty string",
PutResponseBodyInAttribute.getName());
+ put_response_body_in_attribute_.reset();
+ }
+}
- std::string context_name;
- if (context->getProperty(SSLContext.getName(), context_name) &&
!IsNullOrEmpty(context_name)) {
- std::shared_ptr<core::controller::ControllerService> service =
context->getControllerService(context_name);
- if (!service) {
- logger_->log_error("Couldn't find controller service with name '%s'",
context_name);
+std::unique_ptr<minifi::extensions::curl::HTTPClient>
InvokeHTTP::createHTTPClientFromPropertiesAndMembers(const
core::ProcessContext& context) const {
+ std::string method;
+ if (!context.getProperty(Method.getName(), method))
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Method property missing or
invalid");
+
+ std::string url;
+ if (!context.getProperty(URL.getName(), url))
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "URL property missing or
invalid");
+
+ std::shared_ptr<minifi::controllers::SSLContextService> ssl_context_service;
+ if (auto ssl_context_name = context.getProperty(SSLContext)) {
+ if (auto service = context.getControllerService(*ssl_context_name)) {
+ ssl_context_service =
std::dynamic_pointer_cast<minifi::controllers::SSLContextService>(service);
+ if (!ssl_context_service)
+ logger_->log_error("Controller service '%s' is not an
SSLContextService", *ssl_context_name);
} else {
- ssl_context_service_ =
std::dynamic_pointer_cast<minifi::controllers::SSLContextService>(service);
- if (!ssl_context_service_) {
- logger_->log_error("Controller service '%s' is not an
SSLContextService", context_name);
- }
+ logger_->log_error("Couldn't find controller service with name '%s'",
*ssl_context_name);
}
}
- std::string use_chunked_encoding = "false";
- if (!context->getProperty(UseChunkedEncoding.getName(),
use_chunked_encoding)) {
- logger_->log_debug("%s attribute is missing, so default value of %s will
be used", UseChunkedEncoding.getName(), UseChunkedEncoding.getValue());
- }
+ auto client = std::make_unique<minifi::extensions::curl::HTTPClient>();
+ client->initialize(std::move(method), std::move(url),
std::move(ssl_context_service));
+ setupClientTimeouts(*client, context);
+ setupClientProxy(*client, context);
+ setupClientFollowRedirects(*client, context);
+ setupClientPeerVerification(*client, context);
+ setupClientContentType(*client, context, send_body_);
+ setupClientTransferEncoding(*client, use_chunked_encoding_);
- use_chunked_encoding_ =
utils::StringUtils::toBool(use_chunked_encoding).value_or(false);
+ return client;
+}
- std::string disable_peer_verification;
- disable_peer_verification_ =
(context->getProperty(DisablePeerVerification.getName(),
disable_peer_verification) &&
utils::StringUtils::toBool(disable_peer_verification).value_or(false));
- proxy_ = {};
- context->getProperty(ProxyHost.getName(), proxy_.host);
- std::string port_str;
- if (context->getProperty(ProxyPort.getName(), port_str) &&
!port_str.empty()) {
- core::Property::StringToInt(port_str, proxy_.port);
- }
- context->getProperty(ProxyUsername.getName(), proxy_.username);
- context->getProperty(ProxyPassword.getName(), proxy_.password);
- context->getProperty(FollowRedirects.getName(), follow_redirects_);
- context->getProperty(SendMessageBody.getName(), send_body_);
+void InvokeHTTP::onSchedule(const std::shared_ptr<core::ProcessContext>&
context, const std::shared_ptr<core::ProcessSessionFactory>&
/*sessionFactory*/) {
+ gsl_Expects(context);
- invalid_http_header_field_handling_strategy_ =
utils::parseEnumProperty<InvalidHTTPHeaderFieldHandlingOption>(*context,
InvalidHTTPHeaderFieldHandlingStrategy);
+ setupMembersFromProperties(*context);
+ client_queue_ =
gsl::make_not_null(utils::ResourceQueue<extensions::curl::HTTPClient>::create(getMaxConcurrentTasks(),
logger_));
}
-bool InvokeHTTP::shouldEmitFlowFile() const {
- return ("POST" == method_ || "PUT" == method_ || "PATCH" == method_);
+bool InvokeHTTP::shouldEmitFlowFile(minifi::extensions::curl::HTTPClient&
client) {
Review Comment:
its static now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
