Test all events queue APIs

Author: Mauro Passerino

rclcpp/include/rclcpp/experimental/buffers/bounded_events_queue.hpp

@@ -18,7 +18,7 @@
 #include <queue>
 #include <utility>
 
-#include "rclcpp/executors/events_queue.hpp"
+#include "rclcpp/experimental/buffers/events_queue.hpp"
 
 namespace rclcpp
 {

rclcpp/include/rclcpp/experimental/buffers/simple_events_queue.hpp

@@ -18,7 +18,7 @@
 #include <queue>
 #include <utility>
 
-#include "rclcpp/executors/events_queue.hpp"
+#include "rclcpp/experimental/buffers/events_queue.hpp"
 
 namespace rclcpp
 {

rclcpp/test/rclcpp/executors/test_events_queue.cpp

@@ -12,18 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 #include <chrono>
-#include <memory>
+#include <queue>
+
 #include <gtest/gtest.h>
 
-#include "rclcpp/executors/events_executor.hpp"
+#include "rclcpp/experimental/buffers/simple_events_queue.hpp"
 #include "rclcpp/experimental/buffers/bounded_events_queue.hpp"
 
-#include "test_msgs/msg/empty.hpp"
-
 using namespace std::chrono_literals;
 
-using rclcpp::executors::EventsExecutor;
-
 class TestEventsQueue : public ::testing::Test
 {
 public:
@@ -38,109 +35,103 @@ class TestEventsQueue : public ::testing::Test
   }
 };
 
-TEST_F(TestEventsQueue, BoundedQueue)
+TEST_F(TestEventsQueue, SimpleQueueTest)
 {
-  // Create BoundedEventsQueue and set limit to 10 events.
-  auto bounded_queue = std::make_unique<rclcpp::experimental::buffers::BoundedEventsQueue>(10);
+  // Create a SimpleEventsQueue and a local queue
+  auto simple_queue = std::make_unique<rclcpp::experimental::buffers::SimpleEventsQueue>();
+  std::queue<rmw_listener_event_t> local_events_queue;
+
+  // Make sure the queue is empty after init
+  simple_queue->init();
+  EXPECT_TRUE(simple_queue->empty());
 
-  // Create an events executor using the bounded queue
-  EventsExecutor executor_sub(std::move(bounded_queue));
+  // Push 11 messages
+  for (int i = 0; i < 11; i++) {
+    rmw_listener_event_t stub_event;
+    simple_queue->push(stub_event);
+  }
 
-  // Create a subscription node
-  auto node_sub = std::make_shared<rclcpp::Node>("node_sub");
+  // Pop one message
+  simple_queue->pop();
 
-  size_t callback_count = 0;
+  local_events_queue = simple_queue->get_all_events();
 
-  auto subscription =
-    node_sub->create_subscription<test_msgs::msg::Empty>(
-      "topic",
-      rclcpp::QoS(10),
-      [&](test_msgs::msg::Empty::SharedPtr) {
-        callback_count++;
-      });
+  // We should have (11 - 1) events in the local queue
+  size_t local_queue_size = local_events_queue.size();
 
-  // Add susbscription node to the executor, so the queue can start getting events
-  executor_sub.add_node(node_sub);
+  // The local queue size should be 10
+  EXPECT_EQ(10u, local_queue_size);
 
-  // Create a publisher node
-  auto node_pub = std::make_shared<rclcpp::Node>("node_pub");
-  auto publisher = node_pub->create_publisher<test_msgs::msg::Empty>("topic", rclcpp::QoS(10));
+  // The simple queue should be empty after taking all events
+  EXPECT_TRUE(simple_queue->empty());
 
-  // Let subscriptions executor spin to execute any previous events
-  // not related to the subscription, so we start with an empty queue
-  executor_sub.spin_some(10ms);
+  // Lets push an event into the queue and get it back
+  rmw_listener_event_t push_event = {simple_queue.get(), SUBSCRIPTION_EVENT};
 
-  // Publish 11 messages, the eleventh msg should prune the queue
-  // and we should end up with only one event on it
-  for (int i = 0; i < 11; i++) {
-    publisher->publish(std::make_unique<test_msgs::msg::Empty>());
-    std::this_thread::sleep_for(1ms);
-  }
+  simple_queue->push(push_event);
 
-  // Let subscriptions executor spin
-  executor_sub.spin_some(10ms);
+  rmw_listener_event_t front_event = simple_queue->front();
 
-  // The callback count should be 1
-  EXPECT_EQ(1u, callback_count);
+  // The events should be equal
+  EXPECT_EQ(push_event.entity, front_event.entity);
+  EXPECT_EQ(push_event.type, front_event.type);
+}
 
-  // Reset callback count
-  callback_count = 0;
 
-  // Publish 5 messages
-  for (int i = 0; i < 5; i++) {
-    publisher->publish(std::make_unique<test_msgs::msg::Empty>());
-    std::this_thread::sleep_for(1ms);
+TEST_F(TestEventsQueue, BoundedQueueTest)
+{
+  // Create a BoundedEventsQueue with limit of 10 events and a local events queue
+  auto bounded_queue = std::make_unique<rclcpp::experimental::buffers::BoundedEventsQueue>(10);
+  std::queue<rmw_listener_event_t> local_events_queue;
+
+  // Make sure the queue is empty after init
+  bounded_queue->init();
+  EXPECT_TRUE(bounded_queue->empty());
+
+  // Push 11 messages, the eleventh msg should prune the queue
+  // and we should end up with only one event on it
+  for (int i = 0; i < 11; i++) {
+    rmw_listener_event_t stub_event;
+    bounded_queue->push(stub_event);
   }
 
-  // Let subscriptions executor spin
-  executor_sub.spin_some(10ms);
+  local_events_queue = bounded_queue->get_all_events();
 
-  // The callback count should be 5, the queue shouldn't have been pruned
-  EXPECT_EQ(5u, callback_count);
-}
+  size_t local_queue_size = local_events_queue.size();
 
-TEST_F(TestEventsQueue, SimpleQueue)
-{
-  // Create SimpleEventsQueue
-  auto simple_queue = std::make_unique<rclcpp::experimental::buffers::SimpleEventsQueue>();
+  // The queue size should be 1
+  EXPECT_EQ(1u, local_queue_size);
+
+  // The bounded queue should be empty after taking all events
+  EXPECT_TRUE(bounded_queue->empty());
 
-  // Create an events executor using the simple queue
-  EventsExecutor executor_sub(std::move(simple_queue));
+  // Push 5 messages
+  for (int i = 0; i < 5; i++) {
+    rmw_listener_event_t stub_event;
+    bounded_queue->push(stub_event);
+  }
 
-  // Create a subscription node with QoS->depth = 10
-  auto node_sub = std::make_shared<rclcpp::Node>("node_sub");
+  // Pop one message
+  bounded_queue->pop();
 
-  size_t callback_count = 0;
+  local_events_queue = bounded_queue->get_all_events();
 
-  auto subscription =
-    node_sub->create_subscription<test_msgs::msg::Empty>(
-      "topic",
-      rclcpp::QoS(10),
-      [&](test_msgs::msg::Empty::SharedPtr) {
-        callback_count++;
-      });
+  local_queue_size = local_events_queue.size();
 
-  // Add susbscription node to the executor, so the queue can start getting events
-  executor_sub.add_node(node_sub);
+  // The local queue size should be 4 as the bounded queue shouldn't have been pruned
+  EXPECT_EQ(4u, local_queue_size);
 
-  // Create a publisher node
-  auto node_pub = std::make_shared<rclcpp::Node>("node_pub");
-  auto publisher = node_pub->create_publisher<test_msgs::msg::Empty>("topic", rclcpp::QoS(10));
+  // The bounded queue should be empty after taking all events
+  EXPECT_TRUE(bounded_queue->empty());
 
-  // Let subscriptions executor spin to execute any previous events
-  // not related to the subscription, so we start with an empty queue
-  executor_sub.spin_some(10ms);
+  // Lets push an event into the queue and get it back
+  rmw_listener_event_t push_event = {bounded_queue.get(), WAITABLE_EVENT};
 
-  // Publish 11 messages, but as the subscription has only 10 as queue depth,
-  // we should lost a message (the events queue finish with 11 messages but one is not valid)
-  for (int i = 0; i < 11; i++) {
-    publisher->publish(std::make_unique<test_msgs::msg::Empty>());
-    std::this_thread::sleep_for(1ms);
-  }
+  bounded_queue->push(push_event);
 
-  // Let subscriptions executor spin
-  executor_sub.spin_some(10ms);
+  rmw_listener_event_t front_event = bounded_queue->front();
 
-  // The callback count should be 10
-  EXPECT_EQ(10u, callback_count);
+  // The events should be the equal
+  EXPECT_EQ(push_event.entity, front_event.entity);
+  EXPECT_EQ(push_event.type, front_event.type);
 }