End to end compression/decompression mode tests

Signed-off-by: Ivan Santiago Paunovic <[email protected]>

Author: ivanpauno

test/domain_bridge/test_domain_bridge_end_to_end.cpp

@@ -21,15 +21,20 @@
 #include "rclcpp/context.hpp"
 #include "rclcpp/executors.hpp"
 #include "rclcpp/node.hpp"
+#include "rclcpp/serialization.hpp"
 #include "test_msgs/msg/basic_types.hpp"
 
+#include "domain_bridge/compress_messages.hpp"
 #include "domain_bridge/domain_bridge.hpp"
+#include "domain_bridge/msg/compressed_msg.hpp"
+
 
 static constexpr std::size_t kDomain1{1u};
 static constexpr std::size_t kDomain2{2u};
 
 using namespace std::chrono_literals;
 
+// TODO(ivanpauno): We could use parameterized testing here ...
 class TestDomainBridgeEndToEnd : public ::testing::Test
 {
 protected:
@@ -178,3 +183,72 @@ TEST_F(TestDomainBridgeEndToEnd, remap_topic_name_with_substitution)
   bridge.add_to_executor(spinner.get_executor());
   EXPECT_TRUE(poll_condition([&got_message]() {return got_message.load();}, 3s));
 }
+
+TEST_F(TestDomainBridgeEndToEnd, compress_mode)
+{
+  const std::string topic_name("test_compress");
+  const std::string domain_bridge_name("my_test_name");
+
+  std::atomic<bool> got_message = false;
+
+  // Create a publisher on domain 1
+  auto pub = node_1_->create_publisher<test_msgs::msg::BasicTypes>(topic_name, 1);
+  auto sub = node_2_->create_subscription<domain_bridge::msg::CompressedMsg>(
+    topic_name,
+    1,
+    [&got_message](const domain_bridge::msg::CompressedMsg &) {got_message = true;});
+
+  // Bridge the publisher topic to domain 2 with a remap option
+  domain_bridge::DomainBridgeOptions domain_bridge_options;
+  domain_bridge_options.name(domain_bridge_name);
+  domain_bridge_options.mode(domain_bridge::DomainBridgeOptions::Mode::Compress);
+  domain_bridge::DomainBridge bridge{domain_bridge_options};
+  bridge.bridge_topic(
+    topic_name, "test_msgs/msg/BasicTypes", kDomain1, kDomain2);
+
+  EXPECT_TRUE(poll_condition([sub]() {return sub->get_publisher_count() == 1u;}, 3s));
+  pub->publish(test_msgs::msg::BasicTypes{});
+  ScopedAsyncSpinner spinner{context_1_};
+  spinner.get_executor().add_node(node_1_);
+  spinner.get_executor().add_node(node_2_);
+  bridge.add_to_executor(spinner.get_executor());
+  EXPECT_TRUE(poll_condition([&got_message]() {return got_message.load();}, 3s));
+}
+
+TEST_F(TestDomainBridgeEndToEnd, decompress_mode)
+{
+  const std::string topic_name("test_decompress");
+  const std::string domain_bridge_name("my_test_name");
+
+  std::atomic<bool> got_message = false;
+
+  // Create a publisher on domain 1
+  auto pub = node_1_->create_publisher<domain_bridge::msg::CompressedMsg>(topic_name, 1);
+  auto sub = node_2_->create_subscription<test_msgs::msg::BasicTypes>(
+    topic_name,
+    1,
+    [&got_message](const test_msgs::msg::BasicTypes &) {got_message = true;});
+
+  // Bridge the publisher topic to domain 2 with a remap option
+  domain_bridge::DomainBridgeOptions domain_bridge_options;
+  domain_bridge_options.name(domain_bridge_name);
+  domain_bridge_options.mode(domain_bridge::DomainBridgeOptions::Mode::Decompress);
+  domain_bridge::DomainBridge bridge{domain_bridge_options};
+  bridge.bridge_topic(
+    topic_name, "test_msgs/msg/BasicTypes", kDomain1, kDomain2);
+
+  EXPECT_TRUE(poll_condition([sub]() {return sub->get_publisher_count() == 1u;}, 3s));
+  rclcpp::Serialization<test_msgs::msg::BasicTypes> serializer;
+  test_msgs::msg::BasicTypes msg;
+  rclcpp::SerializedMessage serialized_msg;
+  serializer.serialize_message(&msg, &serialized_msg);
+  domain_bridge::msg::CompressedMsg compressed_msg;
+  std::unique_ptr<ZSTD_CCtx, size_t (*)(ZSTD_CCtx *)> cctx{ZSTD_createCCtx(), &ZSTD_freeCCtx};
+  compressed_msg.data = domain_bridge::compress_message(cctx.get(), std::move(serialized_msg));
+  pub->publish(compressed_msg);
+  ScopedAsyncSpinner spinner{context_1_};
+  spinner.get_executor().add_node(node_1_);
+  spinner.get_executor().add_node(node_2_);
+  bridge.add_to_executor(spinner.get_executor());
+  EXPECT_TRUE(poll_condition([&got_message]() {return got_message.load();}, 3s));
+}