Google Git
Sign in
apache/iggy/refs/heads/harness-single-node-default/./foreign/cpp/tests/e2e/message.cpp
blob: e20ef6008fee3ceaa379b038c37c4b5f218eb0f8 [file]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
#include <cstdint>
#include <string>
#include <thread>
#include <gtest/gtest.h>
#include "lib.rs.h"
#include "tests/e2e/test_helpers.hpp"
class LowLevelE2E_Message : public E2ETestFixture {};
TEST_F(LowLevelE2E_Message, SendAndPollMessagesRoundTrip) {
RecordProperty("description", "Sends 10 messages and polls them back, verifying count, offsets, and payloads.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 10; i++) {
auto msg = iggy::ffi::make_message(to_payload("test message " + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
iggy::ffi::SendMessagesResponse sent;
ASSERT_NO_THROW(sent = client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0),
"partition_id", partition_id_bytes(0), std::move(messages)));
ASSERT_TRUE(sent.confirmations.empty())
<< "The legacy server reports no offsets, so the confirmation list must stay empty, got "
<< sent.confirmations.size();
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 100, false);
ASSERT_EQ(polled.partition_id, 0u) << "Polled partition_id mismatches the partition we sent to";
ASSERT_EQ(polled.count, 10u);
ASSERT_EQ(polled.messages.size(), 10u);
for (std::uint32_t i = 0; i < 10; i++) {
ASSERT_EQ(polled.messages[i].offset, static_cast<std::uint64_t>(i));
std::string expected = "test message " + std::to_string(i);
std::string actual(polled.messages[i].payload.begin(), polled.messages[i].payload.end());
ASSERT_EQ(actual, expected) << "Payload mismatch at offset " << i;
}
}
TEST_F(LowLevelE2E_Message, PollMessagesVerifyMessageIds) {
RecordProperty("description", "Verifies that polled message IDs match the sent IDs.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
auto msg = iggy::ffi::make_message(to_payload("id-test-message"), rust::Vec<iggy::ffi::HeaderEntry>());
msg.id_lo = 42;
msg.id_hi = 0;
messages.push_back(std::move(msg));
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 100, false);
ASSERT_EQ(polled.messages.size(), 1u);
ASSERT_EQ(polled.messages[0].id_lo, 42u);
ASSERT_EQ(polled.messages[0].id_hi, 0u);
}
TEST_F(LowLevelE2E_Message, PollMessagesFromEmptyPartition) {
RecordProperty("description", "Verifies polling from an empty partition returns zero messages.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 100, false);
ASSERT_EQ(polled.count, 0u);
ASSERT_EQ(polled.messages.size(), 0u);
}
TEST_F(LowLevelE2E_Message, SendMessagesBeforeLoginThrows) {
RecordProperty("description", "Verifies send_messages throws when not authenticated.");
iggy::ffi::Client *client = GetLoggedOutClient();
ASSERT_NO_THROW(client->connect());
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
auto msg = iggy::ffi::make_message(to_payload("should-fail"), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
ASSERT_THROW(client->send_messages(make_numeric_identifier(1), make_numeric_identifier(1), "partition_id",
partition_id_bytes(0), std::move(messages)),
std::exception);
ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
ASSERT_NO_THROW(client->disconnect());
rust::Vec<iggy::ffi::IggyMessageToSend> disconnected_messages;
auto disconnected_msg =
iggy::ffi::make_message(to_payload("should-still-fail"), rust::Vec<iggy::ffi::HeaderEntry>());
disconnected_messages.push_back(std::move(disconnected_msg));
ASSERT_THROW(client->send_messages(make_numeric_identifier(1), make_numeric_identifier(1), "partition_id",
partition_id_bytes(0), std::move(disconnected_messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, SendMessagesWithInvalidStreamId) {
RecordProperty("description", "Throws when sending messages with an invalid stream identifier.");
iggy::ffi::Client *client = GetLoggedInClient();
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
auto msg = iggy::ffi::make_message(to_payload("test"), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
iggy::ffi::Identifier invalid_id;
invalid_id.kind = "invalid";
invalid_id.length = 0;
ASSERT_THROW(client->send_messages(invalid_id, make_numeric_identifier(1), "partition_id", partition_id_bytes(0),
std::move(messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, SendMessagesToNonExistentStream) {
RecordProperty("description", "Throws when sending messages to a non-existent stream.");
iggy::ffi::Client *client = GetLoggedInClient();
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
auto msg = iggy::ffi::make_message(to_payload("test"), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
ASSERT_THROW(client->send_messages(make_string_identifier("nonexistent-stream-12345"), make_numeric_identifier(0),
"partition_id", partition_id_bytes(0), std::move(messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, SendMessagesWithInvalidPartitioningKind) {
RecordProperty("description", "Throws when sending messages with an invalid partitioning kind.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
auto msg = iggy::ffi::make_message(to_payload("test"), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "invalid_kind",
partition_id_bytes(0), std::move(messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, SendMessagesWithInvalidPartitioningValue) {
RecordProperty("description", "Throws when sending messages with insufficient partitioning value bytes.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
auto msg = iggy::ffi::make_message(to_payload("test"), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
rust::Vec<std::uint8_t> short_bytes;
short_bytes.push_back(0x00);
short_bytes.push_back(0x01);
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
std::move(short_bytes), std::move(messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, SendMessagesToSpecificPartitionVerified) {
RecordProperty("description",
"Verifies messages sent to a specific partition are only retrievable from that partition.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 3, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 5; i++) {
auto msg = iggy::ffi::make_message(to_payload("partition-test-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled_part0 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0,
"consumer", make_numeric_identifier(1), "offset", 0, 100, false);
ASSERT_EQ(polled_part0.partition_id, 0u);
ASSERT_EQ(polled_part0.count, 5u);
auto polled_part1 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 1,
"consumer", make_numeric_identifier(1), "offset", 0, 100, false);
ASSERT_EQ(polled_part1.partition_id, 1u);
ASSERT_EQ(polled_part1.count, 0u);
}
TEST_F(LowLevelE2E_Message, SendEmptyMessageVectorThrows) {
RecordProperty("description", "Throws when sending an empty message vector.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> empty_messages;
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(empty_messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, SendMessageWithEmptyPayloadThrows) {
RecordProperty("description", "Throws when sending a message with an empty payload.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
rust::Vec<std::uint8_t> empty_payload;
auto msg = iggy::ffi::make_message(std::move(empty_payload), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, SendMessageWithOversizedPayloadThrows) {
RecordProperty("description", "Throws when sending a message exceeding maximum payload size.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
// Build a payload one byte over the SDK's max payload size (64 MB).
constexpr std::uint32_t kOversizedPayloadBytes = 64'000'001u;
rust::Vec<std::uint8_t> oversized_payload;
oversized_payload.reserve(kOversizedPayloadBytes);
for (std::uint32_t i = 0; i < kOversizedPayloadBytes; i++) {
oversized_payload.push_back(0x41);
}
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
auto msg = iggy::ffi::make_message(std::move(oversized_payload), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, SendMessagesPreservesOrder) {
RecordProperty("description", "Verifies messages are stored and retrieved in the order they were sent.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 50; i++) {
auto msg =
iggy::ffi::make_message(to_payload("order-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 100, false);
ASSERT_EQ(polled.count, 50u);
for (std::uint32_t i = 0; i < 50; i++) {
ASSERT_EQ(polled.messages[i].offset, static_cast<std::uint64_t>(i));
std::string expected = "order-" + std::to_string(i);
std::string actual(polled.messages[i].payload.begin(), polled.messages[i].payload.end());
EXPECT_EQ(actual, expected) << "Payload mismatch at offset " << i;
}
}
TEST_F(LowLevelE2E_Message, SendMessagesWithDuplicateIds) {
RecordProperty("description", "Verifies sending multiple messages with the same ID succeeds.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 3; i++) {
auto msg =
iggy::ffi::make_message(to_payload("dup-id-msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
msg.id_lo = 99;
msg.id_hi = 0;
messages.push_back(std::move(msg));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0),
"partition_id", partition_id_bytes(0), std::move(messages)));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 100, false);
ASSERT_EQ(polled.count, 3u);
for (std::size_t i = 0; i < polled.messages.size(); i++) {
EXPECT_EQ(polled.messages[i].id_lo, 99u);
}
}
TEST_F(LowLevelE2E_Message, SendMessagesWithVariousPayloads) {
RecordProperty("description",
"Verifies various payload types including null bytes, UTF-8, and binary data are preserved.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<std::uint8_t> payload_null;
payload_null.push_back(0x00);
payload_null.push_back(0x01);
payload_null.push_back(0x00);
payload_null.push_back(0xFF);
rust::Vec<std::uint8_t> payload_binary;
payload_binary.push_back(0xDE);
payload_binary.push_back(0xAD);
payload_binary.push_back(0xBE);
payload_binary.push_back(0xEF);
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
auto msg0 = iggy::ffi::make_message(to_payload("simple ascii"), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg0));
auto msg1 = iggy::ffi::make_message(std::move(payload_null), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg1));
auto msg2 = iggy::ffi::make_message(to_payload("héllo wörld"), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg2));
auto msg3 = iggy::ffi::make_message(std::move(payload_binary), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg3));
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 100, false);
ASSERT_EQ(polled.count, 4u);
std::string ascii_actual(polled.messages[0].payload.begin(), polled.messages[0].payload.end());
EXPECT_EQ(ascii_actual, "simple ascii");
ASSERT_EQ(polled.messages[1].payload.size(), 4u);
EXPECT_EQ(polled.messages[1].payload[0], 0x00);
EXPECT_EQ(polled.messages[1].payload[1], 0x01);
EXPECT_EQ(polled.messages[1].payload[2], 0x00);
EXPECT_EQ(polled.messages[1].payload[3], 0xFF);
std::string utf8_actual(polled.messages[2].payload.begin(), polled.messages[2].payload.end());
EXPECT_EQ(utf8_actual, "héllo wörld");
ASSERT_EQ(polled.messages[3].payload.size(), 4u);
EXPECT_EQ(polled.messages[3].payload[0], 0xDE);
EXPECT_EQ(polled.messages[3].payload[1], 0xAD);
EXPECT_EQ(polled.messages[3].payload[2], 0xBE);
EXPECT_EQ(polled.messages[3].payload[3], 0xEF);
}
TEST_F(LowLevelE2E_Message, SendAndPollMessageWithTypedHeadersRoundTrip) {
RecordProperty(
"description",
"Sends one message per typed header kind and verifies payload, IDs, header contents, and encoded size.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
struct ExpectedHeaderMessage {
const char *key;
iggy::ffi::HeaderKind value_kind;
const rust::Vec<std::uint8_t> *value;
};
const rust::Vec<std::uint8_t> raw_value{0xDE, 0xAD, 0xBE, 0xEF}; // raw bytes 0xDEADBEEF
const rust::Vec<std::uint8_t> string_value = to_payload("hello"); // UTF-8 string "hello"
const rust::Vec<std::uint8_t> bool_value{0x01}; // bool true
const rust::Vec<std::uint8_t> int8_value{0xFB}; // int8 -5
const rust::Vec<std::uint8_t> int16_value{0x2E, 0xFB}; // int16 -1234
const rust::Vec<std::uint8_t> int32_value{0xEB, 0x32, 0xA4, 0xF8}; // int32 -123456789
const rust::Vec<std::uint8_t> int64_value{0x79, 0x29, 0xED, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF}; // int64 -1234567
const rust::Vec<std::uint8_t> int128_value{
0x00, 0xFF, 0xEE, 0xDD, 0xCC, 0xBB, 0xAA, 0x99,
0x88, 0x77, 0x66, 0x55, 0x44, 0x33, 0x22, 0x11}; // int128 0x112233445566778899AABBCCDDEEFF00
const rust::Vec<std::uint8_t> uint8_value{0xFA}; // uint8 250
const rust::Vec<std::uint8_t> uint16_value{0xD2, 0x04}; // uint16 1234
const rust::Vec<std::uint8_t> uint32_value{0x78, 0x56, 0x34, 0x12}; // uint32 0x12345678
const rust::Vec<std::uint8_t> uint64_value{0x88, 0x77, 0x66, 0x55,
0x44, 0x33, 0x22, 0x11}; // uint64 0x1122334455667788
const rust::Vec<std::uint8_t> uint128_value{
0x10, 0x32, 0x54, 0x76, 0x98, 0xBA, 0xDC, 0xFE,
0xEF, 0xCD, 0xAB, 0x89, 0x67, 0x45, 0x23, 0x01}; // uint128 0x0123456789ABCDEFFEDCBA9876543210
const rust::Vec<std::uint8_t> float32_value{0x00, 0x00, 0x80, 0x3F}; // float32 1.0
const rust::Vec<std::uint8_t> float64_value{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xF0, 0x3F}; // float64 1.0
const ExpectedHeaderMessage expected_messages[] = {
{"raw", iggy::ffi::HeaderKind::Raw, &raw_value},
{"string", iggy::ffi::HeaderKind::String, &string_value},
{"bool", iggy::ffi::HeaderKind::Bool, &bool_value},
{"int8", iggy::ffi::HeaderKind::Int8, &int8_value},
{"int16", iggy::ffi::HeaderKind::Int16, &int16_value},
{"int32", iggy::ffi::HeaderKind::Int32, &int32_value},
{"int64", iggy::ffi::HeaderKind::Int64, &int64_value},
{"int128", iggy::ffi::HeaderKind::Int128, &int128_value},
{"uint8", iggy::ffi::HeaderKind::Uint8, &uint8_value},
{"uint16", iggy::ffi::HeaderKind::Uint16, &uint16_value},
{"uint32", iggy::ffi::HeaderKind::Uint32, &uint32_value},
{"uint64", iggy::ffi::HeaderKind::Uint64, &uint64_value},
{"uint128", iggy::ffi::HeaderKind::Uint128, &uint128_value},
{"float32", iggy::ffi::HeaderKind::Float32, &float32_value},
{"float64", iggy::ffi::HeaderKind::Float64, &float64_value},
};
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (const auto &expected : expected_messages) {
rust::Vec<iggy::ffi::HeaderEntry> headers;
headers.push_back(make_header_entry(make_header_field(iggy::ffi::HeaderKind::String, to_payload(expected.key)),
make_header_field(expected.value_kind, *expected.value)));
messages.push_back(
iggy::ffi::make_message(to_payload(std::string("payload-") + expected.key), std::move(headers)));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0),
"partition_id", partition_id_bytes(0), std::move(messages)));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 100, false);
constexpr std::size_t expected_message_count = sizeof(expected_messages) / sizeof(expected_messages[0]);
ASSERT_EQ(polled.count, expected_message_count);
ASSERT_EQ(polled.messages.size(), expected_message_count);
for (std::size_t i = 0; i < expected_message_count; ++i) {
const auto &expected = expected_messages[i];
const auto &polled_message = polled.messages[i];
EXPECT_EQ(std::string(polled_message.payload.begin(), polled_message.payload.end()),
std::string("payload-") + expected.key);
ASSERT_EQ(polled_message.user_headers.size(), 1u);
EXPECT_EQ(polled_message.user_headers_length,
static_cast<std::uint32_t>(10 + std::string(expected.key).size() + expected.value->size()));
EXPECT_TRUE(has_header(polled_message.user_headers, static_cast<std::uint8_t>(iggy::ffi::HeaderKind::String),
to_payload(expected.key), static_cast<std::uint8_t>(expected.value_kind),
*expected.value));
}
}
TEST_F(LowLevelE2E_Message, SendMessageWithDuplicateTypedHeaderKeysThrows) {
RecordProperty("description", "Throws when a single message contains duplicate typed header keys.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::HeaderEntry> headers;
headers.push_back(make_header_entry(make_header_field(iggy::ffi::HeaderKind::String, to_payload("dup-key")),
make_header_field(iggy::ffi::HeaderKind::String, to_payload("first"))));
headers.push_back(make_header_entry(make_header_field(iggy::ffi::HeaderKind::String, to_payload("dup-key")),
make_header_field(iggy::ffi::HeaderKind::String, to_payload("second"))));
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
messages.push_back(iggy::ffi::make_message(to_payload("payload"), std::move(headers)));
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, SendMessageWithWrongFixedWidthHeaderBytesThrows) {
RecordProperty("description", "Throws when a typed header uses a fixed-width kind with the wrong byte count.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
struct FixedWidthHeaderCase {
const char *key;
iggy::ffi::HeaderKind kind;
};
const FixedWidthHeaderCase fixed_width_header_cases[] = {
{"broken-bool", iggy::ffi::HeaderKind::Bool}, {"broken-int8", iggy::ffi::HeaderKind::Int8},
{"broken-int16", iggy::ffi::HeaderKind::Int16}, {"broken-int32", iggy::ffi::HeaderKind::Int32},
{"broken-int64", iggy::ffi::HeaderKind::Int64}, {"broken-int128", iggy::ffi::HeaderKind::Int128},
{"broken-uint8", iggy::ffi::HeaderKind::Uint8}, {"broken-uint16", iggy::ffi::HeaderKind::Uint16},
{"broken-uint32", iggy::ffi::HeaderKind::Uint32}, {"broken-uint64", iggy::ffi::HeaderKind::Uint64},
{"broken-uint128", iggy::ffi::HeaderKind::Uint128}, {"broken-float32", iggy::ffi::HeaderKind::Float32},
{"broken-float64", iggy::ffi::HeaderKind::Float64},
};
for (const auto &test_case : fixed_width_header_cases) {
SCOPED_TRACE(test_case.key);
rust::Vec<iggy::ffi::HeaderEntry> headers;
rust::Vec<std::uint8_t> broken_width_value;
broken_width_value.push_back(0x01);
broken_width_value.push_back(0x02);
broken_width_value.push_back(0x03);
headers.push_back(make_header_entry(make_header_field(iggy::ffi::HeaderKind::String, to_payload(test_case.key)),
make_header_field(test_case.kind, std::move(broken_width_value))));
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
messages.push_back(iggy::ffi::make_message(to_payload("payload"), std::move(headers)));
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0),
"partition_id", partition_id_bytes(0), std::move(messages)),
std::exception);
}
}
TEST_F(LowLevelE2E_Message, SendMessageWithInvalidTypedHeaderKindThrows) {
RecordProperty("description", "Throws when a typed header uses an unsupported header kind code.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
iggy::ffi::HeaderField invalid_key;
invalid_key.kind = 255;
invalid_key.value = to_payload("bad-kind");
rust::Vec<iggy::ffi::HeaderEntry> headers;
headers.push_back(make_header_entry(std::move(invalid_key),
make_header_field(iggy::ffi::HeaderKind::String, to_payload("value"))));
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
messages.push_back(iggy::ffi::make_message(to_payload("payload"), std::move(headers)));
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, SendMessageWithInvalidTypedHeaderSizesThrows) {
RecordProperty("description", "Throws when typed header key or value sizes violate Rust header size constraints.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::HeaderEntry> empty_key_headers;
empty_key_headers.push_back(
make_header_entry(make_header_field(iggy::ffi::HeaderKind::String, rust::Vec<std::uint8_t>()),
make_header_field(iggy::ffi::HeaderKind::String, to_payload("value"))));
rust::Vec<iggy::ffi::IggyMessageToSend> empty_key_messages;
empty_key_messages.push_back(iggy::ffi::make_message(to_payload("payload"), std::move(empty_key_headers)));
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(empty_key_messages)),
std::exception);
rust::Vec<iggy::ffi::HeaderEntry> empty_raw_value_headers;
empty_raw_value_headers.push_back(
make_header_entry(make_header_field(iggy::ffi::HeaderKind::String, to_payload("key")),
make_header_field(iggy::ffi::HeaderKind::Raw, rust::Vec<std::uint8_t>())));
rust::Vec<iggy::ffi::IggyMessageToSend> empty_raw_value_messages;
empty_raw_value_messages.push_back(
iggy::ffi::make_message(to_payload("payload"), std::move(empty_raw_value_headers)));
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(empty_raw_value_messages)),
std::exception);
rust::Vec<std::uint8_t> oversized_value_bytes;
for (std::size_t index = 0; index < 256; ++index) {
oversized_value_bytes.push_back(static_cast<std::uint8_t>(index));
}
rust::Vec<iggy::ffi::HeaderEntry> oversized_value_headers;
oversized_value_headers.push_back(
make_header_entry(make_header_field(iggy::ffi::HeaderKind::String, to_payload("key")),
make_header_field(iggy::ffi::HeaderKind::Raw, std::move(oversized_value_bytes))));
rust::Vec<iggy::ffi::IggyMessageToSend> oversized_value_messages;
oversized_value_messages.push_back(
iggy::ffi::make_message(to_payload("payload"), std::move(oversized_value_headers)));
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(oversized_value_messages)),
std::exception);
rust::Vec<std::uint8_t> oversized_string_bytes;
for (std::size_t index = 0; index < 256; ++index) {
oversized_string_bytes.push_back('a');
}
rust::Vec<iggy::ffi::HeaderEntry> oversized_string_headers;
oversized_string_headers.push_back(
make_header_entry(make_header_field(iggy::ffi::HeaderKind::String, to_payload("key")),
make_header_field(iggy::ffi::HeaderKind::String, std::move(oversized_string_bytes))));
rust::Vec<iggy::ffi::IggyMessageToSend> oversized_string_messages;
oversized_string_messages.push_back(
iggy::ffi::make_message(to_payload("payload"), std::move(oversized_string_headers)));
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(oversized_string_messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, SendMessageAtUserHeadersSizeBoundary) {
RecordProperty("description",
"Throws when encoded user headers exceed 100_000 bytes and succeeds at exactly 100_000 bytes.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
constexpr std::uint32_t kMaxUserHeadersBytes = 100'000u;
constexpr std::uint32_t kFullHeaderEncodedBytes = 267u; // 10 bytes framing + 2-byte key + 255-byte value
constexpr std::uint32_t kFullHeaderCount = 374u;
constexpr std::uint32_t kTailExactValueBytes = 130u;
constexpr std::uint32_t kTailOversizedValueBytes = 131u;
rust::Vec<iggy::ffi::HeaderEntry> oversized_headers;
for (std::uint32_t index = 0; index < kFullHeaderCount; ++index) {
rust::Vec<std::uint8_t> key_bytes;
key_bytes.push_back(static_cast<std::uint8_t>(index & 0xFF));
key_bytes.push_back(static_cast<std::uint8_t>((index >> 8) & 0xFF));
rust::Vec<std::uint8_t> value_bytes;
for (std::uint32_t value_index = 0; value_index < 255u; ++value_index) {
value_bytes.push_back(static_cast<std::uint8_t>(value_index));
}
oversized_headers.push_back(
make_header_entry(make_header_field(iggy::ffi::HeaderKind::Raw, std::move(key_bytes)),
make_header_field(iggy::ffi::HeaderKind::Raw, std::move(value_bytes))));
}
rust::Vec<std::uint8_t> oversized_tail_key_bytes;
oversized_tail_key_bytes.push_back(static_cast<std::uint8_t>(kFullHeaderCount & 0xFF));
oversized_tail_key_bytes.push_back(static_cast<std::uint8_t>((kFullHeaderCount >> 8) & 0xFF));
rust::Vec<std::uint8_t> oversized_tail_value_bytes;
for (std::uint32_t value_index = 0; value_index < kTailOversizedValueBytes; ++value_index) {
oversized_tail_value_bytes.push_back(static_cast<std::uint8_t>(value_index));
}
oversized_headers.push_back(
make_header_entry(make_header_field(iggy::ffi::HeaderKind::Raw, std::move(oversized_tail_key_bytes)),
make_header_field(iggy::ffi::HeaderKind::Raw, std::move(oversized_tail_value_bytes))));
rust::Vec<iggy::ffi::IggyMessageToSend> oversized_messages;
oversized_messages.push_back(
iggy::ffi::make_message(to_payload("oversized-user-headers"), std::move(oversized_headers)));
ASSERT_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(oversized_messages)),
std::exception);
rust::Vec<iggy::ffi::HeaderEntry> exact_headers;
for (std::uint32_t index = 0; index < kFullHeaderCount; ++index) {
rust::Vec<std::uint8_t> key_bytes;
key_bytes.push_back(static_cast<std::uint8_t>(index & 0xFF));
key_bytes.push_back(static_cast<std::uint8_t>((index >> 8) & 0xFF));
rust::Vec<std::uint8_t> value_bytes;
for (std::uint32_t value_index = 0; value_index < 255u; ++value_index) {
value_bytes.push_back(static_cast<std::uint8_t>(value_index));
}
exact_headers.push_back(
make_header_entry(make_header_field(iggy::ffi::HeaderKind::Raw, std::move(key_bytes)),
make_header_field(iggy::ffi::HeaderKind::Raw, std::move(value_bytes))));
}
rust::Vec<std::uint8_t> exact_tail_key_bytes;
exact_tail_key_bytes.push_back(static_cast<std::uint8_t>(kFullHeaderCount & 0xFF));
exact_tail_key_bytes.push_back(static_cast<std::uint8_t>((kFullHeaderCount >> 8) & 0xFF));
rust::Vec<std::uint8_t> exact_tail_value_bytes;
for (std::uint32_t value_index = 0; value_index < kTailExactValueBytes; ++value_index) {
exact_tail_value_bytes.push_back(static_cast<std::uint8_t>(value_index));
}
exact_headers.push_back(
make_header_entry(make_header_field(iggy::ffi::HeaderKind::Raw, std::move(exact_tail_key_bytes)),
make_header_field(iggy::ffi::HeaderKind::Raw, std::move(exact_tail_value_bytes))));
rust::Vec<iggy::ffi::IggyMessageToSend> exact_messages;
exact_messages.push_back(iggy::ffi::make_message(to_payload("exact-user-headers"), std::move(exact_headers)));
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0),
"partition_id", partition_id_bytes(0), std::move(exact_messages)));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 10, false);
ASSERT_EQ(polled.count, 1u);
ASSERT_EQ(polled.messages.size(), 1u);
EXPECT_EQ(std::string(polled.messages[0].payload.begin(), polled.messages[0].payload.end()), "exact-user-headers");
EXPECT_EQ(polled.messages[0].user_headers_length, kMaxUserHeadersBytes);
EXPECT_EQ(polled.messages[0].user_headers.size(), kFullHeaderCount + 1u);
EXPECT_EQ(kFullHeaderCount * kFullHeaderEncodedBytes + 10u + 2u + kTailExactValueBytes, kMaxUserHeadersBytes);
}
TEST_F(LowLevelE2E_Message, PollMessagesBeforeLoginThrows) {
RecordProperty("description", "Throws when polling messages before authentication.");
iggy::ffi::Client *client = GetLoggedOutClient();
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->poll_messages(make_numeric_identifier(1), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 10, false),
std::exception);
ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
ASSERT_NO_THROW(client->disconnect());
ASSERT_THROW(client->poll_messages(make_numeric_identifier(1), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 10, false),
std::exception);
}
TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidStreamIdThrows) {
RecordProperty("description", "Throws when polling messages with an invalid stream identifier.");
iggy::ffi::Client *client = GetLoggedInClient();
iggy::ffi::Identifier invalid_id;
invalid_id.kind = "invalid";
invalid_id.length = 0;
ASSERT_THROW(client->poll_messages(invalid_id, make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 10, false),
std::exception);
}
TEST_F(LowLevelE2E_Message, PollMessagesFromNonExistentStreamThrows) {
RecordProperty("description", "Throws when polling messages from a non-existent stream.");
iggy::ffi::Client *client = GetLoggedInClient();
ASSERT_THROW(client->poll_messages(make_string_identifier("nonexistent-stream-poll"), make_numeric_identifier(0), 0,
"consumer", make_numeric_identifier(1), "offset", 0, 10, false),
std::exception);
}
TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidConsumerKindThrows) {
RecordProperty("description", "Throws when polling messages with an invalid consumer kind.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "invalid",
make_numeric_identifier(1), "offset", 0, 10, false),
std::exception);
}
TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidStrategyKindThrows) {
RecordProperty("description", "Throws when polling messages with an invalid polling strategy kind.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "invalid", 0, 10, false),
std::exception);
}
TEST_F(LowLevelE2E_Message, PollMessagesCountLessThanAvailable) {
RecordProperty("description", "Returns only the requested count when fewer messages are requested than available.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 10; i++) {
auto msg = iggy::ffi::make_message(to_payload("msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 5, false);
ASSERT_EQ(polled.count, 5u);
ASSERT_EQ(polled.messages.size(), 5u);
}
TEST_F(LowLevelE2E_Message, PollMessagesWithLargeOffset) {
RecordProperty("description", "Returns zero messages when polling with an offset beyond available messages.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 5; i++) {
auto msg = iggy::ffi::make_message(to_payload("msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 999999, 100, false);
ASSERT_EQ(polled.count, 0u);
ASSERT_EQ(polled.messages.size(), 0u);
}
TEST_F(LowLevelE2E_Message, PollMessagesFirstStrategy) {
RecordProperty("description", "Verifies first polling strategy returns messages from the beginning.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 10; i++) {
auto msg = iggy::ffi::make_message(to_payload("msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "first", 0, 3, false);
ASSERT_EQ(polled.count, 3u);
ASSERT_EQ(polled.messages.size(), 3u);
EXPECT_EQ(polled.messages[0].offset, 0u);
for (std::uint32_t i = 0; i < 3; i++) {
EXPECT_EQ(polled.messages[i].offset, static_cast<std::uint64_t>(i));
std::string expected = "msg-" + std::to_string(i);
std::string actual(polled.messages[i].payload.begin(), polled.messages[i].payload.end());
EXPECT_EQ(actual, expected) << "Payload mismatch at offset " << i;
}
}
TEST_F(LowLevelE2E_Message, PollMessagesLastStrategy) {
RecordProperty("description", "Verifies last polling strategy returns messages from the end.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 10; i++) {
auto msg = iggy::ffi::make_message(to_payload("msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "last", 0, 3, false);
ASSERT_EQ(polled.count, 3u);
ASSERT_EQ(polled.messages.size(), 3u);
EXPECT_EQ(polled.messages[0].offset, 7u);
EXPECT_EQ(polled.messages[2].offset, 9u);
for (std::uint32_t i = 0; i < 3; i++) {
std::string expected = "msg-" + std::to_string(7 + i);
std::string actual(polled.messages[i].payload.begin(), polled.messages[i].payload.end());
EXPECT_EQ(actual, expected) << "Payload mismatch at index " << i;
}
}
TEST_F(LowLevelE2E_Message, PollMessagesNextStrategyNoAutoCommit) {
RecordProperty("description",
"Verifies next strategy without auto-commit returns the same messages on repeated calls.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 5; i++) {
auto msg = iggy::ffi::make_message(to_payload("msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled1 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "next", 0, 100, false);
ASSERT_EQ(polled1.count, 5u);
auto polled2 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "next", 0, 100, false);
ASSERT_EQ(polled2.count, 5u);
for (std::uint32_t i = 0; i < 5; i++) {
EXPECT_EQ(polled1.messages[i].offset, static_cast<std::uint64_t>(i));
std::string expected = "msg-" + std::to_string(i);
std::string actual(polled1.messages[i].payload.begin(), polled1.messages[i].payload.end());
EXPECT_EQ(actual, expected) << "polled1 payload mismatch at index " << i;
}
for (std::uint32_t i = 0; i < 5; i++) {
EXPECT_EQ(polled2.messages[i].offset, static_cast<std::uint64_t>(i));
std::string expected = "msg-" + std::to_string(i);
std::string actual(polled2.messages[i].payload.begin(), polled2.messages[i].payload.end());
EXPECT_EQ(actual, expected) << "polled2 payload mismatch at index " << i;
}
}
TEST_F(LowLevelE2E_Message, PollMessagesNextStrategyAutoCommit) {
RecordProperty("description", "Verifies next strategy with auto-commit advances the offset on subsequent polls.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 10; i++) {
auto msg = iggy::ffi::make_message(to_payload("msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled1 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "next", 0, 5, true);
ASSERT_EQ(polled1.count, 5u);
EXPECT_EQ(polled1.messages[0].offset, 0u);
EXPECT_EQ(polled1.messages[4].offset, 4u);
auto polled2 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "next", 0, 5, true);
ASSERT_EQ(polled2.count, 5u);
EXPECT_EQ(polled2.messages[0].offset, 5u);
EXPECT_EQ(polled2.messages[4].offset, 9u);
for (std::uint32_t i = 0; i < 5; i++) {
std::string expected1 = "msg-" + std::to_string(i);
std::string actual1(polled1.messages[i].payload.begin(), polled1.messages[i].payload.end());
EXPECT_EQ(actual1, expected1) << "polled1 payload mismatch at index " << i;
}
for (std::uint32_t i = 0; i < 5; i++) {
std::string expected2 = "msg-" + std::to_string(5 + i);
std::string actual2(polled2.messages[i].payload.begin(), polled2.messages[i].payload.end());
EXPECT_EQ(actual2, expected2) << "polled2 payload mismatch at index " << i;
}
auto polled3 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "next", 0, 5, true);
ASSERT_EQ(polled3.count, 0u);
}
TEST_F(LowLevelE2E_Message, PollMessagesConsumerIdIndependence) {
RecordProperty("description", "Verifies different consumer IDs maintain independent offsets.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 5; i++) {
auto msg = iggy::ffi::make_message(to_payload("msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled_c1 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0,
"consumer", make_numeric_identifier(1), "next", 0, 3, true);
ASSERT_EQ(polled_c1.count, 3u);
auto polled_c2 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0,
"consumer", make_numeric_identifier(2), "next", 0, 5, true);
ASSERT_EQ(polled_c2.count, 5u);
auto polled_c1_again = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0,
"consumer", make_numeric_identifier(1), "next", 0, 5, true);
ASSERT_EQ(polled_c1_again.count, 2u);
}
TEST_F(LowLevelE2E_Message, PollMessagesMultipleSendsThenPollOrder) {
RecordProperty("description", "Verifies message ordering is preserved across multiple send batches.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> batch1;
for (std::uint32_t i = 0; i < 5; i++) {
auto msg =
iggy::ffi::make_message(to_payload("batch1-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
batch1.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(batch1));
rust::Vec<iggy::ffi::IggyMessageToSend> batch2;
for (std::uint32_t i = 0; i < 5; i++) {
auto msg =
iggy::ffi::make_message(to_payload("batch2-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
batch2.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(batch2));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 100, false);
ASSERT_EQ(polled.count, 10u);
for (std::uint32_t i = 0; i < 10; i++) {
EXPECT_EQ(polled.messages[i].offset, static_cast<std::uint64_t>(i)) << "Offset mismatch at index " << i;
}
for (std::uint32_t i = 0; i < 5; i++) {
std::string expected = "batch1-" + std::to_string(i);
std::string actual(polled.messages[i].payload.begin(), polled.messages[i].payload.end());
EXPECT_EQ(actual, expected) << "batch1 payload mismatch at index " << i;
}
for (std::uint32_t i = 0; i < 5; i++) {
std::string expected = "batch2-" + std::to_string(i);
std::string actual(polled.messages[5 + i].payload.begin(), polled.messages[5 + i].payload.end());
EXPECT_EQ(actual, expected) << "batch2 payload mismatch at index " << i;
}
}
TEST_F(LowLevelE2E_Message, PollMessagesMultipleCustomIds) {
RecordProperty("description", "Verifies multiple messages with distinct custom IDs are all preserved.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
const std::uint64_t id_values[] = {100, 200, 300, 400, 500};
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 5; i++) {
auto msg = iggy::ffi::make_message(to_payload("msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
msg.id_lo = id_values[i];
msg.id_hi = 0;
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 100, false);
ASSERT_EQ(polled.count, 5u);
for (std::uint32_t i = 0; i < 5; i++) {
EXPECT_EQ(polled.messages[i].id_lo, id_values[i]) << "ID mismatch at index " << i;
EXPECT_EQ(polled.messages[i].id_hi, 0u);
}
}
TEST_F(LowLevelE2E_Message, PollMessagesAfterStreamDeletedThrows) {
RecordProperty("description", "Throws when polling messages after the stream has been deleted.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
auto msg = iggy::ffi::make_message(to_payload("test"), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
std::uint32_t saved_stream_id = stream.id;
client->delete_stream(make_numeric_identifier(saved_stream_id));
ForgetTrackedStream(saved_stream_id);
ASSERT_THROW(client->poll_messages(make_numeric_identifier(saved_stream_id), make_numeric_identifier(0), 0,
"consumer", make_numeric_identifier(1), "offset", 0, 10, false),
std::exception);
}
TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidPartitionIdThrows) {
RecordProperty("description", "Throws when polling with a non-existent partition ID.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 9999, "consumer",
make_numeric_identifier(1), "offset", 0, 10, false),
std::exception);
}
TEST_F(LowLevelE2E_Message, PollMessagesWithCountZeroThrows) {
RecordProperty("description", "Throws when polling with count=0.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 0, false),
std::exception);
}
TEST_F(LowLevelE2E_Message, PollMessagesWithoutSpecifyingPartition) {
RecordProperty("description",
"Verifies polling with partition_id=u32::MAX defaults to partition 0 and returns messages.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 5; i++) {
auto msg = iggy::ffi::make_message(to_payload("msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), UINT32_MAX,
"consumer", make_numeric_identifier(1), "offset", 0, 100, false);
// The Rust side maps UINT32_MAX to None, so the server picks a partition. With a single
// partition topic that should always be partition 0.
ASSERT_EQ(polled.partition_id, 0u) << "u32::MAX sentinel did not map to None — partition_id sentinel regression?";
ASSERT_EQ(polled.count, 5u);
ASSERT_EQ(polled.messages.size(), 5u);
for (std::uint32_t i = 0; i < 5; i++) {
std::string expected = "msg-" + std::to_string(i);
std::string actual(polled.messages[i].payload.begin(), polled.messages[i].payload.end());
EXPECT_EQ(actual, expected) << "Payload mismatch at index " << i;
}
}
TEST_F(LowLevelE2E_Message, PollMessagesTimestampStrategy) {
RecordProperty("description",
"Verifies timestamp polling strategy returns messages with timestamp >= the specified value.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> batch1;
for (std::uint32_t i = 0; i < 5; i++) {
auto msg =
iggy::ffi::make_message(to_payload("batch1-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
batch1.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(batch1));
std::this_thread::sleep_for(std::chrono::milliseconds(100));
rust::Vec<iggy::ffi::IggyMessageToSend> batch2;
for (std::uint32_t i = 0; i < 5; i++) {
auto msg =
iggy::ffi::make_message(to_payload("batch2-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
batch2.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(batch2));
auto all = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 100, false);
ASSERT_EQ(all.count, 10u);
// IggyTimestamp::now() is microsecond-resolution and we slept 100ms between batches; a gap
// smaller than half that window means the test has degraded into a tautology on busy CI.
constexpr std::uint64_t kMinTimestampGapMicros = 50'000;
std::uint64_t batch1_timestamp = all.messages[0].timestamp;
std::uint64_t batch2_timestamp = all.messages[5].timestamp;
ASSERT_GT(batch2_timestamp, batch1_timestamp);
ASSERT_GE(batch2_timestamp - batch1_timestamp, kMinTimestampGapMicros)
<< "Timestamp gap collapsed (" << (batch2_timestamp - batch1_timestamp)
<< "us) — test no longer exercises timestamp filtering";
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(2), "timestamp", batch2_timestamp, 100, false);
ASSERT_GE(polled.count, 5u);
// The server contract is `timestamp >= polling_strategy_value`. If a batch1 message lands on
// exactly the same microsecond as batch2's first message, the count can legitimately exceed 5,
// so verify by prefix rather than indexing each message against `batch2-N`.
for (std::size_t i = 0; i < polled.messages.size(); i++) {
EXPECT_GE(polled.messages[i].timestamp, batch2_timestamp)
<< "Message at index " << i << " has earlier timestamp";
std::string actual(polled.messages[i].payload.begin(), polled.messages[i].payload.end());
EXPECT_TRUE(actual.rfind("batch1-", 0) == 0 || actual.rfind("batch2-", 0) == 0)
<< "Polled message at index " << i << " has unexpected payload: " << actual;
}
}
TEST_F(LowLevelE2E_Message, PollMessagesMonotonicOffsets) {
RecordProperty("description",
"Verifies offsets are monotonically increasing and continuous across multiple polls.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 20; i++) {
auto msg =
iggy::ffi::make_message(to_payload("mono-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
std::uint64_t expected_offset = 0;
for (int chunk = 0; chunk < 4; chunk++) {
auto polled =
client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", expected_offset, 5, false);
ASSERT_EQ(polled.count, 5u) << "Chunk " << chunk;
ASSERT_EQ(polled.messages.size(), 5u) << "Chunk " << chunk;
for (std::size_t i = 0; i < polled.messages.size(); i++) {
EXPECT_EQ(polled.messages[i].offset, expected_offset) << "Chunk " << chunk << " index " << i;
expected_offset++;
}
}
ASSERT_EQ(expected_offset, 20u);
}
TEST_F(LowLevelE2E_Message, SendMessagesLargeBatch) {
RecordProperty("description", "Verifies sending a large batch of 1000 messages succeeds and all are retrievable.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 1000; i++) {
auto msg =
iggy::ffi::make_message(to_payload("batch-msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0),
"partition_id", partition_id_bytes(0), std::move(messages)));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 1000, false);
ASSERT_EQ(polled.count, 1000u);
ASSERT_EQ(polled.messages.size(), 1000u);
EXPECT_EQ(polled.messages[0].offset, 0u);
EXPECT_EQ(polled.messages[999].offset, 999u);
}
TEST_F(LowLevelE2E_Message, SendMessagesWithInvalidTopicIdThrows) {
RecordProperty("description", "Throws when sending messages with an invalid topic identifier.");
iggy::ffi::Client *client = GetLoggedInClient();
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
auto msg = iggy::ffi::make_message(to_payload("test"), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
iggy::ffi::Identifier invalid_id;
invalid_id.kind = "invalid";
invalid_id.length = 0;
ASSERT_THROW(client->send_messages(make_numeric_identifier(1), invalid_id, "partition_id", partition_id_bytes(0),
std::move(messages)),
std::exception);
}
TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidTopicIdThrows) {
RecordProperty("description", "Throws when polling messages with an invalid topic identifier.");
iggy::ffi::Client *client = GetLoggedInClient();
iggy::ffi::Identifier invalid_id;
invalid_id.kind = "invalid";
invalid_id.length = 0;
ASSERT_THROW(client->poll_messages(make_numeric_identifier(1), invalid_id, 0, "consumer",
make_numeric_identifier(1), "offset", 0, 10, false),
std::exception);
}
TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidConsumerIdThrows) {
RecordProperty("description", "Throws when polling messages with an invalid consumer identifier.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
iggy::ffi::Identifier invalid_id;
invalid_id.kind = "invalid";
invalid_id.length = 0;
ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
invalid_id, "offset", 0, 10, false),
std::exception);
}
TEST_F(LowLevelE2E_Message, ConsumerGroupCreateJoinAndPollMessages) {
RecordProperty("description",
"Creates a consumer group, joins it, sends messages, and polls them using consumer_group kind.");
const std::string stream_name = GetRandomName();
iggy::ffi::Client *client = GetLoggedInClient();
client->create_stream(stream_name);
auto stream = client->get_stream(make_string_identifier(stream_name));
TrackStream(stream.id);
const std::string topic_name = GetRandomName();
client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
"server_default");
const std::string group_name = GetRandomName();
auto group =
client->create_consumer_group(make_numeric_identifier(stream.id), make_numeric_identifier(0), group_name);
ASSERT_EQ(group.members_count, 0u);
ASSERT_NO_THROW(client->join_consumer_group(make_numeric_identifier(stream.id), make_numeric_identifier(0),
make_numeric_identifier(group.id)));
auto group_after_join = client->get_consumer_group(make_numeric_identifier(stream.id), make_numeric_identifier(0),
make_numeric_identifier(group.id));
ASSERT_EQ(group_after_join.members_count, 1u);
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 10; i++) {
auto msg =
iggy::ffi::make_message(to_payload("cg-msg-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0,
"consumer_group", make_numeric_identifier(group.id), "offset", 0, 100, false);
ASSERT_EQ(polled.count, 10u);
ASSERT_EQ(polled.messages.size(), 10u);
for (std::uint32_t i = 0; i < 10; i++) {
std::string expected = "cg-msg-" + std::to_string(i);
std::string actual(polled.messages[i].payload.begin(), polled.messages[i].payload.end());
EXPECT_EQ(actual, expected) << "Payload mismatch at offset " << i;
}
ASSERT_NO_THROW(client->leave_consumer_group(make_numeric_identifier(stream.id), make_numeric_identifier(0),
make_numeric_identifier(group.id)));
auto group_after_leave = client->get_consumer_group(make_numeric_identifier(stream.id), make_numeric_identifier(0),
make_numeric_identifier(group.id));
ASSERT_EQ(group_after_leave.members_count, 0u);
}