Add message IDs to header.

ehpor · ehpor · commit 1eb0c2a01b19 · 2026-02-21T13:52:03.000+09:00
These are the message IDs for all topics and supertopics.
Also rename all "frame" to "message" to avoid inconsistent naming.
diff --git a/catkit_core/LocalMessageBroker.cpp b/catkit_core/LocalMessageBroker.cpp
@@ -511,7 +511,7 @@ Message LocalMessageBroker::PrepareMessageImpl(std::string_view topic, size_t pa
 	header->producer_pid = GetProcessId();
 	header->producer_timestamp = 0;
 
-	header->partial_frame_id = 0;
+	header->partial_message_id = 0;
 	header->start_byte = 0;
 	header->end_byte = payload_size;
 
@@ -548,15 +548,15 @@ Message LocalMessageBroker::PublishMessage(Message message, bool is_final)
 		std::uint64_t first_id = topic_header->first_frame_id.load(std::memory_order_relaxed);
 
 		std::uint64_t frame_id;
-		if (message.m_Header->partial_frame_id == 0)
+		if (message.m_Header->partial_message_id == 0)
 		{
 			// Get a frame ID.
 			frame_id = topic_header->ReserveNextMessageId();
 		}
 		else
 		{
 			frame_id = message.GetFrameId();
-			message.m_Header->partial_frame_id++;
+			message.m_Header->partial_message_id++;
 		}
 
 		// Check if we need to remove an old frame from the topic.
@@ -809,9 +809,7 @@ TopicHeader *LocalMessageBroker::GetTopicHeader(std::string_view topic)
 	// The topic header doesn't exist, so create it.
 	TopicHeader temp_topic_header;
 
-	temp_topic_header.next_frame_id = 0;
-	temp_topic_header.first_frame_id = 0;
-	temp_topic_header.last_frame_id = 0;
+	temp_topic_header.availability = 0;
 	temp_topic_header.frame_rate = 0.0;
 
 	topic_header = (TopicHeader *) m_TopicHeaders->Insert(topic, &temp_topic_header);
diff --git a/catkit_core/LocalMessageBroker.h b/catkit_core/LocalMessageBroker.h
@@ -2,7 +2,7 @@
 #define LOCAL_MESSAGE_BROKER_H
 
 #include "HashMap.h"
-#include "Event.h".
+#include "Event.h"
 #include "HybridPoolAllocator.h"
 #include "PoolAllocator.h"
 #include "SharedMemory.h"
diff --git a/catkit_core/MessageBroker.cpp b/catkit_core/MessageBroker.cpp
@@ -4,8 +4,8 @@
 
 #include "LocalMessageBroker.h"
 
-Message::Message(MessageHeader *header, void *payload, std::uint64_t frame_id)
-	: m_Header(header), m_Payload(payload), m_FrameId(frame_id)
+Message::Message(MessageHeader *header, void *payload)
+	: m_Header(header), m_Payload(payload)
 {
 }
 
@@ -19,14 +19,14 @@ const Uuid &Message::GetPayloadId() const
 	return m_Header->payload_id;
 }
 
-std::uint64_t Message::GetFrameId() const
+std::uint64_t Message::GetMessageId() const
 {
-	return m_FrameId;
+	return m_Header->message_ids[0];
 }
 
-std::uint16_t Message::GetPartialFrameId() const
+std::uint16_t Message::GetPartialMessageId() const
 {
-	return m_Header->partial_frame_id;
+	return m_Header->partial_message_id;
 }
 
 const Uuid &Message::GetTraceId() const
@@ -143,33 +143,33 @@ void Message::SetEndByte(std::uint64_t end_byte)
 	m_Header->end_byte = end_byte;
 }
 
