spetz commented on code in PR #4397:
URL: https://github.com/apache/iggy/pull/4397#discussion_r4446951907
##########
foreign/cpp/include/iggy.hpp:
##########
@@ -5738,6 +5808,305 @@ class PollingStrategy final {
std::uint64_t polling_strategy_value_;
};
+/**
+ * @brief Selects the destination partition for a batch of messages.
+ *
+ * Balanced() distributes batches across the topic's partitions. PartitionId()
+ * selects one partition explicitly. MessagesKey() hashes the supplied bytes
+ * modulo the topic's partition count, so changing that count can change the
+ * destination for a key.
+ *
+ * The selected strategy applies to the entire SendMessages() call. Validation
+ * is deferred until the batch is sent. In particular, a message key must
+ * contain between 1 and 255 bytes.
+ */
+class Partitioning final {
+ public:
+ /**
+ * @brief Selects partitions using the client's balanced strategy.
+ * @return Balanced partitioning with no value payload.
+ */
+ static Partitioning Balanced() { return Partitioning("balanced", {}); }
Review Comment:
No test calls `Partitioning::Balanced()` or `Partitioning::MessagesKey()`.
Every `SendMessages` call in `foreign/cpp/tests` and `bdd/cpp` uses
`PartitionId`, so the `balanced` and `messages_key` kinds and the 1 to 255 byte
key rule at `foreign/cpp/src/client.rs:452` never run from C++. The only send
round trip checks the confirmation count and `PartitionId()`
(`foreign/cpp/tests/e2e/message.cpp:82`), so a wrong `BaseOffset()`,
`StreamId()` or `TopicId()` mapping also passes.
##########
foreign/cpp/src/client.cpp:
##########
@@ -553,6 +570,105 @@ Stats IggyBlockingClient::GetStats() {
return RethrowAsIggyException([this] { return
Stats::FromFfi(Handle()->get_stats()); });
}
+SendMessagesResponse IggyBlockingClient::SendMessages(const Identifier &stream,
+ const Identifier &topic,
+ const Partitioning
&partitioning,
+ const
std::vector<IggyMessageToSend> &messages) {
+ return RethrowAsIggyException([this, &stream, &topic, &partitioning,
&messages](ffi::ErrorInfo &error_info) {
+ rust::Vec<ffi::IggyMessageToSend> ffi_messages;
+ ffi_messages.reserve(messages.size());
+ for (const auto &message : messages) {
+ ffi_messages.push_back(message.ToFfi());
Review Comment:
`SendMessages` takes the batch by const reference, so `ToFfi` copies every
payload and header byte into a `rust::Vec` with `push_back`
(`foreign/cpp/src/type_conversions.cpp:208`). In cxx 1.0.205 each `push_back`
makes four out-of-line calls into Rust (`size`, `reserve_total`, `data`,
`set_len`), so a 1 MB batch makes about four million calls across the bridge
before anything is sent. Master had no caller of `IggyMessageToSend::ToFfi`, so
this PR is what puts that loop on the main send path. The vector `Create`
overload tells callers to move the payload to avoid a copy
(`foreign/cpp/include/iggy.hpp:1060`), but the send copies it again anyway.
`SendBinaryRequest` pays the same per-byte cost through `ToFfiBytes`.
##########
foreign/cpp/include/iggy.hpp:
##########
@@ -7427,6 +7798,181 @@ class IggyBlockingClient final {
*/
Stats GetStats();
+ /**
+ * @brief Sends a non-empty batch of messages to one topic.
+ *
+ * The partitioning strategy selects one destination for the entire batch.
+ * Message and partitioning constraints are validated before the request is
+ * sent. The Rust SDK assigns generated IDs to messages whose ID is zero;
+ * the C++ input objects are not modified.
+ *
+ * Delivery is at least once. A failed or unknown transport outcome can
+ * leave the batch committed, so retrying may append duplicates. Successful
+ * responses can contain no confirmations; see SendMessagesResponse.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Destination topic, addressed by numeric ID or name.
+ * @param partitioning Strategy used to select the destination partition.
+ * @param messages Messages to send. The collection must not be empty.
+ * @return Available per-partition commit confirmations.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier, partitioning strategy, message, or batch is
+ * invalid; the stream, topic, or partition does not exist; the
+ * caller lacks send permission; or the request fails.
+ */
+ SendMessagesResponse SendMessages(const Identifier &stream,
+ const Identifier &topic,
+ const Partitioning &partitioning,
+ const std::vector<IggyMessageToSend>
&messages);
+
+ /**
+ * @brief Reads one partition with a message count and optional on-read
commit.
+ *
+ * A partition is required. For additional read limits or a continuation,
+ * use the overload taking PollOptions.
+ */
+ PolledMessages PollMessages(const Identifier &stream,
Review Comment:
This overload takes the consumer fourth and a `bool auto_commit`, while
every other consumer call in the class, including the two existing
`PollMessages` overloads at `foreign/cpp/include/iggy.hpp:7573` and
`foreign/cpp/include/iggy.hpp:7585`, takes the consumer first and a
`CommitPolicy`. Its body at `foreign/cpp/src/client.cpp:596` only builds
`PollOptions().SetCount(count)` with a commit policy and forwards, so it adds
no capability, and once released the class keeps two argument orders for one
name. Its doc also hides that `true` means `CommitPolicy::OnRead`, so a caller
never learns that a lost reply blocks the partition with `PartitionLaneBlocked`
until `AbandonOperation()`, which the other overload documents at
`foreign/cpp/include/iggy.hpp:7558`.
##########
foreign/cpp/BUILD.bazel:
##########
@@ -61,7 +64,12 @@ genrule(
env -u PWD \
CARGO_TARGET_DIR="$$CARGO_TARGET_DIR" \
RUSTC="$$RUSTC" \
- "$$CARGO" build --quiet $$FLAGS
+ "$$CARGO" build --quiet --lib $$FLAGS
+
+ env -u PWD \
+ CARGO_TARGET_DIR="$$CARGO_TARGET_DIR" \
+ RUSTC="$$RUSTC" \
+ "$$CARGO" rustc --quiet --tests $$FLAGS --
--emit="link=$$OUT_TESTS"
Review Comment:
`rust_unit_tests` is an output of the same genrule that builds
`libiggy_cpp.a`, so every target that depends on `//:iggy-cpp` now also
compiles and links the Rust test binary. That includes the release build in CI
(`.github/actions/cpp-bazel/pre-merge/action.yml:73`) and the BDD image
(`bdd/cpp/Dockerfile:36`), which never run it, and a compile error in a
`#[cfg(test)]` module now fails the library build as well.
##########
foreign/cpp/tests/e2e/client.cpp:
##########
@@ -2843,264 +2730,319 @@ TEST_F(E2E_Client,
GetClientsReflectsAdditionalSession) {
EXPECT_TRUE(found_after);
}
-TEST_F(LowLevelE2E_Client, GetClusterMetadataBeforeLoginThrows) {
- iggy::ffi::ErrorInfo error_info{};
+TEST_F(E2E_Client, GetClusterMetadataBeforeLoginThrows) {
RecordProperty("description",
"Rejects get_cluster_metadata before connect, after connect
but before login, and after disconnect, "
"and serves it once authenticated.");
- iggy::ffi::Client *client = GetLoggedOutClient();
+ auto client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(client->get_cluster_metadata(), std::exception);
- ASSERT_NO_THROW(client->connect());
+ ASSERT_THROW(client.GetClusterMetadata(), std::exception);
+ ASSERT_NO_THROW(client.Connect());
// The roster is private, so the read is auth-gated. No pre-login read is
// needed: a client that dialed a backup logs in there and the server
// forwards the register to the primary.
- ASSERT_THROW(client->get_cluster_metadata(), std::exception);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy", error_info));
- iggy::ffi::ClusterMetadata metadata{};
- ASSERT_NO_THROW({ metadata = client->get_cluster_metadata(); });
- ASSERT_GE(metadata.nodes.size(), 1u);
- ASSERT_NO_THROW(client->disconnect());
- ASSERT_THROW(client->get_cluster_metadata(), std::exception);
+ ASSERT_THROW(client.GetClusterMetadata(), std::exception);
+ ASSERT_NO_THROW(client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW({
+ const auto metadata = client.GetClusterMetadata();
+ ASSERT_GE(metadata.Nodes().size(), 1u);
+ });
+ ASSERT_NO_THROW(client.Disconnect());
+ ASSERT_THROW(client.GetClusterMetadata(), std::exception);
}
-TEST_F(LowLevelE2E_Client, GetClusterMetadataReturnsSingleNodeMetadata) {
+TEST_F(E2E_Client, GetClusterMetadataReturnsSingleNodeMetadata) {
RecordProperty("description",
"Returns the expected single-node cluster metadata shape
from the default test server.");
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- iggy::ffi::ClusterMetadata metadata{};
- ASSERT_NO_THROW({ metadata = client->get_cluster_metadata(); });
+ const auto metadata = client.GetClusterMetadata();
- EXPECT_EQ(static_cast<std::string>(metadata.name), "single-node");
- ASSERT_EQ(metadata.nodes.size(), 1u);
+ EXPECT_EQ(metadata.Name(), "single-node");
+ ASSERT_EQ(metadata.Nodes().size(), 1u);
- const auto &node = metadata.nodes[0];
- EXPECT_FALSE(static_cast<std::string>(node.name).empty());
- EXPECT_FALSE(static_cast<std::string>(node.ip).empty());
- EXPECT_EQ(static_cast<std::string>(node.role), "leader");
- EXPECT_EQ(static_cast<std::string>(node.status), "healthy");
- EXPECT_NE(node.endpoints.tcp, 0u);
- EXPECT_NE(node.endpoints.http, 0u);
+ const auto &node = metadata.Nodes()[0];
+ EXPECT_EQ(node.ReplicaId(), 0u);
+ EXPECT_FALSE(node.Name().empty());
+ EXPECT_FALSE(node.Ip().empty());
+ EXPECT_EQ(node.Role(), "leader");
+ EXPECT_EQ(node.Status(), "healthy");
+ EXPECT_NE(node.Endpoints().Tcp(), 0u);
+ EXPECT_NE(node.Endpoints().Http(), 0u);
}
-TEST_F(LowLevelE2E_Client, GetClusterMetadataIsStableAcrossBackToBackCalls) {
+TEST_F(E2E_Client, GetClusterMetadataIsStableAcrossBackToBackCalls) {
RecordProperty("description", "Returns stable single-node cluster metadata
across back-to-back calls.");
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- iggy::ffi::ClusterMetadata first_metadata{};
- iggy::ffi::ClusterMetadata second_metadata{};
- ASSERT_NO_THROW({
- first_metadata = client->get_cluster_metadata();
- second_metadata = client->get_cluster_metadata();
- });
+ const auto first_metadata = client.GetClusterMetadata();
+ const auto second_metadata = client.GetClusterMetadata();
- EXPECT_EQ(static_cast<std::string>(first_metadata.name),
static_cast<std::string>(second_metadata.name));
- ASSERT_EQ(first_metadata.nodes.size(), 1u);
- ASSERT_EQ(second_metadata.nodes.size(), 1u);
+ EXPECT_EQ(first_metadata.Name(), second_metadata.Name());
+ ASSERT_EQ(first_metadata.Nodes().size(), 1u);
+ ASSERT_EQ(second_metadata.Nodes().size(), 1u);
- const auto &first_node = first_metadata.nodes[0];
- const auto &second_node = second_metadata.nodes[0];
- EXPECT_EQ(static_cast<std::string>(first_node.name),
static_cast<std::string>(second_node.name));
- EXPECT_EQ(static_cast<std::string>(first_node.ip),
static_cast<std::string>(second_node.ip));
- EXPECT_EQ(static_cast<std::string>(first_node.role),
static_cast<std::string>(second_node.role));
- EXPECT_EQ(static_cast<std::string>(first_node.status),
static_cast<std::string>(second_node.status));
- EXPECT_EQ(first_node.endpoints.tcp, second_node.endpoints.tcp);
- EXPECT_EQ(first_node.endpoints.quic, second_node.endpoints.quic);
- EXPECT_EQ(first_node.endpoints.http, second_node.endpoints.http);
- EXPECT_EQ(first_node.endpoints.websocket, second_node.endpoints.websocket);
+ const auto &first_node = first_metadata.Nodes()[0];
+ const auto &second_node = second_metadata.Nodes()[0];
+ EXPECT_EQ(first_node.ReplicaId(), second_node.ReplicaId());
+ EXPECT_EQ(first_node.Name(), second_node.Name());
+ EXPECT_EQ(first_node.Ip(), second_node.Ip());
+ EXPECT_EQ(first_node.Role(), second_node.Role());
+ EXPECT_EQ(first_node.Status(), second_node.Status());
+ EXPECT_EQ(first_node.Endpoints().Tcp(), second_node.Endpoints().Tcp());
+ EXPECT_EQ(first_node.Endpoints().Quic(), second_node.Endpoints().Quic());
+ EXPECT_EQ(first_node.Endpoints().Http(), second_node.Endpoints().Http());
+ EXPECT_EQ(first_node.Endpoints().Websocket(),
second_node.Endpoints().Websocket());
}
-TEST_F(LowLevelE2E_Client, PingSucceedsForNewConnection) {
+TEST_F(E2E_Client, PingSucceedsForNewConnection) {
RecordProperty("description", "Successfully pings the server from a fresh
unauthenticated client session.");
- iggy::ffi::Client *client = GetLoggedOutClient();
+ auto client = GetLoggedOutHighLevelClient();
// The VSR client has no lazy connect; ping still needs no authentication.
- ASSERT_NO_THROW(client->connect());
- ASSERT_NO_THROW(client->ping());
+ ASSERT_NO_THROW(client.Connect());
+ ASSERT_NO_THROW(client.Ping());
}
-TEST_F(LowLevelE2E_Client,
HeartbeatIntervalReturnsDefaultValueForNewConnection) {
+TEST_F(E2E_Client, HeartbeatIntervalReturnsDefaultValueForNewConnection) {
RecordProperty("description",
"Returns the default heartbeat interval in microseconds for
a fresh unauthenticated client.");
- constexpr std::uint64_t default_heartbeat_micros = 5'000'000ull;
- iggy::ffi::Client *client = GetLoggedOutClient();
+ constexpr std::int64_t default_heartbeat_micros = 5'000'000;
+ auto client =
GetLoggedOutHighLevelClient();
- const auto heartbeat_interval = client->heartbeat_interval();
- EXPECT_EQ(heartbeat_interval, default_heartbeat_micros);
+ EXPECT_EQ(client.HeartbeatInterval().count(), default_heartbeat_micros);
}
-TEST_F(LowLevelE2E_Client,
HeartbeatIntervalReturnsConfiguredValueFromConnectionString) {
+TEST_F(E2E_Client,
HeartbeatIntervalReturnsConfiguredValueFromConnectionString) {
RecordProperty("description",
"Returns the configured heartbeat interval in microseconds
from the connection string.");
- constexpr std::uint64_t configured_heartbeat_micros = 10'000'000ull;
- iggy::ffi::Client *client = nullptr;
- ASSERT_NO_THROW(
- { client =
iggy::ffi::from_connection_string("iggy://iggy:[email protected]:8090?heartbeat_interval=10s");
});
- ASSERT_NE(client, nullptr);
- TrackClient(client);
+ constexpr std::int64_t configured_heartbeat_micros = 10'000'000;
+ auto client =
+
iggy::IggyBlockingClient::FromConnectionString("iggy://iggy:[email protected]:8090?heartbeat_interval=10s");
- const auto heartbeat_interval = client->heartbeat_interval();
- EXPECT_EQ(heartbeat_interval, configured_heartbeat_micros);
+ EXPECT_EQ(client.HeartbeatInterval().count(), configured_heartbeat_micros);
}
-TEST_F(LowLevelE2E_Client, SnapshotBeforeLoginThrows) {
- iggy::ffi::ErrorInfo error_info{};
+TEST_F(E2E_Client, SnapshotBeforeLoginThrows) {
RecordProperty("description",
"Rejects snapshot before connect, after connect but before
login, and after disconnect.");
- iggy::ffi::Client *client = GetLoggedOutClient();
+ auto client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"test"})),
std::exception);
+ ASSERT_THROW(client.Snapshot(iggy::SnapshotCompression::Deflated(),
{iggy::SystemSnapshotType::Test()}),
+ std::exception);
- ASSERT_NO_THROW(client->connect());
- ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"test"})),
std::exception);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy", error_info));
- ASSERT_NO_THROW(client->disconnect());
- ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"test"})),
std::exception);
+ ASSERT_NO_THROW(client.Connect());
+ ASSERT_THROW(client.Snapshot(iggy::SnapshotCompression::Deflated(),
{iggy::SystemSnapshotType::Test()}),
+ std::exception);
+ ASSERT_NO_THROW(client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(client.Disconnect());
+ ASSERT_THROW(client.Snapshot(iggy::SnapshotCompression::Deflated(),
{iggy::SystemSnapshotType::Test()}),
+ std::exception);
}
-TEST_F(LowLevelE2E_Client, SnapshotAllCombinedWithOtherTypeThrows) {
+TEST_F(E2E_Client, SnapshotAllCombinedWithOtherTypeThrows) {
RecordProperty("description", "Rejects combining the all snapshot type
with any other snapshot type.");
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"all",
"test"})), std::exception);
+ ASSERT_THROW(client.Snapshot(iggy::SnapshotCompression::Deflated(),
+ {iggy::SystemSnapshotType::All(),
iggy::SystemSnapshotType::Test()}),
+ std::exception);
}
-TEST_F(LowLevelE2E_Client, SnapshotWithEmptySnapshotTypesThrows) {
+TEST_F(E2E_Client, SnapshotWithEmptySnapshotTypesThrows) {
RecordProperty("description", "Rejects an empty snapshot type list in the
wrapper before sending.");
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- rust::Vec<rust::String> snapshot_types;
- ASSERT_THROW(client->snapshot("deflated", snapshot_types), std::exception);
+ ASSERT_THROW(client.Snapshot(iggy::SnapshotCompression::Deflated(),
std::vector<iggy::SystemSnapshotType>{}),
+ std::exception);
}
-TEST_F(LowLevelE2E_Client, SnapshotReturnsNonEmptyBytes) {
+TEST_F(E2E_Client, SnapshotReturnsNonEmptyBytes) {
RecordProperty("description", "Returns a non-empty snapshot for a valid
compression and snapshot type.");
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- rust::Vec<std::uint8_t> snapshot_bytes;
- ASSERT_NO_THROW({ snapshot_bytes = client->snapshot("deflated",
make_snapshot_types({"test"})); });
+ std::vector<std::uint8_t> snapshot_bytes;
+ ASSERT_NO_THROW({
+ snapshot_bytes =
client.Snapshot(iggy::SnapshotCompression::Deflated(),
{iggy::SystemSnapshotType::Test()});
+ });
EXPECT_FALSE(snapshot_bytes.empty());
}
-TEST_F(LowLevelE2E_Client, SnapshotWithInvalidCompressionThrows) {
- RecordProperty("description",
- "Rejects empty or invalid snapshot compression values in
the wrapper before sending.");
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_THROW(client->snapshot("", make_snapshot_types({"test"})),
std::exception);
- ASSERT_THROW(client->snapshot("invalid-compression",
make_snapshot_types({"test"})), std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, SnapshotWithInvalidSnapshotTypeThrows) {
- RecordProperty("description", "Rejects invalid snapshot type values in the
wrapper before sending.");
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_THROW(client->snapshot("deflated",
make_snapshot_types({"not-a-real-type"})), std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, SendBinaryRequestPingReturnsEmptyBytes) {
+TEST_F(E2E_Client, SendBinaryRequestPingReturnsEmptyBytes) {
RecordProperty("description", "Returns an empty response body for a raw
ping command with an empty payload.");
constexpr std::uint32_t ping_command_code = 1;
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- rust::Vec<std::uint8_t> empty_payload;
- rust::Vec<std::uint8_t> response;
- ASSERT_NO_THROW({
- iggy::ffi::ErrorInfo error_info{};
- response = client->send_binary_request(ping_command_code,
empty_payload, error_info);
- });
+ const std::vector<std::uint8_t> empty_payload;
+ std::vector<std::uint8_t> response;
+ ASSERT_NO_THROW({ response = client.SendBinaryRequest(ping_command_code,
empty_payload); });
EXPECT_TRUE(response.empty());
}
-TEST_F(LowLevelE2E_Client, SendBinaryRequestGetStatsReturnsNonEmptyBytes) {
+TEST_F(E2E_Client, SendBinaryRequestGetStatsReturnsNonEmptyBytes) {
RecordProperty("description",
"Returns a non-empty response body for a raw get-stats
command with an empty payload.");
constexpr std::uint32_t get_stats_command_code = 10;
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client =
GetLoggedInHighLevelClient();
- rust::Vec<std::uint8_t> empty_payload;
- rust::Vec<std::uint8_t> response;
- ASSERT_NO_THROW({
- iggy::ffi::ErrorInfo error_info{};
- response = client->send_binary_request(get_stats_command_code,
empty_payload, error_info);
- });
+ const std::vector<std::uint8_t> empty_payload;
+ std::vector<std::uint8_t> response;
+ ASSERT_NO_THROW({ response =
client.SendBinaryRequest(get_stats_command_code, empty_payload); });
EXPECT_FALSE(response.empty());
}
-TEST_F(LowLevelE2E_Client, SendBinaryRequestLoginUserCodeThrows) {
- iggy::ffi::ErrorInfo error_info{};
+TEST_F(E2E_Client, SendBinaryRequestLoginUserCodeThrows) {
RecordProperty("description",
"Rejects the login-user session-control code client-side
before it reaches the server.");
constexpr std::uint32_t login_user_command_code = 38;
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client =
GetLoggedInHighLevelClient();
- rust::Vec<std::uint8_t> empty_payload;
- ASSERT_THROW(client->send_binary_request(login_user_command_code,
empty_payload, error_info), std::exception);
+ const std::vector<std::uint8_t> empty_payload;
+ ASSERT_THROW(client.SendBinaryRequest(login_user_command_code,
empty_payload), std::exception);
}
-TEST_F(LowLevelE2E_Client, SendBinaryRequestUnknownCommandCodeThrows) {
- iggy::ffi::ErrorInfo error_info{};
+TEST_F(E2E_Client, SendBinaryRequestUnknownCommandCodeThrows) {
RecordProperty("description", "Rejects an unknown command code with an
invalid-command error from the server.");
constexpr std::uint32_t unknown_command_code = 60000;
- constexpr std::uint32_t invalid_command_code = 3;
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client =
GetLoggedInHighLevelClient();
+
+ const std::vector<std::uint8_t> empty_payload;
+ ASSERT_THROW(client.SendBinaryRequest(unknown_command_code,
empty_payload), std::exception);
+}
+
+TEST_F(E2E_Client, ConsumerOffsetStoreGetDeleteRoundTrip) {
+ RecordProperty("description", "Stores a consumer offset, reads it back,
and deletes it.");
+ const std::string stream_name = GetRandomName();
+ auto client = GetLoggedInHighLevelClient();
+
+ client.CreateStream(stream_name);
+ const auto stream =
client.GetStream(iggy::Identifier::String(stream_name));
+ TrackStream(stream.Id());
+ const std::string topic_name = GetRandomName();
+ client.CreateTopic(iggy::Identifier::Numeric(stream.Id()), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1));
+
+ std::vector<iggy::IggyMessageToSend> messages;
+ for (std::uint32_t i = 0; i < 5; i++) {
+ messages.push_back(
+ iggy::IggyMessageToSend::Create("offset-" + std::to_string(i),
std::vector<iggy::HeaderEntry>()));
+ }
+ client.SendMessages(iggy::Identifier::Numeric(stream.Id()),
iggy::Identifier::Numeric(0),
+ iggy::Partitioning::PartitionId(0), messages);
+
+ const auto consumer =
iggy::Consumer::Single(iggy::Identifier::String("offset-consumer"));
+ ASSERT_NO_THROW(client.StoreConsumerOffset(consumer,
iggy::Identifier::Numeric(stream.Id()),
+ iggy::Identifier::Numeric(0),
0, 2));
+
+ auto stored =
+ client.GetConsumerOffset(consumer,
iggy::Identifier::Numeric(stream.Id()), iggy::Identifier::Numeric(0), 0);
+ EXPECT_EQ(stored.PartitionId(), 0u);
+ EXPECT_EQ(stored.StoredOffset(), 2u);
+
+ ASSERT_NO_THROW(
+ client.DeleteConsumerOffset(consumer,
iggy::Identifier::Numeric(stream.Id()), iggy::Identifier::Numeric(0), 0));
+
+ ASSERT_THROW(
+ client.GetConsumerOffset(consumer,
iggy::Identifier::Numeric(stream.Id()), iggy::Identifier::Numeric(0), 0),
+ iggy::IggyException);
+}
+
+TEST_F(E2E_Client, ConsumerOffsetKeepsDistinctConsumersApart) {
+ RecordProperty("description", "Each consumer owns its stored offset on the
same partition.");
+ const std::string stream_name = GetRandomName();
+ auto client = GetLoggedInHighLevelClient();
+
+ client.CreateStream(stream_name);
+ const auto stream =
client.GetStream(iggy::Identifier::String(stream_name));
+ TrackStream(stream.Id());
+ const std::string topic_name = GetRandomName();
+ client.CreateTopic(iggy::Identifier::Numeric(stream.Id()), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1));
+
+ std::vector<iggy::IggyMessageToSend> messages;
+ for (std::uint32_t i = 0; i < 5; i++) {
+ messages.push_back(
+ iggy::IggyMessageToSend::Create("offset-" + std::to_string(i),
std::vector<iggy::HeaderEntry>()));
+ }
+ client.SendMessages(iggy::Identifier::Numeric(stream.Id()),
iggy::Identifier::Numeric(0),
+ iggy::Partitioning::PartitionId(0), messages);
+
+ const auto consumer_a =
iggy::Consumer::Single(iggy::Identifier::String("consumer-a"));
+ const auto consumer_b =
iggy::Consumer::Single(iggy::Identifier::String("consumer-b"));
+ client.StoreConsumerOffset(consumer_a,
iggy::Identifier::Numeric(stream.Id()), iggy::Identifier::Numeric(0), 0, 1);
+ client.StoreConsumerOffset(consumer_b,
iggy::Identifier::Numeric(stream.Id()), iggy::Identifier::Numeric(0), 0, 3);
- rust::Vec<std::uint8_t> empty_payload;
- ASSERT_THROW(client->send_binary_request(unknown_command_code,
empty_payload, error_info), std::exception);
- EXPECT_EQ(error_info.code, invalid_command_code);
- EXPECT_EQ(std::string(error_info.kind), "invalid_command");
- EXPECT_FALSE(error_info.has_operation_id);
- EXPECT_EQ(error_info.operation_id, 0u);
+ auto offset_a =
+ client.GetConsumerOffset(consumer_a,
iggy::Identifier::Numeric(stream.Id()), iggy::Identifier::Numeric(0), 0);
+ auto offset_b =
+ client.GetConsumerOffset(consumer_b,
iggy::Identifier::Numeric(stream.Id()), iggy::Identifier::Numeric(0), 0);
+
+ EXPECT_EQ(offset_a.StoredOffset(), 1u);
+ EXPECT_EQ(offset_b.StoredOffset(), 3u);
}
-TEST_F(LowLevelE2E_Client,
SendBinaryRequestToReplicaGetStatsReturnsNonEmptyBytes) {
+TEST_F(E2E_Client, StoreConsumerOffsetRejectsMissingPartitionId) {
+ RecordProperty("description", "Storing an offset needs an explicit
partition.");
+ const std::string stream_name = GetRandomName();
+ auto client = GetLoggedInHighLevelClient();
+
+ client.CreateStream(stream_name);
+ const auto stream =
client.GetStream(iggy::Identifier::String(stream_name));
+ TrackStream(stream.Id());
+ const std::string topic_name = GetRandomName();
+ client.CreateTopic(iggy::Identifier::Numeric(stream.Id()), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1));
+
+ const auto consumer =
iggy::Consumer::Single(iggy::Identifier::String("offset-consumer"));
+ ASSERT_THROW(client.StoreConsumerOffset(consumer,
iggy::Identifier::Numeric(stream.Id()),
Review Comment:
The topic has one empty partition, and the server rejects offset 1 there
with `InvalidOffset` even when the call names partition 0
(`core/partitions/src/iggy_partition.rs:5565`, which
`foreign/cpp/tests/e2e/consumer_group.cpp:1203` relies on). The test checks
only the exception type, so it passes whether or not a missing partition is
refused, and a wrapper that sent partition 0 for `std::nullopt` would still be
green.
##########
foreign/cpp/include/iggy.hpp:
##########
@@ -780,6 +790,17 @@ class Permissions final {
* A consumer offset belongs either to an individual consumer or to a consumer
* group. Create a value with Single() or Group(), then pass it to the consumer
* offset operations on IggyBlockingClient.
+ *
+ * Two callers naming the same identifier share one offset, so under
+ * PollingStrategy::Next() with auto-commit each sees only the messages the
+ * other has not read yet. Give every independent consumer its own name.
+ *
+ * @code{.cpp}
+ * const auto polled{client.PollMessages(
Review Comment:
The example polls as consumer group `my-group` without joining it, so the
server rejects the call with `ConsumerGroupPartitionNotOwned` (5009). A group
poll passes the fence at `core/server/src/dispatch/partition.rs:1297` only for
a member that owns the partition, which is why the e2e group poll joins and
waits for the assignment first (`foreign/cpp/tests/e2e/message.cpp:1460`). The
paragraph above describes two callers sharing one offset under `Next()` with
auto-commit, which holds for `Consumer::Single` and not for a group, so a
reader who copies the example gets an exception instead of the split the text
promises.
##########
foreign/cpp/src/client.cpp:
##########
@@ -280,14 +280,32 @@ TopicDetails IggyBlockingClient::CreateTopic(const
Identifier &stream,
}
if (auto value = options.Durability()) {
ffi_options.has_durability = true;
- ffi_options.durability = std::string(iggy::to_string(*value));
+ switch (*value) {
Review Comment:
These two switches stand in for the public `iggy::to_string(Durability)`
that this PR deletes from `iggy.hpp`, which breaks every caller of it and is
not mentioned in the description. `Durability` is now the only option enum in
the header without a `to_string`, and the field right below still calls
`to_string(PartitionResizePolicy)` (`foreign/cpp/src/client.cpp:315`).
##########
foreign/cpp/include/iggy.hpp:
##########
@@ -7427,6 +7798,181 @@ class IggyBlockingClient final {
*/
Stats GetStats();
+ /**
+ * @brief Sends a non-empty batch of messages to one topic.
+ *
+ * The partitioning strategy selects one destination for the entire batch.
+ * Message and partitioning constraints are validated before the request is
+ * sent. The Rust SDK assigns generated IDs to messages whose ID is zero;
+ * the C++ input objects are not modified.
+ *
+ * Delivery is at least once. A failed or unknown transport outcome can
+ * leave the batch committed, so retrying may append duplicates. Successful
+ * responses can contain no confirmations; see SendMessagesResponse.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Destination topic, addressed by numeric ID or name.
+ * @param partitioning Strategy used to select the destination partition.
+ * @param messages Messages to send. The collection must not be empty.
+ * @return Available per-partition commit confirmations.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier, partitioning strategy, message, or batch is
+ * invalid; the stream, topic, or partition does not exist; the
+ * caller lacks send permission; or the request fails.
+ */
+ SendMessagesResponse SendMessages(const Identifier &stream,
+ const Identifier &topic,
+ const Partitioning &partitioning,
+ const std::vector<IggyMessageToSend>
&messages);
+
+ /**
+ * @brief Reads one partition with a message count and optional on-read
commit.
+ *
+ * A partition is required. For additional read limits or a continuation,
+ * use the overload taking PollOptions.
+ */
+ PolledMessages PollMessages(const Identifier &stream,
+ const Identifier &topic,
+ std::uint32_t partition_id,
+ const Consumer &consumer,
+ const PollingStrategy &strategy,
+ std::uint32_t count,
+ bool auto_commit);
+
+ /**
+ * @brief Retrieves the server's option catalog for one resource type.
+ *
+ * The catalog describes accepted create-option keys, their HeaderKind
+ * encodings, defaults, and descriptions. A supported scope with no option
+ * keys returns an empty collection.
+ *
+ * @param scope Resource scope. Must be `topic`, `stream`, or `user`.
+ * @return Option specifications for the requested scope.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * @p scope is invalid; or the request fails.
+ */
+ std::vector<OptionSpec> DescribeOptions(std::string scope);
Review Comment:
Master already has `DescribeOptions(OptionsScope)`
(`foreign/cpp/include/iggy.hpp:7766`), and this overload repeats its body
(`foreign/cpp/src/client.cpp:603` against `foreign/cpp/src/client.cpp:550`)
with a string the compiler cannot check. A misspelled scope now reaches the
bridge and fails at run time, as `foreign/cpp/tests/e2e/topic.cpp:374` shows,
where the enum made that impossible. The description lists `DescribeOptions`
among the added functions, but it existed before this PR.
##########
foreign/cpp/tests/e2e/client.cpp:
##########
@@ -2843,264 +2730,319 @@ TEST_F(E2E_Client,
GetClientsReflectsAdditionalSession) {
EXPECT_TRUE(found_after);
}
-TEST_F(LowLevelE2E_Client, GetClusterMetadataBeforeLoginThrows) {
- iggy::ffi::ErrorInfo error_info{};
+TEST_F(E2E_Client, GetClusterMetadataBeforeLoginThrows) {
RecordProperty("description",
"Rejects get_cluster_metadata before connect, after connect
but before login, and after disconnect, "
"and serves it once authenticated.");
- iggy::ffi::Client *client = GetLoggedOutClient();
+ auto client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(client->get_cluster_metadata(), std::exception);
- ASSERT_NO_THROW(client->connect());
+ ASSERT_THROW(client.GetClusterMetadata(), std::exception);
+ ASSERT_NO_THROW(client.Connect());
// The roster is private, so the read is auth-gated. No pre-login read is
// needed: a client that dialed a backup logs in there and the server
// forwards the register to the primary.
- ASSERT_THROW(client->get_cluster_metadata(), std::exception);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy", error_info));
- iggy::ffi::ClusterMetadata metadata{};
- ASSERT_NO_THROW({ metadata = client->get_cluster_metadata(); });
- ASSERT_GE(metadata.nodes.size(), 1u);
- ASSERT_NO_THROW(client->disconnect());
- ASSERT_THROW(client->get_cluster_metadata(), std::exception);
+ ASSERT_THROW(client.GetClusterMetadata(), std::exception);
+ ASSERT_NO_THROW(client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW({
+ const auto metadata = client.GetClusterMetadata();
+ ASSERT_GE(metadata.Nodes().size(), 1u);
+ });
+ ASSERT_NO_THROW(client.Disconnect());
+ ASSERT_THROW(client.GetClusterMetadata(), std::exception);
}
-TEST_F(LowLevelE2E_Client, GetClusterMetadataReturnsSingleNodeMetadata) {
+TEST_F(E2E_Client, GetClusterMetadataReturnsSingleNodeMetadata) {
RecordProperty("description",
"Returns the expected single-node cluster metadata shape
from the default test server.");
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- iggy::ffi::ClusterMetadata metadata{};
- ASSERT_NO_THROW({ metadata = client->get_cluster_metadata(); });
+ const auto metadata = client.GetClusterMetadata();
- EXPECT_EQ(static_cast<std::string>(metadata.name), "single-node");
- ASSERT_EQ(metadata.nodes.size(), 1u);
+ EXPECT_EQ(metadata.Name(), "single-node");
+ ASSERT_EQ(metadata.Nodes().size(), 1u);
- const auto &node = metadata.nodes[0];
- EXPECT_FALSE(static_cast<std::string>(node.name).empty());
- EXPECT_FALSE(static_cast<std::string>(node.ip).empty());
- EXPECT_EQ(static_cast<std::string>(node.role), "leader");
- EXPECT_EQ(static_cast<std::string>(node.status), "healthy");
- EXPECT_NE(node.endpoints.tcp, 0u);
- EXPECT_NE(node.endpoints.http, 0u);
+ const auto &node = metadata.Nodes()[0];
+ EXPECT_EQ(node.ReplicaId(), 0u);
+ EXPECT_FALSE(node.Name().empty());
+ EXPECT_FALSE(node.Ip().empty());
+ EXPECT_EQ(node.Role(), "leader");
+ EXPECT_EQ(node.Status(), "healthy");
+ EXPECT_NE(node.Endpoints().Tcp(), 0u);
+ EXPECT_NE(node.Endpoints().Http(), 0u);
}
-TEST_F(LowLevelE2E_Client, GetClusterMetadataIsStableAcrossBackToBackCalls) {
+TEST_F(E2E_Client, GetClusterMetadataIsStableAcrossBackToBackCalls) {
RecordProperty("description", "Returns stable single-node cluster metadata
across back-to-back calls.");
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- iggy::ffi::ClusterMetadata first_metadata{};
- iggy::ffi::ClusterMetadata second_metadata{};
- ASSERT_NO_THROW({
- first_metadata = client->get_cluster_metadata();
- second_metadata = client->get_cluster_metadata();
- });
+ const auto first_metadata = client.GetClusterMetadata();
+ const auto second_metadata = client.GetClusterMetadata();
- EXPECT_EQ(static_cast<std::string>(first_metadata.name),
static_cast<std::string>(second_metadata.name));
- ASSERT_EQ(first_metadata.nodes.size(), 1u);
- ASSERT_EQ(second_metadata.nodes.size(), 1u);
+ EXPECT_EQ(first_metadata.Name(), second_metadata.Name());
+ ASSERT_EQ(first_metadata.Nodes().size(), 1u);
+ ASSERT_EQ(second_metadata.Nodes().size(), 1u);
- const auto &first_node = first_metadata.nodes[0];
- const auto &second_node = second_metadata.nodes[0];
- EXPECT_EQ(static_cast<std::string>(first_node.name),
static_cast<std::string>(second_node.name));
- EXPECT_EQ(static_cast<std::string>(first_node.ip),
static_cast<std::string>(second_node.ip));
- EXPECT_EQ(static_cast<std::string>(first_node.role),
static_cast<std::string>(second_node.role));
- EXPECT_EQ(static_cast<std::string>(first_node.status),
static_cast<std::string>(second_node.status));
- EXPECT_EQ(first_node.endpoints.tcp, second_node.endpoints.tcp);
- EXPECT_EQ(first_node.endpoints.quic, second_node.endpoints.quic);
- EXPECT_EQ(first_node.endpoints.http, second_node.endpoints.http);
- EXPECT_EQ(first_node.endpoints.websocket, second_node.endpoints.websocket);
+ const auto &first_node = first_metadata.Nodes()[0];
+ const auto &second_node = second_metadata.Nodes()[0];
+ EXPECT_EQ(first_node.ReplicaId(), second_node.ReplicaId());
+ EXPECT_EQ(first_node.Name(), second_node.Name());
+ EXPECT_EQ(first_node.Ip(), second_node.Ip());
+ EXPECT_EQ(first_node.Role(), second_node.Role());
+ EXPECT_EQ(first_node.Status(), second_node.Status());
+ EXPECT_EQ(first_node.Endpoints().Tcp(), second_node.Endpoints().Tcp());
+ EXPECT_EQ(first_node.Endpoints().Quic(), second_node.Endpoints().Quic());
+ EXPECT_EQ(first_node.Endpoints().Http(), second_node.Endpoints().Http());
+ EXPECT_EQ(first_node.Endpoints().Websocket(),
second_node.Endpoints().Websocket());
}
-TEST_F(LowLevelE2E_Client, PingSucceedsForNewConnection) {
+TEST_F(E2E_Client, PingSucceedsForNewConnection) {
RecordProperty("description", "Successfully pings the server from a fresh
unauthenticated client session.");
- iggy::ffi::Client *client = GetLoggedOutClient();
+ auto client = GetLoggedOutHighLevelClient();
// The VSR client has no lazy connect; ping still needs no authentication.
- ASSERT_NO_THROW(client->connect());
- ASSERT_NO_THROW(client->ping());
+ ASSERT_NO_THROW(client.Connect());
+ ASSERT_NO_THROW(client.Ping());
}
-TEST_F(LowLevelE2E_Client,
HeartbeatIntervalReturnsDefaultValueForNewConnection) {
+TEST_F(E2E_Client, HeartbeatIntervalReturnsDefaultValueForNewConnection) {
RecordProperty("description",
"Returns the default heartbeat interval in microseconds for
a fresh unauthenticated client.");
- constexpr std::uint64_t default_heartbeat_micros = 5'000'000ull;
- iggy::ffi::Client *client = GetLoggedOutClient();
+ constexpr std::int64_t default_heartbeat_micros = 5'000'000;
+ auto client =
GetLoggedOutHighLevelClient();
- const auto heartbeat_interval = client->heartbeat_interval();
- EXPECT_EQ(heartbeat_interval, default_heartbeat_micros);
+ EXPECT_EQ(client.HeartbeatInterval().count(), default_heartbeat_micros);
}
-TEST_F(LowLevelE2E_Client,
HeartbeatIntervalReturnsConfiguredValueFromConnectionString) {
+TEST_F(E2E_Client,
HeartbeatIntervalReturnsConfiguredValueFromConnectionString) {
RecordProperty("description",
"Returns the configured heartbeat interval in microseconds
from the connection string.");
- constexpr std::uint64_t configured_heartbeat_micros = 10'000'000ull;
- iggy::ffi::Client *client = nullptr;
- ASSERT_NO_THROW(
- { client =
iggy::ffi::from_connection_string("iggy://iggy:[email protected]:8090?heartbeat_interval=10s");
});
- ASSERT_NE(client, nullptr);
- TrackClient(client);
+ constexpr std::int64_t configured_heartbeat_micros = 10'000'000;
+ auto client =
+
iggy::IggyBlockingClient::FromConnectionString("iggy://iggy:[email protected]:8090?heartbeat_interval=10s");
- const auto heartbeat_interval = client->heartbeat_interval();
- EXPECT_EQ(heartbeat_interval, configured_heartbeat_micros);
+ EXPECT_EQ(client.HeartbeatInterval().count(), configured_heartbeat_micros);
}
-TEST_F(LowLevelE2E_Client, SnapshotBeforeLoginThrows) {
- iggy::ffi::ErrorInfo error_info{};
+TEST_F(E2E_Client, SnapshotBeforeLoginThrows) {
RecordProperty("description",
"Rejects snapshot before connect, after connect but before
login, and after disconnect.");
- iggy::ffi::Client *client = GetLoggedOutClient();
+ auto client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"test"})),
std::exception);
+ ASSERT_THROW(client.Snapshot(iggy::SnapshotCompression::Deflated(),
{iggy::SystemSnapshotType::Test()}),
+ std::exception);
- ASSERT_NO_THROW(client->connect());
- ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"test"})),
std::exception);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy", error_info));
- ASSERT_NO_THROW(client->disconnect());
- ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"test"})),
std::exception);
+ ASSERT_NO_THROW(client.Connect());
+ ASSERT_THROW(client.Snapshot(iggy::SnapshotCompression::Deflated(),
{iggy::SystemSnapshotType::Test()}),
+ std::exception);
+ ASSERT_NO_THROW(client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(client.Disconnect());
+ ASSERT_THROW(client.Snapshot(iggy::SnapshotCompression::Deflated(),
{iggy::SystemSnapshotType::Test()}),
+ std::exception);
}
-TEST_F(LowLevelE2E_Client, SnapshotAllCombinedWithOtherTypeThrows) {
+TEST_F(E2E_Client, SnapshotAllCombinedWithOtherTypeThrows) {
RecordProperty("description", "Rejects combining the all snapshot type
with any other snapshot type.");
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"all",
"test"})), std::exception);
+ ASSERT_THROW(client.Snapshot(iggy::SnapshotCompression::Deflated(),
+ {iggy::SystemSnapshotType::All(),
iggy::SystemSnapshotType::Test()}),
+ std::exception);
}
-TEST_F(LowLevelE2E_Client, SnapshotWithEmptySnapshotTypesThrows) {
+TEST_F(E2E_Client, SnapshotWithEmptySnapshotTypesThrows) {
RecordProperty("description", "Rejects an empty snapshot type list in the
wrapper before sending.");
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- rust::Vec<rust::String> snapshot_types;
- ASSERT_THROW(client->snapshot("deflated", snapshot_types), std::exception);
+ ASSERT_THROW(client.Snapshot(iggy::SnapshotCompression::Deflated(),
std::vector<iggy::SystemSnapshotType>{}),
+ std::exception);
}
-TEST_F(LowLevelE2E_Client, SnapshotReturnsNonEmptyBytes) {
+TEST_F(E2E_Client, SnapshotReturnsNonEmptyBytes) {
RecordProperty("description", "Returns a non-empty snapshot for a valid
compression and snapshot type.");
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- rust::Vec<std::uint8_t> snapshot_bytes;
- ASSERT_NO_THROW({ snapshot_bytes = client->snapshot("deflated",
make_snapshot_types({"test"})); });
+ std::vector<std::uint8_t> snapshot_bytes;
+ ASSERT_NO_THROW({
+ snapshot_bytes =
client.Snapshot(iggy::SnapshotCompression::Deflated(),
{iggy::SystemSnapshotType::Test()});
+ });
EXPECT_FALSE(snapshot_bytes.empty());
}
-TEST_F(LowLevelE2E_Client, SnapshotWithInvalidCompressionThrows) {
- RecordProperty("description",
- "Rejects empty or invalid snapshot compression values in
the wrapper before sending.");
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_THROW(client->snapshot("", make_snapshot_types({"test"})),
std::exception);
- ASSERT_THROW(client->snapshot("invalid-compression",
make_snapshot_types({"test"})), std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, SnapshotWithInvalidSnapshotTypeThrows) {
- RecordProperty("description", "Rejects invalid snapshot type values in the
wrapper before sending.");
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_THROW(client->snapshot("deflated",
make_snapshot_types({"not-a-real-type"})), std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, SendBinaryRequestPingReturnsEmptyBytes) {
+TEST_F(E2E_Client, SendBinaryRequestPingReturnsEmptyBytes) {
RecordProperty("description", "Returns an empty response body for a raw
ping command with an empty payload.");
constexpr std::uint32_t ping_command_code = 1;
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- rust::Vec<std::uint8_t> empty_payload;
- rust::Vec<std::uint8_t> response;
- ASSERT_NO_THROW({
- iggy::ffi::ErrorInfo error_info{};
- response = client->send_binary_request(ping_command_code,
empty_payload, error_info);
- });
+ const std::vector<std::uint8_t> empty_payload;
+ std::vector<std::uint8_t> response;
+ ASSERT_NO_THROW({ response = client.SendBinaryRequest(ping_command_code,
empty_payload); });
EXPECT_TRUE(response.empty());
}
-TEST_F(LowLevelE2E_Client, SendBinaryRequestGetStatsReturnsNonEmptyBytes) {
+TEST_F(E2E_Client, SendBinaryRequestGetStatsReturnsNonEmptyBytes) {
RecordProperty("description",
"Returns a non-empty response body for a raw get-stats
command with an empty payload.");
constexpr std::uint32_t get_stats_command_code = 10;
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client =
GetLoggedInHighLevelClient();
- rust::Vec<std::uint8_t> empty_payload;
- rust::Vec<std::uint8_t> response;
- ASSERT_NO_THROW({
- iggy::ffi::ErrorInfo error_info{};
- response = client->send_binary_request(get_stats_command_code,
empty_payload, error_info);
- });
+ const std::vector<std::uint8_t> empty_payload;
+ std::vector<std::uint8_t> response;
+ ASSERT_NO_THROW({ response =
client.SendBinaryRequest(get_stats_command_code, empty_payload); });
EXPECT_FALSE(response.empty());
}
-TEST_F(LowLevelE2E_Client, SendBinaryRequestLoginUserCodeThrows) {
- iggy::ffi::ErrorInfo error_info{};
+TEST_F(E2E_Client, SendBinaryRequestLoginUserCodeThrows) {
RecordProperty("description",
"Rejects the login-user session-control code client-side
before it reaches the server.");
constexpr std::uint32_t login_user_command_code = 38;
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client =
GetLoggedInHighLevelClient();
- rust::Vec<std::uint8_t> empty_payload;
- ASSERT_THROW(client->send_binary_request(login_user_command_code,
empty_payload, error_info), std::exception);
+ const std::vector<std::uint8_t> empty_payload;
+ ASSERT_THROW(client.SendBinaryRequest(login_user_command_code,
empty_payload), std::exception);
}
-TEST_F(LowLevelE2E_Client, SendBinaryRequestUnknownCommandCodeThrows) {
- iggy::ffi::ErrorInfo error_info{};
+TEST_F(E2E_Client, SendBinaryRequestUnknownCommandCodeThrows) {
RecordProperty("description", "Rejects an unknown command code with an
invalid-command error from the server.");
constexpr std::uint32_t unknown_command_code = 60000;
- constexpr std::uint32_t invalid_command_code = 3;
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client =
GetLoggedInHighLevelClient();
+
+ const std::vector<std::uint8_t> empty_payload;
+ ASSERT_THROW(client.SendBinaryRequest(unknown_command_code,
empty_payload), std::exception);
Review Comment:
The README now promises `Code()` and `Kind()` for `SendMessages` and
`SendBinaryRequest` (`foreign/cpp/README.md:110`), but this port dropped the
checks of code 3, kind `invalid_command` and the empty operation ID that the
bridge version of the test made. No test reads `Code()` or `Kind()` from an
exception of either call, so a wrapper that stops passing `ErrorInfo` to the
bridge still passes the suite.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
