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 2021/10/30 20:22:40 UTC
[GitHub] [nifi-minifi-cpp] szaszm opened a new pull request #1208: MINIFICPP-1678 Create PutUDP processor (no tests or docs yet)
szaszm opened a new pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208
https://issues.apache.org/jira/browse/MINIFICPP-1678
Quickly hacked together a PoC. If there is interest, I can pursue this further.
Manually tested on linux only.
---
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:
- [x] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [x] 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.
- [x] 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:
- [x] 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)?
- [x] If applicable, have you updated the LICENSE file?
- [x] If applicable, have you updated the NOTICE file?
### For documentation related changes:
- [x] 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] szaszm commented on a change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r777601496
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,112 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include "SingleInputTestController.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/expected.h"
+#include "utils/StringUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+struct DatagramListener {
+ DatagramListener(const char* const hostname, const char* const port)
+ :resolved_names_{utils::net::resolveHost(hostname, port, utils::net::IpProtocol::Udp).value()},
+ open_socket_{utils::net::open_socket(resolved_names_.get())
+ | utils::valueOrElse([=]() -> utils::net::OpenSocketResult { throw std::runtime_error{utils::StringUtils::join_pack("Failed to connect to ", hostname, " on port ", port)}; })}
+ {
+ const auto bind_result = bind(open_socket_.socket_.get(), open_socket_.selected_name->ai_addr, open_socket_.selected_name->ai_addrlen);
+ if (bind_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("bind: ", utils::net::get_last_socket_error().message())};
+ }
+ }
+
+ struct ReceiveResult {
+ std::string remote_address;
+ std::string message;
+ };
+
+ [[nodiscard]] ReceiveResult receive(const size_t max_message_size = 8192) const {
+ ReceiveResult result;
+ result.message.resize(max_message_size);
+ sockaddr_storage remote_address{};
+ socklen_t addrlen = sizeof(remote_address);
+ const auto recv_result = recvfrom(open_socket_.socket_.get(), result.message.data(), result.message.size(), 0, std::launder(reinterpret_cast<sockaddr*>(&remote_address)), &addrlen);
Review comment:
It shouldn't be needed, but since the code most likely violates the aliasing rules, I wanted to reduce the possibility of a future compiler messing it up with some optimization.
The Sockets API was made in the 90s with no regards to aliasing rules in C and C++, and back then it didn't really matter, because compilers were much simpler and rarely if ever took advantage of optimization opportunities that arose as a consequence of having differently typed pointers in the same context, and those can not point to the same memory location.
In the intended usage of the sockets API, one is supposed to initialize a memory location as `struct sockaddr_storage`, then fill it in with a function that works with `struct sockaddr_in` or `struct sockaddr_in6` (for IPv4 and IPv6 respectively) while passing it around as `struct sockaddr*`. These structures have compatible layouts, but this doesn't make the practice legal C or C++.
You're correct that `std::launder` is not related, but I was hoping that the optimization barrier it creates overlaps with those that are required to make this aliasing violation work in that hypothetical future compiler that takes advantage of this UB. There was a similar usage of `std::launder` in libc++ "just to be safe": https://reviews.llvm.org/D47607.
I can't give a good explanation of the proper usage of `std::launder`, because I'm not too familiar with the object model changes of C++17 that introduced it. I think it's used to obtain a valid pointer from an invalidated pointer to the same memory location, or something like this.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor (no tests or docs yet)
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r758328323
##########
File path: CPPLINT.cfg
##########
@@ -1,2 +1,2 @@
set noparent
-filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check
+filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check,-whitespace/braces
Review comment:
I found one more occurrence in TailFileTests.cpp:589. It's still not common, so I could change it, but a linter rule with many false positives is not very useful in my opinion.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r771550804
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,171 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include <unordered_map>
+#include "range/v3/view/transform.hpp"
+#include "range/v3/range/conversion.hpp"
+#include "TestBase.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+class SingleInputTestController : public TestController {
Review comment:
In the end I spent a few hours on `RouteTextController` and saw that performing this would require a rewrite of the tests, which I don't want to do.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r777617491
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,112 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include "SingleInputTestController.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/expected.h"
+#include "utils/StringUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+struct DatagramListener {
+ DatagramListener(const char* const hostname, const char* const port)
+ :resolved_names_{utils::net::resolveHost(hostname, port, utils::net::IpProtocol::Udp).value()},
+ open_socket_{utils::net::open_socket(resolved_names_.get())
+ | utils::valueOrElse([=]() -> utils::net::OpenSocketResult { throw std::runtime_error{utils::StringUtils::join_pack("Failed to connect to ", hostname, " on port ", port)}; })}
+ {
+ const auto bind_result = bind(open_socket_.socket_.get(), open_socket_.selected_name->ai_addr, open_socket_.selected_name->ai_addrlen);
+ if (bind_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("bind: ", utils::net::get_last_socket_error().message())};
+ }
+ }
+
+ struct ReceiveResult {
+ std::string remote_address;
+ std::string message;
+ };
+
+ [[nodiscard]] ReceiveResult receive(const size_t max_message_size = 8192) const {
+ ReceiveResult result;
+ result.message.resize(max_message_size);
+ sockaddr_storage remote_address{};
+ socklen_t addrlen = sizeof(remote_address);
+ const auto recv_result = recvfrom(open_socket_.socket_.get(), result.message.data(), result.message.size(), 0, std::launder(reinterpret_cast<sockaddr*>(&remote_address)), &addrlen);
+ if (recv_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("recvfrom: ", utils::net::get_last_socket_error().message())};
+ }
+ result.message.resize(gsl::narrow<size_t>(recv_result));
+ result.remote_address = utils::net::sockaddr_ntop(std::launder(reinterpret_cast<sockaddr*>(&remote_address)));
+ return result;
+ }
+
+ std::unique_ptr<addrinfo, utils::net::addrinfo_deleter> resolved_names_;
+ utils::net::OpenSocketResult open_socket_;
+};
+} // namespace
+
+// Testing the failure relationship is not required, because since UDP in general without guarantees, flow files are always routed to success, unless there is
+// some weird IO error with the content repo.
+TEST_CASE("PutUDP", "[putudp]") {
+ const auto putudp = std::make_shared<PutUDP>("PutUDP");
+ auto random_engine = std::mt19937{std::random_device{}()}; // NOLINT: "Missing space before { [whitespace/braces] [5]"
+ // most systems use ports 32768 - 65535 as ephemeral ports, so avoid binding to those
+ const auto port = std::uniform_int_distribution<uint16_t>{10000, 32768 - 1}(random_engine);
+ const auto port_str = std::to_string(port);
+
+ test::SingleInputTestController controller{putudp};
+ LogTestController::getInstance().setTrace<PutUDP>();
+ LogTestController::getInstance().setTrace<core::ProcessContext>();
+ LogTestController::getInstance().setLevelByClassName(spdlog::level::trace, "org::apache::nifi::minifi::core::ProcessContextExpr");
+ putudp->setProperty(PutUDP::Hostname, "${literal('localhost')}");
+ putudp->setProperty(PutUDP::Port, utils::StringUtils::join_pack("${literal('", port_str, "')}"));
+
+ DatagramListener listener{"localhost", port_str.c_str()};
+
+ {
+ const char* const message = "first message: hello";
+ const auto result = controller.trigger(message);
+ const auto& success_flow_files = result.at(PutUDP::Success);
+ REQUIRE(success_flow_files.size() == 1);
+ REQUIRE(result.at(PutUDP::Failure).empty());
+ REQUIRE(controller.plan->getContent(success_flow_files[0]) == message);
+ auto receive_result = listener.receive();
+ REQUIRE(receive_result.message == message);
+ REQUIRE(!receive_result.remote_address.empty());
+ }
+
+ {
Review comment:
Once `UniqueSocketHandle` has a destructor, `DatagramListener` will close the socket on destruction, so we will still only have two sockets open at the same time. Does that take very long, or is there another reason for not wanting to open a port 3 times instead of 2?
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r777622464
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,112 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include "SingleInputTestController.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/expected.h"
+#include "utils/StringUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+struct DatagramListener {
+ DatagramListener(const char* const hostname, const char* const port)
+ :resolved_names_{utils::net::resolveHost(hostname, port, utils::net::IpProtocol::Udp).value()},
+ open_socket_{utils::net::open_socket(resolved_names_.get())
+ | utils::valueOrElse([=]() -> utils::net::OpenSocketResult { throw std::runtime_error{utils::StringUtils::join_pack("Failed to connect to ", hostname, " on port ", port)}; })}
+ {
+ const auto bind_result = bind(open_socket_.socket_.get(), open_socket_.selected_name->ai_addr, open_socket_.selected_name->ai_addrlen);
+ if (bind_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("bind: ", utils::net::get_last_socket_error().message())};
+ }
+ }
+
+ struct ReceiveResult {
+ std::string remote_address;
+ std::string message;
+ };
+
+ [[nodiscard]] ReceiveResult receive(const size_t max_message_size = 8192) const {
+ ReceiveResult result;
+ result.message.resize(max_message_size);
+ sockaddr_storage remote_address{};
+ socklen_t addrlen = sizeof(remote_address);
+ const auto recv_result = recvfrom(open_socket_.socket_.get(), result.message.data(), result.message.size(), 0, std::launder(reinterpret_cast<sockaddr*>(&remote_address)), &addrlen);
+ if (recv_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("recvfrom: ", utils::net::get_last_socket_error().message())};
+ }
+ result.message.resize(gsl::narrow<size_t>(recv_result));
+ result.remote_address = utils::net::sockaddr_ntop(std::launder(reinterpret_cast<sockaddr*>(&remote_address)));
+ return result;
+ }
+
+ std::unique_ptr<addrinfo, utils::net::addrinfo_deleter> resolved_names_;
+ utils::net::OpenSocketResult open_socket_;
+};
+} // namespace
+
+// Testing the failure relationship is not required, because since UDP in general without guarantees, flow files are always routed to success, unless there is
+// some weird IO error with the content repo.
+TEST_CASE("PutUDP", "[putudp]") {
+ const auto putudp = std::make_shared<PutUDP>("PutUDP");
+ auto random_engine = std::mt19937{std::random_device{}()}; // NOLINT: "Missing space before { [whitespace/braces] [5]"
+ // most systems use ports 32768 - 65535 as ephemeral ports, so avoid binding to those
+ const auto port = std::uniform_int_distribution<uint16_t>{10000, 32768 - 1}(random_engine);
+ const auto port_str = std::to_string(port);
+
+ test::SingleInputTestController controller{putudp};
+ LogTestController::getInstance().setTrace<PutUDP>();
+ LogTestController::getInstance().setTrace<core::ProcessContext>();
+ LogTestController::getInstance().setLevelByClassName(spdlog::level::trace, "org::apache::nifi::minifi::core::ProcessContextExpr");
+ putudp->setProperty(PutUDP::Hostname, "${literal('localhost')}");
+ putudp->setProperty(PutUDP::Port, utils::StringUtils::join_pack("${literal('", port_str, "')}"));
+
+ DatagramListener listener{"localhost", port_str.c_str()};
+
+ {
+ const char* const message = "first message: hello";
+ const auto result = controller.trigger(message);
+ const auto& success_flow_files = result.at(PutUDP::Success);
+ REQUIRE(success_flow_files.size() == 1);
+ REQUIRE(result.at(PutUDP::Failure).empty());
+ REQUIRE(controller.plan->getContent(success_flow_files[0]) == message);
+ auto receive_result = listener.receive();
+ REQUIRE(receive_result.message == message);
+ REQUIRE(!receive_result.remote_address.empty());
+ }
+
+ {
Review comment:
I just wanted to minimize the chance of collision and avoid SO_REUSEADDR, because [it can cause problems](https://stackoverflow.com/a/3233022). And in general it's not nice that unit tests open ports, but it's hard to avoid here.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor (no tests or docs yet)
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r760332401
##########
File path: CPPLINT.cfg
##########
@@ -1,2 +1,2 @@
set noparent
-filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check
+filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check,-whitespace/braces
Review comment:
I'll push a change to NOLINT in the next batch of changes, hopefully this week.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor (no tests or docs yet)
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r760326181
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,171 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include <unordered_map>
+#include "range/v3/view/transform.hpp"
+#include "range/v3/range/conversion.hpp"
+#include "TestBase.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+class SingleInputTestController : public TestController {
Review comment:
I think it's better if we move it in the first reuse, and I would rather move it into its own header. I can convert RouteText to use this if you and @adamdebreceni (author of RouteText) think it would be useful, and that would be a first reuse.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r764509172
##########
File path: libminifi/include/utils/expected.h
##########
@@ -0,0 +1,176 @@
+/**
+ * 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 <type_traits>
+#include <utility>
+#include "nonstd/expected.hpp"
+#include "utils/detail/MonadicOperationWrappers.h"
+#include "utils/GeneralUtils.h"
+#include "utils/meta/detected.h"
+
+namespace org::apache::nifi::minifi::utils {
+namespace detail {
+
+template<typename T>
+inline constexpr bool is_expected_v = false;
+
+template<typename V, typename E>
+inline constexpr bool is_expected_v<nonstd::expected<V, E>> = true;
+
+// map implementation
+template<typename Expected, typename F, typename = std::enable_if_t<is_expected_v<utils::remove_cvref_t<Expected>>>>
+auto operator|(Expected&& object, map_wrapper<F> f) {
Review comment:
I didn't bother with noexcept as it would've required even more metaprogramming, I would've needed to basically duplicate the implementation inside the noexcept specifier. There is a lot of code duplication inside the implementation, but I think this is the simplest way to do it. I could use `if constexpr` instead of SFINAE a lot, so I think it should also compile faster, but I don't want to implement them twice to measure.
--
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 pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#issuecomment-1006847442
In the last commit I made some changes to simplify the flow of control and data in PutUDP::onTrigger. Since I submitted this after approvals, please have a look at it if you have the time.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r777608909
##########
File path: libminifi/include/utils/net/Socket.h
##########
@@ -0,0 +1,104 @@
+/**
+ * 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 <string>
+#include <system_error>
+#ifdef WIN32
+#ifndef WIN32_LEAN_AND_MEAN
+#define WIN32_LEAN_AND_MEAN
+#endif /* WIN32_LEAN_AND_MEAN */
+#include <WinSock2.h>
+#else
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netdb.h>
+#include <unistd.h>
+#endif /* WIN32 */
+#include "nonstd/expected.hpp"
+#include "utils/gsl.h"
+
+namespace org::apache::nifi::minifi::utils::net {
+#ifdef WIN32
+using SocketDescriptor = SOCKET;
+using ip4addr = in_addr;
+inline constexpr SocketDescriptor InvalidSocket = INVALID_SOCKET;
+constexpr int SocketError = SOCKET_ERROR;
+#else
+using SocketDescriptor = int;
+using ip4addr = in_addr_t;
+#undef INVALID_SOCKET
+inline constexpr SocketDescriptor InvalidSocket = -1;
+#undef SOCKET_ERROR
+inline constexpr int SocketError = -1;
+#endif /* WIN32 */
+
+/**
+ * Return the last socket error code, based on errno on posix and WSAGetLastError() on windows.
+ */
+std::error_code get_last_socket_error();
+
+inline void close_socket(SocketDescriptor sockfd) {
+#ifdef WIN32
+ closesocket(sockfd);
+#else
+ ::close(sockfd);
+#endif
+}
+
+class UniqueSocketHandle {
+ public:
+ explicit UniqueSocketHandle(SocketDescriptor owner_sockfd) noexcept
+ :owner_sockfd_(owner_sockfd)
+ {}
+
Review comment:
It was an oversight, thanks for raising this.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r777609050
##########
File path: libminifi/src/utils/net/Socket.cpp
##########
@@ -0,0 +1,72 @@
+/**
+ * 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/Socket.h"
+#include "Exception.h"
+#include <cstring>
+#include <system_error>
+#ifdef WIN32
+#ifndef WIN32_LEAN_AND_MEAN
+#define WIN32_LEAN_AND_MEAN
+#endif /* WIN32_LEAN_AND_MEAN */
+#include <ws2tcpip.h>
+#else
+#include <arpa/inet.h>
+#endif /* WIN32 */
+
+namespace org::apache::nifi::minifi::utils::net {
+std::error_code get_last_socket_error() {
+#ifdef WIN32
+ const auto error_code = WSAGetLastError();
+#else
+ const auto error_code = errno;
+#endif /* WIN32 */
+ return {error_code, std::system_category()};
+}
+
+nonstd::expected<OpenSocketResult, std::error_code> open_socket(const addrinfo* const getaddrinfo_result) {
+ for (const addrinfo* it = getaddrinfo_result; it; it = it->ai_next) {
+ const auto fd = socket(it->ai_family, it->ai_socktype, it->ai_protocol);
+ if (fd != utils::net::InvalidSocket) return OpenSocketResult{UniqueSocketHandle{fd}, gsl::make_not_null(it)};
+ }
+ return nonstd::make_unexpected(get_last_socket_error());
+}
+
+std::string sockaddr_ntop(const sockaddr* const sa) {
+ std::string result;
+ if (sa->sa_family == AF_INET) {
+ sockaddr_in sa_in{};
+ std::memcpy(reinterpret_cast<void*>(&sa_in), sa, sizeof(sockaddr_in));
Review comment:
Good point, removing.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r780130483
##########
File path: libminifi/include/core/ProcessContext.h
##########
@@ -113,10 +113,13 @@ class ProcessContext : public controller::ControllerServiceLookup, public core::
}
template<typename T>
- bool getProperty(const std::string &name, T &value) const {
+ std::enable_if_t<!std::is_convertible_v<T&, FlowFile&>, bool> getProperty(const std::string &name, T &value) const {
Review comment:
I think that the added requirement on this method has merit, but currently nothing verifies it, IMO a test that checks that `ctx.getProperty("blah", flowfile)` does not compile (or something along these lines) would help cement its purpose
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r777608142
##########
File path: libminifi/src/utils/net/DNS.cpp
##########
@@ -0,0 +1,93 @@
+/**
+ * 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/DNS.h"
+#include "Exception.h"
+#include "utils/StringUtils.h"
+
+#ifdef WIN32
+#ifndef WIN32_LEAN_AND_MEAN
+#define WIN32_LEAN_AND_MEAN
+#endif
+#include <winsock2.h>
+#include <ws2tcpip.h>
+#include "utils/net/Socket.h"
+#else
+#include <netdb.h>
+#include <cstring>
+#endif /* WIN32 */
+
+namespace org::apache::nifi::minifi::utils::net {
+
+namespace {
+
+#ifndef WIN32
+class addrinfo_category : public std::error_category {
+ public:
+ [[nodiscard]] const char* name() const noexcept override { return "addrinfo"; }
+
+ [[nodiscard]] std::string message(int value) const override {
+ return gai_strerror(value);
+ }
+};
+
+const addrinfo_category& get_addrinfo_category() {
+ static addrinfo_category instance;
+ return instance;
+}
+#endif
+
+std::error_code get_last_getaddrinfo_err_code(int getaddrinfo_result) {
+#ifdef WIN32
+ (void)getaddrinfo_result; // against unused warnings on windows
+ return std::error_code{WSAGetLastError(), std::system_category()};
+#else
+ return std::error_code{getaddrinfo_result, get_addrinfo_category()};
+#endif /* WIN32 */
+}
+} // namespace
+
+void addrinfo_deleter::operator()(addrinfo* const p) const noexcept {
+ freeaddrinfo(p);
+}
+
+nonstd::expected<std::unique_ptr<addrinfo, addrinfo_deleter>, std::error_code> resolveHost(const char* const hostname, const char* const port, const IpProtocol protocol, const bool need_canonname) {
+ addrinfo hints{};
+ memset(&hints, 0, sizeof hints); // make sure the struct is empty
+ hints.ai_family = AF_UNSPEC;
+ hints.ai_socktype = protocol == IpProtocol::Tcp ? SOCK_STREAM : SOCK_DGRAM;
+ hints.ai_flags = need_canonname ? AI_CANONNAME : 0;
+ if (!hostname)
+ hints.ai_flags |= AI_PASSIVE;
+ hints.ai_protocol = [protocol]() -> int {
+ switch (protocol) {
+ case IpProtocol::Tcp: return IPPROTO_TCP;
+ case IpProtocol::Udp: return IPPROTO_UDP;
Review comment:
I prefer to hide this detail from the interface as long as possible. I meant to use this enumerator as a way to hide the BSD/POSIX Sockets API macros from the API.
If there is ever going to be another usage, then this goal will need to be reevaluated and probably the interface of both usages redesigned.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r764506451
##########
File path: extensions/expression-language/ProcessContextExpr.cpp
##########
@@ -54,9 +56,11 @@ bool ProcessContextExpr::getDynamicProperty(const Property &property, std::strin
ProcessContext::getDynamicProperty(name, expression_str);
logger_->log_debug("Compiling expression for %s/%s: %s", getProcessorNode()->getName(), name, expression_str);
dynamic_property_expressions_.emplace(name, expression::compile(expression_str));
+ expression_strs_.insert_or_assign(name, expression_str);
}
minifi::expression::Parameters p(shared_from_this(), flow_file);
value = dynamic_property_expressions_[name](p).asString();
+ logger_->log_debug(R"(expression "%s" of dynamic property "%s" evaluated to: %s)", expression_strs_[name], name, value);
Review comment:
This addition helped debugging expression language properties.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r773773939
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,112 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include "SingleInputTestController.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/expected.h"
+#include "utils/StringUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+struct DatagramListener {
+ DatagramListener(const char* const hostname, const char* const port)
+ :resolved_names_{utils::net::resolveHost(hostname, port, utils::net::IpProtocol::Udp).value()},
+ open_socket_{utils::net::open_socket(resolved_names_.get())
+ | utils::valueOrElse([=]() -> utils::net::OpenSocketResult { throw std::runtime_error{utils::StringUtils::join_pack("Failed to connect to ", hostname, " on port ", port)}; })}
+ {
+ const auto bind_result = bind(open_socket_.socket_.get(), open_socket_.selected_name->ai_addr, open_socket_.selected_name->ai_addrlen);
+ if (bind_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("bind: ", utils::net::get_last_socket_error().message())};
+ }
+ }
+
+ struct ReceiveResult {
+ std::string remote_address;
+ std::string message;
+ };
+
+ [[nodiscard]] ReceiveResult receive(const size_t max_message_size = 8192) const {
+ ReceiveResult result;
+ result.message.resize(max_message_size);
+ sockaddr_storage remote_address{};
+ socklen_t addrlen = sizeof(remote_address);
+ const auto recv_result = recvfrom(open_socket_.socket_.get(), result.message.data(), result.message.size(), 0, std::launder(reinterpret_cast<sockaddr*>(&remote_address)), &addrlen);
Review comment:
I have never seen `std::launder` before. Can you explain (or give a link to an explanation of) why it is needed here?
I have looked at some examples, but they all talked about placement new and const members or inheritance, which don't seem to apply here.
##########
File path: extensions/standard-processors/processors/PutUDP.h
##########
@@ -0,0 +1,54 @@
+/**
+ * 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 <cstddef>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "Processor.h"
+#include "utils/Export.h"
+
+namespace org::apache::nifi::minifi::core::logging { class Logger; }
+
+namespace org::apache::nifi::minifi::processors {
+class PutUDP final : public core::Processor {
+ public:
+ EXTENSIONAPI static const core::Property Hostname;
+ EXTENSIONAPI static const core::Property Port;
+
+ EXTENSIONAPI static const core::Relationship Success;
+ EXTENSIONAPI static const core::Relationship Failure;
+
+ explicit PutUDP(const std::string& name, const utils::Identifier& uuid = {});
+ PutUDP(const PutUDP&) = delete;
+ PutUDP& operator=(const PutUDP&) = delete;
+ ~PutUDP() final;
+
+ void initialize() final;
+ void notifyStop() final;
+ void onSchedule(core::ProcessContext*, core::ProcessSessionFactory *) final;
+ void onTrigger(core::ProcessContext*, core::ProcessSession*) final;
+
+ core::annotation::Input getInputRequirement() const noexcept final { return core::annotation::Input::INPUT_REQUIRED; }
+ bool isSingleThreaded() const noexcept final { return true; /* for now */ }
+ private:
+ std::string hostname_;
+ std::string port_; // Can be a service name, like http or ssh
Review comment:
this comment would be useful in PROCESSORS.md and the manifest, so I would move it to the description of `PutUDP::Port`
##########
File path: libminifi/src/utils/net/DNS.cpp
##########
@@ -0,0 +1,93 @@
+/**
+ * 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/DNS.h"
+#include "Exception.h"
+#include "utils/StringUtils.h"
+
+#ifdef WIN32
+#ifndef WIN32_LEAN_AND_MEAN
+#define WIN32_LEAN_AND_MEAN
+#endif
+#include <winsock2.h>
+#include <ws2tcpip.h>
+#include "utils/net/Socket.h"
+#else
+#include <netdb.h>
+#include <cstring>
+#endif /* WIN32 */
+
+namespace org::apache::nifi::minifi::utils::net {
+
+namespace {
+
+#ifndef WIN32
+class addrinfo_category : public std::error_category {
+ public:
+ [[nodiscard]] const char* name() const noexcept override { return "addrinfo"; }
+
+ [[nodiscard]] std::string message(int value) const override {
+ return gai_strerror(value);
+ }
+};
+
+const addrinfo_category& get_addrinfo_category() {
+ static addrinfo_category instance;
+ return instance;
+}
+#endif
+
+std::error_code get_last_getaddrinfo_err_code(int getaddrinfo_result) {
+#ifdef WIN32
+ (void)getaddrinfo_result; // against unused warnings on windows
+ return std::error_code{WSAGetLastError(), std::system_category()};
+#else
+ return std::error_code{getaddrinfo_result, get_addrinfo_category()};
+#endif /* WIN32 */
+}
+} // namespace
+
+void addrinfo_deleter::operator()(addrinfo* const p) const noexcept {
+ freeaddrinfo(p);
+}
+
+nonstd::expected<std::unique_ptr<addrinfo, addrinfo_deleter>, std::error_code> resolveHost(const char* const hostname, const char* const port, const IpProtocol protocol, const bool need_canonname) {
+ addrinfo hints{};
+ memset(&hints, 0, sizeof hints); // make sure the struct is empty
+ hints.ai_family = AF_UNSPEC;
+ hints.ai_socktype = protocol == IpProtocol::Tcp ? SOCK_STREAM : SOCK_DGRAM;
+ hints.ai_flags = need_canonname ? AI_CANONNAME : 0;
+ if (!hostname)
+ hints.ai_flags |= AI_PASSIVE;
+ hints.ai_protocol = [protocol]() -> int {
+ switch (protocol) {
+ case IpProtocol::Tcp: return IPPROTO_TCP;
+ case IpProtocol::Udp: return IPPROTO_UDP;
Review comment:
I'm not sure if it's worth the extra code in `IpProtocol` as this is currently only used here, but
```c++
hints.ai_socktype = protocol.sockType();
...
hints.ai_protocol = protocol.raw();
```
would be nicer.
##########
File path: libminifi/include/utils/net/Socket.h
##########
@@ -0,0 +1,104 @@
+/**
+ * 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 <string>
+#include <system_error>
+#ifdef WIN32
+#ifndef WIN32_LEAN_AND_MEAN
+#define WIN32_LEAN_AND_MEAN
+#endif /* WIN32_LEAN_AND_MEAN */
+#include <WinSock2.h>
+#else
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netdb.h>
+#include <unistd.h>
+#endif /* WIN32 */
+#include "nonstd/expected.hpp"
+#include "utils/gsl.h"
+
+namespace org::apache::nifi::minifi::utils::net {
+#ifdef WIN32
+using SocketDescriptor = SOCKET;
+using ip4addr = in_addr;
+inline constexpr SocketDescriptor InvalidSocket = INVALID_SOCKET;
+constexpr int SocketError = SOCKET_ERROR;
+#else
+using SocketDescriptor = int;
+using ip4addr = in_addr_t;
+#undef INVALID_SOCKET
+inline constexpr SocketDescriptor InvalidSocket = -1;
+#undef SOCKET_ERROR
+inline constexpr int SocketError = -1;
+#endif /* WIN32 */
+
+/**
+ * Return the last socket error code, based on errno on posix and WSAGetLastError() on windows.
+ */
+std::error_code get_last_socket_error();
+
+inline void close_socket(SocketDescriptor sockfd) {
+#ifdef WIN32
+ closesocket(sockfd);
+#else
+ ::close(sockfd);
+#endif
+}
+
+class UniqueSocketHandle {
+ public:
+ explicit UniqueSocketHandle(SocketDescriptor owner_sockfd) noexcept
+ :owner_sockfd_(owner_sockfd)
+ {}
+
Review comment:
Why is there no destructor? I would expect one which calls `close_socket`.
##########
File path: libminifi/src/utils/net/Socket.cpp
##########
@@ -0,0 +1,72 @@
+/**
+ * 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/Socket.h"
+#include "Exception.h"
+#include <cstring>
+#include <system_error>
+#ifdef WIN32
+#ifndef WIN32_LEAN_AND_MEAN
+#define WIN32_LEAN_AND_MEAN
+#endif /* WIN32_LEAN_AND_MEAN */
+#include <ws2tcpip.h>
+#else
+#include <arpa/inet.h>
+#endif /* WIN32 */
+
+namespace org::apache::nifi::minifi::utils::net {
+std::error_code get_last_socket_error() {
+#ifdef WIN32
+ const auto error_code = WSAGetLastError();
+#else
+ const auto error_code = errno;
+#endif /* WIN32 */
+ return {error_code, std::system_category()};
+}
+
+nonstd::expected<OpenSocketResult, std::error_code> open_socket(const addrinfo* const getaddrinfo_result) {
+ for (const addrinfo* it = getaddrinfo_result; it; it = it->ai_next) {
+ const auto fd = socket(it->ai_family, it->ai_socktype, it->ai_protocol);
+ if (fd != utils::net::InvalidSocket) return OpenSocketResult{UniqueSocketHandle{fd}, gsl::make_not_null(it)};
+ }
+ return nonstd::make_unexpected(get_last_socket_error());
+}
+
+std::string sockaddr_ntop(const sockaddr* const sa) {
+ std::string result;
+ if (sa->sa_family == AF_INET) {
+ sockaddr_in sa_in{};
+ std::memcpy(reinterpret_cast<void*>(&sa_in), sa, sizeof(sockaddr_in));
Review comment:
do we need this `reinterpret_cast`? any pointer type can be implicitly converted to `void*`
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,112 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include "SingleInputTestController.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/expected.h"
+#include "utils/StringUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+struct DatagramListener {
+ DatagramListener(const char* const hostname, const char* const port)
+ :resolved_names_{utils::net::resolveHost(hostname, port, utils::net::IpProtocol::Udp).value()},
+ open_socket_{utils::net::open_socket(resolved_names_.get())
+ | utils::valueOrElse([=]() -> utils::net::OpenSocketResult { throw std::runtime_error{utils::StringUtils::join_pack("Failed to connect to ", hostname, " on port ", port)}; })}
+ {
+ const auto bind_result = bind(open_socket_.socket_.get(), open_socket_.selected_name->ai_addr, open_socket_.selected_name->ai_addrlen);
+ if (bind_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("bind: ", utils::net::get_last_socket_error().message())};
+ }
+ }
+
+ struct ReceiveResult {
+ std::string remote_address;
+ std::string message;
+ };
+
+ [[nodiscard]] ReceiveResult receive(const size_t max_message_size = 8192) const {
+ ReceiveResult result;
+ result.message.resize(max_message_size);
+ sockaddr_storage remote_address{};
+ socklen_t addrlen = sizeof(remote_address);
+ const auto recv_result = recvfrom(open_socket_.socket_.get(), result.message.data(), result.message.size(), 0, std::launder(reinterpret_cast<sockaddr*>(&remote_address)), &addrlen);
+ if (recv_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("recvfrom: ", utils::net::get_last_socket_error().message())};
+ }
+ result.message.resize(gsl::narrow<size_t>(recv_result));
+ result.remote_address = utils::net::sockaddr_ntop(std::launder(reinterpret_cast<sockaddr*>(&remote_address)));
+ return result;
+ }
+
+ std::unique_ptr<addrinfo, utils::net::addrinfo_deleter> resolved_names_;
+ utils::net::OpenSocketResult open_socket_;
+};
+} // namespace
+
+// Testing the failure relationship is not required, because since UDP in general without guarantees, flow files are always routed to success, unless there is
+// some weird IO error with the content repo.
+TEST_CASE("PutUDP", "[putudp]") {
+ const auto putudp = std::make_shared<PutUDP>("PutUDP");
+ auto random_engine = std::mt19937{std::random_device{}()}; // NOLINT: "Missing space before { [whitespace/braces] [5]"
+ // most systems use ports 32768 - 65535 as ephemeral ports, so avoid binding to those
+ const auto port = std::uniform_int_distribution<uint16_t>{10000, 32768 - 1}(random_engine);
+ const auto port_str = std::to_string(port);
+
+ test::SingleInputTestController controller{putudp};
+ LogTestController::getInstance().setTrace<PutUDP>();
+ LogTestController::getInstance().setTrace<core::ProcessContext>();
+ LogTestController::getInstance().setLevelByClassName(spdlog::level::trace, "org::apache::nifi::minifi::core::ProcessContextExpr");
+ putudp->setProperty(PutUDP::Hostname, "${literal('localhost')}");
+ putudp->setProperty(PutUDP::Port, utils::StringUtils::join_pack("${literal('", port_str, "')}"));
+
+ DatagramListener listener{"localhost", port_str.c_str()};
+
+ {
+ const char* const message = "first message: hello";
+ const auto result = controller.trigger(message);
+ const auto& success_flow_files = result.at(PutUDP::Success);
+ REQUIRE(success_flow_files.size() == 1);
+ REQUIRE(result.at(PutUDP::Failure).empty());
+ REQUIRE(controller.plan->getContent(success_flow_files[0]) == message);
+ auto receive_result = listener.receive();
+ REQUIRE(receive_result.message == message);
+ REQUIRE(!receive_result.remote_address.empty());
+ }
+
+ {
Review comment:
these two parts could be `SECTION`s instead of just blocks so that
- they can be given a description, and
- they are run separately instead of after each other
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r777603062
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,112 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include "SingleInputTestController.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/expected.h"
+#include "utils/StringUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+struct DatagramListener {
+ DatagramListener(const char* const hostname, const char* const port)
+ :resolved_names_{utils::net::resolveHost(hostname, port, utils::net::IpProtocol::Udp).value()},
+ open_socket_{utils::net::open_socket(resolved_names_.get())
+ | utils::valueOrElse([=]() -> utils::net::OpenSocketResult { throw std::runtime_error{utils::StringUtils::join_pack("Failed to connect to ", hostname, " on port ", port)}; })}
+ {
+ const auto bind_result = bind(open_socket_.socket_.get(), open_socket_.selected_name->ai_addr, open_socket_.selected_name->ai_addrlen);
+ if (bind_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("bind: ", utils::net::get_last_socket_error().message())};
+ }
+ }
+
+ struct ReceiveResult {
+ std::string remote_address;
+ std::string message;
+ };
+
+ [[nodiscard]] ReceiveResult receive(const size_t max_message_size = 8192) const {
+ ReceiveResult result;
+ result.message.resize(max_message_size);
+ sockaddr_storage remote_address{};
+ socklen_t addrlen = sizeof(remote_address);
+ const auto recv_result = recvfrom(open_socket_.socket_.get(), result.message.data(), result.message.size(), 0, std::launder(reinterpret_cast<sockaddr*>(&remote_address)), &addrlen);
+ if (recv_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("recvfrom: ", utils::net::get_last_socket_error().message())};
+ }
+ result.message.resize(gsl::narrow<size_t>(recv_result));
+ result.remote_address = utils::net::sockaddr_ntop(std::launder(reinterpret_cast<sockaddr*>(&remote_address)));
+ return result;
+ }
+
+ std::unique_ptr<addrinfo, utils::net::addrinfo_deleter> resolved_names_;
+ utils::net::OpenSocketResult open_socket_;
+};
+} // namespace
+
+// Testing the failure relationship is not required, because since UDP in general without guarantees, flow files are always routed to success, unless there is
+// some weird IO error with the content repo.
+TEST_CASE("PutUDP", "[putudp]") {
+ const auto putudp = std::make_shared<PutUDP>("PutUDP");
+ auto random_engine = std::mt19937{std::random_device{}()}; // NOLINT: "Missing space before { [whitespace/braces] [5]"
+ // most systems use ports 32768 - 65535 as ephemeral ports, so avoid binding to those
+ const auto port = std::uniform_int_distribution<uint16_t>{10000, 32768 - 1}(random_engine);
+ const auto port_str = std::to_string(port);
+
+ test::SingleInputTestController controller{putudp};
+ LogTestController::getInstance().setTrace<PutUDP>();
+ LogTestController::getInstance().setTrace<core::ProcessContext>();
+ LogTestController::getInstance().setLevelByClassName(spdlog::level::trace, "org::apache::nifi::minifi::core::ProcessContextExpr");
+ putudp->setProperty(PutUDP::Hostname, "${literal('localhost')}");
+ putudp->setProperty(PutUDP::Port, utils::StringUtils::join_pack("${literal('", port_str, "')}"));
+
+ DatagramListener listener{"localhost", port_str.c_str()};
+
+ {
+ const char* const message = "first message: hello";
+ const auto result = controller.trigger(message);
+ const auto& success_flow_files = result.at(PutUDP::Success);
+ REQUIRE(success_flow_files.size() == 1);
+ REQUIRE(result.at(PutUDP::Failure).empty());
+ REQUIRE(controller.plan->getContent(success_flow_files[0]) == message);
+ auto receive_result = listener.receive();
+ REQUIRE(receive_result.message == message);
+ REQUIRE(!receive_result.remote_address.empty());
+ }
+
+ {
Review comment:
I want to reuse the listener so that the test only needs one port from the host system (+ one for the processor), not two.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r764506130
##########
File path: NOTICE
##########
@@ -58,6 +58,8 @@ This software includes third party software subject to the following copyrights:
- IANA timezone database - public domain
- date (HowardHinnant/date) - notices below
- range-v3 - Eric Niebler and other contributors
+- expected-lite - Copyright (C) 2016-2020 Martin Moene.
+- TartanLlama/expected - public domain, thanks to Sy Brand
Review comment:
I copied the tests for the monadic extensions, not the implementation code itself.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor (no tests or docs yet)
Posted by GitBox <gi...@apache.org>.
martinzink commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r758257122
##########
File path: CPPLINT.cfg
##########
@@ -1,2 +1,2 @@
set noparent
-filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check
+filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check,-whitespace/braces
Review comment:
This will disable this whitespace/braces checks everywhere, we could fix that false positive in a much smaller scope by leaving a comment // NOLINT(whitespace/braces) at the of the PutUDPTests.cpp:134 line
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor (no tests or docs yet)
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r760353833
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,171 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include <unordered_map>
+#include "range/v3/view/transform.hpp"
+#include "range/v3/range/conversion.hpp"
+#include "TestBase.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+class SingleInputTestController : public TestController {
Review comment:
I agree that using this class could greatly simplify many test cases that use `GetFile`/`PutFile` to read/write flowfiles while testing some other processor (`ListenHTTPTests`, `ExpressionLanguageTests`, `PutSFTPTests`, `ExtractTextTests`, `FocusArchiveTests`, `TensorFlowTests` seem to be the ones in this category), the straightforward cases could fit into this PR (maybe `RouteTextTests`) but we could do them in a separate PR(s)
--
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 closed pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
adamdebreceni closed pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r779721702
##########
File path: libminifi/include/core/ProcessContext.h
##########
@@ -113,10 +113,13 @@ class ProcessContext : public controller::ControllerServiceLookup, public core::
}
template<typename T>
- bool getProperty(const std::string &name, T &value) const {
+ std::enable_if_t<!std::is_convertible_v<T&, FlowFile&>, bool> getProperty(const std::string &name, T &value) const {
Review comment:
I just realized the bug you pointed out in the parentheses. Should I change it to exclude shared_ptr<FlowFile>, too, or just remove the condition?
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r779437831
##########
File path: libminifi/include/core/ProcessContext.h
##########
@@ -113,10 +113,13 @@ class ProcessContext : public controller::ControllerServiceLookup, public core::
}
template<typename T>
- bool getProperty(const std::string &name, T &value) const {
+ std::enable_if_t<!std::is_convertible_v<T&, FlowFile&>, bool> getProperty(const std::string &name, T &value) const {
Review comment:
I assume this changed to disambiguate it from the newly added overload, but since `Property` and `std::string` are not interconvertible it is not necessary (even if they were `T` would be resolved to `std::shared_ptr<FlowFile>` and satisfy the non-convertible condition)
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r777625820
##########
File path: extensions/standard-processors/processors/PutUDP.h
##########
@@ -0,0 +1,54 @@
+/**
+ * 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 <cstddef>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "Processor.h"
+#include "utils/Export.h"
+
+namespace org::apache::nifi::minifi::core::logging { class Logger; }
+
+namespace org::apache::nifi::minifi::processors {
+class PutUDP final : public core::Processor {
+ public:
+ EXTENSIONAPI static const core::Property Hostname;
+ EXTENSIONAPI static const core::Property Port;
+
+ EXTENSIONAPI static const core::Relationship Success;
+ EXTENSIONAPI static const core::Relationship Failure;
+
+ explicit PutUDP(const std::string& name, const utils::Identifier& uuid = {});
+ PutUDP(const PutUDP&) = delete;
+ PutUDP& operator=(const PutUDP&) = delete;
+ ~PutUDP() final;
+
+ void initialize() final;
+ void notifyStop() final;
+ void onSchedule(core::ProcessContext*, core::ProcessSessionFactory *) final;
+ void onTrigger(core::ProcessContext*, core::ProcessSession*) final;
+
+ core::annotation::Input getInputRequirement() const noexcept final { return core::annotation::Input::INPUT_REQUIRED; }
+ bool isSingleThreaded() const noexcept final { return true; /* for now */ }
+ private:
+ std::string hostname_;
+ std::string port_; // Can be a service name, like http or ssh
Review comment:
moved in 30cc7b4
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r777601496
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,112 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include "SingleInputTestController.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/expected.h"
+#include "utils/StringUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+struct DatagramListener {
+ DatagramListener(const char* const hostname, const char* const port)
+ :resolved_names_{utils::net::resolveHost(hostname, port, utils::net::IpProtocol::Udp).value()},
+ open_socket_{utils::net::open_socket(resolved_names_.get())
+ | utils::valueOrElse([=]() -> utils::net::OpenSocketResult { throw std::runtime_error{utils::StringUtils::join_pack("Failed to connect to ", hostname, " on port ", port)}; })}
+ {
+ const auto bind_result = bind(open_socket_.socket_.get(), open_socket_.selected_name->ai_addr, open_socket_.selected_name->ai_addrlen);
+ if (bind_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("bind: ", utils::net::get_last_socket_error().message())};
+ }
+ }
+
+ struct ReceiveResult {
+ std::string remote_address;
+ std::string message;
+ };
+
+ [[nodiscard]] ReceiveResult receive(const size_t max_message_size = 8192) const {
+ ReceiveResult result;
+ result.message.resize(max_message_size);
+ sockaddr_storage remote_address{};
+ socklen_t addrlen = sizeof(remote_address);
+ const auto recv_result = recvfrom(open_socket_.socket_.get(), result.message.data(), result.message.size(), 0, std::launder(reinterpret_cast<sockaddr*>(&remote_address)), &addrlen);
Review comment:
It shouldn't be needed, but since the code most likely violates the aliasing rules, I wanted to reduce the possibility of a future compiler messing it up with some optimization.
The BSD/POSIX Sockets API was made in the 90s with no regards to aliasing rules in C and C++, and back then it didn't really matter, because compilers were much simpler and rarely if ever took advantage of optimization opportunities that arose as a consequence of having differently typed pointers in the same context, and those can not point to the same memory location.
In the intended usage of the sockets API, one is supposed to initialize a memory location as `struct sockaddr_storage`, then fill it in with a function that works with `struct sockaddr_in` or `struct sockaddr_in6` (for IPv4 and IPv6 respectively) while passing it around as `struct sockaddr*`. These structures have compatible layouts, but this doesn't make the practice legal C or C++.
You're correct that `std::launder` is not related, but I was hoping that the optimization barrier it creates overlaps with those that are required to make this aliasing violation work in that hypothetical future compiler that takes advantage of this UB. There was a similar usage of `std::launder` in libc++ "just to be safe": https://reviews.llvm.org/D47607.
I can't give a good explanation of the proper usage of `std::launder`, because I'm not too familiar with the object model changes of C++17 that introduced it. I think it's used to obtain a valid pointer from an invalidated pointer to the same memory location, or something like this.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r777617491
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,112 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include "SingleInputTestController.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/expected.h"
+#include "utils/StringUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+struct DatagramListener {
+ DatagramListener(const char* const hostname, const char* const port)
+ :resolved_names_{utils::net::resolveHost(hostname, port, utils::net::IpProtocol::Udp).value()},
+ open_socket_{utils::net::open_socket(resolved_names_.get())
+ | utils::valueOrElse([=]() -> utils::net::OpenSocketResult { throw std::runtime_error{utils::StringUtils::join_pack("Failed to connect to ", hostname, " on port ", port)}; })}
+ {
+ const auto bind_result = bind(open_socket_.socket_.get(), open_socket_.selected_name->ai_addr, open_socket_.selected_name->ai_addrlen);
+ if (bind_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("bind: ", utils::net::get_last_socket_error().message())};
+ }
+ }
+
+ struct ReceiveResult {
+ std::string remote_address;
+ std::string message;
+ };
+
+ [[nodiscard]] ReceiveResult receive(const size_t max_message_size = 8192) const {
+ ReceiveResult result;
+ result.message.resize(max_message_size);
+ sockaddr_storage remote_address{};
+ socklen_t addrlen = sizeof(remote_address);
+ const auto recv_result = recvfrom(open_socket_.socket_.get(), result.message.data(), result.message.size(), 0, std::launder(reinterpret_cast<sockaddr*>(&remote_address)), &addrlen);
+ if (recv_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("recvfrom: ", utils::net::get_last_socket_error().message())};
+ }
+ result.message.resize(gsl::narrow<size_t>(recv_result));
+ result.remote_address = utils::net::sockaddr_ntop(std::launder(reinterpret_cast<sockaddr*>(&remote_address)));
+ return result;
+ }
+
+ std::unique_ptr<addrinfo, utils::net::addrinfo_deleter> resolved_names_;
+ utils::net::OpenSocketResult open_socket_;
+};
+} // namespace
+
+// Testing the failure relationship is not required, because since UDP in general without guarantees, flow files are always routed to success, unless there is
+// some weird IO error with the content repo.
+TEST_CASE("PutUDP", "[putudp]") {
+ const auto putudp = std::make_shared<PutUDP>("PutUDP");
+ auto random_engine = std::mt19937{std::random_device{}()}; // NOLINT: "Missing space before { [whitespace/braces] [5]"
+ // most systems use ports 32768 - 65535 as ephemeral ports, so avoid binding to those
+ const auto port = std::uniform_int_distribution<uint16_t>{10000, 32768 - 1}(random_engine);
+ const auto port_str = std::to_string(port);
+
+ test::SingleInputTestController controller{putudp};
+ LogTestController::getInstance().setTrace<PutUDP>();
+ LogTestController::getInstance().setTrace<core::ProcessContext>();
+ LogTestController::getInstance().setLevelByClassName(spdlog::level::trace, "org::apache::nifi::minifi::core::ProcessContextExpr");
+ putudp->setProperty(PutUDP::Hostname, "${literal('localhost')}");
+ putudp->setProperty(PutUDP::Port, utils::StringUtils::join_pack("${literal('", port_str, "')}"));
+
+ DatagramListener listener{"localhost", port_str.c_str()};
+
+ {
+ const char* const message = "first message: hello";
+ const auto result = controller.trigger(message);
+ const auto& success_flow_files = result.at(PutUDP::Success);
+ REQUIRE(success_flow_files.size() == 1);
+ REQUIRE(result.at(PutUDP::Failure).empty());
+ REQUIRE(controller.plan->getContent(success_flow_files[0]) == message);
+ auto receive_result = listener.receive();
+ REQUIRE(receive_result.message == message);
+ REQUIRE(!receive_result.remote_address.empty());
+ }
+
+ {
Review comment:
Once `UniqueSocketHandle` has a destructor, `DatagramListener` will close the socket on destruction, so we will still only have two sockets open at the same time. Does that take a very long, or is there another reason for not wanting to open a port 3 times instead of 2?
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor (no tests or docs yet)
Posted by GitBox <gi...@apache.org>.
martinzink commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r760072345
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,171 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include <unordered_map>
+#include "range/v3/view/transform.hpp"
+#include "range/v3/range/conversion.hpp"
+#include "TestBase.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+class SingleInputTestController : public TestController {
Review comment:
I really like this and I feel this could be useful in the future. Could you move this to TestBase.h for easier use in other PR-s aswell?
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor (no tests or docs yet)
Posted by GitBox <gi...@apache.org>.
martinzink commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r759227291
##########
File path: CPPLINT.cfg
##########
@@ -1,2 +1,2 @@
set noparent
-filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check
+filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check,-whitespace/braces
Review comment:
Yeah, I agree if the linter rule is unreliable than its useless. Since this is not just a one-off, I am fine either way.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r779714817
##########
File path: libminifi/include/core/ProcessContext.h
##########
@@ -113,10 +113,13 @@ class ProcessContext : public controller::ControllerServiceLookup, public core::
}
template<typename T>
- bool getProperty(const std::string &name, T &value) const {
+ std::enable_if_t<!std::is_convertible_v<T&, FlowFile&>, bool> getProperty(const std::string &name, T &value) const {
Review comment:
I think I just generally don't like using `std::string` for a property and didn't want to rely on that type as a discriminator. It compiles with this restriction removed, but I think it's helpful to fail early in this case if someone tries to use a missing overload `(property_name, flow_file) -> optional<T>`.
It also adds complexity to the code, so I can remove it if you think that would be an improvement.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor (no tests or docs yet)
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r753433091
##########
File path: PROCESSORS.md
##########
@@ -1458,6 +1459,28 @@ In the list below, the names of required properties appear in bold. Any other pr
| - | - |
|success|After a successful SQL update operation, the incoming FlowFile sent here|
+## PutUDP
+
+### Description
+
+The PutUDP processor receives a FlowFile and packages the FlowFile content into a single UDP datagram packet which is then transmitted to the configured UDP server.
+The processor doesn't guarantee a successful transfer, even if the flow file is routed to the success relationship.
Review comment:
The implementation doesn't check OS buffer sizes and other things to keep the implementation simple. We wouldn't be able to guarantee delivery anyway with UDP, so I thought this was an OK tradeoff. We can extend the processor later if needed.
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,171 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include <unordered_map>
+#include "range/v3/view/transform.hpp"
+#include "range/v3/range/conversion.hpp"
+#include "TestBase.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+class SingleInputTestController : public TestController {
Review comment:
This was inspired by `RouteTextController`: it also uses a sourceless connection to push custom flow files into the processor and destinationless connections to get them out.
##########
File path: CPPLINT.cfg
##########
@@ -1,2 +1,2 @@
set noparent
-filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check
+filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check,-whitespace/braces
Review comment:
I got a false positive at PutUDPTests.cpp:134 when instantiating the random generator.
```
/home/szaszm/nifi-minifi-cpp-2/extensions/standard-processors/tests/unit/PutUDPTests.cpp:133: Missing space before { [whitespace/braces] [5]
```
##########
File path: PROCESSORS.md
##########
@@ -1458,6 +1459,28 @@ In the list below, the names of required properties appear in bold. Any other pr
| - | - |
|success|After a successful SQL update operation, the incoming FlowFile sent here|
+## PutUDP
+
+### Description
+
+The PutUDP processor receives a FlowFile and packages the FlowFile content into a single UDP datagram packet which is then transmitted to the configured UDP server.
+The processor doesn't guarantee a successful transfer, even if the flow file is routed to the success relationship.
+### 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 |
+| -------- | ------------- | ---------------- | ----------------------------------------------- |
+| **Host** | localhost | | The ip address or hostname of the destination. |
+| **Port** | | | The port on the destination. |
Review comment:
No expression language support for now, although it wouldn't be too hard to add. I wanted to keep the implementation minimal until there is demand for more advanced features.
not implemented nifi properties:
- Max Size of Socket Send Buffer: I didn't bother with OS buffer sizes, defaults should work in most cases.
- Idle Connection Expiration: Not implemented because there are no connections, so there is no point in keeping track of timeouts to then do nothing when it expires.
- timeout: Doesn't do anything even in nifi
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor (no tests or docs yet)
Posted by GitBox <gi...@apache.org>.
martinzink commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r758257122
##########
File path: CPPLINT.cfg
##########
@@ -1,2 +1,2 @@
set noparent
-filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check
+filter=-runtime/reference,-runtime/string,-build/c++11,-build/include_subdir,-whitespace/forcolon,-build/namespaces_literals,-readability/check,-whitespace/braces
Review comment:
This will disable this whitespace/braces checks everywhere, if there is only one false positive we could fix that in a much smaller scope by leaving a comment // NOLINT(whitespace/braces) at the of the PutUDPTests.cpp:134 line
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r777609848
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,112 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include "SingleInputTestController.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/expected.h"
+#include "utils/StringUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+struct DatagramListener {
+ DatagramListener(const char* const hostname, const char* const port)
+ :resolved_names_{utils::net::resolveHost(hostname, port, utils::net::IpProtocol::Udp).value()},
+ open_socket_{utils::net::open_socket(resolved_names_.get())
+ | utils::valueOrElse([=]() -> utils::net::OpenSocketResult { throw std::runtime_error{utils::StringUtils::join_pack("Failed to connect to ", hostname, " on port ", port)}; })}
+ {
+ const auto bind_result = bind(open_socket_.socket_.get(), open_socket_.selected_name->ai_addr, open_socket_.selected_name->ai_addrlen);
+ if (bind_result == utils::net::SocketError) {
+ throw std::runtime_error{utils::StringUtils::join_pack("bind: ", utils::net::get_last_socket_error().message())};
+ }
+ }
+
+ struct ReceiveResult {
+ std::string remote_address;
+ std::string message;
+ };
+
+ [[nodiscard]] ReceiveResult receive(const size_t max_message_size = 8192) const {
+ ReceiveResult result;
+ result.message.resize(max_message_size);
+ sockaddr_storage remote_address{};
+ socklen_t addrlen = sizeof(remote_address);
+ const auto recv_result = recvfrom(open_socket_.socket_.get(), result.message.data(), result.message.size(), 0, std::launder(reinterpret_cast<sockaddr*>(&remote_address)), &addrlen);
Review comment:
makes sense, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor (no tests or docs yet)
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r760368931
##########
File path: extensions/standard-processors/tests/unit/PutUDPTests.cpp
##########
@@ -0,0 +1,171 @@
+/**
+ *
+ * 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 <memory>
+#include <new>
+#include <random>
+#include <string>
+#include <unordered_map>
+#include "range/v3/view/transform.hpp"
+#include "range/v3/range/conversion.hpp"
+#include "TestBase.h"
+#include "PutUDP.h"
+#include "utils/net/DNS.h"
+#include "utils/net/Socket.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+namespace {
+class SingleInputTestController : public TestController {
Review comment:
Ok, then I'll extract the common parts from this and `RouteTextController` and leave the rest for future refactoring efforts.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r779717436
##########
File path: libminifi/include/core/ProcessContext.h
##########
@@ -113,10 +113,13 @@ class ProcessContext : public controller::ControllerServiceLookup, public core::
}
template<typename T>
- bool getProperty(const std::string &name, T &value) const {
+ std::enable_if_t<!std::is_convertible_v<T&, FlowFile&>, bool> getProperty(const std::string &name, T &value) const {
Review comment:
Ideally all templates would have proper strict requirements (e.g. in this case convertibility through PropertyValue), but it's tedious to declare all of these until we're able to use C++20 concepts.
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r780130483
##########
File path: libminifi/include/core/ProcessContext.h
##########
@@ -113,10 +113,13 @@ class ProcessContext : public controller::ControllerServiceLookup, public core::
}
template<typename T>
- bool getProperty(const std::string &name, T &value) const {
+ std::enable_if_t<!std::is_convertible_v<T&, FlowFile&>, bool> getProperty(const std::string &name, T &value) const {
Review comment:
I think that the added requirement on this method has merit, but currently nothing verifies it, IMO a test that checks that `ctx.getProperty("blah", flowfile)` does not compile (or something along these lines) would help cement its purpose
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r780342657
##########
File path: libminifi/include/core/ProcessContext.h
##########
@@ -113,10 +113,13 @@ class ProcessContext : public controller::ControllerServiceLookup, public core::
}
template<typename T>
- bool getProperty(const std::string &name, T &value) const {
+ std::enable_if_t<!std::is_convertible_v<T&, FlowFile&>, bool> getProperty(const std::string &name, T &value) const {
Review comment:
Added a test in 04ef808
--
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 change in pull request #1208: MINIFICPP-1678 Create PutUDP processor
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1208:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1208#discussion_r780342657
##########
File path: libminifi/include/core/ProcessContext.h
##########
@@ -113,10 +113,13 @@ class ProcessContext : public controller::ControllerServiceLookup, public core::
}
template<typename T>
- bool getProperty(const std::string &name, T &value) const {
+ std::enable_if_t<!std::is_convertible_v<T&, FlowFile&>, bool> getProperty(const std::string &name, T &value) const {
Review comment:
Added a test in 04ef808
--
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