Check for nullptrs when pushing new messages to the message cache

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);
+}