Merge pull request #38 from mauropasse/mauro/clean-events-executor

Implement no-op on IPC subscription

rclcpp/include/rclcpp/any_subscription_callback.hpp

@@ -185,6 +185,12 @@ class AnySubscriptionCallback
     ConstMessageSharedPtr message, const rclcpp::MessageInfo & message_info)
   {
     TRACEPOINT(callback_start, static_cast<const void *>(this), true);
+
+    // If the message is not valid, return.
+    if(!message) {
+      return;
+    }
+
     if (const_shared_ptr_callback_) {
       const_shared_ptr_callback_(message);
     } else if (const_shared_ptr_with_info_callback_) {
@@ -208,6 +214,12 @@ class AnySubscriptionCallback
     MessageUniquePtr message, const rclcpp::MessageInfo & message_info)
   {
     TRACEPOINT(callback_start, static_cast<const void *>(this), true);
+
+    // If the message is not valid, return.
+    if(!message) {
+      return;
+    }
+
     if (shared_ptr_callback_) {
       typename std::shared_ptr<MessageT> shared_message = std::move(message);
       shared_ptr_callback_(shared_message);

rclcpp/include/rclcpp/experimental/buffers/ring_buffer_implementation.hpp

@@ -91,7 +91,12 @@ class RingBufferImplementation : public BufferImplementationBase<BufferT>
 
     if (!has_data_()) {
       RCLCPP_ERROR(rclcpp::get_logger("rclcpp"), "Calling dequeue on empty intra-process buffer");
-      throw std::runtime_error("Calling dequeue on empty intra-process buffer");
+      // This situation can happen on the EventsExecutor if we have more events in the queue
+      // than messages in the history cache (set by the qos_policies.depth of the subscription)
+      // For example if we set depth=1 and we get 2 messages really fast (so no time for processing),
+      // we could end up with 2 events in the queue but only 1 msg is actually stored on the buffer.
+      // In this case we return an empty buffer.
+      return BufferT();
     }
 
     auto request = std::move(ring_buffer_[read_index_]);