feat(cpp): add messaging FFI functions for C++ SDK

[seokjin0414]

Summary

Closes #2965 (Phase 1 of 2).

• Add get_streams, send_messages, poll_messages FFI functions to C++ SDK
• Add shared structs: Stream, IggyMessageToSend, IggyMessagePolled, PolledMessages
• Add 6 e2e tests for message round-trip
• All additions are purely additive — no existing FFI changes

Discussed with @slbotbm on Discord — PR split and message struct design confirmed.

Test plan

• cargo check / cargo clippy / cargo fmt clean
• 6 e2e tests in tests/message/low_level_e2e.cpp (Bazel + iggy-server)

[slbotbm]

@seokjin0414 Thanks for the PR!

Here is my first round of review:
The CI is failing. Please check which test is failing. Also, the cpp code is likely not linted properly, so please do that as well.

---

Let's unify the Message struct, instead of having separate structs for sending and polling. The unified struct could look like:

struct Message {
      checksum: u64,
      id_lo: u64,
      id_hi: u64,
      offset: u64,
      timestamp: u64,
      origin_timestamp: u64,
      user_headers_length: u32,
      payload_length: u32,
      reserved: u64,
      payload: Vec<u8>,
      user_headers: Vec<u8>,
  }

Define a function using impl Message that would allow the user to create a message to send. This has to be exported to the cpp side. If the bdd tests only require the payload to be set, you can just define a function called from(payload: Vec<u8>).

---

You are using map(Into::into) in many places. Let's use from to be more explicit about what is being converted to what.

---

let mut iggy_messages: Vec<IggyMessage> = messages
            .into_iter()
            .map(|m| {
                let id = ((m.id_hi as u128) << 64) | (m.id_lo as u128);
                let payload = Bytes::from(m.payload);
                let msg = if id > 0 {
                    IggyMessage::builder().id(id).payload(payload).build()
                } else {
                    IggyMessage::builder().payload(payload).build()
                };
                msg.map_err(|error| format!("Could not build message: {error}"))
            })
            .collect::<Result<Vec<_>, _>>()?;

In the above code, please define TryFrom<ffi::Message> for IggyMessage for the conversion and use that.

---

Let's rename strategy_kind and strategy_value to partitioning_strategy_kind and partitioning_strategy_value respectively so that there is no ambiguity about its meaning.

[seokjin0414]

@slbotbm Thanks for the review! Addressed all feedback in c8f9adf:

1. CI fix: clang-format applied, fragile s.id > 0 assertion removed (75c2a28)
2. Unified Message struct: Merged IggyMessageToSend/IggyMessagePolled into single Message with all header fields. Added new_message(payload) factory function exported to C++
3. TryFrom: Defined TryFrom<ffi::Message> for IggyMessage and use it in send_messages
4. Explicit from: Replaced map(Into::into) with map(ffi::Stream::from) / map(ffi::Message::from)
5. Naming: Renamed to polling_strategy_kind/polling_strategy_value — these refer to the polling strategy (offset/timestamp/first/last/next), not partitioning. Partitioning params (partitioning_kind/partitioning_value) are on send_messages separately

[slbotbm]

@seokjin0414 Thanks for the quick turn-around.

I may not have been clear in my review previously, but I wanted you to implement something more like

impl ffi::Message {
      pub fn new_message(&mut self, payload: Vec<u8>) -> Result<(), String> {
          if payload.is_empty() {
              return Err("Could not create message: payload must not be empty".to_string());
          }

          let payload_length = payload.len() as u32;

          *self = Self {
              checksum: 0,
              id_lo: 0,
              id_hi: 0,
              offset: 0,
              timestamp: 0,
              origin_timestamp: 0,
              user_headers_length: 0,
              payload_length,
              reserved: 0,
              payload,
              user_headers: Vec::new(),
          };

          Ok(())
      }
}

The above interface would allow us to call Message.new_message(...) from the cpp side while avoiding ambiguity.

---

GetStreamsReturnsEmptyAfterCleanup and GetStreamsReturnsStreamAfterCreation test the get_streams function, but they are in the messages/ folder. Please move them to their appropriate folders.

---

The amount of testing is insufficient for the three functions.
Please think about the following and add tests:

• get_streams:

  • For GetStreamsReturnsStreamAfterCreation, verify everything that is verifiable, not just the name. Also create additional streams and topics inside them, and verify everything.
  • Does the sdk allow calling get_streams before login and connect?
  • Does calling get_stream and get_streams when there is only one one stream return the same stream with the same details?
  • What happens when you call get_streams repeatedly?
  • Lastly, in GetStreamsReturnsEmptyAfterCleanup, you are deleting every stream without ever creating any. Instead of this, let's create some unique streams, delete those, and then verify whether 0 is returned or not. Ignore this please.

