[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #921: MINIFICPP-1388 Introduce buffer for HTTP requests in ListenHTTP processor

[GitBox]

fgerlits commented on a change in pull request #921:
URL: https://github.com/apache/nifi-minifi-cpp/pull/921#discussion_r504112349



##########
File path: extensions/civetweb/processors/ListenHTTP.cpp
##########
@@ -212,51 +235,80 @@ void ListenHTTP::onSchedule(core::ProcessContext 
*context, core::ProcessSessionF
 ListenHTTP::~ListenHTTP() = default;
 
 void ListenHTTP::onTrigger(core::ProcessContext *context, core::ProcessSession 
*session) {
-  std::shared_ptr<core::FlowFile> flow_file = session->get();
+  logger_->log_debug("OnTrigger ListenHTTP");
+  processIncomingFlowFile(session);
+  processRequestBuffer(session);
+}
 
-  // Do nothing if there are no incoming files
+void ListenHTTP::processIncomingFlowFile(core::ProcessSession *session) {
+  std::shared_ptr<core::FlowFile> flow_file = session->get();
   if (!flow_file) {
     return;
   }
 
   std::string type;
   flow_file->getAttribute("http.type", type);
 
-  if (type == "response_body") {
-
-    if (handler_) {
-      struct response_body response { "", "", "" };
-      ResponseBodyReadCallback cb(&response.body);
-      flow_file->getAttribute("filename", response.uri);
-      flow_file->getAttribute("mime.type", response.mime_type);
-      if (response.mime_type.empty()) {
-        logger_->log_warn("Using default mime type of application/octet-stream 
for response body file: %s", response.uri);
-        response.mime_type = "application/octet-stream";
-      }
-      session->read(flow_file, &cb);
-      handler_->set_response_body(std::move(response));
+  if (type == "response_body" && handler_) {
+    response_body response;
+    ResponseBodyReadCallback cb(&response.body);
+    flow_file->getAttribute("filename", response.uri);
+    flow_file->getAttribute("mime.type", response.mime_type);
+    if (response.mime_type.empty()) {
+      logger_->log_warn("Using default mime type of application/octet-stream 
for response body file: %s", response.uri);
+      response.mime_type = "application/octet-stream";
     }
+    session->read(flow_file, &cb);
+    handler_->setResponseBody(std::move(response));
   }
 
   session->remove(flow_file);
 }
 
-ListenHTTP::Handler::Handler(std::string base_uri, core::ProcessContext 
*context, core::ProcessSessionFactory *session_factory, std::string 
&&auth_dn_regex, std::string &&header_as_attrs_regex)
+void 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;
+    if (!handler_->request_buffer.tryDequeue(flow_file_buffer_pair)) {
+      break;
+    }
+
+    auto flow_file = flow_file_buffer_pair.first;
+    session->add(flow_file);
+
+    if (flow_file_buffer_pair.second) {
+      WriteCallback callback(std::move(flow_file_buffer_pair.second));
+      session->write(flow_file, &callback);
+    }
+
+    session->transfer(flow_file, Success);
+  }
+
+  logger_->log_debug("ListenHTTP transferred %d flow files from HTTP request 
buffer", flow_file_count);
+}
+
+ListenHTTP::Handler::Handler(std::string base_uri, core::ProcessContext 
*context, std::string &&auth_dn_regex, std::string &&header_as_attrs_regex, 
std::size_t buffer_size)
     : base_uri_(std::move(base_uri)),
       auth_dn_regex_(std::move(auth_dn_regex)),
       headers_as_attrs_regex_(std::move(header_as_attrs_regex)),
-      logger_(logging::LoggerFactory<ListenHTTP::Handler>::getLogger()) {
-  process_context_ = context;
-  session_factory_ = session_factory;
+      process_context_(context),
+      logger_(logging::LoggerFactory<ListenHTTP::Handler>::getLogger()),
+      buffer_size_(buffer_size) {
 }
 
-void ListenHTTP::Handler::send_error_response(struct mg_connection *conn) {
+void ListenHTTP::Handler::sendHttp500(mg_connection* const conn) {
   mg_printf(conn, "HTTP/1.1 500 Internal Server Error\r\n"
-            "Content-Type: text/html\r\n"
-            "Content-Length: 0\r\n\r\n");
+                  "Content-Type: text/html\r\n"
+                  "Content-Length: 0\r\n\r\n");
 }
 
-void ListenHTTP::Handler::set_header_attributes(const mg_request_info 
*req_info, const std::shared_ptr<core::FlowFile> &flow_file) const {
+void ListenHTTP::Handler::sendHttp503(mg_connection* const conn) {
+  mg_printf(conn, "HTTP/1.1 503 Service Unavailable\r\n"
+                  "Content-Type: text/html\r\n"
+                  "Content-Length: 0\r\n\r\n");

Review comment:
       thanks




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org