From f1850fe0a8a4044a0f90f96f498ecb0a3bf7959f Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Fri, 17 Apr 2026 16:19:50 +0200 Subject: [PATCH 01/13] add zmq interface --- CMakeLists.txt | 16 +++ config/ZmqClient.yaml | 17 +++ config/ZmqServer.yaml | 19 +++ include/network_interfaces/zmq_interface.hpp | 94 +++++++++++++ network_interface_plugins.xml | 8 ++ package.xml | 1 + src/network_interfaces/zmq_interface.cpp | 135 +++++++++++++++++++ 7 files changed, 290 insertions(+) create mode 100644 config/ZmqClient.yaml create mode 100644 config/ZmqServer.yaml create mode 100644 include/network_interfaces/zmq_interface.hpp create mode 100644 src/network_interfaces/zmq_interface.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index ac0fa25..cecccb6 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -20,6 +20,7 @@ find_package(pluginlib REQUIRED) find_package(Boost REQUIRED COMPONENTS system) find_package(PkgConfig REQUIRED) pkg_check_modules(ZSTD REQUIRED libzstd) +find_package(zmqpp_vendor REQUIRED) if(BUILD_TESTING) find_package(launch_testing_ament_cmake REQUIRED) @@ -55,6 +56,10 @@ add_library(tcp_interface SHARED src/network_interfaces/tcp_interface.cpp ) +add_library(zmq_interface SHARED + src/network_interfaces/zmq_interface.cpp +) + target_link_libraries(network_bridge PUBLIC ${std_msgs_TARGETS} ${tf2_msgs_TARGETS} @@ -76,6 +81,16 @@ target_link_libraries(tcp_interface PUBLIC ${Boost_LIBRARIES} ) +ament_target_dependencies(zmq_interface + rclcpp + pluginlib + zmqpp_vendor +) + +target_link_libraries(zmq_interface + zmqpp +) + pluginlib_export_plugin_description_file(network_bridge network_interface_plugins.xml) install(TARGETS @@ -86,6 +101,7 @@ install(TARGETS install(TARGETS udp_interface tcp_interface + zmq_interface DESTINATION lib/ ) diff --git a/config/ZmqClient.yaml b/config/ZmqClient.yaml new file mode 100644 index 0000000..4d09a8f --- /dev/null +++ b/config/ZmqClient.yaml @@ -0,0 +1,17 @@ +# Configuration for the ZMQ Client (PULL) +/**/network_bridge_client: + ros__parameters: + network_interface: "network_bridge::ZmqInterface" + + ZmqInterface: + role: "client" + remote_address: "127.0.0.1" + port: 5555 + + default_rate: 10.0 + default_zstd_level: 3 + publish_stale_data: False + + # We use a publish namespace so that it doesn't clash with the server's subscription on the same machine + subscribe_namespace: "" + publish_namespace: "/client_side" diff --git a/config/ZmqServer.yaml b/config/ZmqServer.yaml new file mode 100644 index 0000000..d4a06e4 --- /dev/null +++ b/config/ZmqServer.yaml @@ -0,0 +1,19 @@ +# Configuration for the ZMQ Server (PUSH) +/**/network_bridge_server: + ros__parameters: + network_interface: "network_bridge::ZmqInterface" + + ZmqInterface: + role: "server" + port: 5555 + + default_rate: 10.0 + default_zstd_level: 3 + publish_stale_data: False + + # The server subscribes to this topic and pushes it to ZMQ + topics: + - "/test_topic" + + subscribe_namespace: "" + publish_namespace: "" diff --git a/include/network_interfaces/zmq_interface.hpp b/include/network_interfaces/zmq_interface.hpp new file mode 100644 index 0000000..b086f16 --- /dev/null +++ b/include/network_interfaces/zmq_interface.hpp @@ -0,0 +1,94 @@ +/* +============================================================================== +MIT License + +Copyright (c) 2024 Ethan M Brown +Copyright (c) 2026 PAL Robotics + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. +============================================================================== +*/ + +#pragma once + +#include +#include +#include +#include + +#include + +#include "network_interfaces/network_interface_base.hpp" + +namespace network_bridge { + +/** + * @class ZmqInterface + * @brief Represents a ZMQ network interface. + * + * The ZmqInterface class is a concrete implementation of the NetworkInterface + * abstract class. It provides functionality for opening, closing, receiving and + * writing data to a ZMQ interface. It also handles receiving data + * asynchronously and provides error handling capabilities. + */ +class ZmqInterface : public NetworkInterface { +public: + ZmqInterface() : NetworkInterface() { + ready_ = false; + failed_ = false; + } + + virtual ~ZmqInterface() { close(); } + +protected: + /** + * @brief Initializes interface by loading parameters. + * + * Called from NetworkInterface::initialize() + */ + void initialize_() override; + +public: + bool has_failed() const override; + bool is_ready() const override; + void open() override; + void close() override; + void write(const std::vector &data) override; + +protected: + void load_parameters(); + void setup_server(); + void setup_client(); + void receive_thread(); + +private: + zmqpp::context context_; + std::shared_ptr socket_; + + std::string role_; + std::string remote_address_; + int port_; + std::atomic ready_; + std::atomic failed_; + std::atomic shutting_down_; + + std::thread packet_thread_; +}; + +} // namespace network_bridge diff --git a/network_interface_plugins.xml b/network_interface_plugins.xml index 3ed802c..df92cd9 100644 --- a/network_interface_plugins.xml +++ b/network_interface_plugins.xml @@ -14,4 +14,12 @@ + + + + + A ZeroMQ network interface for the network bridge. + + + diff --git a/package.xml b/package.xml index c68d187..927833a 100644 --- a/package.xml +++ b/package.xml @@ -24,6 +24,7 @@ tf2_msgs tf2_ros pluginlib + zmqpp_vendor diff --git a/src/network_interfaces/zmq_interface.cpp b/src/network_interfaces/zmq_interface.cpp new file mode 100644 index 0000000..31b2064 --- /dev/null +++ b/src/network_interfaces/zmq_interface.cpp @@ -0,0 +1,135 @@ +/* +============================================================================== +MIT License + +Copyright (c) 2024 Ethan M Brown +Copyright (c) 2026 PAL Robotics + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. +============================================================================== +*/ + +#include + +#include "network_interfaces/zmq_interface.hpp" + +namespace network_bridge { + +void ZmqInterface::initialize_() { load_parameters(); } + +void ZmqInterface::load_parameters() { + std::string prefix = "ZmqInterface."; + node_->declare_parameter(prefix + "role", std::string("")); + node_->declare_parameter(prefix + "remote_address", std::string("")); + node_->declare_parameter(prefix + "port", 0); + + node_->get_parameter(prefix + "role", role_); + node_->get_parameter(prefix + "remote_address", remote_address_); + node_->get_parameter(prefix + "port", port_); + + RCLCPP_INFO(node_->get_logger(), "role_: %s", role_.c_str()); + RCLCPP_INFO(node_->get_logger(), "Remote Address: %s", + remote_address_.c_str()); + RCLCPP_INFO(node_->get_logger(), "Remote Port: %d", port_); +} + +void ZmqInterface::open() { + shutting_down_ = false; + failed_ = false; + ready_ = false; + if (role_ == "server") { + setup_server(); + ready_ = true; + } else if (role_ == "client") { + setup_client(); + ready_ = true; + packet_thread_ = + std::thread(std::bind(&ZmqInterface::receive_thread, this)); + } else { + RCLCPP_ERROR(node_->get_logger(), "Invalid role specified: %s", + role_.c_str()); + failed_ = true; + return; + } +} + +bool ZmqInterface::is_ready() const { return ready_ && !failed_; } + +bool ZmqInterface::has_failed() const { return failed_; } + +void ZmqInterface::close() { + if (shutting_down_.exchange(true)) { + return; + } + ready_ = false; + + if (socket_) { + try { + socket_->close(); + } catch (const zmqpp::exception &e) { + RCLCPP_ERROR(node_->get_logger(), "ZMQ exception: %s", e.what()); + } + } + + if (packet_thread_.joinable()) { + packet_thread_.join(); + } + + try { + context_.terminate(); + } catch (const zmqpp::exception &e) { + RCLCPP_ERROR(node_->get_logger(), "ZMQ exception: %s", e.what()); + } +} + +void ZmqInterface::receive_thread() { + zmqpp::poller poller; + poller.add(*socket_, zmqpp::poller::poll_in); + while (!shutting_down_ && rclcpp::ok()) { + if (poller.poll(100)) { + zmqpp::message msg; + socket_->receive(msg); + const void *data = msg.raw_data(0); + size_t size = msg.size(0); + recv_cb_( + std::span(static_cast(data), size)); + } + } +} + +void ZmqInterface::setup_server() { + socket_ = std::make_shared(context_, zmqpp::socket_type::push); + socket_->bind("tcp://*:" + std::to_string(port_)); +} + +void ZmqInterface::setup_client() { + socket_ = std::make_shared(context_, zmqpp::socket_type::pull); + socket_->connect("tcp://" + remote_address_ + ":" + std::to_string(port_)); +} + +void ZmqInterface::write(const std::vector &data) { + zmqpp::message msg; + msg.add_raw(data.data(), data.size()); + socket_->send(msg); +} + +} // namespace network_bridge + +PLUGINLIB_EXPORT_CLASS(network_bridge::ZmqInterface, + network_bridge::NetworkInterface) From 2eeb6b2dc794653b1c7dc47aaa68121b92d4ef39 Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Mon, 20 Apr 2026 10:56:19 +0200 Subject: [PATCH 02/13] Created zmq tests --- CMakeLists.txt | 4 ++ config/{ZmqServer.yaml => Zmq1.yaml} | 6 +- config/{ZmqClient.yaml => Zmq2.yaml} | 7 +- test/test_zmq.py | 102 +++++++++++++++++++++++++++ 4 files changed, 112 insertions(+), 7 deletions(-) rename config/{ZmqServer.yaml => Zmq1.yaml} (80%) rename config/{ZmqClient.yaml => Zmq2.yaml} (58%) create mode 100644 test/test_zmq.py diff --git a/CMakeLists.txt b/CMakeLists.txt index cecccb6..774bf03 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -38,6 +38,10 @@ if(BUILD_TESTING) test/test_tcp.py TIMEOUT 2 # Sets a timeout for the test in seconds ) + add_launch_test( + test/test_zmq.py + TIMEOUT 2 # Sets a timeout for the test in seconds + ) endif() include_directories(include) diff --git a/config/ZmqServer.yaml b/config/Zmq1.yaml similarity index 80% rename from config/ZmqServer.yaml rename to config/Zmq1.yaml index d4a06e4..11c6c05 100644 --- a/config/ZmqServer.yaml +++ b/config/Zmq1.yaml @@ -1,5 +1,5 @@ # Configuration for the ZMQ Server (PUSH) -/**/network_bridge_server: +/**/zmq_bridge_server: ros__parameters: network_interface: "network_bridge::ZmqInterface" @@ -15,5 +15,5 @@ topics: - "/test_topic" - subscribe_namespace: "" - publish_namespace: "" + subscribe_namespace: "/zmq1" + publish_namespace: "/zmq1" diff --git a/config/ZmqClient.yaml b/config/Zmq2.yaml similarity index 58% rename from config/ZmqClient.yaml rename to config/Zmq2.yaml index 4d09a8f..b926930 100644 --- a/config/ZmqClient.yaml +++ b/config/Zmq2.yaml @@ -1,5 +1,5 @@ # Configuration for the ZMQ Client (PULL) -/**/network_bridge_client: +/**/zmq_bridge_client: ros__parameters: network_interface: "network_bridge::ZmqInterface" @@ -12,6 +12,5 @@ default_zstd_level: 3 publish_stale_data: False - # We use a publish namespace so that it doesn't clash with the server's subscription on the same machine - subscribe_namespace: "" - publish_namespace: "/client_side" + subscribe_namespace: "/zmq2" + publish_namespace: "/zmq2" diff --git a/test/test_zmq.py b/test/test_zmq.py new file mode 100644 index 0000000..9c4d28d --- /dev/null +++ b/test/test_zmq.py @@ -0,0 +1,102 @@ +import time +import unittest + +from ament_index_python.packages import get_package_share_directory +import launch +import launch.actions +import launch_ros.actions +import launch_testing +import rclpy +from rclpy.node import Node +from rclpy.task import Future + +from std_msgs.msg import String + + +def generate_test_description(): + config = get_package_share_directory("network_bridge") + "/config/" + zmq1 = launch_ros.actions.Node( + package="network_bridge", + executable="network_bridge", + name="zmq_bridge_server", + output="screen", + parameters=[config + "Zmq1.yaml"], + arguments=["--ros-args", "--log-level", "debug", "--log-level", "rcl:=info"], + ) + + zmq2 = launch_ros.actions.Node( + package="network_bridge", + executable="network_bridge", + name="zmq_bridge_client", + output="screen", + parameters=[config + "Zmq2.yaml"], + arguments=["--ros-args", "--log-level", "debug", "--log-level", "rcl:=info"], + ) + + return launch.LaunchDescription( + [ + zmq1, + launch.actions.TimerAction(period=0.1, actions=[zmq2]), + launch_testing.actions.ReadyToTest(), + ] + ) + + +class ZmqTestNode(Node): + + def __init__(self): + super().__init__("test_node") + self.test_message_received = Future() + self.received_msg = None + self.publisher = self.create_publisher(String, "/zmq1/test_topic", 10) + self.subscriber = self.create_subscription( + String, "/zmq2/test_topic", self.listener_callback, 10 + ) + + def publish(self, msg): + self.publisher.publish(msg) + + def listener_callback(self, msg): + self.received_msg = msg + self.test_message_received.set_result(True) + + +class TestZmq(unittest.TestCase): + + @classmethod + def setUpClass(cls): + rclpy.init() + + @classmethod + def tearDownClass(cls): + rclpy.shutdown() + + def test_node_output(self, proc_output): + proc_output.assertWaitFor("Server bound", timeout=0.5) + proc_output.assertWaitFor("Client connected", timeout=0.5) + + node = ZmqTestNode() + time.sleep(0.15) + + test_msg = String() + test_msg.data = "Testing123" + node.publish(test_msg) + + try: + rclpy.spin_until_future_complete( + node, node.test_message_received, timeout_sec=10.0 + ) + self.assertTrue( + node.test_message_received.done(), "Timeout on message receival." + ) + self.assertEqual( + node.received_msg.data, + "Testing123", + "The received message did not match the expected output.", + ) + finally: + node.destroy_node() + + +if __name__ == "__main__": + launch_testing.main() From 8bd05e15e5f1172434fa7e0f24add5d4e935640d Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Mon, 20 Apr 2026 17:10:20 +0200 Subject: [PATCH 03/13] adding info prints for testing purposes --- src/network_interfaces/zmq_interface.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/network_interfaces/zmq_interface.cpp b/src/network_interfaces/zmq_interface.cpp index 31b2064..e99909f 100644 --- a/src/network_interfaces/zmq_interface.cpp +++ b/src/network_interfaces/zmq_interface.cpp @@ -116,11 +116,13 @@ void ZmqInterface::receive_thread() { void ZmqInterface::setup_server() { socket_ = std::make_shared(context_, zmqpp::socket_type::push); socket_->bind("tcp://*:" + std::to_string(port_)); + RCLCPP_INFO(node_->get_logger(), "Server bound to port %d", port_); } void ZmqInterface::setup_client() { socket_ = std::make_shared(context_, zmqpp::socket_type::pull); socket_->connect("tcp://" + remote_address_ + ":" + std::to_string(port_)); + RCLCPP_INFO(node_->get_logger(), "Client connected to port %d", port_); } void ZmqInterface::write(const std::vector &data) { From 7dc3e463018f7df3729b5e049f03ca2e264c4cc6 Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Mon, 20 Apr 2026 17:10:30 +0200 Subject: [PATCH 04/13] linting --- include/network_interfaces/zmq_interface.hpp | 14 +++-- src/network_interfaces/zmq_interface.cpp | 55 ++++++++++++-------- 2 files changed, 42 insertions(+), 27 deletions(-) diff --git a/include/network_interfaces/zmq_interface.hpp b/include/network_interfaces/zmq_interface.hpp index b086f16..8f8154c 100644 --- a/include/network_interfaces/zmq_interface.hpp +++ b/include/network_interfaces/zmq_interface.hpp @@ -36,7 +36,8 @@ SOFTWARE. #include "network_interfaces/network_interface_base.hpp" -namespace network_bridge { +namespace network_bridge +{ /** * @class ZmqInterface @@ -47,14 +48,17 @@ namespace network_bridge { * writing data to a ZMQ interface. It also handles receiving data * asynchronously and provides error handling capabilities. */ -class ZmqInterface : public NetworkInterface { +class ZmqInterface : public NetworkInterface +{ public: - ZmqInterface() : NetworkInterface() { + ZmqInterface() + : NetworkInterface() + { ready_ = false; failed_ = false; } - virtual ~ZmqInterface() { close(); } + virtual ~ZmqInterface() {close();} protected: /** @@ -69,7 +73,7 @@ class ZmqInterface : public NetworkInterface { bool is_ready() const override; void open() override; void close() override; - void write(const std::vector &data) override; + void write(const std::vector & data) override; protected: void load_parameters(); diff --git a/src/network_interfaces/zmq_interface.cpp b/src/network_interfaces/zmq_interface.cpp index e99909f..25ed7e8 100644 --- a/src/network_interfaces/zmq_interface.cpp +++ b/src/network_interfaces/zmq_interface.cpp @@ -29,11 +29,13 @@ SOFTWARE. #include "network_interfaces/zmq_interface.hpp" -namespace network_bridge { +namespace network_bridge +{ -void ZmqInterface::initialize_() { load_parameters(); } +void ZmqInterface::initialize_() {load_parameters();} -void ZmqInterface::load_parameters() { +void ZmqInterface::load_parameters() +{ std::string prefix = "ZmqInterface."; node_->declare_parameter(prefix + "role", std::string("")); node_->declare_parameter(prefix + "remote_address", std::string("")); @@ -44,12 +46,14 @@ void ZmqInterface::load_parameters() { node_->get_parameter(prefix + "port", port_); RCLCPP_INFO(node_->get_logger(), "role_: %s", role_.c_str()); - RCLCPP_INFO(node_->get_logger(), "Remote Address: %s", - remote_address_.c_str()); + RCLCPP_INFO( + node_->get_logger(), "Remote Address: %s", + remote_address_.c_str()); RCLCPP_INFO(node_->get_logger(), "Remote Port: %d", port_); } -void ZmqInterface::open() { +void ZmqInterface::open() +{ shutting_down_ = false; failed_ = false; ready_ = false; @@ -60,20 +64,22 @@ void ZmqInterface::open() { setup_client(); ready_ = true; packet_thread_ = - std::thread(std::bind(&ZmqInterface::receive_thread, this)); + std::thread(std::bind(&ZmqInterface::receive_thread, this)); } else { - RCLCPP_ERROR(node_->get_logger(), "Invalid role specified: %s", - role_.c_str()); + RCLCPP_ERROR( + node_->get_logger(), "Invalid role specified: %s", + role_.c_str()); failed_ = true; return; } } -bool ZmqInterface::is_ready() const { return ready_ && !failed_; } +bool ZmqInterface::is_ready() const {return ready_ && !failed_;} -bool ZmqInterface::has_failed() const { return failed_; } +bool ZmqInterface::has_failed() const {return failed_;} -void ZmqInterface::close() { +void ZmqInterface::close() +{ if (shutting_down_.exchange(true)) { return; } @@ -82,7 +88,7 @@ void ZmqInterface::close() { if (socket_) { try { socket_->close(); - } catch (const zmqpp::exception &e) { + } catch (const zmqpp::exception & e) { RCLCPP_ERROR(node_->get_logger(), "ZMQ exception: %s", e.what()); } } @@ -93,39 +99,43 @@ void ZmqInterface::close() { try { context_.terminate(); - } catch (const zmqpp::exception &e) { + } catch (const zmqpp::exception & e) { RCLCPP_ERROR(node_->get_logger(), "ZMQ exception: %s", e.what()); } } -void ZmqInterface::receive_thread() { +void ZmqInterface::receive_thread() +{ zmqpp::poller poller; poller.add(*socket_, zmqpp::poller::poll_in); while (!shutting_down_ && rclcpp::ok()) { if (poller.poll(100)) { zmqpp::message msg; socket_->receive(msg); - const void *data = msg.raw_data(0); + const void * data = msg.raw_data(0); size_t size = msg.size(0); recv_cb_( - std::span(static_cast(data), size)); + std::span(static_cast(data), size)); } } } -void ZmqInterface::setup_server() { +void ZmqInterface::setup_server() +{ socket_ = std::make_shared(context_, zmqpp::socket_type::push); socket_->bind("tcp://*:" + std::to_string(port_)); RCLCPP_INFO(node_->get_logger(), "Server bound to port %d", port_); } -void ZmqInterface::setup_client() { +void ZmqInterface::setup_client() +{ socket_ = std::make_shared(context_, zmqpp::socket_type::pull); socket_->connect("tcp://" + remote_address_ + ":" + std::to_string(port_)); RCLCPP_INFO(node_->get_logger(), "Client connected to port %d", port_); } -void ZmqInterface::write(const std::vector &data) { +void ZmqInterface::write(const std::vector & data) +{ zmqpp::message msg; msg.add_raw(data.data(), data.size()); socket_->send(msg); @@ -133,5 +143,6 @@ void ZmqInterface::write(const std::vector &data) { } // namespace network_bridge -PLUGINLIB_EXPORT_CLASS(network_bridge::ZmqInterface, - network_bridge::NetworkInterface) +PLUGINLIB_EXPORT_CLASS( + network_bridge::ZmqInterface, + network_bridge::NetworkInterface) From fdee8ef775dc3c44c6c1648827f8909ee67a4ead Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Tue, 21 Apr 2026 00:29:42 +0200 Subject: [PATCH 05/13] added zmq launch file --- launch/zmq.launch.py | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) create mode 100644 launch/zmq.launch.py diff --git a/launch/zmq.launch.py b/launch/zmq.launch.py new file mode 100644 index 0000000..5cf4be9 --- /dev/null +++ b/launch/zmq.launch.py @@ -0,0 +1,28 @@ +from ament_index_python.packages import get_package_share_directory +from launch import LaunchDescription +from launch_ros.actions import Node + + +def generate_launch_description(): + config_dir = get_package_share_directory("network_bridge") + zmq1_config = config_dir + "/config/Zmq1.yaml" + zmq2_config = config_dir + "/config/Zmq2.yaml" + + return LaunchDescription( + [ + Node( + package="network_bridge", + executable="network_bridge", + name="zmq_bridge_server", + output="screen", + parameters=[zmq1_config], + ), + Node( + package="network_bridge", + executable="network_bridge", + name="zmq_bridge_client", + output="screen", + parameters=[zmq2_config], + ), + ] + ) From 1556c4abac5a0179296be8e6ca7056095a8762d1 Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Tue, 21 Apr 2026 00:34:44 +0200 Subject: [PATCH 06/13] Updated README.md Included information about the ZMQ interface. --- README.md | 42 +++++++++++++++++++++++++++++++++++++++--- 1 file changed, 39 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 1598db8..cbc1dc3 100644 --- a/README.md +++ b/README.md @@ -1,22 +1,29 @@ # Network Bridge + [![CI](https://github.com/brow1633/network_bridge/actions/workflows/CI.yml/badge.svg)](https://github.com/brow1633/network_bridge/actions/workflows/CI.yml) -**Network Bridge** is a lightweight ROS2 node designed for robust communication between robotic systems over arbitrary network protocols. Supporting UDP and TCP protocols out of the box, this packages seamlessly bridges ROS2 topics across networks, facilitating effective remote communications between a base station and robotic systems, or between multiple robotic systems. +**Network Bridge** is a lightweight ROS2 node designed for robust communication between robotic systems over arbitrary network protocols. Supporting UDP, TCP, and ZMQ protocols out of the box, this packages seamlessly bridges ROS2 topics across networks, facilitating effective remote communications between a base station and robotic systems, or between multiple robotic systems. ## Installation + ### Installation via apt + Install with: + ``` sudo apt install ros--network-bridge ``` ### Building from Source + Simply clone the repository into your ROS2 workspace and build with `colcon build`. ## Usage ### Demo + #### TCP + ``` ros2 launch network_bridge tcp.launch.py @@ -26,6 +33,7 @@ ros2 topic echo /tcp2/MyDefaultTopic ``` #### UDP + ``` ros2 launch network_bridge udp.launch.py @@ -33,13 +41,29 @@ ros2 topic pub /udp1/MyDefaultTopic std_msgs/msg/String "data: 'Hello World'" ros2 topic echo /udp2/MyDefaultTopic ``` + +#### ZMQ + +``` +ros2 launch network_bridge zmq.launch.py + +ros2 topic pub /zmq1/MyDefaultTopic std_msgs/msg/String "data: 'Hello World'" + +ros2 topic echo /zmq2/MyDefaultTopic +``` + ### Configuration + Simply setup the network interface parameters and list your desired topics to get started. If you are using UDP over cellular data, it is recommended to setup a VPN to facilitate connection. Also, please note that **no encryption** occurs within this package. Currently, if you would like encryption, you must use a VPN. See `config/Udp1.yaml` for a description of all parameters, as well as the TCP example configuration files. + #### Minimal Example + The following configuration examples demonstrate a robot sending a message on `/gps/fix` over UDP to a basestation that will then re-publish the message. This works seamlessly on all message types, so long as they are built and sourced on both ends of the transmission. + #### Robot + ``` /udp_sender: ros__parameters: @@ -52,7 +76,9 @@ The following configuration examples demonstrate a robot sending a message on `/ topics: - "/gps/fix" ``` + #### Base Station + ``` /udp_receiver: ros__parameters: @@ -62,11 +88,14 @@ The following configuration examples demonstrate a robot sending a message on `/ remote_address: "192.168.1.2" send_port: 5001 ``` + #### Special case: TF + The TF topic `/tf` or `/tf_static` are handled as a special cases. The subscriber side will listen to all TF messages, accumulate them (similarly to a TF buffer) and send all of them at the specified rate. The behavior can be disabled or forced using the `is_tf` configuration. If some TFs need to be excluded or if the list of TFs to include is finite, one can use the include and exclude regex parameters. A transform is matched (hence excluded or included) if either the `frame_id` or `child_frame_id` are matching a pattern. + ``` /udp_sender: ros__parameters: @@ -92,21 +121,27 @@ If some TFs need to be excluded or if the list of TFs to include is finite, one ``` ### Choice of protocol + - **UDP**: Use UDP for low-latency, high-throughput communications, where occasional data loss is tolerable. Ideal for real-time telemetry data like sensor streams. - **TCP**: Opt for TCP when data integrity and reliability are critical. This ensures that control commands and state transitions are reliably delivered, though with potentially higher latency. +- **ZMQ**: Use ZMQ for reliable, high-performance messaging patterns. It provides a more robust and flexible communication architecture compared to raw TCP/UDP, abstracting away complex socket management. -Network protocols are implemented as pluginlib plugins, allowing the creation of arbitrary interfaces using the abstract class `include/network_interfaces/network_interface_base.hpp`. Any interface that can send and receive bytes could theoretically be implemented, including protocols that go beyond point-to-point communication, such as ZMQ. Please consider opening a pull request if you implement a new network interface. +Network protocols are implemented as pluginlib plugins, allowing the creation of arbitrary interfaces using the abstract class `include/network_interfaces/network_interface_base.hpp`. Any interface that can send and receive bytes could theoretically be implemented, including protocols that go beyond point-to-point communication. Please consider opening a pull request if you implement a new network interface. ### Tuning + This node can be launched with logger level DEBUG, which provides useful information for tuning the compression, rate and stale message parameters. For each message that is sent, the receiving side will output the number of bytes received, the decompressed size in bytes and the transmission delay. ### Contributing + Thank you for considering contributing! #### Code Formatting + Python code is formatted with `black`, and C++ is formatted with `uncrustify`. #### Pre-commit hooks + To ease the friction of linting, there are pre-commit hooks that you can install: ```bash @@ -120,4 +155,5 @@ pre-commit install # Run on commit automatically which will reformat code automatically when you commit changes. ## Acknowledgements -This package was developed for use in the Indy Autonomous Challenge by the Purdue AI Racing team. Inspiration was taken from mqtt_client (https://github.com/ika-rwth-aachen/mqtt_client/). + +This package was developed for use in the Indy Autonomous Challenge by the Purdue AI Racing team. Inspiration was taken from mqtt_client (). From 0a683f3a1c9c57c81548fe99eb6f5f6013f3592f Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Wed, 22 Apr 2026 15:52:13 +0200 Subject: [PATCH 07/13] fixing linking problems for versions older than humble --- CMakeLists.txt | 4 ---- 1 file changed, 4 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 774bf03..0e47613 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -91,10 +91,6 @@ ament_target_dependencies(zmq_interface zmqpp_vendor ) -target_link_libraries(zmq_interface - zmqpp -) - pluginlib_export_plugin_description_file(network_bridge network_interface_plugins.xml) install(TARGETS From c204af6c95073d7473f63f1ccdbd51c1130dd7b4 Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Wed, 22 Apr 2026 15:52:53 +0200 Subject: [PATCH 08/13] fixing test temporization issues for ZMQ interfaces --- CMakeLists.txt | 2 +- test/test_zmq.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 0e47613..70145d9 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -40,7 +40,7 @@ if(BUILD_TESTING) ) add_launch_test( test/test_zmq.py - TIMEOUT 2 # Sets a timeout for the test in seconds + TIMEOUT 5 # Sets a timeout for the test in seconds ) endif() diff --git a/test/test_zmq.py b/test/test_zmq.py index 9c4d28d..84f3db0 100644 --- a/test/test_zmq.py +++ b/test/test_zmq.py @@ -76,7 +76,7 @@ def test_node_output(self, proc_output): proc_output.assertWaitFor("Client connected", timeout=0.5) node = ZmqTestNode() - time.sleep(0.15) + time.sleep(1.5) test_msg = String() test_msg.data = "Testing123" From 1cb28a9ace823883df63f49c2f63e4521404d963 Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Wed, 22 Apr 2026 15:54:16 +0200 Subject: [PATCH 09/13] socket bind/connection exception management --- src/network_interfaces/zmq_interface.cpp | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/src/network_interfaces/zmq_interface.cpp b/src/network_interfaces/zmq_interface.cpp index 25ed7e8..e442617 100644 --- a/src/network_interfaces/zmq_interface.cpp +++ b/src/network_interfaces/zmq_interface.cpp @@ -123,14 +123,26 @@ void ZmqInterface::receive_thread() void ZmqInterface::setup_server() { socket_ = std::make_shared(context_, zmqpp::socket_type::push); - socket_->bind("tcp://*:" + std::to_string(port_)); + try { + socket_->bind("tcp://*:" + std::to_string(port_)); + } catch (const zmqpp::exception & e) { + RCLCPP_ERROR(node_->get_logger(), "Bind failed: %s", e.what()); + failed_ = true; + return; + } RCLCPP_INFO(node_->get_logger(), "Server bound to port %d", port_); } void ZmqInterface::setup_client() { socket_ = std::make_shared(context_, zmqpp::socket_type::pull); - socket_->connect("tcp://" + remote_address_ + ":" + std::to_string(port_)); + try { + socket_->connect("tcp://" + remote_address_ + ":" + std::to_string(port_)); + } catch (const zmqpp::exception & e) { + RCLCPP_ERROR(node_->get_logger(), "Connect failed: %s", e.what()); + failed_ = true; + return; + } RCLCPP_INFO(node_->get_logger(), "Client connected to port %d", port_); } From 1ae0b00f29c2c5c2da663e53dfc49be9e7ee0089 Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Mon, 4 May 2026 15:55:07 +0200 Subject: [PATCH 10/13] Introduced ZMQ pattern parameter - Defined parameter to select the message forwarding/receiving pattern - Adapted socket connection and binding to use the specified pattern --- config/Zmq1.yaml | 3 ++- config/Zmq2.yaml | 3 ++- include/network_interfaces/zmq_interface.hpp | 1 + src/network_interfaces/zmq_interface.cpp | 14 ++++++++++++-- 4 files changed, 17 insertions(+), 4 deletions(-) diff --git a/config/Zmq1.yaml b/config/Zmq1.yaml index 11c6c05..8480df9 100644 --- a/config/Zmq1.yaml +++ b/config/Zmq1.yaml @@ -1,10 +1,11 @@ -# Configuration for the ZMQ Server (PUSH) +# Configuration for the ZMQ Server (PUSH/PULL or PUB/SUB) /**/zmq_bridge_server: ros__parameters: network_interface: "network_bridge::ZmqInterface" ZmqInterface: role: "server" + # pattern: "pub_sub" # Options: pub_sub (default) or push_pull port: 5555 default_rate: 10.0 diff --git a/config/Zmq2.yaml b/config/Zmq2.yaml index b926930..6ab23cd 100644 --- a/config/Zmq2.yaml +++ b/config/Zmq2.yaml @@ -1,10 +1,11 @@ -# Configuration for the ZMQ Client (PULL) +# Configuration for the ZMQ Client (PUSH/PULL or PUB/SUB) /**/zmq_bridge_client: ros__parameters: network_interface: "network_bridge::ZmqInterface" ZmqInterface: role: "client" + # pattern: "pub_sub" # Options: pub_sub (default) or push_pull remote_address: "127.0.0.1" port: 5555 diff --git a/include/network_interfaces/zmq_interface.hpp b/include/network_interfaces/zmq_interface.hpp index 8f8154c..aa312a1 100644 --- a/include/network_interfaces/zmq_interface.hpp +++ b/include/network_interfaces/zmq_interface.hpp @@ -86,6 +86,7 @@ class ZmqInterface : public NetworkInterface std::shared_ptr socket_; std::string role_; + std::string pattern_; std::string remote_address_; int port_; std::atomic ready_; diff --git a/src/network_interfaces/zmq_interface.cpp b/src/network_interfaces/zmq_interface.cpp index e442617..e340d37 100644 --- a/src/network_interfaces/zmq_interface.cpp +++ b/src/network_interfaces/zmq_interface.cpp @@ -38,14 +38,17 @@ void ZmqInterface::load_parameters() { std::string prefix = "ZmqInterface."; node_->declare_parameter(prefix + "role", std::string("")); + node_->declare_parameter(prefix + "pattern", std::string("pub_sub")); node_->declare_parameter(prefix + "remote_address", std::string("")); node_->declare_parameter(prefix + "port", 0); node_->get_parameter(prefix + "role", role_); + node_->get_parameter(prefix + "pattern", pattern_); node_->get_parameter(prefix + "remote_address", remote_address_); node_->get_parameter(prefix + "port", port_); RCLCPP_INFO(node_->get_logger(), "role_: %s", role_.c_str()); + RCLCPP_INFO(node_->get_logger(), "pattern_: %s", pattern_.c_str()); RCLCPP_INFO( node_->get_logger(), "Remote Address: %s", remote_address_.c_str()); @@ -122,7 +125,9 @@ void ZmqInterface::receive_thread() void ZmqInterface::setup_server() { - socket_ = std::make_shared(context_, zmqpp::socket_type::push); + zmqpp::socket_type type = + (pattern_ == "pub_sub") ? zmqpp::socket_type::pub : zmqpp::socket_type::push; + socket_ = std::make_shared(context_, type); try { socket_->bind("tcp://*:" + std::to_string(port_)); } catch (const zmqpp::exception & e) { @@ -135,7 +140,12 @@ void ZmqInterface::setup_server() void ZmqInterface::setup_client() { - socket_ = std::make_shared(context_, zmqpp::socket_type::pull); + zmqpp::socket_type type = + (pattern_ == "pub_sub") ? zmqpp::socket_type::sub : zmqpp::socket_type::pull; + socket_ = std::make_shared(context_, type); + if (pattern_ == "pub_sub") { + socket_->subscribe(""); + } try { socket_->connect("tcp://" + remote_address_ + ":" + std::to_string(port_)); } catch (const zmqpp::exception & e) { From 8812346e86a57691fc98ad413d7fbf65f356d5d4 Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Mon, 4 May 2026 16:41:13 +0200 Subject: [PATCH 11/13] Introduced tests for both ZMQ patterns --- CMakeLists.txt | 2 +- config/Zmq1PushPull.yaml | 20 ++++++++++++++ config/Zmq2PushPull.yaml | 17 ++++++++++++ test/test_zmq.py | 57 ++++++++++++++++++++++++++++++++++++---- 4 files changed, 90 insertions(+), 6 deletions(-) create mode 100644 config/Zmq1PushPull.yaml create mode 100644 config/Zmq2PushPull.yaml diff --git a/CMakeLists.txt b/CMakeLists.txt index 70145d9..5e982d2 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -40,7 +40,7 @@ if(BUILD_TESTING) ) add_launch_test( test/test_zmq.py - TIMEOUT 5 # Sets a timeout for the test in seconds + TIMEOUT 10 # Sets a timeout for the test in seconds ) endif() diff --git a/config/Zmq1PushPull.yaml b/config/Zmq1PushPull.yaml new file mode 100644 index 0000000..baad6f5 --- /dev/null +++ b/config/Zmq1PushPull.yaml @@ -0,0 +1,20 @@ +# Configuration for the ZMQ Server (PUSH/PULL or PUB/SUB) +/**/zmq_bridge_server_push: + ros__parameters: + network_interface: "network_bridge::ZmqInterface" + + ZmqInterface: + role: "server" + pattern: "push_pull" # Options: pub_sub (default) or push_pull + port: 5556 + + default_rate: 10.0 + default_zstd_level: 3 + publish_stale_data: False + + # The server subscribes to this topic and pushes it to ZMQ + topics: + - "/test_topic" + + subscribe_namespace: "/zmq1/push_pull" + publish_namespace: "/zmq1/push_pull" diff --git a/config/Zmq2PushPull.yaml b/config/Zmq2PushPull.yaml new file mode 100644 index 0000000..613f5f2 --- /dev/null +++ b/config/Zmq2PushPull.yaml @@ -0,0 +1,17 @@ +# Configuration for the ZMQ Client (PUSH/PULL or PUB/SUB) +/**/zmq_bridge_client_pull: + ros__parameters: + network_interface: "network_bridge::ZmqInterface" + + ZmqInterface: + role: "client" + pattern: "push_pull" # Options: pub_sub (default) or push_pull + remote_address: "127.0.0.1" + port: 5556 + + default_rate: 10.0 + default_zstd_level: 3 + publish_stale_data: False + + subscribe_namespace: "/zmq2/push_pull" + publish_namespace: "/zmq2/push_pull" diff --git a/test/test_zmq.py b/test/test_zmq.py index 84f3db0..f5d0d58 100644 --- a/test/test_zmq.py +++ b/test/test_zmq.py @@ -33,10 +33,30 @@ def generate_test_description(): arguments=["--ros-args", "--log-level", "debug", "--log-level", "rcl:=info"], ) + zmq1_push_pull = launch_ros.actions.Node( + package="network_bridge", + executable="network_bridge", + name="zmq_bridge_server_push", + output="screen", + parameters=[config + "Zmq1PushPull.yaml"], + arguments=["--ros-args", "--log-level", "debug", "--log-level", "rcl:=info"] + ) + + zmq2_push_pull = launch_ros.actions.Node( + package="network_bridge", + executable="network_bridge", + name="zmq_bridge_client_pull", + output="screen", + parameters=[config + "Zmq2PushPull.yaml"], + arguments=["--ros-args", "--log-level", "debug", "--log-level", "rcl:=info"] + ) + return launch.LaunchDescription( [ zmq1, launch.actions.TimerAction(period=0.1, actions=[zmq2]), + launch.actions.TimerAction(period=0.2, actions=[zmq1_push_pull]), + launch.actions.TimerAction(period=0.3, actions=[zmq2_push_pull]), launch_testing.actions.ReadyToTest(), ] ) @@ -44,13 +64,13 @@ def generate_test_description(): class ZmqTestNode(Node): - def __init__(self): - super().__init__("test_node") + def __init__(self, name, pub_topic, sub_topic): + super().__init__(name) self.test_message_received = Future() self.received_msg = None - self.publisher = self.create_publisher(String, "/zmq1/test_topic", 10) + self.publisher = self.create_publisher(String, pub_topic, 10) self.subscriber = self.create_subscription( - String, "/zmq2/test_topic", self.listener_callback, 10 + String, sub_topic, self.listener_callback, 10 ) def publish(self, msg): @@ -75,7 +95,10 @@ def test_node_output(self, proc_output): proc_output.assertWaitFor("Server bound", timeout=0.5) proc_output.assertWaitFor("Client connected", timeout=0.5) - node = ZmqTestNode() + node = ZmqTestNode( + "zmq_test_node", + "/zmq1/test_topic", + "/zmq2/test_topic") time.sleep(1.5) test_msg = String() @@ -97,6 +120,30 @@ def test_node_output(self, proc_output): finally: node.destroy_node() + node_push_pull = ZmqTestNode( + "zmq_push_pull_test_node", + "/zmq1/push_pull/test_topic", + "/zmq2/push_pull/test_topic") + time.sleep(1.5) + + node_push_pull.publish(test_msg) + + try: + rclpy.spin_until_future_complete( + node_push_pull, + node_push_pull.test_message_received, + timeout_sec=10.0 + ) + self.assertTrue( + node_push_pull.test_message_received.done(), "Timeout on message receival." + ) + self.assertEqual( + node_push_pull.received_msg.data, + "Testing123", + "The received message did not match the expected output.", + ) + finally: + node_push_pull.destroy_node() if __name__ == "__main__": launch_testing.main() From 10038cba801373b94def6b61dc80fe30518f0062 Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Tue, 5 May 2026 14:38:07 +0200 Subject: [PATCH 12/13] removed deprecated ament_target_dependencies from CMakeLists --- CMakeLists.txt | 26 +++++++++++++++++++++----- 1 file changed, 21 insertions(+), 5 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 5e982d2..ce9192b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -85,11 +85,27 @@ target_link_libraries(tcp_interface PUBLIC ${Boost_LIBRARIES} ) -ament_target_dependencies(zmq_interface - rclcpp - pluginlib - zmqpp_vendor -) +if(DEFINED zmqpp_vendor_INCLUDE_DIRS) + target_include_directories(zmq_interface PUBLIC + ${zmqpp_vendor_INCLUDE_DIRS} + ) +endif() + +if(DEFINED zmqpp_vendor_TARGETS) + target_link_libraries(zmq_interface PUBLIC + rclcpp::rclcpp + pluginlib::pluginlib + ${zmqpp_vendor_TARGETS} + ) +endif() + +if(DEFINED zmqpp_vendor_LIBRARIES) + target_link_libraries(zmq_interface PUBLIC + rclcpp::rclcpp + pluginlib::pluginlib + ${zmqpp_vendor_LIBRARIES} + ) +endif() pluginlib_export_plugin_description_file(network_bridge network_interface_plugins.xml) From b04bc137ea8ca0b3357ab0b57e279c4086b5fd32 Mon Sep 17 00:00:00 2001 From: Lorenzo Ferrini Date: Mon, 11 May 2026 11:21:47 +0200 Subject: [PATCH 13/13] updated README.md - Included information about the supported ZMQ messaging patterns. - Put information on network protocols implementation under dedicated subsection --- README.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/README.md b/README.md index cbc1dc3..d4cd007 100644 --- a/README.md +++ b/README.md @@ -126,6 +126,16 @@ If some TFs need to be excluded or if the list of TFs to include is finite, one - **TCP**: Opt for TCP when data integrity and reliability are critical. This ensures that control commands and state transitions are reliably delivered, though with potentially higher latency. - **ZMQ**: Use ZMQ for reliable, high-performance messaging patterns. It provides a more robust and flexible communication architecture compared to raw TCP/UDP, abstracting away complex socket management. +#### ZMQ Communication Patterns + +ZMQ supports multiple messaging patterns. This package currently supports two, configurable via the `pattern` parameter: + +- **PUB/SUB** (`pattern: pub_sub`): One publisher broadcasts messages to multiple subscribers. Best for one-to-many data distribution (e.g., sensor streams to multiple consumers). Subscribers receive only the topics they subscribe to; late-joining subscribers miss messages sent before connection. + +- **PUSH/PULL** (`pattern: push_pull`): One pusher sends messages to a pool of pullers in a round-robin fashion. Best for load-balanced pipelines where each message must be processed by exactly one consumer. Provides back-pressure and queuing, unlike PUB/SUB. + +### Network Protocol Implementation + Network protocols are implemented as pluginlib plugins, allowing the creation of arbitrary interfaces using the abstract class `include/network_interfaces/network_interface_base.hpp`. Any interface that can send and receive bytes could theoretically be implemented, including protocols that go beyond point-to-point communication. Please consider opening a pull request if you implement a new network interface. ### Tuning