-MessageSubscription::MessageSubscription(std::shared_ptr<MessageBroker> message_broker, std::string_view topic, std::uint64_t preferred_next_frame_id, MessageSubscriptionMode mode)
-	: m_MessageBroker(message_broker), m_Topic(topic), m_PreferredNextFrameId(preferred_next_frame_id), m_SubscriptionMode(mode)
+MessageSubscription::MessageSubscription(std::shared_ptr<MessageBroker> message_broker, std::string_view topic, std::uint64_t preferred_next_message_id, MessageSubscriptionMode mode)
+	: m_MessageBroker(message_broker), m_Topic(topic), m_PreferredNextMessageId(preferred_next_message_id), m_SubscriptionMode(mode)
 {
 }
 
 std::optional<Message> MessageSubscription::GetNextMessage(double timeout_in_seconds, EventWaitMethod wait_type, void (*error_check)())
 {
-	auto message = m_MessageBroker->GetNextMessage(m_Topic, m_PreferredNextFrameId, m_SubscriptionMode, timeout_in_seconds, wait_type, error_check);
+	auto message = m_MessageBroker->GetNextMessage(m_Topic, m_PreferredNextMessageId, m_SubscriptionMode, timeout_in_seconds, wait_type, error_check);
 
 	if (!message.has_value())
 		return message;
 
-	// We are going to return a message. Update our frame id for the next call.
-	m_PreferredNextFrameId = message->GetFrameId() + 1;
+	// We are going to return a message. Update our message id for the next call.
+	m_PreferredNextMessageId = message->GetMessageId() + 1;
 
 	return message;
 }
 
 std::optional<Message> MessageSubscription::TryGetNextMessage()
 {
-	auto message = m_MessageBroker->TryGetNextMessage(m_Topic, m_PreferredNextFrameId, m_SubscriptionMode);
+	auto message = m_MessageBroker->TryGetNextMessage(m_Topic, m_PreferredNextMessageId, m_SubscriptionMode);
 
 	if (!message.has_value())
 		return message;
 
-	// We are going to return a message. Update our frame id for the next call.
-	m_PreferredNextFrameId = message->GetFrameId() + 1;
+	// We are going to return a message. Update our message id for the next call.
+	m_PreferredNextMessageId = message->GetMessageId() + 1;
 
 	return message;
 }
@@ -181,7 +181,7 @@ std::optional<size_t> MessageBroker::GetCurrentMessageId(std::string_view topic)
 	if (!message.has_value())
 		return std::nullopt;
 
-	return message.value().GetFrameId();
+	return message.value().GetMessageId();
 }
 
 Message MessageBroker::PrepareMessage(std::string_view topic, size_t payload_size, uint8_t memory_block_id)
@@ -251,14 +251,14 @@ Message MessageBroker::PublishArray(std::string_view topic, ArrayView array, Uui
 MessageSubscription MessageBroker::Subscribe(std::string_view topic, MessageSubscriptionMode mode)
 {
 	auto current_message = GetCurrentMessage(topic);
-	auto starting_frame_id = current_message.has_value() ? current_message->GetFrameId() : 0;
+	auto starting_message_id = current_message.has_value() ? current_message->GetMessageId() : 0;
 
-	return Subscribe(topic, starting_frame_id, mode);
+	return Subscribe(topic, starting_message_id, mode);
 }
 
-MessageSubscription MessageBroker::Subscribe(std::string_view topic, size_t preferred_next_frame_id, MessageSubscriptionMode mode)
+MessageSubscription MessageBroker::Subscribe(std::string_view topic, size_t preferred_next_message_id, MessageSubscriptionMode mode)
 {
-	return MessageSubscription(shared_from_this(), topic, preferred_next_frame_id, mode);
+	return MessageSubscription(shared_from_this(), topic, preferred_next_message_id, mode);
 }
 
 void MessageBroker::PrintDebugInfo() const
diff --git a/catkit_core/MessageBroker.h b/catkit_core/MessageBroker.h
@@ -15,12 +15,13 @@
 #include <optional>
 
 const size_t TOPIC_MAX_KEY_SIZE = 127;
+const size_t TOPIC_MAX_DEPTH = 7;
 const size_t HOST_NAME_SIZE = 64;
 const size_t METADATA_MAX_STRLEN = 8;
 const size_t METADATA_MAX_KEYLEN = 7;
 const size_t MAX_NUM_METADATA_ENTRIES = 12;
 
-const std::uint64_t INVALID_FRAME_ID = 0xFFFFFFFFFFFFFFFF;
+const std::uint64_t INVALID_MESSAGE_ID = 0xFFFFFFFFFFFFFFFF;
 
 enum class MetadataType : std::uint8_t
 {
@@ -68,7 +69,8 @@ struct MessageHeader
 	std::uint64_t start_byte;
 	std::uint64_t end_byte;
 
-	std::uint16_t partial_frame_id;
+	std::uint64_t message_ids[TOPIC_MAX_DEPTH];
+	std::uint16_t partial_message_id;
 
 	std::uint8_t num_metadata_entries;
 	MetadataEntry metadata_entries[MAX_NUM_METADATA_ENTRIES];
@@ -85,14 +87,14 @@ class Message
 	friend class RemoteBrokerServer;
 
 private:
-	Message(MessageHeader *header, void *payload, std::uint64_t frame_id);
+	Message(MessageHeader *header, void *payload);
 
 public:
 	std::string_view GetTopic() const;
 
 	const Uuid &GetPayloadId() const;
-	std::uint64_t GetFrameId() const;
-	std::uint16_t GetPartialFrameId() const;
+	std::uint64_t GetMessageId() const;
+	std::uint16_t GetPartialMessageId() const;
 
 	const Uuid &GetTraceId() const;
 
@@ -122,8 +124,6 @@ class Message
 private:
 	MessageHeader *m_Header;
 	void *m_Payload;
-
-	std::uint64_t m_FrameId;
 };
 
 enum class MessageSubscriptionMode
@@ -143,12 +143,12 @@ class MessageSubscription
 	std::optional<Message> TryGetNextMessage();
 
 private:
-	MessageSubscription(std::shared_ptr<MessageBroker> broker, std::string_view topic, std::uint64_t preferred_next_frame_id, MessageSubscriptionMode mode);
+	MessageSubscription(std::shared_ptr<MessageBroker> broker, std::string_view topic, std::uint64_t preferred_next_message_id, MessageSubscriptionMode mode);
 
 	std::shared_ptr<MessageBroker> m_MessageBroker;
 
 	std::string m_Topic;
-	std::uint64_t m_PreferredNextFrameId;
+	std::uint64_t m_PreferredNextMessageId;
 	MessageSubscriptionMode m_SubscriptionMode;
 };
 
