This is an automated email from the ASF dual-hosted git repository. lordgamez pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 18a51802c3cda5c445c7da9d0383655c4a7e71e1 Author: Marton Szasz <[email protected]> AuthorDate: Tue Jul 26 01:54:03 2022 +0200 MINIFICPP-1889 fix ListenHTTP yield, add reg test Signed-off-by: Gabor Gyimesi <[email protected]> This closes #1375 --- extensions/civetweb/processors/ListenHTTP.cpp | 37 +++++++++++++++------------ extensions/civetweb/processors/ListenHTTP.h | 4 +-- extensions/civetweb/tests/ListenHTTPTests.cpp | 20 +++++++++++++++ 3 files changed, 43 insertions(+), 18 deletions(-) diff --git a/extensions/civetweb/processors/ListenHTTP.cpp b/extensions/civetweb/processors/ListenHTTP.cpp index 6aa0fc750..888186f5a 100644 --- a/extensions/civetweb/processors/ListenHTTP.cpp +++ b/extensions/civetweb/processors/ListenHTTP.cpp @@ -224,16 +224,21 @@ void ListenHTTP::onSchedule(core::ProcessContext *context, core::ProcessSessionF ListenHTTP::~ListenHTTP() = default; -void ListenHTTP::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) { - logger_->log_debug("OnTrigger ListenHTTP"); - processIncomingFlowFile(session); - processRequestBuffer(session); +void ListenHTTP::onTrigger(core::ProcessContext*, core::ProcessSession* session) { + gsl_Expects(session); + logger_->log_trace("OnTrigger ListenHTTP"); + const bool incoming_processed = processIncomingFlowFile(*session); + const bool request_processed = processRequestBuffer(*session); + if (!incoming_processed && !request_processed) { + yield(); + } } -void ListenHTTP::processIncomingFlowFile(core::ProcessSession *session) { - std::shared_ptr<core::FlowFile> flow_file = session->get(); +/// @return Whether there was a flow file processed. +bool ListenHTTP::processIncomingFlowFile(core::ProcessSession &session) { + std::shared_ptr<core::FlowFile> flow_file = session.get(); if (!flow_file) { - return; + return false; } std::string type; @@ -247,14 +252,16 @@ void ListenHTTP::processIncomingFlowFile(core::ProcessSession *session) { logger_->log_warn("Using default mime type of application/octet-stream for response body file: %s", response.uri); response.mime_type = "application/octet-stream"; } - response.body = to_string(session->readBuffer(flow_file)); + response.body = to_string(session.readBuffer(flow_file)); handler_->setResponseBody(std::move(response)); } - session->remove(flow_file); + session.remove(flow_file); + return true; } -void ListenHTTP::processRequestBuffer(core::ProcessSession *session) { +/// @return Whether there was a request processed +bool ListenHTTP::processRequestBuffer(core::ProcessSession& session) { std::size_t flow_file_count = 0; for (; batch_size_ == 0 || batch_size_ > flow_file_count; ++flow_file_count) { FlowFileBufferPair flow_file_buffer_pair; @@ -263,19 +270,17 @@ void ListenHTTP::processRequestBuffer(core::ProcessSession *session) { } auto flow_file = flow_file_buffer_pair.first; - session->add(flow_file); + session.add(flow_file); if (flow_file_buffer_pair.second) { - session->write(flow_file, [request_content = flow_file_buffer_pair.second.get()](const std::shared_ptr<io::BaseStream>& stream) -> int64_t { - const auto write_ret = stream->write(request_content->getBuffer()); - return io::isError(write_ret) ? -1 : gsl::narrow<int64_t>(write_ret); - }); + session.writeBuffer(flow_file, flow_file_buffer_pair.second->getBuffer()); } - session->transfer(flow_file, Success); + session.transfer(flow_file, Success); } logger_->log_debug("ListenHTTP transferred %zu flow files from HTTP request buffer", flow_file_count); + return flow_file_count > 0; } ListenHTTP::Handler::Handler(std::string base_uri, core::ProcessContext *context, std::string &&auth_dn_regex, std::optional<utils::Regex> &&headers_as_attrs_regex) diff --git a/extensions/civetweb/processors/ListenHTTP.h b/extensions/civetweb/processors/ListenHTTP.h index bc3ad4b9d..c91adc2d3 100644 --- a/extensions/civetweb/processors/ListenHTTP.h +++ b/extensions/civetweb/processors/ListenHTTP.h @@ -185,8 +185,8 @@ class ListenHTTP : public core::Processor { private: static const uint64_t DEFAULT_BUFFER_SIZE; - void processIncomingFlowFile(core::ProcessSession *session); - void processRequestBuffer(core::ProcessSession *session); + bool processIncomingFlowFile(core::ProcessSession &session); + bool processRequestBuffer(core::ProcessSession &session); std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<ListenHTTP>::getLogger(); CivetCallbacks callbacks_; diff --git a/extensions/civetweb/tests/ListenHTTPTests.cpp b/extensions/civetweb/tests/ListenHTTPTests.cpp index f8aa0dc8c..e9ee3ee43 100644 --- a/extensions/civetweb/tests/ListenHTTPTests.cpp +++ b/extensions/civetweb/tests/ListenHTTPTests.cpp @@ -36,6 +36,9 @@ #include "FlowController.h" #include "SchedulingAgent.h" #include "core/ProcessGroup.h" +#include "SingleProcessorTestController.h" + +namespace org::apache::nifi::minifi::test { class ListenHTTPTestsFixture { public: @@ -655,3 +658,20 @@ TEST_CASE_METHOD(ListenHTTPTestsFixture, "HTTPS minimum SSL version", "[https]") } #endif #endif + +TEST_CASE("ListenHTTP bored yield", "[listenhttp][bored][yield]") { + using processors::ListenHTTP; + const auto listen_http = std::make_shared<ListenHTTP>("listenhttp"); + SingleProcessorTestController controller{listen_http}; + listen_http->setProperty(ListenHTTP::Port, "0"); + + REQUIRE(!listen_http->isYield()); + const auto output = controller.trigger(); + REQUIRE(std::all_of(std::begin(output), std::end(output), [](const auto& kv) { + const auto& [relationship, flow_files] = kv; + return flow_files.empty(); + })); + REQUIRE(listen_http->isYield()); +} + +} // namespace org::apache::nifi::minifi::test