• send_messages:

  • What happens when you call send_messages before login and connect?
  • What happens when you give invalid topic and / or stream identifiers to the function?
  • What happens when you give non-existent topic and / or stream identifiers to the function?
  • What happens when you give wrong partitioning kind and / or value to the function? (also test the relevant limits)
  • Verify that the messages are being stored in partitions according to the strategy we have strategy we have chosen.
  • What happens when you supply an empty vector to the send_messages?
  • Is there a limit on how many messages can be sent in total, or in a single batch?
  • What happens when you try to send an invalid payload through send_messages (empty payload, payload size > 64,000,000 bytes)
  • When send_messages is called multiple times, is the order of the messages preserved?
  • You have written a test where one custom ID is set and sent to the server. What happens when you set the same ID for multiple messages?
  • Lastly, since we are working with bytes, I would like to you to try sending a variety of payloads, and check whether they succeed or not. (null bytes, utf-8 strings, emojis, etc. - consider using property based testing)

• poll_messages :

  • What happens when you call poll_messages before login and connect?
  • What happens when you give invalid topic, stream, and / or consumer identifiers to the function?
  • What happens when you give non-existent topic, stream, and / or consumer identifiers to the function?
  • What happens when you give a wrong consumer_kind to the function?
  • What happens when you give a wrong polling strategy kind and / or value to the function? (also test the relevant limits)
  • What happens when you give an invalid or non-existent partition ID to the function?
  • What happens when you supply count = 0 to the poll_messages?
  • Is there a limit on how many messages can be requested in a single poll?
  • When the requested count is smaller than the number of available messages, are only that many messages returned?
  • When the requested count is larger than the number of available messages, are only the available messages returned?
  • Verify that polling without specifying a partition behaves as expected.
  • Verify the behavior of each polling strategy: offset, first, last, next, and timestamp.
  • For offset, what happens when the offset is valid, too large, or points exactly to the end?
  • For timestamp, what happens when the timestamp is before the first message, between messages, and after the last message?
  • What happens when auto_commit = false and poll_messages is called repeatedly with next?
  • What happens when auto_commit = true and poll_messages is called repeatedly with next?
  • Are offsets tracked independently for different consumer IDs?
  • Are offsets tracked independently for consumer and consumer_group?
  • When multiple consumers in the same consumer group poll, is the behavior what we expect?
  • When poll_messages is called multiple times after multiple send_messages calls, is the order of the returned messages preserved?
  • Verify that the returned offsets are monotonic and correct across multiple polls.
  • Verify that the returned payloads are preserved exactly for a variety of byte payloads. (null bytes, utf-8 strings, emojis, arbitrary bytes, etc. - consider doing property-based testing)
  • Verify that custom message IDs, if present, are returned correctly for multiple messages, not just one.
  • What happens when you poll from an empty partition, and then poll again after new messages are sent?
  • What happens when you poll after the stream or topic has been deleted?

[slbotbm]

I wrote the above in multiple sittings, so please tell me if I wrote overlapping things.

[hubcio]

One comment from my side: remember that Rust SDK (which you are calling) is also perform these validations and returns respective errors. Make sure that you are not double checking this on rust wrapper side or cpp side.

[seokjin0414]

@slbotbm

• new_message changed to impl Message method, payload validation removed
• get_streams tests moved to tests/stream/
• Added 30 new e2e tests (4 get_streams, 11 send_messages, 15 poll_messages) — consumer group tests excluded pending feat(cpp): Add functions related to consumer groups #2988

[slbotbm]

@seokjin0414 Thanks for the update. Could you please tell me which tests you have not added from the above list? Just copy pasting the questions I had written will do.

[slbotbm]

@hubcio about your comment: would it make sense to add some of these test to the rust sdk instead? I ask because some of the e2e tests can become unit tests on the rust side. I ask because we would also not have to duplicate these tests for the python sdk

[seokjin0414]

@slbotbm
Tests not added:

send_messages:

• Is there a limit on how many messages can be sent in total, or in a single batch? — need to investigate the actual server limit first
• Property-based testing — used manual cases (null bytes, UTF-8, binary) instead

poll_messages:

• Invalid or non-existent partition ID — overlooked, will add
• count = 0 — overlooked, will add
• Limit on how many messages can be requested in a single poll — need to investigate the actual server limit first
• Polling without specifying a partition — overlooked, will add
• Timestamp strategy (before/between/after) — overlooked, will add
• Monotonic offset verification across multiple polls — overlooked, will add
• Consumer vs consumer_group offset independence — pending feat(cpp): Add functions related to consumer groups #2988
• Multiple consumers in same consumer group — pending feat(cpp): Add functions related to consumer groups #2988
• Property-based testing — used manual cases instead