@@ -178,13 +178,13 @@ class MessageBroker : public std::enable_shared_from_this<MessageBroker>
 	Message PublishArray(std::string_view topic, ArrayView array, Uuid trace_id, uint8_t memory_block_id = 0);
 
 	MessageSubscription Subscribe(std::string_view topic, MessageSubscriptionMode mode = MessageSubscriptionMode::NewestOnly);
-	MessageSubscription Subscribe(std::string_view topic, size_t preferred_next_frame_id, MessageSubscriptionMode mode = MessageSubscriptionMode::NewestOnly);
+	MessageSubscription Subscribe(std::string_view topic, size_t preferred_next_message_id, MessageSubscriptionMode mode = MessageSubscriptionMode::NewestOnly);
 
 	virtual void PrintDebugInfo() const;
 
 protected:
-	virtual std::optional<Message> GetNextMessage(std::string_view topic, size_t preferred_next_frame_id, MessageSubscriptionMode mode = MessageSubscriptionMode::NewestOnly, double timeout_in_seconds = -1, EventWaitMethod wait_type = EventWaitMethod::Default, void (*error_check)() = nullptr) = 0;
-	virtual std::optional<Message> TryGetNextMessage(std::string_view topic, size_t preferred_next_frame_id, MessageSubscriptionMode mode = MessageSubscriptionMode::NewestOnly) = 0;
+	virtual std::optional<Message> GetNextMessage(std::string_view topic, size_t preferred_next_message_id, MessageSubscriptionMode mode = MessageSubscriptionMode::NewestOnly, double timeout_in_seconds = -1, EventWaitMethod wait_type = EventWaitMethod::Default, void (*error_check)() = nullptr) = 0;
+	virtual std::optional<Message> TryGetNextMessage(std::string_view topic, size_t preferred_next_message_id, MessageSubscriptionMode mode = MessageSubscriptionMode::NewestOnly) = 0;
 };
 
 #endif // MESSAGE_BROKER_H

Original file line number	Diff line number	Diff line change
`@@ -4,8 +4,8 @@`
`4`	`4`
`5`	`5`	`#include "LocalMessageBroker.h"`
`6`	`6`
`7`		`-Message::Message(MessageHeader header, void payload, std::uint64_t frame_id)`
`8`		`- : m_Header(header), m_Payload(payload), m_FrameId(frame_id)`
	`7`	`+Message::Message(MessageHeader header, void payload)`
	`8`	`+ : m_Header(header), m_Payload(payload)`
`9`	`9`	`{`
`10`	`10`	`}`
`11`	`11`
`@@ -19,14 +19,14 @@ const Uuid &Message::GetPayloadId() const`
`19`	`19`	`return m_Header->payload_id;`
`20`	`20`	`}`
`21`	`21`
`22`		`-std::uint64_t Message::GetFrameId() const`
	`22`	`+std::uint64_t Message::GetMessageId() const`
`23`	`23`	`{`
`24`		`- return m_FrameId;`
	`24`	`+ return m_Header->message_ids[0];`
`25`	`25`	`}`
`26`	`26`
`27`		`-std::uint16_t Message::GetPartialFrameId() const`
	`27`	`+std::uint16_t Message::GetPartialMessageId() const`
`28`	`28`	`{`
`29`		`- return m_Header->partial_frame_id;`
	`29`	`+ return m_Header->partial_message_id;`
`30`	`30`	`}`
`31`	`31`
`32`	`32`	`const Uuid &Message::GetTraceId() const`
`@@ -143,33 +143,33 @@ void Message::SetEndByte(std::uint64_t end_byte)`
`143`	`143`	`m_Header->end_byte = end_byte;`
`144`	`144`	`}`
`145`	`145`
`146`		`-MessageSubscription::MessageSubscription(std::shared_ptr<MessageBroker> message_broker, std::string_view topic, std::uint64_t preferred_next_frame_id, MessageSubscriptionMode mode)`
`147`		`- : m_MessageBroker(message_broker), m_Topic(topic), m_PreferredNextFrameId(preferred_next_frame_id), m_SubscriptionMode(mode)`
	`146`	`+MessageSubscription::MessageSubscription(std::shared_ptr<MessageBroker> message_broker, std::string_view topic, std::uint64_t preferred_next_message_id, MessageSubscriptionMode mode)`
	`147`	`+ : m_MessageBroker(message_broker), m_Topic(topic), m_PreferredNextMessageId(preferred_next_message_id), m_SubscriptionMode(mode)`
