You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2022/06/14 11:38:56 UTC
[GitHub] [nifi-minifi-cpp] lordgamez opened a new pull request, #1350: MINIFICPP-1857 Create ListenTCP processor
lordgamez opened a new pull request, #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350
https://issues.apache.org/jira/browse/MINIFICPP-1857
--------------------------------------
Thank you for submitting a contribution to Apache NiFi - MiNiFi C++.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [ ] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [ ] Does your PR title start with MINIFICPP-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [ ] Has your PR been rebased against the latest commit within the target branch (typically main)?
- [ ] Is your initial contribution a single, squashed commit?
### For code changes:
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the LICENSE file?
- [ ] If applicable, have you updated the NOTICE file?
### For documentation related changes:
- [ ] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check GitHub Actions CI results for build issues and submit an update to your PR as soon as possible.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907092339
##########
PROCESSORS.md:
##########
@@ -1279,6 +1280,24 @@ In the list below, the names of required properties appear in bold. Any other pr
+## ListenTCP
+
+### Description
+
+Listens for incoming TCP connections and reads data from each connection using a line separator as the message demarcator. For each message the processor produces a single FlowFile.
+
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+|-------------------------------|---------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Listening Port** | | | The port to listen on for communication. |
+| **Max Batch Size** | 500 | | The maximum number of messages to process at a time. |
+| **Max Size of Message Queue** | 0 | | Maximum number of messages allowed to be buffered before processing them when the processor is triggered. If the buffer full, the message is ignored. If set to zero the buffer is unlimited. |
Review Comment:
Updated in 28faa0284cea7c53f796f27cd0effebb5d260f49
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907325646
##########
extensions/standard-processors/processors/NetworkListenerProcessor.cpp:
##########
@@ -0,0 +1,79 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "NetworkListenerProcessor.h"
+#include "utils/net/UdpServer.h"
+#include "utils/net/TcpServer.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+NetworkListenerProcessor::~NetworkListenerProcessor() {
+ stopServer();
+}
+
+void NetworkListenerProcessor::onTrigger(const std::shared_ptr<core::ProcessContext>&, const std::shared_ptr<core::ProcessSession>& session) {
+ gsl_Expects(session && 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))
+ break;
+ transferAsFlowFile(received_message, *session);
+ ++logs_processed;
+ }
+}
+
+void NetworkListenerProcessor::startServer(
+ const core::ProcessContext& context, const core::Property& max_batch_size_prop, const core::Property& max_queue_size_prop, const core::Property& port_prop, utils::net::IpProtocol protocol) {
+ gsl_Expects(!server_thread_.joinable() && !server_);
+ context.getProperty(max_batch_size_prop.getName(), max_batch_size_);
+ if (max_batch_size_ < 1)
+ throw Exception(PROCESSOR_EXCEPTION, "Max Batch Size property is invalid");
+
+ uint64_t max_queue_size = 0;
+ context.getProperty(max_queue_size_prop.getName(), max_queue_size);
+ auto max_queue_size_opt = max_queue_size > 0 ? std::optional<uint64_t>(max_queue_size) : std::nullopt;
+
+ int port;
+ context.getProperty(port_prop.getName(), port);
+
+ if (protocol == utils::net::IpProtocol::UDP) {
+ server_ = std::make_unique<utils::net::UdpServer>(max_queue_size_opt, port, logger_);
+ } else if (protocol == utils::net::IpProtocol::TCP) {
+ server_ = std::make_unique<utils::net::TcpServer>(max_queue_size_opt, port, logger_);
+ } else {
+ throw Exception(PROCESSOR_EXCEPTION, "Invalid protocol");
+ }
+
+ server_thread_ = std::thread([this]() { server_->run(); });
+ logger_->log_debug("Started %s server on port %d with %s max queue size and %zu max batch size",
+ protocol.toString(),
+ port,
+ max_queue_size_opt ? std::to_string(*max_queue_size_opt) : "unlimited",
+ max_batch_size_);
+}
+
+void NetworkListenerProcessor::stopServer() {
+ if (server_) {
+ server_->stop();
+ }
+ if (server_thread_.joinable()) {
+ server_thread_.join();
+ }
+ server_.reset();
Review Comment:
I think it's the responsibility of the user of the object to decide if they want to start the server a separate thread entirely or they want to run it on an already created thread.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r906106551
##########
PROCESSORS.md:
##########
@@ -1279,6 +1280,24 @@ In the list below, the names of required properties appear in bold. Any other pr
+## ListenTCP
+
+### Description
+
+Listens for incoming TCP connections and reads data from each connection using a line separator as the message demarcator. For each message the processor produces a single FlowFile.
+
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+|-------------------------------|---------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Listening Port** | | | The port to listen on for communication. |
+| **Max Batch Size** | 500 | | The maximum number of messages to process at a time. |
+| **Max Size of Message Queue** | 0 | | Maximum number of messages allowed to be buffered before processing them when the processor is triggered. If the buffer full, the message is ignored. If set to zero the buffer is unlimited. |
Review Comment:
Good point, the same NiFi processor has a 10000 limit as a default. Should we go with that? Do you think we should also change the default size of the ListenSyslog queue as well?
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907188647
##########
docker/test/integration/minifi/core/TcpClientContainer.py:
##########
@@ -0,0 +1,23 @@
+import logging
+from .Container import Container
+
+
+class TcpClientContainer(Container):
+ def __init__(self, name, vols, network, image_store, command=None):
+ super().__init__(name, 'tcp-client', vols, network, image_store, command)
Review Comment:
We actually do this in `ListenTcpTests.cpp` where we test the processor's functionality. I think it still has some added value as in the docker test we have a system level test instead of the unit level like in ListenTcpTests where we only test the processor separately. This way we can be sure that the processor is registered and works on a system level when defined in a flow configuration and can listen to messages on a separate network dedicated to it in the docker environment.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
fgerlits commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r905172649
##########
PROCESSORS.md:
##########
@@ -1279,6 +1280,24 @@ In the list below, the names of required properties appear in bold. Any other pr
+## ListenTCP
+
+### Description
+
+Listens for incoming TCP connections and reads data from each connection using a line separator as the message demarcator. For each message the processor produces a single FlowFile.
+
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+|-------------------------------|---------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Listening Port** | | | The port to listen on for communication. |
+| **Max Batch Size** | 500 | | The maximum number of messages to process at a time. |
+| **Max Size of Message Queue** | 0 | | Maximum number of messages allowed to be buffered before processing them when the processor is triggered. If the buffer full, the message is ignored. If set to zero the buffer is unlimited. |
Review Comment:
Are we sure we want the default buffer size to be unlimited? I understand that dropping messages is bad, but unlimited memory consumption can be even worse.
##########
docker/test/integration/steps/steps.py:
##########
@@ -416,6 +416,12 @@ def step_impl(context):
context.test.start_splunk()
+# TCP client
+@given('a TCP client is setup to send logs to minifi')
Review Comment:
"set up" should be two words; "setup" is a noun
##########
PROCESSORS.md:
##########
@@ -1279,6 +1280,24 @@ In the list below, the names of required properties appear in bold. Any other pr
+## ListenTCP
+
+### Description
+
+Listens for incoming TCP connections and reads data from each connection using a line separator as the message demarcator. For each message the processor produces a single FlowFile.
+
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+|-------------------------------|---------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Listening Port** | | | The port to listen on for communication. |
+| **Max Batch Size** | 500 | | The maximum number of messages to process at a time. |
+| **Max Size of Message Queue** | 0 | | Maximum number of messages allowed to be buffered before processing them when the processor is triggered. If the buffer full, the message is ignored. If set to zero the buffer is unlimited. |
Review Comment:
typo:
```suggestion
| **Max Size of Message Queue** | 0 | | Maximum number of messages allowed to be buffered before processing them when the processor is triggered. If the buffer is full, the message is ignored. If set to zero the buffer is unlimited. |
```
##########
extensions/standard-processors/processors/ListenSyslog.cpp:
##########
@@ -86,72 +86,20 @@ void ListenSyslog::initialize() {
void ListenSyslog::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>&) {
gsl_Expects(context && !server_thread_.joinable() && !server_);
Review Comment:
Since `server_thread_` and `server_` have been moved to `NetworkListenerProcessor`, I would move these two expectation checks to `NetworkListenerProcessor::startServer()`.
(Also from `ListenTCP::onSchedule()`.)
##########
libminifi/include/utils/net/UdpServer.h:
##########
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <optional>
+#include <memory>
+#include <string>
+
+#include "Server.h"
+#include "utils/MinifiConcurrentQueue.h"
+#include "core/logging/Logger.h"
+#include "core/logging/LoggerConfiguration.h"
+#include "asio/ts/buffer.hpp"
+#include "asio/ts/internet.hpp"
+#include "asio/streambuf.hpp"
+
+namespace org::apache::nifi::minifi::utils::net {
+
+class UdpServer : public Server {
+ public:
+ UdpServer(std::optional<size_t> max_queue_size,
+ uint16_t port,
+ std::shared_ptr<core::logging::Logger> logger);
+
+ private:
+ void doReceive();
+
+ asio::ip::udp::socket socket_;
+ asio::ip::udp::endpoint sender_endpoint_;
+ std::string buffer_;
+
+ static inline constexpr size_t MAX_UDP_PACKET_SIZE = 65535;
Review Comment:
Is this `inline` necessary? I think `static constexpr` member variables are `inline` already.
##########
extensions/standard-processors/tests/unit/ListenTcpTests.cpp:
##########
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <string>
+
+#include "Catch.h"
+#include "processors/ListenTCP.h"
+#include "SingleProcessorTestController.h"
+#include "asio.hpp"
+
+using ListenTCP = org::apache::nifi::minifi::processors::ListenTCP;
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::test {
+
+constexpr uint64_t PORT = 10254;
+using ProcessorTriggerResult = std::unordered_map<core::Relationship, std::vector<std::shared_ptr<core::FlowFile>>>;
+
+void sendMessagesViaTCP(const std::vector<std::string_view>& contents) {
+ asio::io_context io_context;
+ asio::ip::tcp::socket socket(io_context);
+ asio::ip::tcp::endpoint remote_endpoint(asio::ip::address::from_string("127.0.0.1"), PORT);
+ socket.connect(remote_endpoint);
+ std::error_code err;
+ for (auto& content : contents) {
+ std::string tcp_message(content);
+ tcp_message += '\n';
+ socket.send(asio::buffer(tcp_message, tcp_message.size()), 0, err);
+ }
+ REQUIRE(!err);
+ socket.close();
+}
+
+void check_for_attributes(core::FlowFile& flow_file) {
+ CHECK(std::to_string(PORT) == flow_file.getAttribute("tcp.port"));
+ CHECK("127.0.0.1" == flow_file.getAttribute("tcp.sender"));
+}
+
+bool triggerUntil(test::SingleProcessorTestController& controller,
+ const std::unordered_map<core::Relationship, size_t>& expected_quantities,
+ ProcessorTriggerResult& result,
+ const std::chrono::milliseconds max_duration,
+ const std::chrono::milliseconds wait_time = 50ms) {
+ auto start_time = std::chrono::steady_clock::now();
+ while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ for (auto& [relationship, flow_files] : controller.trigger()) {
+ result[relationship].insert(result[relationship].end(), flow_files.begin(), flow_files.end());
+ }
+ bool expected_quantities_met = true;
+ for (const auto& [relationship, expected_quantity] : expected_quantities) {
+ if (result[relationship].size() < expected_quantity) {
+ expected_quantities_met = false;
+ break;
+ }
+ }
+ if (expected_quantities_met)
+ return true;
Review Comment:
just a suggestion, feel free to ignore, but I think this would be nicer:
```suggestion
if (ranges::all_of(expected_quantities, [&result](const auto& kv) {
const auto& [relationship, expected_quantity] = kv;
return result[relationship].size() >= expected_quantity;
})) {
return true;
}
```
##########
extensions/standard-processors/tests/unit/ListenTcpTests.cpp:
##########
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <string>
+
+#include "Catch.h"
+#include "processors/ListenTCP.h"
+#include "SingleProcessorTestController.h"
+#include "asio.hpp"
+
+using ListenTCP = org::apache::nifi::minifi::processors::ListenTCP;
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::test {
+
+constexpr uint64_t PORT = 10254;
+using ProcessorTriggerResult = std::unordered_map<core::Relationship, std::vector<std::shared_ptr<core::FlowFile>>>;
+
+void sendMessagesViaTCP(const std::vector<std::string_view>& contents) {
+ asio::io_context io_context;
+ asio::ip::tcp::socket socket(io_context);
+ asio::ip::tcp::endpoint remote_endpoint(asio::ip::address::from_string("127.0.0.1"), PORT);
+ socket.connect(remote_endpoint);
+ std::error_code err;
+ for (auto& content : contents) {
+ std::string tcp_message(content);
+ tcp_message += '\n';
+ socket.send(asio::buffer(tcp_message, tcp_message.size()), 0, err);
+ }
+ REQUIRE(!err);
+ socket.close();
+}
+
+void check_for_attributes(core::FlowFile& flow_file) {
+ CHECK(std::to_string(PORT) == flow_file.getAttribute("tcp.port"));
+ CHECK("127.0.0.1" == flow_file.getAttribute("tcp.sender"));
+}
+
+bool triggerUntil(test::SingleProcessorTestController& controller,
+ const std::unordered_map<core::Relationship, size_t>& expected_quantities,
+ ProcessorTriggerResult& result,
+ const std::chrono::milliseconds max_duration,
+ const std::chrono::milliseconds wait_time = 50ms) {
+ auto start_time = std::chrono::steady_clock::now();
+ while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ for (auto& [relationship, flow_files] : controller.trigger()) {
+ result[relationship].insert(result[relationship].end(), flow_files.begin(), flow_files.end());
+ }
+ bool expected_quantities_met = true;
+ for (const auto& [relationship, expected_quantity] : expected_quantities) {
+ if (result[relationship].size() < expected_quantity) {
+ expected_quantities_met = false;
+ break;
+ }
+ }
+ if (expected_quantities_met)
+ return true;
+ std::this_thread::sleep_for(wait_time);
+ }
+ return false;
+}
+
+bool countLogOccurrencesUntil(const std::string& pattern,
+ const int occurrences,
+ const std::chrono::milliseconds max_duration,
+ const std::chrono::milliseconds wait_time = 50ms) {
+ auto start_time = std::chrono::steady_clock::now();
+ while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ if (LogTestController::getInstance().countOccurrences(pattern) == occurrences)
+ return true;
+ std::this_thread::sleep_for(wait_time);
+ }
+ return false;
+}
Review Comment:
This is quite generic, and it's also used in `ListenSyslogTests`, so it may be worth moving it to `TestUtils` or `IntegrationTestUtils`.
##########
docker/test/integration/steps/steps.py:
##########
@@ -416,6 +416,12 @@ def step_impl(context):
context.test.start_splunk()
+# TCP client
+@given('a TCP client is setup to send logs to minifi')
Review Comment:
why "logs"? it sends a test message
```suggestion
@given('a TCP client is set up to send a test TCP message to minifi')
```
##########
extensions/standard-processors/processors/NetworkListenerProcessor.cpp:
##########
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "NetworkListenerProcessor.h"
+#include "utils/net/UdpServer.h"
+#include "utils/net/TcpServer.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+NetworkListenerProcessor::~NetworkListenerProcessor() {
+ stopServer();
+}
+
+void NetworkListenerProcessor::onTrigger(const std::shared_ptr<core::ProcessContext>&, const std::shared_ptr<core::ProcessSession>& session) {
+ gsl_Expects(session && 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))
+ break;
+ transferAsFlowFile(received_message, *session);
+ ++logs_processed;
+ }
+}
+
+void NetworkListenerProcessor::startServer(
+ const core::ProcessContext& context, const core::Property& max_batch_size_prop, const core::Property& max_queue_size_prop, const core::Property& port_prop, utils::net::Protocol protocol) {
+ context.getProperty(max_batch_size_prop.getName(), max_batch_size_);
+ if (max_batch_size_ < 1)
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Max Batch Size property is invalid");
+
+ uint64_t max_queue_size = 0;
+ context.getProperty(max_queue_size_prop.getName(), max_queue_size);
+ auto max_queue_size_opt = max_queue_size > 0 ? std::optional<uint64_t>(max_queue_size) : std::nullopt;
+
+ int port;
+ context.getProperty(port_prop.getName(), port);
+
+ if (protocol == utils::net::Protocol::UDP) {
+ server_ = std::make_unique<utils::net::UdpServer>(max_queue_size_opt, port, logger_);
+ } else if (protocol == utils::net::Protocol::TCP) {
+ server_ = std::make_unique<utils::net::TcpServer>(max_queue_size_opt, port, logger_);
+ } else {
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Invalid protocol");
+ }
+
+ server_thread_ = std::thread([this]() { server_->run(); });
+ logger_->log_debug("Started %s server on port %d with %s max queue size and %zu max batch size",
+ protocol.toString(),
+ port,
+ max_queue_size_opt ? std::to_string(*max_queue_size_opt) : "no",
Review Comment:
nitpicking, but I would use "unlimited" instead of "no"
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907095053
##########
libminifi/include/utils/net/UdpServer.h:
##########
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <optional>
+#include <memory>
+#include <string>
+
+#include "Server.h"
+#include "utils/MinifiConcurrentQueue.h"
+#include "core/logging/Logger.h"
+#include "core/logging/LoggerConfiguration.h"
+#include "asio/ts/buffer.hpp"
+#include "asio/ts/internet.hpp"
+#include "asio/streambuf.hpp"
+
+namespace org::apache::nifi::minifi::utils::net {
+
+class UdpServer : public Server {
+ public:
+ UdpServer(std::optional<size_t> max_queue_size,
+ uint16_t port,
+ std::shared_ptr<core::logging::Logger> logger);
+
+ private:
+ void doReceive();
+
+ asio::ip::udp::socket socket_;
+ asio::ip::udp::endpoint sender_endpoint_;
+ std::string buffer_;
+
+ static inline constexpr size_t MAX_UDP_PACKET_SIZE = 65535;
Review Comment:
Removed in 28faa0284cea7c53f796f27cd0effebb5d260f49
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
fgerlits commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907197972
##########
extensions/standard-processors/tests/unit/ListenTcpTests.cpp:
##########
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <string>
+
+#include "Catch.h"
+#include "processors/ListenTCP.h"
+#include "SingleProcessorTestController.h"
+#include "asio.hpp"
+
+using ListenTCP = org::apache::nifi::minifi::processors::ListenTCP;
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::test {
+
+constexpr uint64_t PORT = 10254;
+using ProcessorTriggerResult = std::unordered_map<core::Relationship, std::vector<std::shared_ptr<core::FlowFile>>>;
+
+void sendMessagesViaTCP(const std::vector<std::string_view>& contents) {
+ asio::io_context io_context;
+ asio::ip::tcp::socket socket(io_context);
+ asio::ip::tcp::endpoint remote_endpoint(asio::ip::address::from_string("127.0.0.1"), PORT);
+ socket.connect(remote_endpoint);
+ std::error_code err;
+ for (auto& content : contents) {
+ std::string tcp_message(content);
+ tcp_message += '\n';
+ socket.send(asio::buffer(tcp_message, tcp_message.size()), 0, err);
+ }
+ REQUIRE(!err);
+ socket.close();
+}
+
+void check_for_attributes(core::FlowFile& flow_file) {
+ CHECK(std::to_string(PORT) == flow_file.getAttribute("tcp.port"));
+ CHECK("127.0.0.1" == flow_file.getAttribute("tcp.sender"));
+}
+
+bool triggerUntil(test::SingleProcessorTestController& controller,
+ const std::unordered_map<core::Relationship, size_t>& expected_quantities,
+ ProcessorTriggerResult& result,
+ const std::chrono::milliseconds max_duration,
+ const std::chrono::milliseconds wait_time = 50ms) {
+ auto start_time = std::chrono::steady_clock::now();
+ while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ for (auto& [relationship, flow_files] : controller.trigger()) {
+ result[relationship].insert(result[relationship].end(), flow_files.begin(), flow_files.end());
+ }
+ bool expected_quantities_met = true;
+ for (const auto& [relationship, expected_quantity] : expected_quantities) {
+ if (result[relationship].size() < expected_quantity) {
+ expected_quantities_met = false;
+ break;
+ }
+ }
+ if (expected_quantities_met)
+ return true;
+ std::this_thread::sleep_for(wait_time);
+ }
+ return false;
+}
+
+bool countLogOccurrencesUntil(const std::string& pattern,
+ const int occurrences,
+ const std::chrono::milliseconds max_duration,
+ const std::chrono::milliseconds wait_time = 50ms) {
+ auto start_time = std::chrono::steady_clock::now();
+ while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ if (LogTestController::getInstance().countOccurrences(pattern) == occurrences)
+ return true;
+ std::this_thread::sleep_for(wait_time);
+ }
+ return false;
+}
Review Comment:
This could be done as part of MINIFICPP-1872, but I would prefer the functions in `test/Utils.h` to be in a namespace instead of being global.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907314000
##########
extensions/standard-processors/processors/NetworkListenerProcessor.cpp:
##########
@@ -0,0 +1,79 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "NetworkListenerProcessor.h"
+#include "utils/net/UdpServer.h"
+#include "utils/net/TcpServer.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+NetworkListenerProcessor::~NetworkListenerProcessor() {
+ stopServer();
+}
+
+void NetworkListenerProcessor::onTrigger(const std::shared_ptr<core::ProcessContext>&, const std::shared_ptr<core::ProcessSession>& session) {
+ gsl_Expects(session && 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))
+ break;
+ transferAsFlowFile(received_message, *session);
+ ++logs_processed;
+ }
+}
+
+void NetworkListenerProcessor::startServer(
+ const core::ProcessContext& context, const core::Property& max_batch_size_prop, const core::Property& max_queue_size_prop, const core::Property& port_prop, utils::net::IpProtocol protocol) {
+ gsl_Expects(!server_thread_.joinable() && !server_);
+ context.getProperty(max_batch_size_prop.getName(), max_batch_size_);
+ if (max_batch_size_ < 1)
+ throw Exception(PROCESSOR_EXCEPTION, "Max Batch Size property is invalid");
+
+ uint64_t max_queue_size = 0;
+ context.getProperty(max_queue_size_prop.getName(), max_queue_size);
+ auto max_queue_size_opt = max_queue_size > 0 ? std::optional<uint64_t>(max_queue_size) : std::nullopt;
+
+ int port;
+ context.getProperty(port_prop.getName(), port);
+
+ if (protocol == utils::net::IpProtocol::UDP) {
+ server_ = std::make_unique<utils::net::UdpServer>(max_queue_size_opt, port, logger_);
+ } else if (protocol == utils::net::IpProtocol::TCP) {
+ server_ = std::make_unique<utils::net::TcpServer>(max_queue_size_opt, port, logger_);
+ } else {
+ throw Exception(PROCESSOR_EXCEPTION, "Invalid protocol");
+ }
+
+ server_thread_ = std::thread([this]() { server_->run(); });
+ logger_->log_debug("Started %s server on port %d with %s max queue size and %zu max batch size",
+ protocol.toString(),
+ port,
+ max_queue_size_opt ? std::to_string(*max_queue_size_opt) : "unlimited",
+ max_batch_size_);
+}
+
+void NetworkListenerProcessor::stopServer() {
+ if (server_) {
+ server_->stop();
+ }
+ if (server_thread_.joinable()) {
+ server_thread_.join();
+ }
+ server_.reset();
Review Comment:
it feels like the `server_thread_` should be the responsibility of the `server_` object
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907092561
##########
docker/test/integration/steps/steps.py:
##########
@@ -416,6 +416,12 @@ def step_impl(context):
context.test.start_splunk()
+# TCP client
+@given('a TCP client is setup to send logs to minifi')
Review Comment:
Updated in 28faa0284cea7c53f796f27cd0effebb5d260f49
##########
docker/test/integration/steps/steps.py:
##########
@@ -416,6 +416,12 @@ def step_impl(context):
context.test.start_splunk()
+# TCP client
+@given('a TCP client is setup to send logs to minifi')
Review Comment:
Updated in 28faa0284cea7c53f796f27cd0effebb5d260f49
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907202853
##########
extensions/standard-processors/tests/unit/ListenTcpTests.cpp:
##########
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <string>
+
+#include "Catch.h"
+#include "processors/ListenTCP.h"
+#include "SingleProcessorTestController.h"
+#include "asio.hpp"
+
+using ListenTCP = org::apache::nifi::minifi::processors::ListenTCP;
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::test {
+
+constexpr uint64_t PORT = 10254;
+using ProcessorTriggerResult = std::unordered_map<core::Relationship, std::vector<std::shared_ptr<core::FlowFile>>>;
+
+void sendMessagesViaTCP(const std::vector<std::string_view>& contents) {
+ asio::io_context io_context;
+ asio::ip::tcp::socket socket(io_context);
+ asio::ip::tcp::endpoint remote_endpoint(asio::ip::address::from_string("127.0.0.1"), PORT);
+ socket.connect(remote_endpoint);
+ std::error_code err;
+ for (auto& content : contents) {
+ std::string tcp_message(content);
+ tcp_message += '\n';
+ socket.send(asio::buffer(tcp_message, tcp_message.size()), 0, err);
+ }
+ REQUIRE(!err);
+ socket.close();
+}
+
+void check_for_attributes(core::FlowFile& flow_file) {
+ CHECK(std::to_string(PORT) == flow_file.getAttribute("tcp.port"));
+ CHECK("127.0.0.1" == flow_file.getAttribute("tcp.sender"));
+}
+
+bool triggerUntil(test::SingleProcessorTestController& controller,
+ const std::unordered_map<core::Relationship, size_t>& expected_quantities,
+ ProcessorTriggerResult& result,
+ const std::chrono::milliseconds max_duration,
+ const std::chrono::milliseconds wait_time = 50ms) {
+ auto start_time = std::chrono::steady_clock::now();
+ while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ for (auto& [relationship, flow_files] : controller.trigger()) {
+ result[relationship].insert(result[relationship].end(), flow_files.begin(), flow_files.end());
+ }
+ bool expected_quantities_met = true;
+ for (const auto& [relationship, expected_quantity] : expected_quantities) {
+ if (result[relationship].size() < expected_quantity) {
+ expected_quantities_met = false;
+ break;
+ }
+ }
+ if (expected_quantities_met)
+ return true;
+ std::this_thread::sleep_for(wait_time);
+ }
+ return false;
+}
+
+bool countLogOccurrencesUntil(const std::string& pattern,
+ const int occurrences,
+ const std::chrono::milliseconds max_duration,
+ const std::chrono::milliseconds wait_time = 50ms) {
+ auto start_time = std::chrono::steady_clock::now();
+ while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ if (LogTestController::getInstance().countOccurrences(pattern) == occurrences)
+ return true;
+ std::this_thread::sleep_for(wait_time);
+ }
+ return false;
+}
Review Comment:
I thought about that, but I concluded that it would be better to be part of MINIFICPP-1872 as I didn't want to touch additional files depending on it in this PR
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907288117
##########
extensions/standard-processors/processors/ListenSyslog.cpp:
##########
@@ -53,15 +53,15 @@ const core::Property ListenSyslog::ParseMessages(
const core::Property ListenSyslog::MaxQueueSize(
core::PropertyBuilder::createProperty("Max Size of Message Queue")
->withDescription("Maximum number of Syslog messages allowed to be buffered before processing them when the processor is triggered. "
- "If the buffer full, the message is ignored. If set to zero the buffer is unlimited.")
- ->withDefaultValue<uint64_t>(0)->build());
+ "If the is buffer full, the message is ignored. If set to zero the buffer is unlimited.")
Review Comment:
Good catch, fixed in 2798590a9068fcdf1061f4284f30e3454385ec0c
##########
extensions/standard-processors/processors/ListenTCP.h:
##########
@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include "NetworkListenerProcessor.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+class ListenTCP : public NetworkListenerProcessor {
+ public:
+ explicit ListenTCP(const std::string& name, const utils::Identifier& uuid = {})
+ : NetworkListenerProcessor(name, uuid, core::logging::LoggerFactory<ListenTCP>::getLogger()) {
+ }
+ ListenTCP(const ListenTCP&) = delete;
+ ListenTCP(ListenTCP&&) = delete;
+ ListenTCP& operator=(const ListenTCP&) = delete;
+ ListenTCP& operator=(ListenTCP&&) = delete;
Review Comment:
Good point, updated in 2798590a9068fcdf1061f4284f30e3454385ec0c
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] martinzink commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
martinzink commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r898986178
##########
extensions/standard-processors/processors/ListenTCP.h:
##########
@@ -0,0 +1,64 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include "NetworkListenerProcessor.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+class ListenTCP : public NetworkListenerProcessor {
+ public:
+ explicit ListenTCP(const std::string& name, const utils::Identifier& uuid = {})
+ : NetworkListenerProcessor(name, uuid, core::logging::LoggerFactory<ListenTCP>::getLogger()) {
+ }
+ ListenTCP(const ListenTCP&) = delete;
+ ListenTCP(ListenTCP&&) = delete;
+ ListenTCP& operator=(const ListenTCP&) = delete;
+ ListenTCP& operator=(ListenTCP&&) = delete;
+
+ EXTENSIONAPI static constexpr const char* Description = "Listens for Syslog messages being sent to a given port over TCP or UDP. "
+ "Incoming messages are optionally checked against regular expressions for RFC5424 and RFC3164 formatted messages. "
+ "With parsing enabled the individual parts of the message will be placed as FlowFile attributes and "
+ "valid messages will be transferred to success relationship, while invalid messages will be transferred to invalid relationship. "
+ "With parsing disabled all message will be routed to the success relationship, but it will only contain the sender, protocol, and port attributes";
Review Comment:
This should be updated to the ListenTCP description like in PROCESSORS.md
##########
libminifi/src/utils/net/TcpServer.cpp:
##########
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "utils/net/TcpServer.h"
+
+namespace org::apache::nifi::minifi::utils::net {
+
+TcpSession::TcpSession(asio::io_context& io_context, utils::ConcurrentQueue<Message>& concurrent_queue, std::optional<size_t> max_queue_size, std::shared_ptr<core::logging::Logger> logger)
+ : concurrent_queue_(concurrent_queue),
+ max_queue_size_(max_queue_size),
+ socket_(io_context),
+ logger_(std::move(logger)) {
+}
+
+asio::ip::tcp::socket& TcpSession::getSocket() {
+ return socket_;
+}
+
+void TcpSession::start() {
+ asio::async_read_until(socket_,
+ buffer_,
+ '\n',
+ [self = shared_from_this()](const auto& error_code, size_t) -> void {
+ self->handleReadUntilNewLine(error_code);
+ });
+}
+
+void TcpSession::handleReadUntilNewLine(std::error_code error_code) {
+ if (error_code)
+ return;
+ std::istream is(&buffer_);
+ std::string message;
+ std::getline(is, message);
+ if (!max_queue_size_ || max_queue_size_ > concurrent_queue_.size())
+ concurrent_queue_.enqueue(Message(message, Protocol::TCP, socket_.remote_endpoint().address(), socket_.local_endpoint().port()));
+ else
+ logger_->log_warn("Queue is full. Syslog message ignored.");
Review Comment:
Since it is not a syslog specific class, this and the other warning in UDP server should probably just says TCP/UDP message ignored.
##########
docker/test/integration/features/network_listener.feature:
##########
@@ -0,0 +1,14 @@
+Feature: Minifi C++ can act as a syslog listener
Review Comment:
```suggestion
Feature: Minifi C++ can act as a TCP listener
```
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r899104181
##########
docker/test/integration/features/network_listener.feature:
##########
@@ -0,0 +1,14 @@
+Feature: Minifi C++ can act as a syslog listener
Review Comment:
Thanks, I changed to "network listener" to be able to use this feature in other networking (like UDP) tests in the future. Changed in daedc5003b406d07957791d4d5ded2420888fc3f
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907280099
##########
extensions/standard-processors/processors/ListenTCP.h:
##########
@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include "NetworkListenerProcessor.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+class ListenTCP : public NetworkListenerProcessor {
+ public:
+ explicit ListenTCP(const std::string& name, const utils::Identifier& uuid = {})
+ : NetworkListenerProcessor(name, uuid, core::logging::LoggerFactory<ListenTCP>::getLogger()) {
+ }
+ ListenTCP(const ListenTCP&) = delete;
+ ListenTCP(ListenTCP&&) = delete;
+ ListenTCP& operator=(const ListenTCP&) = delete;
+ ListenTCP& operator=(ListenTCP&&) = delete;
Review Comment:
the base class `Connectable` already has these members deleted
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] szaszm commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r906203581
##########
extensions/standard-processors/processors/NetworkListenerProcessor.cpp:
##########
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "NetworkListenerProcessor.h"
+#include "utils/net/UdpServer.h"
+#include "utils/net/TcpServer.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+NetworkListenerProcessor::~NetworkListenerProcessor() {
+ stopServer();
+}
+
+void NetworkListenerProcessor::onTrigger(const std::shared_ptr<core::ProcessContext>&, const std::shared_ptr<core::ProcessSession>& session) {
+ gsl_Expects(session && 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))
+ break;
+ transferAsFlowFile(received_message, *session);
+ ++logs_processed;
+ }
+}
+
+void NetworkListenerProcessor::startServer(
+ const core::ProcessContext& context, const core::Property& max_batch_size_prop, const core::Property& max_queue_size_prop, const core::Property& port_prop, utils::net::Protocol protocol) {
+ context.getProperty(max_batch_size_prop.getName(), max_batch_size_);
+ if (max_batch_size_ < 1)
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Max Batch Size property is invalid");
Review Comment:
I would use the PROCESSOR_EXCEPTION type here instead, because this is not the scheduler / thread pool.
##########
extensions/standard-processors/processors/ListenSyslog.cpp:
##########
@@ -34,8 +34,8 @@ const core::Property ListenSyslog::ProtocolProperty(
core::PropertyBuilder::createProperty("Protocol")
->withDescription("The protocol for Syslog communication.")
->isRequired(true)
- ->withAllowableValues(Protocol::values())
- ->withDefaultValue(toString(Protocol::UDP))
+ ->withAllowableValues(utils::net::Protocol::values())
+ ->withDefaultValue(toString(utils::net::Protocol::UDP))
Review Comment:
There is an `IpProtocol` enum in utils/net/DNS.h, would it make sense to use that here?
##########
docker/test/integration/minifi/core/TcpClientContainer.py:
##########
@@ -0,0 +1,23 @@
+import logging
+from .Container import Container
+
+
+class TcpClientContainer(Container):
+ def __init__(self, name, vols, network, image_store, command=None):
+ super().__init__(name, 'tcp-client', vols, network, image_store, command)
Review Comment:
I feel like a docker container for a simple TCP client is overkill. We can just open a socket on localhost and connect. What do you think?
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907092740
##########
extensions/standard-processors/processors/ListenSyslog.cpp:
##########
@@ -86,72 +86,20 @@ void ListenSyslog::initialize() {
void ListenSyslog::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>&) {
gsl_Expects(context && !server_thread_.joinable() && !server_);
Review Comment:
Updated in 28faa0284cea7c53f796f27cd0effebb5d260f49
##########
extensions/standard-processors/processors/NetworkListenerProcessor.cpp:
##########
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "NetworkListenerProcessor.h"
+#include "utils/net/UdpServer.h"
+#include "utils/net/TcpServer.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+NetworkListenerProcessor::~NetworkListenerProcessor() {
+ stopServer();
+}
+
+void NetworkListenerProcessor::onTrigger(const std::shared_ptr<core::ProcessContext>&, const std::shared_ptr<core::ProcessSession>& session) {
+ gsl_Expects(session && 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))
+ break;
+ transferAsFlowFile(received_message, *session);
+ ++logs_processed;
+ }
+}
+
+void NetworkListenerProcessor::startServer(
+ const core::ProcessContext& context, const core::Property& max_batch_size_prop, const core::Property& max_queue_size_prop, const core::Property& port_prop, utils::net::Protocol protocol) {
+ context.getProperty(max_batch_size_prop.getName(), max_batch_size_);
+ if (max_batch_size_ < 1)
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Max Batch Size property is invalid");
+
+ uint64_t max_queue_size = 0;
+ context.getProperty(max_queue_size_prop.getName(), max_queue_size);
+ auto max_queue_size_opt = max_queue_size > 0 ? std::optional<uint64_t>(max_queue_size) : std::nullopt;
+
+ int port;
+ context.getProperty(port_prop.getName(), port);
+
+ if (protocol == utils::net::Protocol::UDP) {
+ server_ = std::make_unique<utils::net::UdpServer>(max_queue_size_opt, port, logger_);
+ } else if (protocol == utils::net::Protocol::TCP) {
+ server_ = std::make_unique<utils::net::TcpServer>(max_queue_size_opt, port, logger_);
+ } else {
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Invalid protocol");
+ }
+
+ server_thread_ = std::thread([this]() { server_->run(); });
+ logger_->log_debug("Started %s server on port %d with %s max queue size and %zu max batch size",
+ protocol.toString(),
+ port,
+ max_queue_size_opt ? std::to_string(*max_queue_size_opt) : "no",
Review Comment:
Updated in 28faa0284cea7c53f796f27cd0effebb5d260f49
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907341601
##########
PROCESSORS.md:
##########
@@ -1279,6 +1280,24 @@ In the list below, the names of required properties appear in bold. Any other pr
+## ListenTCP
+
+### Description
+
+Listens for incoming TCP connections and reads data from each connection using a line separator as the message demarcator. For each message the processor produces a single FlowFile.
+
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+|-------------------------------|---------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Listening Port** | | | The port to listen on for communication. |
+| **Max Batch Size** | 500 | | The maximum number of messages to process at a time. |
+| **Max Size of Message Queue** | 10000 | | Maximum number of messages allowed to be buffered before processing them when the processor is triggered. If the buffer is full, the message is ignored. If set to zero the buffer is unlimited. |
Review Comment:
I'm not sure if we have a consensus about it. That's true that if it has a default there will always be a value provided. I usually still mark them required regardless of it having a default or not, if the property always needs to have a value.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907278206
##########
extensions/standard-processors/processors/ListenSyslog.cpp:
##########
@@ -53,15 +53,15 @@ const core::Property ListenSyslog::ParseMessages(
const core::Property ListenSyslog::MaxQueueSize(
core::PropertyBuilder::createProperty("Max Size of Message Queue")
->withDescription("Maximum number of Syslog messages allowed to be buffered before processing them when the processor is triggered. "
- "If the buffer full, the message is ignored. If set to zero the buffer is unlimited.")
- ->withDefaultValue<uint64_t>(0)->build());
+ "If the is buffer full, the message is ignored. If set to zero the buffer is unlimited.")
Review Comment:
the "is" is probably not where intended
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
fgerlits commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r906150903
##########
PROCESSORS.md:
##########
@@ -1279,6 +1280,24 @@ In the list below, the names of required properties appear in bold. Any other pr
+## ListenTCP
+
+### Description
+
+Listens for incoming TCP connections and reads data from each connection using a line separator as the message demarcator. For each message the processor produces a single FlowFile.
+
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+|-------------------------------|---------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Listening Port** | | | The port to listen on for communication. |
+| **Max Batch Size** | 500 | | The maximum number of messages to process at a time. |
+| **Max Size of Message Queue** | 0 | | Maximum number of messages allowed to be buffered before processing them when the processor is triggered. If the buffer full, the message is ignored. If set to zero the buffer is unlimited. |
Review Comment:
Yes, I think a default of 10000 for both sounds good.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907182021
##########
extensions/standard-processors/processors/NetworkListenerProcessor.cpp:
##########
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "NetworkListenerProcessor.h"
+#include "utils/net/UdpServer.h"
+#include "utils/net/TcpServer.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+NetworkListenerProcessor::~NetworkListenerProcessor() {
+ stopServer();
+}
+
+void NetworkListenerProcessor::onTrigger(const std::shared_ptr<core::ProcessContext>&, const std::shared_ptr<core::ProcessSession>& session) {
+ gsl_Expects(session && 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))
+ break;
+ transferAsFlowFile(received_message, *session);
+ ++logs_processed;
+ }
+}
+
+void NetworkListenerProcessor::startServer(
+ const core::ProcessContext& context, const core::Property& max_batch_size_prop, const core::Property& max_queue_size_prop, const core::Property& port_prop, utils::net::Protocol protocol) {
+ context.getProperty(max_batch_size_prop.getName(), max_batch_size_);
+ if (max_batch_size_ < 1)
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Max Batch Size property is invalid");
Review Comment:
Updated in c273419e5fce339e2808399bce2f11188448946a
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907093008
##########
extensions/standard-processors/tests/unit/ListenTcpTests.cpp:
##########
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <string>
+
+#include "Catch.h"
+#include "processors/ListenTCP.h"
+#include "SingleProcessorTestController.h"
+#include "asio.hpp"
+
+using ListenTCP = org::apache::nifi::minifi::processors::ListenTCP;
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::test {
+
+constexpr uint64_t PORT = 10254;
+using ProcessorTriggerResult = std::unordered_map<core::Relationship, std::vector<std::shared_ptr<core::FlowFile>>>;
+
+void sendMessagesViaTCP(const std::vector<std::string_view>& contents) {
+ asio::io_context io_context;
+ asio::ip::tcp::socket socket(io_context);
+ asio::ip::tcp::endpoint remote_endpoint(asio::ip::address::from_string("127.0.0.1"), PORT);
+ socket.connect(remote_endpoint);
+ std::error_code err;
+ for (auto& content : contents) {
+ std::string tcp_message(content);
+ tcp_message += '\n';
+ socket.send(asio::buffer(tcp_message, tcp_message.size()), 0, err);
+ }
+ REQUIRE(!err);
+ socket.close();
+}
+
+void check_for_attributes(core::FlowFile& flow_file) {
+ CHECK(std::to_string(PORT) == flow_file.getAttribute("tcp.port"));
+ CHECK("127.0.0.1" == flow_file.getAttribute("tcp.sender"));
+}
+
+bool triggerUntil(test::SingleProcessorTestController& controller,
+ const std::unordered_map<core::Relationship, size_t>& expected_quantities,
+ ProcessorTriggerResult& result,
+ const std::chrono::milliseconds max_duration,
+ const std::chrono::milliseconds wait_time = 50ms) {
+ auto start_time = std::chrono::steady_clock::now();
+ while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ for (auto& [relationship, flow_files] : controller.trigger()) {
+ result[relationship].insert(result[relationship].end(), flow_files.begin(), flow_files.end());
+ }
+ bool expected_quantities_met = true;
+ for (const auto& [relationship, expected_quantity] : expected_quantities) {
+ if (result[relationship].size() < expected_quantity) {
+ expected_quantities_met = false;
+ break;
+ }
+ }
+ if (expected_quantities_met)
+ return true;
Review Comment:
Updated in 28faa0284cea7c53f796f27cd0effebb5d260f49
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907336619
##########
PROCESSORS.md:
##########
@@ -1279,6 +1280,24 @@ In the list below, the names of required properties appear in bold. Any other pr
+## ListenTCP
+
+### Description
+
+Listens for incoming TCP connections and reads data from each connection using a line separator as the message demarcator. For each message the processor produces a single FlowFile.
+
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+|-------------------------------|---------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Listening Port** | | | The port to listen on for communication. |
+| **Max Batch Size** | 500 | | The maximum number of messages to process at a time. |
+| **Max Size of Message Queue** | 10000 | | Maximum number of messages allowed to be buffered before processing them when the processor is triggered. If the buffer is full, the message is ignored. If set to zero the buffer is unlimited. |
Review Comment:
what is the consensus about a property being "required" and having a default? as far as I know if a property has a default it cannot be "not-provided"
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r899104477
##########
extensions/standard-processors/processors/ListenTCP.h:
##########
@@ -0,0 +1,64 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include "NetworkListenerProcessor.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+class ListenTCP : public NetworkListenerProcessor {
+ public:
+ explicit ListenTCP(const std::string& name, const utils::Identifier& uuid = {})
+ : NetworkListenerProcessor(name, uuid, core::logging::LoggerFactory<ListenTCP>::getLogger()) {
+ }
+ ListenTCP(const ListenTCP&) = delete;
+ ListenTCP(ListenTCP&&) = delete;
+ ListenTCP& operator=(const ListenTCP&) = delete;
+ ListenTCP& operator=(ListenTCP&&) = delete;
+
+ EXTENSIONAPI static constexpr const char* Description = "Listens for Syslog messages being sent to a given port over TCP or UDP. "
+ "Incoming messages are optionally checked against regular expressions for RFC5424 and RFC3164 formatted messages. "
+ "With parsing enabled the individual parts of the message will be placed as FlowFile attributes and "
+ "valid messages will be transferred to success relationship, while invalid messages will be transferred to invalid relationship. "
+ "With parsing disabled all message will be routed to the success relationship, but it will only contain the sender, protocol, and port attributes";
Review Comment:
Good catch, I forgot to change it after the rebase.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907182845
##########
extensions/standard-processors/processors/ListenSyslog.cpp:
##########
@@ -34,8 +34,8 @@ const core::Property ListenSyslog::ProtocolProperty(
core::PropertyBuilder::createProperty("Protocol")
->withDescription("The protocol for Syslog communication.")
->isRequired(true)
- ->withAllowableValues(Protocol::values())
- ->withDefaultValue(toString(Protocol::UDP))
+ ->withAllowableValues(utils::net::Protocol::values())
+ ->withDefaultValue(toString(utils::net::Protocol::UDP))
Review Comment:
Good point, I extracted the `IpProtocol` to be a common smart enum and to be used by all utils::net components in c273419e5fce339e2808399bce2f11188448946a
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] martinzink closed pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
martinzink closed pull request #1350: MINIFICPP-1857 Create ListenTCP processor
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907094793
##########
extensions/standard-processors/tests/unit/ListenTcpTests.cpp:
##########
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <string>
+
+#include "Catch.h"
+#include "processors/ListenTCP.h"
+#include "SingleProcessorTestController.h"
+#include "asio.hpp"
+
+using ListenTCP = org::apache::nifi::minifi::processors::ListenTCP;
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::test {
+
+constexpr uint64_t PORT = 10254;
+using ProcessorTriggerResult = std::unordered_map<core::Relationship, std::vector<std::shared_ptr<core::FlowFile>>>;
+
+void sendMessagesViaTCP(const std::vector<std::string_view>& contents) {
+ asio::io_context io_context;
+ asio::ip::tcp::socket socket(io_context);
+ asio::ip::tcp::endpoint remote_endpoint(asio::ip::address::from_string("127.0.0.1"), PORT);
+ socket.connect(remote_endpoint);
+ std::error_code err;
+ for (auto& content : contents) {
+ std::string tcp_message(content);
+ tcp_message += '\n';
+ socket.send(asio::buffer(tcp_message, tcp_message.size()), 0, err);
+ }
+ REQUIRE(!err);
+ socket.close();
+}
+
+void check_for_attributes(core::FlowFile& flow_file) {
+ CHECK(std::to_string(PORT) == flow_file.getAttribute("tcp.port"));
+ CHECK("127.0.0.1" == flow_file.getAttribute("tcp.sender"));
+}
+
+bool triggerUntil(test::SingleProcessorTestController& controller,
+ const std::unordered_map<core::Relationship, size_t>& expected_quantities,
+ ProcessorTriggerResult& result,
+ const std::chrono::milliseconds max_duration,
+ const std::chrono::milliseconds wait_time = 50ms) {
+ auto start_time = std::chrono::steady_clock::now();
+ while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ for (auto& [relationship, flow_files] : controller.trigger()) {
+ result[relationship].insert(result[relationship].end(), flow_files.begin(), flow_files.end());
+ }
+ bool expected_quantities_met = true;
+ for (const auto& [relationship, expected_quantity] : expected_quantities) {
+ if (result[relationship].size() < expected_quantity) {
+ expected_quantities_met = false;
+ break;
+ }
+ }
+ if (expected_quantities_met)
+ return true;
+ std::this_thread::sleep_for(wait_time);
+ }
+ return false;
+}
+
+bool countLogOccurrencesUntil(const std::string& pattern,
+ const int occurrences,
+ const std::chrono::milliseconds max_duration,
+ const std::chrono::milliseconds wait_time = 50ms) {
+ auto start_time = std::chrono::steady_clock::now();
+ while (std::chrono::steady_clock::now() < start_time + max_duration) {
+ if (LogTestController::getInstance().countOccurrences(pattern) == occurrences)
+ return true;
+ std::this_thread::sleep_for(wait_time);
+ }
+ return false;
+}
Review Comment:
Moved to test/Utils.h and with the `sendMessagesViaTCP` also used in `ListenSyslogTests`. Additionally moved `triggerUntil` to be part of `SingleProcessorTestController` and created a Jira ticket to unify the split up test utilities in [MINIFICPP-1872](https://issues.apache.org/jira/browse/MINIFICPP-1872)
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r907096762
##########
PROCESSORS.md:
##########
@@ -1279,6 +1280,24 @@ In the list below, the names of required properties appear in bold. Any other pr
+## ListenTCP
+
+### Description
+
+Listens for incoming TCP connections and reads data from each connection using a line separator as the message demarcator. For each message the processor produces a single FlowFile.
+
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+|-------------------------------|---------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Listening Port** | | | The port to listen on for communication. |
+| **Max Batch Size** | 500 | | The maximum number of messages to process at a time. |
+| **Max Size of Message Queue** | 0 | | Maximum number of messages allowed to be buffered before processing them when the processor is triggered. If the buffer full, the message is ignored. If set to zero the buffer is unlimited. |
Review Comment:
Updated message queue to 10000 in 9bf72f076f3e3f50bd67a783ca3931b592b578ff
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] lordgamez commented on a diff in pull request #1350: MINIFICPP-1857 Create ListenTCP processor
Posted by GitBox <gi...@apache.org>.
lordgamez commented on code in PR #1350:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1350#discussion_r899104729
##########
libminifi/src/utils/net/TcpServer.cpp:
##########
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "utils/net/TcpServer.h"
+
+namespace org::apache::nifi::minifi::utils::net {
+
+TcpSession::TcpSession(asio::io_context& io_context, utils::ConcurrentQueue<Message>& concurrent_queue, std::optional<size_t> max_queue_size, std::shared_ptr<core::logging::Logger> logger)
+ : concurrent_queue_(concurrent_queue),
+ max_queue_size_(max_queue_size),
+ socket_(io_context),
+ logger_(std::move(logger)) {
+}
+
+asio::ip::tcp::socket& TcpSession::getSocket() {
+ return socket_;
+}
+
+void TcpSession::start() {
+ asio::async_read_until(socket_,
+ buffer_,
+ '\n',
+ [self = shared_from_this()](const auto& error_code, size_t) -> void {
+ self->handleReadUntilNewLine(error_code);
+ });
+}
+
+void TcpSession::handleReadUntilNewLine(std::error_code error_code) {
+ if (error_code)
+ return;
+ std::istream is(&buffer_);
+ std::string message;
+ std::getline(is, message);
+ if (!max_queue_size_ || max_queue_size_ > concurrent_queue_.size())
+ concurrent_queue_.enqueue(Message(message, Protocol::TCP, socket_.remote_endpoint().address(), socket_.local_endpoint().port()));
+ else
+ logger_->log_warn("Queue is full. Syslog message ignored.");
Review Comment:
Good point updated in daedc5003b406d07957791d4d5ded2420888fc3f
##########
extensions/standard-processors/processors/ListenTCP.h:
##########
@@ -0,0 +1,64 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include "NetworkListenerProcessor.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+class ListenTCP : public NetworkListenerProcessor {
+ public:
+ explicit ListenTCP(const std::string& name, const utils::Identifier& uuid = {})
+ : NetworkListenerProcessor(name, uuid, core::logging::LoggerFactory<ListenTCP>::getLogger()) {
+ }
+ ListenTCP(const ListenTCP&) = delete;
+ ListenTCP(ListenTCP&&) = delete;
+ ListenTCP& operator=(const ListenTCP&) = delete;
+ ListenTCP& operator=(ListenTCP&&) = delete;
+
+ EXTENSIONAPI static constexpr const char* Description = "Listens for Syslog messages being sent to a given port over TCP or UDP. "
+ "Incoming messages are optionally checked against regular expressions for RFC5424 and RFC3164 formatted messages. "
+ "With parsing enabled the individual parts of the message will be placed as FlowFile attributes and "
+ "valid messages will be transferred to success relationship, while invalid messages will be transferred to invalid relationship. "
+ "With parsing disabled all message will be routed to the success relationship, but it will only contain the sender, protocol, and port attributes";
Review Comment:
Good catch, I forgot to change it after the rebase. Updated in daedc5003b406d07957791d4d5ded2420888fc3f
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org