Check for nullptrs when pushing new messages to the message cache (#2219)

* fix: add nullptr check when pushing new messages to the message cache

Signed-off-by: Jose Faria <[email protected]>

* refactor: apply changes from review

Signed-off-by: Jose Faria <[email protected]>

* nit: update doxygen comments

Signed-off-by: jncfa <[email protected]>

* Update Doxygen comment for the MessageCacheCircularBuffer::push(msg)

Signed-off-by: Michael Orlov <[email protected]>

---------

Signed-off-by: Jose Faria <[email protected]>
Signed-off-by: jncfa <[email protected]>
Signed-off-by: Michael Orlov <[email protected]>
Co-authored-by: Michael Orlov <[email protected]>

Author: jncfa

rosbag2_cpp/include/rosbag2_cpp/cache/circular_message_cache.hpp

@@ -58,8 +58,8 @@ class ROSBAG2_CPP_PUBLIC CircularMessageCache
 
   /// Puts msg into circular buffer, replacing the oldest msg when buffer is full
   /// \return True if message was successfully pushed, otherwise false.
-  /// NOTE: This will always return true, since the circular buffer by design drops old messages
-  /// when the buffer is full.
+  /// NOTE: Unless message is null or too large for the buffer, this will always return true
+  /// since the circular buffer by design drops old messages when the buffer is full.
   bool push(std::shared_ptr<const rosbag2_storage::SerializedBagMessage> msg) override;
 
   /// Get current buffer to consume.

rosbag2_cpp/include/rosbag2_cpp/cache/message_cache_circular_buffer.hpp

@@ -54,8 +54,12 @@ class ROSBAG2_CPP_PUBLIC MessageCacheCircularBuffer
   explicit MessageCacheCircularBuffer(size_t max_cache_size);
 
   /**
-  * If buffer size has some space left, we push the message regardless of its size,
-  *  but if this results in exceeding buffer size, we begin dropping old messages.
+  * \brief Pushes a SerializedBagMessage into the cache buffer.
+  * \details If buffer size has some space left, we push the message regardless of its size,
+  * but if this results in exceeding buffer size, we begin dropping old messages.
+  * \param msg SerializedBagMessage to add to the buffer.
+  * \return True if message was successfully pushed. Returns false if msg is null or if msg size
+  * exceeds max buffer size.
   */
   bool push(CacheBufferInterface::buffer_element_t msg) override;
 

rosbag2_cpp/src/rosbag2_cpp/cache/circular_message_cache.cpp

@@ -43,10 +43,7 @@ CircularMessageCache::~CircularMessageCache()
 bool CircularMessageCache::push(std::shared_ptr<const rosbag2_storage::SerializedBagMessage> msg)
 {
   std::lock_guard<std::mutex> cache_lock(producer_buffer_mutex_);
-  producer_buffer_->push(msg);
-  // Always return true since circular message cache drops old messages by design and it
-  // shouldn't be counted as a lost messages.
-  return true;
+  return producer_buffer_->push(msg);
 }
 
 std::shared_ptr<CacheBufferInterface> CircularMessageCache::get_consumer_buffer()

rosbag2_cpp/src/rosbag2_cpp/cache/message_cache_circular_buffer.cpp

@@ -32,9 +32,14 @@ MessageCacheCircularBuffer::MessageCacheCircularBuffer(size_t max_cache_size)
 
 bool MessageCacheCircularBuffer::push(CacheBufferInterface::buffer_element_t msg)
 {
+  if (!msg || !msg->serialized_data) {
+    ROSBAG2_CPP_LOG_ERROR("Attempted to push null message into circular buffer. Dropping message!");
+    return false;
+  }
+
   // Drop message if it exceeds the buffer size
   if (msg->serialized_data->buffer_length > max_bytes_size_) {
-    ROSBAG2_CPP_LOG_WARN_STREAM("Last message exceeds snapshot buffer size. Dropping message!");
+    ROSBAG2_CPP_LOG_WARN("Last message exceeds snapshot buffer size. Dropping message!");
     return false;
   }
 

rosbag2_cpp/test/rosbag2_cpp/test_circular_message_cache.cpp

@@ -133,3 +133,12 @@ TEST_F(CircularMessageCacheTest, circular_message_cache_ensure_empty) {
   EXPECT_THAT(circular_message_cache->get_consumer_buffer()->size(), Eq(0u));
   circular_message_cache->release_consumer_buffer();
 }
+
+TEST_F(CircularMessageCacheTest, circular_message_cache_rejects_null_message) {
+  auto circular_message_cache_ = std::make_shared<rosbag2_cpp::cache::CircularMessageCache>(
+  cache_size_);
+
+  bool result = true;
+  ASSERT_NO_THROW(result = circular_message_cache_->push(nullptr));
+  EXPECT_FALSE(result);
+}