I can add the overlooked ones now.

[slbotbm]

@seokjin0414 please go ahead and add the ones you overlooked. I'll review then

[seokjin0414]

@slbotbm
Done. Added the overlooked tests:

Your question	Test added
Invalid or non-existent partition ID	PollMessagesWithInvalidPartitionId — partition 9999 on topic with 1 partition → throws
count = 0	PollMessagesWithCountZero — returns success with 0 messages (server returns empty, not error)
Polling without specifying a partition	PollMessagesWithoutSpecifyingPartition — UINT32_MAX defaults to partition 0
Timestamp strategy (before/between/after)	PollMessagesTimestampStrategy — 2 batches with 100ms gap, verifies >= semantics
Monotonic offset verification	PollMessagesMonotonicOffsets — 20 messages, 4 polls of 5, verifies continuous offsets 0-19
Batch size limit	SendMessagesLargeBatch — 1000 messages in single batch, all retrievable
Poll request limit	PollMessagesLargeCount — count=UINT32_MAX, returns only available 10 messages

Still not added:

• Consumer group tests — pending feat(cpp): Add functions related to consumer groups #2988
• Property-based testing — covered with manual cases (null bytes, UTF-8, binary)

[slbotbm]

@seokjin0414 Thanks for the update! I'll review this by friday/saturday. In the meantime, could you please:

1. rebase this pr and fix conflicts, since feat(cpp): Add functions related to consumer groups #2988 was recently merged
2. Fix the failing CI
3. write unit tests for Message.new_message in tests/messages/unit_tests.cpp? You could refer to https://github.com/apache/iggy/blob/master/foreign/cpp/tests/identifier/unit_tests.cpp to think about what kind of unit tests to write.

[seokjin0414]

@slbotbm

• Renamed 10 tests with Throws suffix
• Added payload/offset verification:
  • PollMessagesFirstStrategy — payload + offset for all 3 messages
  • PollMessagesLastStrategy — payload for messages at indices 7-9
  • PollMessagesNextStrategyNoAutoCommit — payload + offset for both polls
  • PollMessagesNextStrategyAutoCommit — payload for polled1 (0-4) and polled2 (5-9)
  • PollMessagesMultipleSendsThenPollOrder — offset for all 10 messages
  • PollMessagesWithoutSpecifyingPartition — payload for 5 messages
  • PollMessagesTimestampStrategy — payload for batch2 messages
• Deleted PollMessagesLargeCount
• Removed accidental #[allow] from identifier.rs

[slbotbm]

@seokjin0414 I requested some small changes.
Other than this, I noticed

• you were not testing cases for send_messages and poll_messages when invalid topic ID was passed to them.
• No test for invalid consumer identifier for poll_messages

Also, since the consumer group PR was merged, could you add a test for creating a consumer group and polling messages after joining that?

[seokjin0414]

Thank you for the review!

• Added field assertions to GetStreamsReturnsStreamAfterCreation (created_at, size_bytes, messages_count, topics_count) and field comparison to GetStreamsConsistentWithGetStream (created_at, size_bytes). Note: id is excluded from both due to a known cxx Vec<SharedStruct> issue where id returns 0.
• Added SendMessagesWithInvalidTopicIdThrows, PollMessagesWithInvalidTopicIdThrows, PollMessagesWithInvalidConsumerIdThrows
• Added join_consumer_group and leave_consumer_group FFI functions
• Added ConsumerGroupCreateJoinAndPollMessages test covering the full flow: create → join → verify member count → send → poll with consumer_group kind → verify payloads → leave → verify member count drops to 0

[slbotbm]

@seokjin0414 Thanks for the update!

Could you add the following comment in tests/client/low_level_e2e.cpp please:

TODO(slbotbm): Add tests for join_consumer_group() and leave_consumer_group()

Also, you commented:

Note: id is excluded from both due to a known cxx Vec issue where id returns 0.

Do you have an issue you could link to for me to see?

Otherwise lgtm!

[seokjin0414]

@slbotbm
I investigated further it's not a cxx bug
Iggy uses a slab allocator for stream IDs (metadata/reader.rs:202, vacant_key()) so the first stream gets id=0
The original ASSERT_GT(s.id, 0u) was simply wrong for that reason
I've added EXPECT_EQ(list_id, single.id) to GetStreamsConsistentWithGetStream instead which compares the two API results without assuming a specific value