MINIFICPP-2709 Fix site to site receive using HTTP protocol

Author: lordgamez

core-framework/include/http/HTTPStream.h

@@ -48,54 +48,49 @@ class HttpStream : public io::BaseStreamImpl {
   }
 
   const std::shared_ptr<HTTPClient>& getClient() {
-    http_client_future_.get();
+    if (http_client_read_future_.valid()) {
+      http_client_read_future_.get();
+    }
+    if (http_client_write_future_.valid()) {
+      http_client_write_future_.get();
+    }
     return http_client_;
   }
 
   void forceClose() {
-    if (started_) {
-      // lock shouldn't be needed here as call paths currently guarantee
-      // flow, but we should be safe anyway.
-      std::lock_guard<std::mutex> lock(mutex_);
-      close();
-      http_client_->forceClose();
-      if (http_client_future_.valid()) {
-        http_client_future_.get();
-      } else {
-        logger_->log_warn("Future status already cleared for {}, continuing", http_client_->getURL());
+    bool should_close = false;
+    {
+      std::lock_guard<std::mutex> read_lock(read_mutex_);
+      std::lock_guard<std::mutex> write_lock(write_mutex_);
+      should_close = read_started_ || write_started_;
+      if (should_close) {
+        close();
+        http_client_->forceClose();
+        read_started_ = false;
+        write_started_ = false;
       }
+    }
 
-      started_ = false;
+    if (should_close) {
+      if (http_client_read_future_.valid()) {
+        http_client_read_future_.get();
+      }
+      if (http_client_write_future_.valid()) {
+        http_client_write_future_.get();
+      }
     }
   }
 
-  /**
-   * Skip to the specified offset.
-   * @param offset offset to which we will skip
-   */
   void seek(size_t offset) override;
 
   [[nodiscard]] size_t tell() const override;
 
-  [[nodiscard]] size_t size() const override {
-    return written;
-  }
+  [[nodiscard]] size_t size() const override;
 
   using BaseStream::write;
   using BaseStream::read;
 
-  /**
-   * Reads data and places it into buf
-   * @param buf buffer in which we extract data
-   * @param buflen
-   */
   size_t read(std::span<std::byte> buf) override;
-
-  /**
-   * writes value to stream
-   * @param value value to write
-   * @param size size of value
-   */
   size_t write(const uint8_t* value, size_t size) override;
 
   static bool submit_client(const std::shared_ptr<HTTPClient>& client) {
@@ -114,19 +109,16 @@ class HttpStream : public io::BaseStreamImpl {
   }
 
   inline bool isFinished(int seconds = 0) {
-    return http_client_future_.wait_for(std::chrono::seconds(seconds)) == std::future_status::ready
+    return http_client_read_future_.wait_for(std::chrono::seconds(seconds)) == std::future_status::ready
         && getByteOutputReadCallback()
         && getByteOutputReadCallback()->getSize() == 0
         && getByteOutputReadCallback()->waitingOps();
   }
 
-  /**
-   * Waits for more data to become available.
-   */
   bool waitForDataAvailable() {
     do {
       logger_->log_trace("Waiting for more data");
-    } while (http_client_future_.wait_for(std::chrono::seconds(0)) != std::future_status::ready
+    } while (http_client_read_future_.wait_for(std::chrono::seconds(0)) != std::future_status::ready
         && getByteOutputReadCallback()
         && getByteOutputReadCallback()->getSize() == 0);
 
@@ -135,16 +127,15 @@ class HttpStream : public io::BaseStreamImpl {
   }
 
  protected:
-  std::vector<uint8_t> array;
-
   std::shared_ptr<HTTPClient> http_client_;
-  std::future<bool> http_client_future_;
-
-  size_t written{0};
+  std::future<bool> http_client_read_future_;
+  std::future<bool> http_client_write_future_;
 
-  std::mutex mutex_;
+  std::mutex read_mutex_;
+  std::mutex write_mutex_;
 
-  std::atomic<bool> started_{false};
+  std::atomic<bool> read_started_{false};
+  std::atomic<bool> write_started_{false};
 
  private:
   utils::ByteOutputCallback* getByteOutputReadCallback() {

core-framework/src/http/HTTPStream.cpp

@@ -40,30 +40,28 @@ void HttpStream::close() {
 }
 
 void HttpStream::seek(size_t /*offset*/) {
-  // seek is an unnecessary part of this implementation
   throw std::logic_error{"HttpStream::seek is unimplemented"};
 }
 
 size_t HttpStream::tell() const {
-  // tell is an unnecessary part of this implementation
   throw std::logic_error{"HttpStream::tell is unimplemented"};
 }
 
-// data stream overrides
+[[nodiscard]] size_t HttpStream::size() const {
+  throw std::logic_error{"HttpStream::size is unimplemented"};
+}
 
 size_t HttpStream::write(const uint8_t* value, size_t size) {
   if (size == 0) return 0;
   if (IsNullOrEmpty(value)) {
     return io::STREAM_ERROR;
   }
-  if (!started_) {
-    std::lock_guard<std::mutex> lock(mutex_);
-    if (!started_) {
-      auto callback = std::make_unique<HttpStreamingCallback>();
-      http_client_->setUploadCallback(std::move(callback));
-      http_client_future_ = std::async(std::launch::async, submit_client, http_client_);
-      started_ = true;
-    }
+  std::lock_guard<std::mutex> lock(write_mutex_);
+  if (!write_started_) {
+    auto callback = std::make_unique<HttpStreamingCallback>();
+    http_client_->setUploadCallback(std::move(callback));
+    http_client_write_future_ = std::async(std::launch::async, submit_client, http_client_);
+    write_started_ = true;
   }
   if (auto http_callback = dynamic_cast<HttpStreamingCallback*>(http_client_->getUploadCallback()))
     http_callback->process(value, size);
@@ -75,14 +73,12 @@ size_t HttpStream::write(const uint8_t* value, size_t size) {
 size_t HttpStream::read(std::span<std::byte> buf) {
   if (buf.empty()) { return 0; }
   if (!IsNullOrEmpty(buf)) {
-    if (!started_) {
-      std::lock_guard<std::mutex> lock(mutex_);
-      if (!started_) {
-        auto read_callback = std::make_unique<HTTPReadByteOutputCallback>(66560, true);
-        http_client_future_ = std::async(std::launch::async, submit_read_client, http_client_, read_callback.get());
-        http_client_->setReadCallback(std::move(read_callback));
-        started_ = true;
-      }
+    std::lock_guard<std::mutex> lock(read_mutex_);
+    if (!read_started_) {
+      auto read_callback = std::make_unique<HTTPReadByteOutputCallback>(66560, true);
+      http_client_read_future_ = std::async(std::launch::async, submit_read_client, http_client_, read_callback.get());
+      http_client_->setReadCallback(std::move(read_callback));
+      read_started_ = true;
     }
     return gsl::not_null(getByteOutputReadCallback())->readFully(reinterpret_cast<char*>(buf.data()), buf.size());
   } else {

docker/test/integration/minifi/flow_serialization/Minifi_flow_json_serializer.py

@@ -133,6 +133,7 @@ def serialize_node(self, connectable, root, visited):
                         'targetUri': group.url,
                         'communicationsTimeout': '30 sec',
                         'yieldDuration': '3 sec',
+                        'transportProtocol': group.transport_protocol,
                         'outputPorts': []
                     }
 

docker/test/integration/minifi/flow_serialization/Minifi_flow_yaml_serializer.py

@@ -136,6 +136,7 @@ def serialize_node(self, connectable, res=None, visited=None):
                         'url': group.url,
                         'timeout': '30 sec',
                         'yield period': '3 sec',
+                        'transport protocol': group.transport_protocol,
                         'Output Ports': []
                     }
 

libminifi/include/sitetosite/HttpSiteToSiteClient.h

@@ -77,6 +77,9 @@ class HttpSiteToSiteClient final : public SiteToSiteClient {
   void deleteTransaction(const utils::Identifier& transaction_id) override;
   void tearDown() override;
 
+ protected:
+  std::pair<uint64_t, uint64_t> readFlowFiles(const std::shared_ptr<Transaction>& transaction, core::ProcessSession& session) override;
+
  private:
   void setSiteToSiteHeaders(minifi::http::HTTPClient& client);
   void closeTransaction(const utils::Identifier &transaction_id);

libminifi/include/sitetosite/SiteToSiteClient.h

@@ -25,6 +25,7 @@
 #include <utility>
 #include <vector>
 #include <optional>
+#include <expected>
 
 #include "Peer.h"
 #include "SiteToSite.h"
@@ -171,6 +172,9 @@ class SiteToSiteClient {
   std::atomic<std::chrono::milliseconds> batch_duration_{0s};
   std::atomic<std::chrono::milliseconds> timeout_{0s};
 
+ protected:
+  virtual std::pair<uint64_t, uint64_t> readFlowFiles(const std::shared_ptr<Transaction>& transaction, core::ProcessSession& session);
+
  private:
   struct ReceiveFlowFileHeaderResult {
     std::map<std::string, std::string> attributes;
@@ -187,9 +191,8 @@ class SiteToSiteClient {
   bool completeReceive(const std::shared_ptr<Transaction>& transaction, const utils::Identifier& transaction_id);
   bool completeSend(const std::shared_ptr<Transaction>& transaction, const utils::Identifier& transaction_id, core::ProcessContext& context);
 
-  bool readFlowFileHeaderData(io::InputStream& stream, const std::string& transaction_id, SiteToSiteClient::ReceiveFlowFileHeaderResult& result);
-  std::optional<ReceiveFlowFileHeaderResult> receiveFlowFileHeader(io::InputStream& stream, const std::shared_ptr<Transaction>& transaction);
-  std::pair<uint64_t, uint64_t> readFlowFiles(const std::shared_ptr<Transaction>& transaction, core::ProcessSession& session);
+  std::expected<void, std::string> readFlowFileHeaderData(io::InputStream& stream, const std::string& transaction_id, SiteToSiteClient::ReceiveFlowFileHeaderResult& result);
+  std::expected<ReceiveFlowFileHeaderResult, std::string> receiveFlowFileHeader(io::InputStream& stream, const std::shared_ptr<Transaction>& transaction);
 
   std::shared_ptr<core::logging::Logger> logger_{core::logging::LoggerFactory<SiteToSiteClient>::getLogger()};
 };

libminifi/src/sitetosite/HttpSiteToSiteClient.cpp

@@ -141,7 +141,6 @@ std::shared_ptr<Transaction> HttpSiteToSiteClient::createTransaction(TransferDir
   } else {
     transaction_client = openConnectionForReceive(transaction);
     transaction->setDataAvailable(true);
-    // 201 tells us that data is available. 200 would mean that nothing is available.
   }
   gsl_Assert(transaction_client);
 
@@ -352,13 +351,11 @@ void HttpSiteToSiteClient::closeTransaction(const utils::Identifier &transaction
 
   logger_->log_debug("Received {} response code from delete", client->getResponseCode());
 
-  if (client->getResponseCode() == 400) {
+  if (client->getResponseCode() >= 400) {
     std::string error(client->getResponseBody().data(), client->getResponseBody().size());
 
-    logger_->log_warn("400 received: {}", error);
-    std::stringstream message;
-    message << "Received " << client->getResponseCode() << " from " << uri.str();
-    throw Exception(SITE2SITE_EXCEPTION, message.str());
+    logger_->log_warn("{} received: {}", client->getResponseCode(), error);
+    throw Exception(SITE2SITE_EXCEPTION, fmt::format("Received {} from {}", client->getResponseCode(), uri.str()));
   }
 
   transaction->close();
@@ -388,4 +385,33 @@ void HttpSiteToSiteClient::setSiteToSiteHeaders(minifi::http::HTTPClient& client
   }
 }
 
+std::pair<uint64_t, uint64_t> HttpSiteToSiteClient::readFlowFiles(const std::shared_ptr<Transaction>& transaction, core::ProcessSession& session) {
+  auto http_stream = dynamic_cast<http::HttpStream*>(peer_->getStream());
+  if (!http_stream) {
+    throw Exception(SITE2SITE_EXCEPTION, "Reading flow files failed: stream cannot be cast to HTTP stream");
+  }
+
+  try {
+    return SiteToSiteClient::readFlowFiles(transaction, session);
+  } catch (const Exception& e) {
+    auto response_code = http_stream->getClientRef()->getResponseCode();
+
+    // 200 tells us that there is no content to read, so we should not treat it as an error.
+    // The read fails in this case because the stream does not contain a valid response body with the expected format.
+    // Unfortunately there is no way to get the response code before trying to read, so we have to let it fail and check the response code afterwards.
+    if (response_code == 200) {
+      logger_->log_debug("Response code 200, no content to read");
+      transaction->setDataAvailable(false);
+      transaction->setState(TransactionState::TRANSACTION_CANCELED);
+      current_code_ = ResponseCode::CANCEL_TRANSACTION;
+      return {0, 0};
+    }
+    throw;
+  }
+
+  if (auto response_code = http_stream->getClientRef()->getResponseCode(); response_code >= 400) {
+    throw Exception(SITE2SITE_EXCEPTION, fmt::format("Error received while reading flow files: {}", response_code));
+  }
+}
+
 }  // namespace org::apache::nifi::minifi::sitetosite