`148`	`148`	`{`
`149`	`149`	`}`
`150`	`150`
`151`	`151`	`std::optional<Message> MessageSubscription::GetNextMessage(double timeout_in_seconds, EventWaitMethod wait_type, void (*error_check)())`
`152`	`152`	`{`
`153`		`- auto message = m_MessageBroker->GetNextMessage(m_Topic, m_PreferredNextFrameId, m_SubscriptionMode, timeout_in_seconds, wait_type, error_check);`
	`153`	`+ auto message = m_MessageBroker->GetNextMessage(m_Topic, m_PreferredNextMessageId, m_SubscriptionMode, timeout_in_seconds, wait_type, error_check);`
`154`	`154`
`155`	`155`	`if (!message.has_value())`
`156`	`156`	`return message;`
`157`	`157`
`158`		`- // We are going to return a message. Update our frame id for the next call.`
`159`		`- m_PreferredNextFrameId = message->GetFrameId() + 1;`
	`158`	`+ // We are going to return a message. Update our message id for the next call.`
	`159`	`+ m_PreferredNextMessageId = message->GetMessageId() + 1;`
`160`	`160`
`161`	`161`	`return message;`
`162`	`162`	`}`
`163`	`163`
`164`	`164`	`std::optional<Message> MessageSubscription::TryGetNextMessage()`
`165`	`165`	`{`
`166`		`- auto message = m_MessageBroker->TryGetNextMessage(m_Topic, m_PreferredNextFrameId, m_SubscriptionMode);`
	`166`	`+ auto message = m_MessageBroker->TryGetNextMessage(m_Topic, m_PreferredNextMessageId, m_SubscriptionMode);`
`167`	`167`
`168`	`168`	`if (!message.has_value())`
`169`	`169`	`return message;`
`170`	`170`
`171`		`- // We are going to return a message. Update our frame id for the next call.`
`172`		`- m_PreferredNextFrameId = message->GetFrameId() + 1;`
	`171`	`+ // We are going to return a message. Update our message id for the next call.`
	`172`	`+ m_PreferredNextMessageId = message->GetMessageId() + 1;`
`173`	`173`
`174`	`174`	`return message;`
`175`	`175`	`}`
`@@ -181,7 +181,7 @@ std::optional<size_t> MessageBroker::GetCurrentMessageId(std::string_view topic)`
`181`	`181`	`if (!message.has_value())`
`182`	`182`	`return std::nullopt;`
`183`	`183`
`184`		`- return message.value().GetFrameId();`
	`184`	`+ return message.value().GetMessageId();`
`185`	`185`	`}`
`186`	`186`
`187`	`187`	`Message MessageBroker::PrepareMessage(std::string_view topic, size_t payload_size, uint8_t memory_block_id)`
`@@ -251,14 +251,14 @@ Message MessageBroker::PublishArray(std::string_view topic, ArrayView array, Uui`
`251`	`251`	`MessageSubscription MessageBroker::Subscribe(std::string_view topic, MessageSubscriptionMode mode)`
`252`	`252`	`{`
`253`	`253`	`auto current_message = GetCurrentMessage(topic);`
`254`		`- auto starting_frame_id = current_message.has_value() ? current_message->GetFrameId() : 0;`
	`254`	`+ auto starting_message_id = current_message.has_value() ? current_message->GetMessageId() : 0;`
`255`	`255`
`256`		`- return Subscribe(topic, starting_frame_id, mode);`
	`256`	`+ return Subscribe(topic, starting_message_id, mode);`
`257`	`257`	`}`
`258`	`258`
`259`		`-MessageSubscription MessageBroker::Subscribe(std::string_view topic, size_t preferred_next_frame_id, MessageSubscriptionMode mode)`
	`259`	`+MessageSubscription MessageBroker::Subscribe(std::string_view topic, size_t preferred_next_message_id, MessageSubscriptionMode mode)`
`260`	`260`	`{`
`261`		`- return MessageSubscription(shared_from_this(), topic, preferred_next_frame_id, mode);`
	`261`	`+ return MessageSubscription(shared_from_this(), topic, preferred_next_message_id, mode);`
`262`	`262`	`}`
`263`	`263`
`264`	`264`	`void MessageBroker::PrintDebugInfo() const`