Update data callback tests to account for all published samples. (#1900)

* Update data callback tests to account for all published samples.

Signed-off-by: Andrea Sorbini <asorbini@rti.com>

Author: asorbini

rclcpp/test/rclcpp/test_client.cpp

@@ -351,12 +351,15 @@ TEST_F(TestClient, on_new_response_callback) {
   auto client_node = std::make_shared<rclcpp::Node>("client_node", "ns");
   auto server_node = std::make_shared<rclcpp::Node>("server_node", "ns");
 
-  auto client = client_node->create_client<test_msgs::srv::Empty>("test_service");
+  rmw_qos_profile_t client_qos = rmw_qos_profile_services_default;
+  client_qos.depth = 3;
+  auto client = client_node->create_client<test_msgs::srv::Empty>("test_service", client_qos);
   std::atomic<size_t> server_requests_count {0};
   auto server_callback = [&server_requests_count](
     const test_msgs::srv::Empty::Request::SharedPtr,
     test_msgs::srv::Empty::Response::SharedPtr) {server_requests_count++;};
-  auto server = server_node->create_service<test_msgs::srv::Empty>("test_service", server_callback);
+  auto server = server_node->create_service<test_msgs::srv::Empty>(
+    "test_service", server_callback, client_qos);
   auto request = std::make_shared<test_msgs::srv::Empty::Request>();
 
   std::atomic<size_t> c1 {0};
@@ -423,7 +426,7 @@ TEST_F(TestClient, on_new_response_callback) {
   start = std::chrono::steady_clock::now();
   do {
     std::this_thread::sleep_for(100ms);
-  } while (c3 == 0 && std::chrono::steady_clock::now() - start < 10s);
+  } while (c3 < 3 && std::chrono::steady_clock::now() - start < 10s);
 
   EXPECT_EQ(c1.load(), 1u);
   EXPECT_EQ(c2.load(), 1u);

rclcpp/test/rclcpp/test_service.cpp

@@ -246,13 +246,17 @@ TEST_F(TestService, on_new_request_callback) {
   auto server_callback =
     [](const test_msgs::srv::Empty::Request::SharedPtr,
       test_msgs::srv::Empty::Response::SharedPtr) {FAIL();};
-  auto server = node->create_service<test_msgs::srv::Empty>("~/test_service", server_callback);
+  rmw_qos_profile_t service_qos = rmw_qos_profile_services_default;
+  service_qos.depth = 3;
+  auto server = node->create_service<test_msgs::srv::Empty>(
+    "~/test_service", server_callback, service_qos);
 
   std::atomic<size_t> c1 {0};
   auto increase_c1_cb = [&c1](size_t count_msgs) {c1 += count_msgs;};
   server->set_on_new_request_callback(increase_c1_cb);
 
-  auto client = node->create_client<test_msgs::srv::Empty>("~/test_service");
+  auto client = node->create_client<test_msgs::srv::Empty>(
+    "~/test_service", service_qos);
   {
     auto request = std::make_shared<test_msgs::srv::Empty::Request>();
     client->async_send_request(request);
@@ -298,7 +302,7 @@ TEST_F(TestService, on_new_request_callback) {
   start = std::chrono::steady_clock::now();
   do {
     std::this_thread::sleep_for(100ms);
-  } while (c3 == 0 && std::chrono::steady_clock::now() - start < 10s);
+  } while (c3 < 3 && std::chrono::steady_clock::now() - start < 10s);
 
   EXPECT_EQ(c1.load(), 1u);
   EXPECT_EQ(c2.load(), 1u);

rclcpp/test/rclcpp/test_subscription.cpp

@@ -454,7 +454,7 @@ TEST_F(TestSubscription, on_new_message_callback) {
   auto increase_c1_cb = [&c1](size_t count_msgs) {c1 += count_msgs;};
   sub->set_on_new_message_callback(increase_c1_cb);
 
-  auto pub = node->create_publisher<test_msgs::msg::Empty>("~/test_take", 1);
+  auto pub = node->create_publisher<test_msgs::msg::Empty>("~/test_take", 3);
   {
     test_msgs::msg::Empty msg;
     pub->publish(msg);
@@ -500,7 +500,7 @@ TEST_F(TestSubscription, on_new_message_callback) {
   start = std::chrono::steady_clock::now();
   do {
     std::this_thread::sleep_for(100ms);
-  } while (c3 == 0 && std::chrono::steady_clock::now() - start < 10s);
+  } while (c3 < 3 && std::chrono::steady_clock::now() - start < 10s);
 
   EXPECT_EQ(c1.load(), 1u);
   EXPECT_EQ(c2.load(), 1u);