MINIFICPP-2701 ListenUDP udp.sender.port property

PROCESSORS.md

@@ -1565,10 +1565,11 @@ In the list below, the names of required properties appear in bold. Any other pr
 
 ### Output Attributes
 
-| Attribute  | Relationship | Description                                  |
-|------------|--------------|----------------------------------------------|
-| udp.port   |              | The sending port the messages were received. |
-| udp.sender |              | The sending host of the messages.            |
+| Attribute       | Relationship | Description                                             |
+|-----------------|--------------|---------------------------------------------------------|
+| udp.port        |              | The listening port on which the messages were received. |
+| udp.sender      |              | The sending host of the messages.                       |
+| udp.sender.port |              | The sending port of the messages.                       |
 
 
 ## ListFile

core-framework/include/utils/MinifiConcurrentQueue.h

@@ -17,15 +17,17 @@
 #pragma once
 
 #include <algorithm>
+#include <atomic>
 #include <chrono>
+#include <condition_variable>
 #include <deque>
 #include <mutex>
-#include <condition_variable>
-#include <utility>
+#include <optional>
 #include <stdexcept>
-#include <atomic>
+#include <utility>
 
 #include "utils/TryMoveCall.h"
+#include "minifi-cpp/utils/gsl.h"
 
 namespace org::apache::nifi::minifi::utils {
 
@@ -56,6 +58,13 @@ class ConcurrentQueue {
     return tryDequeueImpl(lck, out);
   }
 
+  std::optional<T> tryDequeue() {
+    std::optional<T> result = std::nullopt;
+    const bool consume_result = consume([&result](T value) { result = std::move(value); });
+    gsl_Assert(consume_result == result.has_value());
+    return result;
+  }
+
   template<typename Functor>
   bool consume(Functor&& fun) {
     std::unique_lock<std::mutex> lck(mtx_);

extension-framework/include/utils/net/Message.h

@@ -25,20 +25,21 @@
 namespace org::apache::nifi::minifi::utils::net {
 
 struct Message {
- public:
-  Message() = default;
-  Message(std::string message_data, IpProtocol protocol, asio::ip::address sender_address, asio::ip::port_type server_port)
+  Message() = delete;
+  Message(std::string message_data, IpProtocol protocol, asio::ip::address remote_address, asio::ip::port_type remote_port, asio::ip::port_type local_port)
       : message_data(std::move(message_data)),
       protocol(protocol),
-      server_port(server_port),
-      sender_address(std::move(sender_address)) {
+      remote_address(std::move(remote_address)),
+      remote_port(remote_port),
+      local_port(local_port) {
   }
 
   bool is_partial = false;
   std::string message_data;
   IpProtocol protocol;
-  asio::ip::port_type server_port;
-  asio::ip::address sender_address;
+  asio::ip::address remote_address;
+  asio::ip::port_type remote_port;
+  asio::ip::port_type local_port;
 };
 
 }  // namespace org::apache::nifi::minifi::utils::net

extension-framework/include/utils/net/Server.h

@@ -47,9 +47,13 @@ class Server {
   bool queueEmpty() {
     return concurrent_queue_.empty();
   }
-  bool tryDequeue(utils::net::Message& received_message) {
-    return concurrent_queue_.tryDequeue(received_message);
+  std::optional<utils::net::Message> tryDequeue() {
+    return concurrent_queue_.tryDequeue();
   }
+  Server(const Server&) = delete;
+  Server(Server&&) = delete;
+  Server& operator=(const Server&) = delete;
+  Server& operator=(Server&&) = delete;
   virtual ~Server() {
     stop();
   }

extension-framework/include/utils/net/TcpServer.h

@@ -41,10 +41,10 @@ class TcpServer : public Server {
  protected:
   asio::awaitable<void> doReceive() override;
 
-  asio::awaitable<void> insecureSession(asio::ip::tcp::socket socket, asio::ip::address remote_address, asio::ip::port_type local_port);
-  asio::awaitable<void> secureSession(asio::ip::tcp::socket socket, asio::ip::address remote_address, asio::ip::port_type local_port);
+  asio::awaitable<void> insecureSession(asio::ip::tcp::socket socket, asio::ip::address remote_address, asio::ip::port_type remote_port, asio::ip::port_type local_port);
+  asio::awaitable<void> secureSession(asio::ip::tcp::socket socket, asio::ip::address remote_address, asio::ip::port_type remote_port, asio::ip::port_type local_port);
 
-  asio::awaitable<void> readLoop(auto& socket, const auto& remote_address, const auto& local_port);
+  asio::awaitable<void> readLoop(auto& socket, asio::ip::address remote_address, asio::ip::port_type remote_port, asio::ip::port_type local_port);
 
   bool consume_delimiter_;
   const std::string delimiter_;

extension-framework/include/utils/net/UdpServer.h

@@ -18,11 +18,9 @@
 
 #include <optional>
 #include <memory>
-#include <string>
 #include <asio/awaitable.hpp>
 
 #include "Server.h"
-#include "utils/MinifiConcurrentQueue.h"
 #include "minifi-cpp/core/logging/Logger.h"
 #include "core/logging/LoggerFactory.h"
 

extension-framework/src/utils/net/TcpServer.cpp

@@ -34,20 +34,23 @@ asio::awaitable<void> TcpServer::doReceive() {
       continue;
     }
     std::error_code error;
-    auto remote_address = socket.lowest_layer().remote_endpoint(error).address();
-    if (error)
+    auto remote_endpoint = socket.lowest_layer().remote_endpoint(error);
+    if (error) {
       logger_->log_warn("Error during fetching remote endpoint: {}", error.message());
+    }
     auto local_port = socket.lowest_layer().local_endpoint(error).port();
-    if (error)
+    if (error) {
       logger_->log_warn("Error during fetching local endpoint: {}", error.message());
-    if (ssl_data_)
-      co_spawn(io_context_, secureSession(std::move(socket), std::move(remote_address), local_port), asio::detached);
-    else
-      co_spawn(io_context_, insecureSession(std::move(socket), std::move(remote_address), local_port), asio::detached);
+    }
+    if (ssl_data_) {
+      co_spawn(io_context_, secureSession(std::move(socket), remote_endpoint.address(), remote_endpoint.port(), local_port), asio::detached);
+    } else {
+      co_spawn(io_context_, insecureSession(std::move(socket), remote_endpoint.address(), remote_endpoint.port(), local_port), asio::detached);
+    }
   }
 }
 
-asio::awaitable<void> TcpServer::readLoop(auto& socket, const auto& remote_address, const auto& local_port) {
+asio::awaitable<void> TcpServer::readLoop(auto& socket, asio::ip::address remote_address, asio::ip::port_type remote_port, asio::ip::port_type local_port) {
   std::string read_message;
   while (true) {
     auto [read_error, bytes_read] = co_await asio::async_read_until(socket, asio::dynamic_buffer(read_message), delimiter_, use_nothrow_awaitable);  // NOLINT
@@ -65,16 +68,16 @@ asio::awaitable<void> TcpServer::readLoop(auto& socket, const auto& remote_addre
 
     if (!max_queue_size_ || max_queue_size_ > concurrent_queue_.size()) {
       auto message_str = read_message.substr(0, bytes_read - (consume_delimiter_ ? delimiter_.size() : 0));
-      concurrent_queue_.enqueue(Message(std::move(message_str), IpProtocol::TCP, remote_address, local_port));
+      concurrent_queue_.enqueue(Message(std::move(message_str), IpProtocol::TCP, remote_address, remote_port, local_port));
     } else {
       logger_->log_warn("Queue is full. TCP message ignored.");
     }
     read_message.erase(0, bytes_read);
   }
 }
 
-asio::awaitable<void> TcpServer::insecureSession(asio::ip::tcp::socket socket, asio::ip::address remote_address, asio::ip::port_type local_port) {
-  co_return co_await readLoop(socket, remote_address, local_port);  // NOLINT
+asio::awaitable<void> TcpServer::insecureSession(asio::ip::tcp::socket socket, asio::ip::address remote_address, asio::ip::port_type remote_port, asio::ip::port_type local_port) {
+  co_return co_await readLoop(socket, remote_address, remote_port, local_port);  // NOLINT
 }
 
 namespace {
@@ -95,7 +98,7 @@ asio::ssl::context setupSslContext(SslServerOptions& ssl_data) {
 }
 }  // namespace
 
-asio::awaitable<void> TcpServer::secureSession(asio::ip::tcp::socket socket, asio::ip::address remote_address, asio::ip::port_type local_port) {
+asio::awaitable<void> TcpServer::secureSession(asio::ip::tcp::socket socket, asio::ip::address remote_address, asio::ip::port_type remote_port, asio::ip::port_type local_port) {
   gsl_Expects(ssl_data_);
   auto ssl_context = setupSslContext(*ssl_data_);
   SslSocket ssl_socket(std::move(socket), ssl_context);
@@ -104,7 +107,7 @@ asio::awaitable<void> TcpServer::secureSession(asio::ip::tcp::socket socket, asi
     logger_->log_warn("Handshake with {} failed due to {}", remote_address, handshake_error.message());
     co_return;
   }
-  co_await readLoop(ssl_socket, remote_address, local_port);  // NOLINT
+  co_await readLoop(ssl_socket, remote_address, remote_port, local_port);
 
   asio::error_code ec;
   std::ignore = ssl_socket.lowest_layer().cancel(ec);

extension-framework/src/utils/net/UdpServer.cpp

@@ -15,8 +15,6 @@
  * limitations under the License.
  */
 #include "utils/net/UdpServer.h"
-#include "asio/use_awaitable.hpp"
-#include "asio/detached.hpp"
 #include "utils/net/AsioCoro.h"
 
 namespace org::apache::nifi::minifi::utils::net {
@@ -43,10 +41,11 @@ asio::awaitable<void> UdpServer::doReceive() {
       continue;
     }
     buffer.resize(bytes_received);
-    if (!max_queue_size_ || max_queue_size_ > concurrent_queue_.size())
-      concurrent_queue_.enqueue(utils::net::Message(std::move(buffer), IpProtocol::UDP, sender_endpoint.address(), socket.local_endpoint().port()));
-    else
+    if (!max_queue_size_ || max_queue_size_ > concurrent_queue_.size()) {
+      concurrent_queue_.enqueue(utils::net::Message(std::move(buffer), IpProtocol::UDP, sender_endpoint.address(), sender_endpoint.port(), socket.local_endpoint().port()));
+    } else {
       logger_->log_warn("Queue is full. UDP message ignored.");
+    }
   }
 }
 

extensions/standard-processors/processors/GetTCP.cpp

@@ -117,7 +117,7 @@ void GetTCP::notifyStop() {
 void GetTCP::transferAsFlowFile(const utils::net::Message& message, core::ProcessSession& session) {
   auto flow_file = session.create();
   session.writeBuffer(flow_file, message.message_data);
-  flow_file->setAttribute(GetTCP::SourceEndpoint.name, fmt::format("{}:{}", message.sender_address.to_string(), std::to_string(message.server_port)));
+  flow_file->setAttribute(GetTCP::SourceEndpoint.name, fmt::format("{}:{}", message.remote_address.to_string(), std::to_string(message.remote_port)));
   if (message.is_partial)
     session.transfer(flow_file, Partial);
   else
@@ -128,11 +128,12 @@ void GetTCP::onTrigger(core::ProcessContext&, core::ProcessSession& session) {
   gsl_Expects(max_batch_size_ > 0);
   size_t logs_processed = 0;
   while (!client_->queueEmpty() && logs_processed < max_batch_size_) {
-    utils::net::Message received_message;
-    if (!client_->tryDequeue(received_message))
+    if (const auto received_message = client_->tryDequeue()) {
+      transferAsFlowFile(received_message.value(), session);
+      ++logs_processed;
+    } else {
       break;
-    transferAsFlowFile(received_message, session);
-    ++logs_processed;
+    }
   }
 }
 
@@ -175,8 +176,8 @@ bool GetTCP::TcpClient::queueEmpty() const {
   return concurrent_queue_.empty();
 }
 
-bool GetTCP::TcpClient::tryDequeue(utils::net::Message& received_message) {
-  return concurrent_queue_.tryDequeue(received_message);
+std::optional<utils::net::Message> GetTCP::TcpClient::tryDequeue() {
+  return concurrent_queue_.tryDequeue();
 }
 
 asio::awaitable<std::error_code> GetTCP::TcpClient::readLoop(auto& socket) {
@@ -203,7 +204,10 @@ asio::awaitable<std::error_code> GetTCP::TcpClient::readLoop(auto& socket) {
       continue;
 
     if (!max_queue_size_ || max_queue_size_ > concurrent_queue_.size()) {
-      utils::net::Message message{read_message.substr(0, bytes_read), utils::net::IpProtocol::TCP, socket.lowest_layer().remote_endpoint().address(), socket.lowest_layer().remote_endpoint().port()};
+      const auto remote_endpoint = socket.lowest_layer().remote_endpoint();
+      const auto local_endpoint = socket.lowest_layer().local_endpoint();
+      utils::net::Message message{read_message.substr(0, bytes_read), utils::net::IpProtocol::TCP, remote_endpoint.address(),
+        remote_endpoint.port(), local_endpoint.port()};
       if (previous_didnt_end_with_delimiter || current_doesnt_end_with_delimiter)
         message.is_partial = true;
       concurrent_queue_.enqueue(std::move(message));

extensions/standard-processors/processors/GetTCP.h

@@ -147,14 +147,17 @@ class GetTCP : public core::ProcessorImpl {
         std::optional<size_t> max_message_size,
         std::vector<utils::net::ConnectionId> connections,
         std::shared_ptr<core::logging::Logger> logger);
-
+    TcpClient(const TcpClient&) = delete;
+    TcpClient(TcpClient&&) = delete;
+    TcpClient& operator=(const TcpClient&) = delete;
+    TcpClient& operator=(TcpClient&&) = delete;
     ~TcpClient();
 
     void run();
     void stop();
 
     bool queueEmpty() const;
-    bool tryDequeue(utils::net::Message& received_message);
+    std::optional<utils::net::Message> tryDequeue();
 
    private:
     asio::awaitable<void> doReceiveFrom(const utils::net::ConnectionId& connection_id);

extensions/standard-processors/processors/ListenSyslog.cpp

@@ -95,8 +95,8 @@ void ListenSyslog::transferAsFlowFile(const utils::net::Message& message, core::
 
   session.writeBuffer(flow_file, message.message_data);
   flow_file->setAttribute("syslog.protocol", std::string{magic_enum::enum_name(message.protocol)});
-  flow_file->setAttribute("syslog.port", std::to_string(message.server_port));
-  flow_file->setAttribute("syslog.sender", message.sender_address.to_string());
+  flow_file->setAttribute("syslog.port", std::to_string(message.local_port));
+  flow_file->setAttribute("syslog.sender", message.remote_address.to_string());
   session.transfer(flow_file, valid ? Success : Invalid);
 }
 

extensions/standard-processors/processors/ListenTCP.cpp

@@ -43,8 +43,8 @@ void ListenTCP::onSchedule(core::ProcessContext& context, core::ProcessSessionFa
 void ListenTCP::transferAsFlowFile(const utils::net::Message& message, core::ProcessSession& session) {
   auto flow_file = session.create();
   session.writeBuffer(flow_file, message.message_data);
-  flow_file->setAttribute("tcp.port", std::to_string(message.server_port));
-  flow_file->setAttribute("tcp.sender", message.sender_address.to_string());
+  flow_file->setAttribute("tcp.port", std::to_string(message.local_port));
+  flow_file->setAttribute("tcp.sender", message.remote_address.to_string());
   session.transfer(flow_file, Success);
 }
 

extensions/standard-processors/processors/ListenUDP.cpp

@@ -34,8 +34,9 @@ void ListenUDP::onSchedule(core::ProcessContext& context, core::ProcessSessionFa
 void ListenUDP::transferAsFlowFile(const utils::net::Message& message, core::ProcessSession& session) {
   auto flow_file = session.create();
   session.writeBuffer(flow_file, message.message_data);
-  flow_file->setAttribute("udp.port", std::to_string(message.server_port));
-  flow_file->setAttribute("udp.sender", message.sender_address.to_string());
+  flow_file->setAttribute(ListeningPort.name, std::to_string(message.local_port));
+  flow_file->setAttribute(SenderPort.name, std::to_string(message.remote_port));
+  flow_file->setAttribute(Sender.name, message.remote_address.to_string());
   session.transfer(flow_file, Success);
 }
 

extensions/standard-processors/processors/ListenUDP.h

@@ -63,9 +63,10 @@ class ListenUDP : public NetworkListenerProcessor {
   EXTENSIONAPI static constexpr auto Success = core::RelationshipDefinition{"success", "Messages received successfully will be sent out this relationship."};
   EXTENSIONAPI static constexpr auto Relationships = std::array{Success};
 
-  EXTENSIONAPI static constexpr auto PortOutputAttribute = core::OutputAttributeDefinition<0>{"udp.port", {}, "The sending port the messages were received."};
+  EXTENSIONAPI static constexpr auto ListeningPort = core::OutputAttributeDefinition<0>{"udp.port", {}, "The listening port on which the messages were received."};
   EXTENSIONAPI static constexpr auto Sender = core::OutputAttributeDefinition<0>{"udp.sender", {}, "The sending host of the messages."};
-  EXTENSIONAPI static constexpr auto OutputAttributes = std::array<core::OutputAttributeReference, 2>{PortOutputAttribute, Sender};
+  EXTENSIONAPI static constexpr auto SenderPort = core::OutputAttributeDefinition<0>{"udp.sender.port", {}, "The sending port of the messages."};
+  EXTENSIONAPI static constexpr auto OutputAttributes = std::to_array<core::OutputAttributeReference>({ListeningPort, Sender, SenderPort});
 
   void initialize() override;
   void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& session_factory) override;

extensions/standard-processors/processors/NetworkListenerProcessor.cpp

@@ -30,11 +30,12 @@ void NetworkListenerProcessor::onTrigger(core::ProcessContext&, core::ProcessSes
   gsl_Expects(max_batch_size_ > 0);
   size_t logs_processed = 0;
   while (!server_->queueEmpty() && logs_processed < max_batch_size_) {
-    utils::net::Message received_message;
-    if (!server_->tryDequeue(received_message))
+    if (const auto received_message = server_->tryDequeue()) {
+      transferAsFlowFile(received_message.value(), session);
+      ++logs_processed;
+    } else {
       break;
-    transferAsFlowFile(received_message, session);
-    ++logs_processed;
+    }
   }
 }
 

extensions/standard-processors/processors/NetworkListenerProcessor.h

@@ -18,20 +18,22 @@
 
 #include <memory>
 #include <string>
-#include <utility>
+#include <thread>
 
 #include "core/ProcessorImpl.h"
-#include "minifi-cpp/core/logging/Logger.h"
 #include "minifi-cpp/core/ProcessContext.h"
 #include "core/ProcessSession.h"
-#include "minifi-cpp/core/Property.h"
 #include "utils/net/Server.h"
 
 namespace org::apache::nifi::minifi::processors {
 
 class NetworkListenerProcessor : public core::ProcessorImpl {
  public:
   using ProcessorImpl::ProcessorImpl;
+  NetworkListenerProcessor(const NetworkListenerProcessor&) = delete;
+  NetworkListenerProcessor(NetworkListenerProcessor&&) = delete;
+  NetworkListenerProcessor& operator=(const NetworkListenerProcessor&) = delete;
+  NetworkListenerProcessor& operator=(NetworkListenerProcessor&&) = delete;
   ~NetworkListenerProcessor() override;
 
   void onTrigger(core::ProcessContext& context, core::ProcessSession& session) override;