This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 6f1d5485d feat(cpp): add functions related to client lifecycle, and
message headers (#3610)
6f1d5485d is described below
commit 6f1d5485d06c3c43afd2510e432b2f7c1a3ac41a
Author: Rimuksh Kansal <[email protected]>
AuthorDate: Fri Jul 17 18:05:14 2026 +0900
feat(cpp): add functions related to client lifecycle, and message headers
(#3610)
---
.../features/step_definitions/messaging_steps.cpp | 5 +-
foreign/cpp/Cargo.toml | 1 +
foreign/cpp/MODULE.bazel | 7 -
foreign/cpp/MODULE.bazel.lock | 2 +-
foreign/cpp/src/client.rs | 115 +++-
foreign/cpp/src/lib.rs | 138 +++-
foreign/cpp/src/messages.rs | 7 +-
foreign/cpp/src/type_conversion.rs | 291 ++++++++-
foreign/cpp/tests/client/low_level_e2e.cpp | 691 ++++++++++++++++++++-
foreign/cpp/tests/common/test_helpers.hpp | 33 +
foreign/cpp/tests/consumer_group/low_level_e2e.cpp | 34 +
foreign/cpp/tests/message/low_level_e2e.cpp | 443 ++++++++++++-
foreign/cpp/tests/message/unit_tests.cpp | 14 +-
foreign/cpp/tests/partition/low_level_e2e.cpp | 8 +
foreign/cpp/tests/stream/low_level_e2e.cpp | 50 +-
foreign/cpp/tests/topic/low_level_e2e.cpp | 61 +-
16 files changed, 1788 insertions(+), 112 deletions(-)
diff --git a/bdd/cpp/features/step_definitions/messaging_steps.cpp
b/bdd/cpp/features/step_definitions/messaging_steps.cpp
index e58592bca..df4afc7db 100644
--- a/bdd/cpp/features/step_definitions/messaging_steps.cpp
+++ b/bdd/cpp/features/step_definitions/messaging_steps.cpp
@@ -117,8 +117,9 @@ WHEN("^I send ([0-9]+) messages to stream ([0-9]+), topic
([0-9]+), partition ([
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (int index = 0; index < message_count; ++index) {
- iggy::ffi::IggyMessageToSend message =
-
iggy::ffi::make_message(bdd::to_payload(bdd::expected_payload(static_cast<std::uint32_t>(index))));
+ iggy::ffi::IggyMessageToSend message = iggy::ffi::make_message(
+
bdd::to_payload(bdd::expected_payload(static_cast<std::uint32_t>(index))),
+ rust::Vec<iggy::ffi::HeaderEntry>());
// Assign an explicit, 1-based id so the last-sent/last-polled
comparison is meaningful.
message.id_lo = static_cast<std::uint64_t>(index + 1);
messages.push_back(std::move(message));
diff --git a/foreign/cpp/Cargo.toml b/foreign/cpp/Cargo.toml
index c5a9c1674..ca0e8ad6e 100644
--- a/foreign/cpp/Cargo.toml
+++ b/foreign/cpp/Cargo.toml
@@ -30,6 +30,7 @@ crate-type = ["staticlib"]
bytes = "1.12.0"
cxx = "1.0.194"
iggy = { path = "../../core/sdk" }
+iggy_binary_protocol = { path = "../../core/binary_protocol" }
iggy_common = { path = "../../core/common" }
# Explicitly enable the runtime + I/O drivers required by
`Runtime::enable_all()` in lib.rs.
# Listing the features here insulates this crate from upstream SDK feature
changes.
diff --git a/foreign/cpp/MODULE.bazel b/foreign/cpp/MODULE.bazel
index 8e0bdddb6..4e28a2851 100644
--- a/foreign/cpp/MODULE.bazel
+++ b/foreign/cpp/MODULE.bazel
@@ -25,13 +25,6 @@ bazel_dep(name = "platforms", version = "1.1.0")
bazel_dep(name = "googletest", version = "1.17.0.bcr.2")
bazel_dep(name = "rules_rust", version = "0.70.0")
-rust = use_extension("@rules_rust//rust:extensions.bzl", "rust")
-rust.toolchain(
- edition = "2024",
- versions = ["1.96.0"],
-)
-use_repo(rust, "rust_toolchains")
-
rust_host_tools = use_extension("@rules_rust//rust:extensions.bzl",
"rust_host_tools")
rust_host_tools.host_tools(
name = "rs_host_tools",
diff --git a/foreign/cpp/MODULE.bazel.lock b/foreign/cpp/MODULE.bazel.lock
index afa030326..f8d8ddc5c 100644
--- a/foreign/cpp/MODULE.bazel.lock
+++ b/foreign/cpp/MODULE.bazel.lock
@@ -487,7 +487,7 @@
},
"@@rules_rust+//crate_universe/private:internal_extensions.bzl%cu_nr": {
"general": {
- "bzlTransitiveDigest": "XJ4D0erEmT3eLY1MntKrXiRPz/VTypO8oNZjgaY/ff8=",
+ "bzlTransitiveDigest": "0UoEjDQrkPOE9U+EQx4lrz7qpp6evMn/oeanXvUNkJs=",
"usagesDigest": "ZmL90WEq2B6/NJ8rtHAqdnDPn+/9xG/GWR5K4UU4tyo=",
"recordedInputs": [
"REPO_MAPPING:bazel_features+,bazel_features_globals
bazel_features++version_extension+bazel_features_globals",
diff --git a/foreign/cpp/src/client.rs b/foreign/cpp/src/client.rs
index 0452518c7..4051d40fb 100644
--- a/foreign/cpp/src/client.rs
+++ b/foreign/cpp/src/client.rs
@@ -17,11 +17,12 @@
use crate::{RUNTIME, ffi};
use iggy::prelude::{
- Client as IggyConnectionClient, CompressionAlgorithm as
RustCompressionAlgorithm, Consumer,
- ConsumerGroupClient, ConsumerOffsetClient, Identifier as RustIdentifier,
- IggyClient as RustIggyClient, IggyClientBuilder as RustIggyClientBuilder,
- IggyExpiry as RustIggyExpiry, IggyMessage, IggyTimestamp, MaxTopicSize as
RustMaxTopicSize,
- MessageClient, PartitionClient, Partitioning, PollingStrategy,
+ Client as IggyConnectionClient, ClusterClient,
+ CompressionAlgorithm as RustCompressionAlgorithm, Consumer,
ConsumerGroupClient,
+ ConsumerOffsetClient, Identifier as RustIdentifier, IggyClient as
RustIggyClient,
+ IggyClientBuilder as RustIggyClientBuilder, IggyExpiry as RustIggyExpiry,
IggyMessage,
+ IggyTimestamp, MaxTopicSize as RustMaxTopicSize, MessageClient,
PartitionClient, Partitioning,
+ Permissions as RustPermissions, PollingStrategy, SegmentClient,
SnapshotCompression as RustSnapshotCompression, StreamClient, SystemClient
as RustSystemClient,
SystemSnapshotType as RustSystemSnapshotType, TopicClient, UserClient,
};
@@ -125,6 +126,26 @@ impl Client {
})
}
+ pub fn disconnect(&self) -> Result<(), String> {
+ RUNTIME.block_on(async {
+ self.inner
+ .disconnect()
+ .await
+ .map_err(|error| format!("Could not disconnect: {error}"))?;
+ Ok(())
+ })
+ }
+
+ pub fn shutdown(&self) -> Result<(), String> {
+ RUNTIME.block_on(async {
+ self.inner
+ .shutdown()
+ .await
+ .map_err(|error| format!("Could not shutdown client:
{error}"))?;
+ Ok(())
+ })
+ }
+
pub fn get_streams(&self) -> Result<Vec<ffi::Stream>, String> {
RUNTIME.block_on(async {
let streams = self
@@ -633,6 +654,33 @@ impl Client {
})
}
+ pub fn delete_segments(
+ &self,
+ stream_id: ffi::Identifier,
+ topic_id: ffi::Identifier,
+ partition_id: u32,
+ segments_count: u32,
+ ) -> Result<(), String> {
+ let rust_stream_id =
RustIdentifier::try_from(stream_id).map_err(|error| {
+ format!("Could not delete segments: invalid stream identifier:
{error}")
+ })?;
+ let rust_topic_id = RustIdentifier::try_from(topic_id).map_err(|error|
{
+ format!("Could not delete segments: invalid topic identifier:
{error}")
+ })?;
+
+ RUNTIME.block_on(async {
+ self.inner
+ .delete_segments(&rust_stream_id, &rust_topic_id,
partition_id, segments_count)
+ .await
+ .map_err(|error| {
+ format!(
+ "Could not delete {segments_count} segments for topic
'{rust_topic_id}' on stream '{rust_stream_id}', partition '{partition_id}':
{error}"
+ )
+ })?;
+ Ok(())
+ })
+ }
+
pub fn create_consumer_group(
&self,
stream_id: ffi::Identifier,
@@ -1024,6 +1072,63 @@ impl Client {
Ok(bytes)
})
}
+
+ pub fn get_cluster_metadata(&self) -> Result<ffi::ClusterMetadata, String>
{
+ RUNTIME.block_on(async {
+ let metadata = self
+ .inner
+ .get_cluster_metadata()
+ .await
+ .map_err(|error| format!("Could not get cluster metadata:
{error}"))?;
+ Ok(ffi::ClusterMetadata::from(metadata))
+ })
+ }
+
+ pub fn update_permissions(
+ &self,
+ user_id: ffi::Identifier,
+ has_permissions: bool,
+ permissions: ffi::Permissions,
+ ) -> Result<(), String> {
+ let rust_user_id = RustIdentifier::try_from(user_id).map_err(|error| {
+ format!("Could not update permissions: invalid user identifier:
{error}")
+ })?;
+ let rust_permissions = has_permissions
+ .then(|| RustPermissions::try_from(permissions))
+ .transpose()
+ .map_err(|error| format!("Could not update permissions:
{error}"))?;
+
+ RUNTIME.block_on(async {
+ self.inner
+ .update_permissions(&rust_user_id, rust_permissions)
+ .await
+ .map_err(|error| {
+ format!("Could not update permissions for user
'{rust_user_id}': {error}")
+ })?;
+ Ok(())
+ })
+ }
+
+ pub fn change_password(
+ &self,
+ user_id: ffi::Identifier,
+ current_password: String,
+ new_password: String,
+ ) -> Result<(), String> {
+ let rust_user_id = RustIdentifier::try_from(user_id).map_err(|error| {
+ format!("Could not change password: invalid user identifier:
{error}")
+ })?;
+
+ RUNTIME.block_on(async {
+ self.inner
+ .change_password(&rust_user_id, ¤t_password,
&new_password)
+ .await
+ .map_err(|error| {
+ format!("Could not change password for user
'{rust_user_id}': {error}")
+ })?;
+ Ok(())
+ })
+ }
}
pub unsafe fn delete_connection(client: *mut Client) -> Result<(), String> {
diff --git a/foreign/cpp/src/lib.rs b/foreign/cpp/src/lib.rs
index febcd9db5..458e280b6 100644
--- a/foreign/cpp/src/lib.rs
+++ b/foreign/cpp/src/lib.rs
@@ -88,11 +88,40 @@ mod ffi {
topics_count: u32,
}
+ #[repr(u8)]
+ enum HeaderKind {
+ Raw = 1,
+ String = 2,
+ Bool = 3,
+ Int8 = 4,
+ Int16 = 5,
+ Int32 = 6,
+ Int64 = 7,
+ Int128 = 8,
+ Uint8 = 9,
+ Uint16 = 10,
+ Uint32 = 11,
+ Uint64 = 12,
+ Uint128 = 13,
+ Float32 = 14,
+ Float64 = 15,
+ }
+
+ struct HeaderField {
+ kind: u8,
+ value: Vec<u8>,
+ }
+
+ struct HeaderEntry {
+ key: HeaderField,
+ value: HeaderField,
+ }
+
struct IggyMessageToSend {
id_lo: u64,
id_hi: u64,
payload: Vec<u8>,
- user_headers: Vec<u8>,
+ user_headers: Vec<HeaderEntry>,
}
struct IggyMessagePolled {
@@ -106,7 +135,7 @@ mod ffi {
payload_length: u32,
reserved: u64,
payload: Vec<u8>,
- user_headers: Vec<u8>,
+ user_headers: Vec<HeaderEntry>,
}
struct PolledMessages {
@@ -222,6 +251,71 @@ mod ffi {
total_disk_space: u64,
}
+ struct TransportEndpoints {
+ tcp: u16,
+ quic: u16,
+ http: u16,
+ websocket: u16,
+ }
+
+ struct ClusterNode {
+ name: String,
+ ip: String,
+ endpoints: TransportEndpoints,
+ role: String,
+ status: String,
+ }
+
+ struct ClusterMetadata {
+ name: String,
+ nodes: Vec<ClusterNode>,
+ }
+
+ struct GlobalPermissions {
+ manage_servers: bool,
+ read_servers: bool,
+ manage_users: bool,
+ read_users: bool,
+ manage_streams: bool,
+ read_streams: bool,
+ manage_topics: bool,
+ read_topics: bool,
+ poll_messages: bool,
+ send_messages: bool,
+ }
+
+ struct TopicPermissions {
+ manage_topic: bool,
+ read_topic: bool,
+ poll_messages: bool,
+ send_messages: bool,
+ }
+
+ struct TopicPermissionEntry {
+ topic_id: u32,
+ permissions: TopicPermissions,
+ }
+
+ struct StreamPermissions {
+ manage_stream: bool,
+ read_stream: bool,
+ manage_topics: bool,
+ read_topics: bool,
+ poll_messages: bool,
+ send_messages: bool,
+ topics: Vec<TopicPermissionEntry>,
+ }
+
+ struct StreamPermissionEntry {
+ stream_id: u32,
+ permissions: StreamPermissions,
+ }
+
+ struct Permissions {
+ global: GlobalPermissions,
+ streams: Vec<StreamPermissionEntry>,
+ }
+
extern "Rust" {
type Client;
type Consumer;
@@ -357,7 +451,7 @@ mod ffi {
auto_commit: bool,
) -> Result<PolledMessages>;
- fn make_message(payload: Vec<u8>) -> IggyMessageToSend;
+ fn make_message(payload: Vec<u8>, user_headers: Vec<HeaderEntry>) ->
IggyMessageToSend;
#[allow(clippy::too_many_arguments)]
fn send_messages(
@@ -388,22 +482,42 @@ mod ffi {
) -> Result<Vec<u8>>;
// Future functions
- // fn disconnect(self: &Client) -> Result<()>;
- // fn shutdown(self: &Client) -> Result<()>;
+ fn disconnect(self: &Client) -> Result<()>;
+ fn shutdown(self: &Client) -> Result<()>;
// fn subscribe_events(self: &Client) -> Result<()>;
- // fn delete_segments(self: &Client, stream_id: Identifier, topic_id:
Identifier, partition_id: u32, segments_count: u32) -> Result<()>;
+ fn delete_segments(
+ self: &Client,
+ stream_id: Identifier,
+ topic_id: Identifier,
+ partition_id: u32,
+ segments_count: u32,
+ ) -> Result<()>;
// fn get_user(self: &Client, user_id: Identifier) -> Result<()>;
// fn get_users(self: &Client) -> Result<()>;
// fn create_user(self: &Client, username: String, password: String,
status: u8) -> Result<()>;
// fn delete_user(self: &Client, user_id: Identifier) -> Result<()>;
// fn update_user(self: &Client, user_id: Identifier, username:
String, status: u8) -> Result<()>;
- // fn update_permissions(self: &Client, user_id: Identifier,
permissions: Vec<u8>) -> Result<()>;
- // fn change_password(self: &Client, user_id: Identifier,
current_password: String, new_password: String) -> Result<()>;
- // fn get_cluster_metadata(self: &Client) -> Result<()>;
- // fn get_personal_access_tokens(self: &Client) -> Result<()>;
- // fn create_personal_access_token(self: &Client, name: String,
expiry: u64) -> Result<()>;
+ fn update_permissions(
+ self: &Client,
+ user_id: Identifier,
+ has_permissions: bool,
+ permissions: Permissions,
+ ) -> Result<()>;
+ fn change_password(
+ self: &Client,
+ user_id: Identifier,
+ current_password: String,
+ new_password: String,
+ ) -> Result<()>;
+ fn get_cluster_metadata(self: &Client) -> Result<ClusterMetadata>;
+ // fn get_personal_access_tokens(self: &Client) ->
Result<Vec<PersonalAccessTokenInfo>>;
+ // fn create_personal_access_token(
+ // self: &Client,
+ // name: String,
+ // expiry: u64,
+ // ) -> Result<RawPersonalAccessToken>;
// fn delete_personal_access_token(self: &Client, name: String) ->
Result<()>;
- // fn login_with_personal_access_token(self: &Client, token: String)
-> Result<()>;
+ // fn login_with_personal_access_token(self: &Client, token: String)
-> Result<IdentityInfo>;
unsafe fn delete_client(client: *mut Client) -> Result<()>;
diff --git a/foreign/cpp/src/messages.rs b/foreign/cpp/src/messages.rs
index f39cdc1d4..13a4150de 100644
--- a/foreign/cpp/src/messages.rs
+++ b/foreign/cpp/src/messages.rs
@@ -17,11 +17,14 @@
use crate::ffi;
-pub fn make_message(payload: Vec<u8>) -> ffi::IggyMessageToSend {
+pub fn make_message(
+ payload: Vec<u8>,
+ user_headers: Vec<ffi::HeaderEntry>,
+) -> ffi::IggyMessageToSend {
ffi::IggyMessageToSend {
id_lo: 0,
id_hi: 0,
payload,
- user_headers: Vec::new(),
+ user_headers,
}
}
diff --git a/foreign/cpp/src/type_conversion.rs
b/foreign/cpp/src/type_conversion.rs
index dd38c1547..8a47f7c40 100644
--- a/foreign/cpp/src/type_conversion.rs
+++ b/foreign/cpp/src/type_conversion.rs
@@ -23,13 +23,19 @@ use iggy::prelude::{
PolledMessages as RustPolledMessages, Stream as RustStream, StreamDetails
as RustStreamDetails,
Topic as RustTopic, TopicDetails as RustTopicDetails, Validatable,
};
+use iggy_binary_protocol::WireUserHeaders;
use iggy_common::{
CacheMetrics as RustCacheMetrics, CacheMetricsKey as RustCacheMetricsKey,
ClientInfo as RustClientInfo, ClientInfoDetails as RustClientInfoDetails,
+ ClusterMetadata as RustClusterMetadata, ClusterNode as RustClusterNode,
ConsumerGroup as RustConsumerGroup, ConsumerGroupInfo as
RustConsumerGroupInfo,
ConsumerGroupMember as RustConsumerGroupMember, ConsumerOffsetInfo as
RustConsumerOffsetInfo,
- Stats as RustStats,
+ GlobalPermissions as RustGlobalPermissions, HeaderEntry as RustHeaderEntry,
+ HeaderField as RustHeaderField, HeaderKind as RustHeaderKind, Permissions
as RustPermissions,
+ Stats as RustStats, StreamPermissions as RustStreamPermissions,
+ TopicPermissions as RustTopicPermissions, TransportEndpoints as
RustTransportEndpoints,
};
+use std::collections::BTreeMap;
impl From<RustIdentifier> for ffi::Identifier {
fn from(identifier: RustIdentifier) -> Self {
@@ -188,6 +194,119 @@ impl From<RustStats> for ffi::Stats {
}
}
+impl From<RustTransportEndpoints> for ffi::TransportEndpoints {
+ fn from(endpoints: RustTransportEndpoints) -> Self {
+ ffi::TransportEndpoints {
+ tcp: endpoints.tcp,
+ quic: endpoints.quic,
+ http: endpoints.http,
+ websocket: endpoints.websocket,
+ }
+ }
+}
+
+impl From<RustClusterNode> for ffi::ClusterNode {
+ fn from(node: RustClusterNode) -> Self {
+ ffi::ClusterNode {
+ name: node.name,
+ ip: node.ip,
+ endpoints: ffi::TransportEndpoints::from(node.endpoints),
+ role: node.role.to_string(),
+ status: node.status.to_string(),
+ }
+ }
+}
+
+impl From<RustClusterMetadata> for ffi::ClusterMetadata {
+ fn from(metadata: RustClusterMetadata) -> Self {
+ ffi::ClusterMetadata {
+ name: metadata.name,
+ nodes: metadata
+ .nodes
+ .into_iter()
+ .map(ffi::ClusterNode::from)
+ .collect(),
+ }
+ }
+}
+
+impl From<ffi::GlobalPermissions> for RustGlobalPermissions {
+ fn from(permissions: ffi::GlobalPermissions) -> Self {
+ RustGlobalPermissions {
+ manage_servers: permissions.manage_servers,
+ read_servers: permissions.read_servers,
+ manage_users: permissions.manage_users,
+ read_users: permissions.read_users,
+ manage_streams: permissions.manage_streams,
+ read_streams: permissions.read_streams,
+ manage_topics: permissions.manage_topics,
+ read_topics: permissions.read_topics,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ }
+ }
+}
+
+impl From<ffi::TopicPermissions> for RustTopicPermissions {
+ fn from(permissions: ffi::TopicPermissions) -> Self {
+ RustTopicPermissions {
+ manage_topic: permissions.manage_topic,
+ read_topic: permissions.read_topic,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ }
+ }
+}
+
+impl TryFrom<ffi::StreamPermissions> for RustStreamPermissions {
+ type Error = String;
+
+ fn try_from(permissions: ffi::StreamPermissions) -> Result<Self,
Self::Error> {
+ let mut topics = BTreeMap::new();
+ for entry in permissions.topics {
+ let topic_id = entry.topic_id as usize;
+ if topics
+ .insert(topic_id,
RustTopicPermissions::from(entry.permissions))
+ .is_some()
+ {
+ return Err(format!("duplicate topic permission ID:
{topic_id}"));
+ }
+ }
+ let topics = (!topics.is_empty()).then_some(topics);
+
+ Ok(RustStreamPermissions {
+ manage_stream: permissions.manage_stream,
+ read_stream: permissions.read_stream,
+ manage_topics: permissions.manage_topics,
+ read_topics: permissions.read_topics,
+ poll_messages: permissions.poll_messages,
+ send_messages: permissions.send_messages,
+ topics,
+ })
+ }
+}
+
+impl TryFrom<ffi::Permissions> for RustPermissions {
+ type Error = String;
+
+ fn try_from(permissions: ffi::Permissions) -> Result<Self, Self::Error> {
+ let mut streams = BTreeMap::new();
+ for entry in permissions.streams {
+ let stream_id = entry.stream_id as usize;
+ let stream_permissions =
RustStreamPermissions::try_from(entry.permissions)?;
+ if streams.insert(stream_id, stream_permissions).is_some() {
+ return Err(format!("duplicate stream permission ID:
{stream_id}"));
+ }
+ }
+ let streams = (!streams.is_empty()).then_some(streams);
+
+ Ok(RustPermissions {
+ global: RustGlobalPermissions::from(permissions.global),
+ streams,
+ })
+ }
+}
+
impl From<RustPartition> for ffi::Partition {
fn from(partition: RustPartition) -> Self {
ffi::Partition {
@@ -319,6 +438,32 @@ impl From<RustIggyMessage> for ffi::IggyMessagePolled {
let id_bytes = message.header.id.to_le_bytes();
let id_lo = u64::from_le_bytes(id_bytes[0..8].try_into().unwrap());
let id_hi = u64::from_le_bytes(id_bytes[8..16].try_into().unwrap());
+ let user_headers = match message.user_headers {
+ Some(raw_headers) => {
+ // Keep polling forward-compatible with future header kinds.
Unlike the Rust SDK's
+ // typed decoder, this structural decoder preserves unknown
kinds and values whose
+ // lengths do not match their fixed-width kind.
+ match WireUserHeaders::from_bytes(raw_headers) {
+ Ok(wire_headers) => wire_headers
+ .iter()
+ .map(|entry| ffi::HeaderEntry {
+ key: ffi::HeaderField {
+ kind: entry.key_kind.0,
+ value: entry.key.to_vec(),
+ },
+ value: ffi::HeaderField {
+ kind: entry.value_kind.0,
+ value: entry.value.to_vec(),
+ },
+ })
+ .collect(),
+ // A malformed header must not make its message or poll
batch unreadable.
+ Err(_) => Vec::new(),
+ }
+ }
+ None => Vec::new(),
+ };
+
ffi::IggyMessagePolled {
checksum: message.header.checksum,
id_lo,
@@ -330,29 +475,155 @@ impl From<RustIggyMessage> for ffi::IggyMessagePolled {
payload_length: message.header.payload_length,
reserved: message.header.reserved,
payload: message.payload.to_vec(),
- user_headers: message
- .user_headers
- .map(|headers| headers.to_vec())
- .unwrap_or_default(),
+ user_headers,
}
}
}
+impl TryFrom<ffi::HeaderEntry> for RustHeaderEntry {
+ type Error = String;
+
+ fn try_from(entry: ffi::HeaderEntry) -> Result<Self, Self::Error> {
+ Ok(RustHeaderEntry {
+ key: decode_field(entry.key.kind, entry.key.value)?,
+ value: decode_field(entry.value.kind, entry.value.value)?,
+ })
+ }
+}
+
+fn decode_field<T>(kind: u8, value: Vec<u8>) -> Result<RustHeaderField<T>,
String> {
+ let kind = RustHeaderKind::from_code(kind)
+ .map_err(|error| format!("Could not convert header field: {error}"))?;
+
+ match kind {
+ RustHeaderKind::Raw => RustHeaderField::try_from(value)
+ .map_err(|error| format!("Could not convert header field:
{error}")),
+ RustHeaderKind::String => {
+ let value = String::from_utf8(value)
+ .map_err(|_| "Could not convert header field: invalid UTF-8
string".to_string())?;
+ RustHeaderField::try_from(value)
+ .map_err(|error| format!("Could not convert header field:
{error}"))
+ }
+ RustHeaderKind::Bool => match value.as_slice() {
+ [0] => Ok(RustHeaderField::from(false)),
+ [1] => Ok(RustHeaderField::from(true)),
+ _ => {
+ Err("Could not convert header field: bool values must encode
as 0 or 1".to_string())
+ }
+ },
+ RustHeaderKind::Int8 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(i8::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: int8 values require exactly 1
bytes, got {}",
+ value.len()
+ )),
+ },
+ RustHeaderKind::Int16 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(i16::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: int16 values require exactly
2 bytes, got {}",
+ value.len()
+ )),
+ },
+ RustHeaderKind::Int32 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(i32::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: int32 values require exactly
4 bytes, got {}",
+ value.len()
+ )),
+ },
+ RustHeaderKind::Int64 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(i64::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: int64 values require exactly
8 bytes, got {}",
+ value.len()
+ )),
+ },
+ RustHeaderKind::Int128 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(i128::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: int128 values require exactly
16 bytes, got {}",
+ value.len()
+ )),
+ },
+ RustHeaderKind::Uint8 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(u8::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: uint8 values require exactly
1 bytes, got {}",
+ value.len()
+ )),
+ },
+ RustHeaderKind::Uint16 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(u16::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: uint16 values require exactly
2 bytes, got {}",
+ value.len()
+ )),
+ },
+ RustHeaderKind::Uint32 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(u32::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: uint32 values require exactly
4 bytes, got {}",
+ value.len()
+ )),
+ },
+ RustHeaderKind::Uint64 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(u64::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: uint64 values require exactly
8 bytes, got {}",
+ value.len()
+ )),
+ },
+ RustHeaderKind::Uint128 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(u128::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: uint128 values require
exactly 16 bytes, got {}",
+ value.len()
+ )),
+ },
+ RustHeaderKind::Float32 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(f32::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: float32 values require
exactly 4 bytes, got {}",
+ value.len()
+ )),
+ },
+ RustHeaderKind::Float64 => match value.try_into() {
+ Ok(bytes) => Ok(RustHeaderField::from(f64::from_le_bytes(bytes))),
+ Err(value) => Err(format!(
+ "Could not convert header field: float64 values require
exactly 8 bytes, got {}",
+ value.len()
+ )),
+ },
+ }
+}
+
impl TryFrom<ffi::IggyMessageToSend> for RustIggyMessage {
type Error = String;
fn try_from(message: ffi::IggyMessageToSend) -> Result<Self, Self::Error> {
- if !message.user_headers.is_empty() {
- return Err(
- "Could not convert message: user_headers are not yet supported
in the C++ SDK"
- .to_string(),
- );
+ // TODO: Document in the C++ SDK that user headers are unordered and
keys must be unique.
+ // The BTreeMap sorts entries by kind and value, discards Vec
insertion order on send and
+ // poll, and rejects duplicate keys.
+ let mut user_headers = BTreeMap::new();
+ for entry in message.user_headers {
+ let header_entry = RustHeaderEntry::try_from(entry)
+ .map_err(|error| format!("Could not convert message user
headers: {error}"))?;
+ if user_headers
+ .insert(header_entry.key, header_entry.value)
+ .is_some()
+ {
+ return Err(
+ "Could not convert message user headers: duplicate header
key".to_string(),
+ );
+ }
}
let id = ((message.id_hi as u128) << 64) | (message.id_lo as u128);
let payload = Bytes::from(message.payload);
RustIggyMessage::builder()
.id(id)
.payload(payload)
+
.maybe_user_headers((!user_headers.is_empty()).then_some(user_headers))
.build()
.map_err(|error| format!("Could not convert message: {error}"))
}
diff --git a/foreign/cpp/tests/client/low_level_e2e.cpp
b/foreign/cpp/tests/client/low_level_e2e.cpp
index 177fdc855..590c8f597 100644
--- a/foreign/cpp/tests/client/low_level_e2e.cpp
+++ b/foreign/cpp/tests/client/low_level_e2e.cpp
@@ -19,6 +19,8 @@
// TODO(slbotbm): Add tests for store_consumer_offset, get_consumer_offset,
and delete_consumer_offset functions
// attached to client after implementing consumer group functions
+// TODO(slbotbm): Add tests for update_permissions after creating create_user,
get_user, etc. functions
+#include <cstddef>
#include <cstdint>
#include <string>
#include <unordered_set>
@@ -141,6 +143,87 @@ TEST_F(LowLevelE2E_Client,
LogoutErrorsWhenCalledMoreThanOnce) {
ASSERT_THROW(client->logout_user(), std::exception);
}
+TEST_F(LowLevelE2E_Client, ChangePasswordBeforeLoginThrows) {
+ RecordProperty("description",
+ "Rejects change_password before connect, after connect but
before login, and after disconnect.");
+ iggy::ffi::Client *client = GetLoggedOutClient();
+ const auto user_id = make_string_identifier("iggy");
+ const std::string old_password = "iggy";
+ const std::string new_password = "iggy-updated-secret";
+
+ ASSERT_THROW(client->change_password(user_id, old_password, new_password),
std::exception);
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_THROW(client->change_password(user_id, old_password, new_password),
std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->change_password(user_id, old_password, new_password),
std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, ChangePasswordWithInvalidCurrentPasswordThrows) {
+ RecordProperty("description", "Rejects change_password when the provided
current password is incorrect.");
+ iggy::ffi::Client *client = GetLoggedInClient();
+ const auto user_id = make_string_identifier("iggy");
+ const std::string wrong_password = "not-the-current-password";
+ const std::string new_password = "iggy-updated-secret";
+
+ ASSERT_THROW(client->change_password(user_id, wrong_password,
new_password), std::exception);
+ ASSERT_NO_THROW(client->logout_user());
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+}
+
+TEST_F(LowLevelE2E_Client, ChangePasswordWithInvalidNewPasswordThrows) {
+ RecordProperty("description",
+ "Rejects change_password when the replacement password
violates client-side length bounds.");
+ iggy::ffi::Client *client = GetLoggedInClient();
+ const auto user_id = make_string_identifier("iggy");
+ const std::string old_password = "iggy";
+ const std::string too_short = "";
+ const std::string too_long(256, 'a');
+
+ ASSERT_THROW(client->change_password(user_id, old_password, too_short),
std::exception);
+ ASSERT_THROW(client->change_password(user_id, old_password, too_long),
std::exception);
+ ASSERT_NO_THROW(client->logout_user());
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+}
+
+TEST_F(LowLevelE2E_Client, ChangePasswordForWrongUserThrows) {
+ RecordProperty("description", "Rejects change_password when targeting a
user that does not exist.");
+ iggy::ffi::Client *client = GetLoggedInClient();
+ const auto wrong_user_id =
make_string_identifier(GetRandomName());
+ const std::string current_password = "iggy";
+ const std::string replacement_secret = "iggy-updated-secret";
+
+ ASSERT_THROW(client->change_password(wrong_user_id, current_password,
replacement_secret), std::exception);
+ ASSERT_NO_THROW(client->logout_user());
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+}
+
+TEST_F(LowLevelE2E_Client, ChangePasswordUpdatesCredentialsAndCanBeRestored) {
+ RecordProperty("description",
+ "Changes the password for the current user, updates login
behavior, and restores the original "
+ "password before the test exits.");
+ iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *second_client = GetLoggedOutClient();
+ iggy::ffi::Client *third_client = GetLoggedOutClient();
+ const auto user_id = make_string_identifier("iggy");
+ const std::string old_password = "iggy";
+ const std::string new_password = "iggy-updated-secret";
+ bool password_changed = false;
+
+ ASSERT_NO_THROW(client->change_password(user_id, old_password,
new_password));
+ password_changed = true;
+
+ EXPECT_THROW(second_client->login_user("iggy", old_password),
std::exception);
+ EXPECT_NO_THROW(second_client->login_user("iggy", new_password));
+
+ if (password_changed) {
+ EXPECT_NO_THROW(client->change_password(user_id, new_password,
old_password));
+ password_changed = false;
+ }
+
+ EXPECT_NO_THROW(third_client->login_user("iggy", old_password));
+}
+
TEST_F(LowLevelE2E_Client, DeleteWhileUnauthenticatedAfterFailedLogin) {
RecordProperty("description", "Allows client cleanup after a failed login
leaves the connection unauthenticated.");
iggy::ffi::Client *client = nullptr;
@@ -153,6 +236,247 @@ TEST_F(LowLevelE2E_Client,
DeleteWhileUnauthenticatedAfterFailedLogin) {
client = nullptr;
}
+TEST_F(LowLevelE2E_Client, ConnectLoginThenDisconnect) {
+ RecordProperty("description",
+ "Connects, logs in, disconnects successfully, and rejects
authenticated operations afterward.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->get_me(), std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, DisconnectWithoutConnect) {
+ RecordProperty("description", "Allows disconnect to be called on a client
that was never explicitly connected.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->disconnect());
+}
+
+TEST_F(LowLevelE2E_Client, DisconnectWithoutLogin) {
+ RecordProperty("description", "Allows disconnect after connect even when
no user has authenticated.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->get_stats(), std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, DisconnectThenReconnectWithoutRelogin) {
+ RecordProperty("description",
+ "Requires logging in again after a disconnect and reconnect
before authenticated operations work.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_THROW(client->get_me(), std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, DisconnectAfterFailedLogin) {
+ RecordProperty("description", "Allows disconnect after a failed login
attempt leaves the client unauthenticated.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_THROW(client->login_user("biggy", "biggy"), std::exception);
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->get_me(), std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, ConnectLoginThenShutdown) {
+ RecordProperty("description",
+ "Connects, logs in, shuts down successfully, and rejects
further operations afterward.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->ping());
+ ASSERT_NO_THROW(client->shutdown());
+ ASSERT_THROW(client->get_me(), std::exception);
+ ASSERT_THROW(client->get_stats(), std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, ShutdownWithoutConnect) {
+ RecordProperty("description", "Allows shutdown to be called on a client
that was never explicitly connected.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->shutdown());
+}
+
+TEST_F(LowLevelE2E_Client, ShutdownWithoutLogin) {
+ RecordProperty("description", "Allows shutdown after connect even when no
user has authenticated.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_NO_THROW(client->shutdown());
+ ASSERT_THROW(client->get_stats(), std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, ShutdownAfterFailedLogin) {
+ RecordProperty("description", "Allows shutdown after a failed login
attempt leaves the client unauthenticated.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_THROW(client->login_user("biggy", "biggy"), std::exception);
+ ASSERT_NO_THROW(client->shutdown());
+ ASSERT_THROW(client->get_me(), std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, RepeatedShutdownCallsHaveStableBehavior) {
+ RecordProperty("description", "Keeps repeated shutdown calls stable across
duplicate invocations.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->shutdown());
+ ASSERT_NO_THROW(client->shutdown());
+ ASSERT_THROW(client->get_me(), std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, ShutdownThenConnectThrows) {
+ RecordProperty("description", "Rejects reconnecting a client after
shutdown transitions it to a terminal state.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->shutdown());
+ ASSERT_THROW(client->connect(), std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, ShutdownThenLoginThrows) {
+ RecordProperty("description",
+ "Rejects logging in again after shutdown, even when login
would normally auto-connect.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->shutdown());
+ ASSERT_THROW(client->login_user("iggy", "iggy"), std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, GetClientsReflectsSessionRemovalAfterShutdown) {
+ RecordProperty("description",
+ "Removes a shut down authenticated session from subsequent
get_clients and get_client results.");
+ iggy::ffi::Client *first_client = GetLoggedInClient();
+ iggy::ffi::Client *second_client = GetLoggedInClient();
+
+ iggy::ffi::ClientInfoDetails first_me{};
+ rust::Vec<iggy::ffi::ClientInfo> clients_after_shutdown;
+ ASSERT_NO_THROW({ first_me = first_client->get_me(); });
+
+ ASSERT_NO_THROW(first_client->shutdown());
+ ASSERT_NO_THROW({ clients_after_shutdown = second_client->get_clients();
});
+
+ bool found_first = false;
+ for (const auto &client : clients_after_shutdown) {
+ if (client.client_id == first_me.client_id) {
+ found_first = true;
+ break;
+ }
+ }
+
+ EXPECT_FALSE(found_first);
+ ASSERT_THROW(second_client->get_client(first_me.client_id),
std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, GetClientsReflectsSessionRemovalAfterDisconnect) {
+ RecordProperty("description",
+ "Removes a disconnected authenticated session from
subsequent get_clients and get_client results.");
+ iggy::ffi::Client *first_client = GetLoggedInClient();
+ iggy::ffi::Client *second_client = GetLoggedInClient();
+
+ iggy::ffi::ClientInfoDetails first_me{};
+ rust::Vec<iggy::ffi::ClientInfo> clients_after_disconnect;
+ ASSERT_NO_THROW({ first_me = first_client->get_me(); });
+
+ ASSERT_NO_THROW(first_client->disconnect());
+ ASSERT_NO_THROW({ clients_after_disconnect = second_client->get_clients();
});
+
+ bool found_first = false;
+ for (const auto &client : clients_after_disconnect) {
+ if (client.client_id == first_me.client_id) {
+ found_first = true;
+ break;
+ }
+ }
+
+ EXPECT_FALSE(found_first);
+ ASSERT_THROW(second_client->get_client(first_me.client_id),
std::exception);
+}
+
+TEST_F(LowLevelE2E_Client,
GetClientsReflectsLoggedOutSessionAsUnauthenticated) {
+ RecordProperty("description",
+ "Keeps a logged out session visible in get_clients and
get_client, but marks it unauthenticated.");
+ iggy::ffi::Client *first_client = GetLoggedInClient();
+ iggy::ffi::Client *second_client = GetLoggedInClient();
+
+ iggy::ffi::ClientInfoDetails first_me{};
+ iggy::ffi::ClientInfoDetails logged_out_client{};
+ rust::Vec<iggy::ffi::ClientInfo> clients_after_logout;
+ ASSERT_NO_THROW({ first_me = first_client->get_me(); });
+
+ ASSERT_NO_THROW(first_client->logout_user());
+ ASSERT_NO_THROW({
+ clients_after_logout = second_client->get_clients();
+ logged_out_client = second_client->get_client(first_me.client_id);
+ });
+
+ bool found_first = false;
+ for (const auto &client : clients_after_logout) {
+ if (client.client_id != first_me.client_id) {
+ continue;
+ }
+
+ found_first = true;
+ EXPECT_FALSE(client.has_user_id);
+ EXPECT_EQ(static_cast<std::string>(client.address),
static_cast<std::string>(first_me.address));
+ EXPECT_EQ(static_cast<std::string>(client.transport),
static_cast<std::string>(first_me.transport));
+ break;
+ }
+
+ EXPECT_TRUE(found_first);
+ EXPECT_EQ(logged_out_client.client_id, first_me.client_id);
+ EXPECT_FALSE(logged_out_client.has_user_id);
+ EXPECT_EQ(static_cast<std::string>(logged_out_client.address),
static_cast<std::string>(first_me.address));
+ EXPECT_EQ(static_cast<std::string>(logged_out_client.transport),
static_cast<std::string>(first_me.transport));
+}
+
TEST_F(LowLevelE2E_Client, LoginWithoutConnect) {
RecordProperty("description", "Supports login without an explicit prior
connect call.");
iggy::ffi::Client *client = nullptr;
@@ -174,6 +498,18 @@ TEST_F(LowLevelE2E_Client, ConnectWithoutLoginThenDelete) {
client = nullptr;
}
+TEST_F(LowLevelE2E_Client, DeleteWithoutDisconnect) {
+ RecordProperty("description", "Allows deleting a connected and
authenticated client without disconnecting first.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(iggy::ffi::delete_client(client));
+ client = nullptr;
+}
+
TEST_F(LowLevelE2E_Client, RepeatedClientMethodCallsHaveStableBehavior) {
RecordProperty("description",
"Keeps repeated connect, login, and delete calls stable
across duplicate invocations.");
@@ -191,6 +527,20 @@ TEST_F(LowLevelE2E_Client,
RepeatedClientMethodCallsHaveStableBehavior) {
ASSERT_NO_THROW(iggy::ffi::delete_client(client));
}
+TEST_F(LowLevelE2E_Client, RepeatedDisconnectCallsHaveStableBehavior) {
+ RecordProperty("description", "Keeps repeated disconnect calls stable
across duplicate invocations.");
+ iggy::ffi::Client *client = nullptr;
+ ASSERT_NO_THROW({ client = iggy::ffi::new_connection(""); });
+ ASSERT_NE(client, nullptr);
+ TrackClient(client);
+
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->get_me(), std::exception);
+}
+
TEST_F(LowLevelE2E_Client, DeleteNullConnectionIsNoop) {
RecordProperty("description", "Treats deleting a null client pointer as a
no-op.");
iggy::ffi::Client *client = nullptr;
@@ -198,12 +548,16 @@ TEST_F(LowLevelE2E_Client, DeleteNullConnectionIsNoop) {
}
TEST_F(LowLevelE2E_Client, GetStatsBeforeLoginThrows) {
- RecordProperty("description", "Rejects get_stats before connect, and after
connect but before login.");
+ RecordProperty("description",
+ "Rejects get_stats before connect, after connect but before
login, and after disconnect.");
iggy::ffi::Client *client = GetLoggedOutClient();
ASSERT_THROW(client->get_stats(), std::exception);
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->get_stats(), std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->get_stats(), std::exception);
}
TEST_F(LowLevelE2E_Client, FlushUnsavedBufferSucceedsForExistingPartition) {
@@ -220,7 +574,7 @@ TEST_F(LowLevelE2E_Client,
FlushUnsavedBufferSucceedsForExistingPartition) {
0, "server_default"));
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- messages.push_back(iggy::ffi::make_message(to_payload("flush-me")));
+ messages.push_back(iggy::ffi::make_message(to_payload("flush-me"),
rust::Vec<iggy::ffi::HeaderEntry>()));
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id),
make_numeric_identifier(0),
"partition_id",
partition_id_bytes(0), std::move(messages)));
@@ -247,7 +601,8 @@ TEST_F(LowLevelE2E_Client,
FlushUnsavedBufferSucceedsForExistingEmptyPartition)
TEST_F(LowLevelE2E_Client, FlushUnsavedBufferBeforeLoginThrows) {
RecordProperty("description",
- "Throws when flush_unsaved_buffer is called before connect,
and after connect but before login.");
+ "Throws when flush_unsaved_buffer is called before connect,
after connect but before login, and "
+ "after disconnect.");
iggy::ffi::Client *client = GetLoggedOutClient();
ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(1),
make_numeric_identifier(1), 0, true),
@@ -255,6 +610,10 @@ TEST_F(LowLevelE2E_Client,
FlushUnsavedBufferBeforeLoginThrows) {
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(1),
make_numeric_identifier(1), 0, true),
std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(1),
make_numeric_identifier(1), 0, true),
+ std::exception);
}
TEST_F(LowLevelE2E_Client, FlushUnsavedBufferOnNonExistentStreamThrows) {
@@ -341,7 +700,7 @@ TEST_F(LowLevelE2E_Client, FlushUnsavedBufferTwiceSucceeds)
{
0, "server_default"));
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- messages.push_back(iggy::ffi::make_message(to_payload("flush-twice")));
+ messages.push_back(iggy::ffi::make_message(to_payload("flush-twice"),
rust::Vec<iggy::ffi::HeaderEntry>()));
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id),
make_numeric_identifier(0),
"partition_id",
partition_id_bytes(0), std::move(messages)));
@@ -372,6 +731,245 @@ TEST_F(LowLevelE2E_Client,
FlushUnsavedBufferWithInvalidPartitionIdsThrows) {
}
}
+TEST_F(LowLevelE2E_Client, DeleteSegmentsBeforeLoginThrows) {
+ RecordProperty("description",
+ "Rejects delete_segments before connect, after connect but
before login, and after disconnect.");
+ const std::string stream_name = GetRandomName();
+ const std::string topic_name = GetRandomName();
+ iggy::ffi::Client *setup_client = GetLoggedInClient();
+
+ ASSERT_NO_THROW(setup_client->create_stream(stream_name));
+ TrackStream(stream_name);
+
ASSERT_NO_THROW(setup_client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", 0,
+ "never_expire", 0,
"server_default"));
+
+ iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
+
ASSERT_THROW(unauthenticated_client->delete_segments(make_string_identifier(stream_name),
+
make_string_identifier(topic_name), 0, 1),
+ std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->connect());
+
ASSERT_THROW(unauthenticated_client->delete_segments(make_string_identifier(stream_name),
+
make_string_identifier(topic_name), 0, 1),
+ std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+
ASSERT_THROW(unauthenticated_client->delete_segments(make_string_identifier(stream_name),
+
make_string_identifier(topic_name), 0, 1),
+ std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, DeleteSegmentsOnNonExistentStreamThrows) {
+ RecordProperty("description", "Throws when deleting segments from a stream
that does not exist.");
+ const std::string stream_name = GetRandomName();
+ const std::string topic_name = GetRandomName();
+ const std::string missing_stream_name = GetRandomName();
+ iggy::ffi::Client *client = GetLoggedInClient();
+
+ ASSERT_NO_THROW(client->create_stream(stream_name));
+ TrackStream(stream_name);
+ ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", 0, "never_expire",
+ 0, "server_default"));
+
+ ASSERT_THROW(
+ client->delete_segments(make_string_identifier(missing_stream_name),
make_string_identifier(topic_name), 0, 1),
+ std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, DeleteSegmentsOnNonExistentTopicThrows) {
+ RecordProperty("description", "Throws when deleting segments from a topic
that does not exist.");
+ const std::string stream_name = GetRandomName();
+ const std::string topic_name = GetRandomName();
+ const std::string missing_topic_name = GetRandomName();
+ iggy::ffi::Client *client = GetLoggedInClient();
+
+ ASSERT_NO_THROW(client->create_stream(stream_name));
+ TrackStream(stream_name);
+ ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", 0, "never_expire",
+ 0, "server_default"));
+
+ ASSERT_THROW(
+ client->delete_segments(make_string_identifier(stream_name),
make_string_identifier(missing_topic_name), 0, 1),
+ std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, DeleteSegmentsOnNonExistentPartitionThrows) {
+ RecordProperty("description", "Throws when deleting segments from a
partition that does not exist.");
+ const std::string stream_name = GetRandomName();
+ const std::string topic_name = GetRandomName();
+ iggy::ffi::Client *client = GetLoggedInClient();
+
+ ASSERT_NO_THROW(client->create_stream(stream_name));
+ TrackStream(stream_name);
+ ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", 0, "never_expire",
+ 0, "server_default"));
+
+ ASSERT_THROW(
+ client->delete_segments(make_string_identifier(stream_name),
make_string_identifier(topic_name), 999, 1),
+ std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, DeleteSegmentsWithZeroCountIsNoOp) {
+ RecordProperty("description", "Treats delete_segments with count 0 as a
no-op.");
+ const std::string stream_name = GetRandomName();
+ const std::string topic_name = GetRandomName();
+ iggy::ffi::Client *client = GetLoggedInClient();
+
+ ASSERT_NO_THROW(client->create_stream(stream_name));
+ TrackStream(stream_name);
+ ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", 0, "never_expire",
+ 0, "server_default"));
+
+ std::uint32_t stream_id = 0;
+ std::uint32_t topic_id = 0;
+ ASSERT_NO_THROW({
+ const auto stream_details =
client->get_stream(make_string_identifier(stream_name));
+ ASSERT_EQ(stream_details.topics.size(), 1u);
+ stream_id = stream_details.id;
+ topic_id = stream_details.topics.front().id;
+ });
+
+ rust::Vec<iggy::ffi::IggyMessageToSend> messages;
+ for (std::uint32_t i = 0; i < 5; ++i) {
+ messages.push_back(iggy::ffi::make_message(to_payload("zero-count-" +
std::to_string(i)),
+
rust::Vec<iggy::ffi::HeaderEntry>()));
+ }
+ ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_id),
make_numeric_identifier(topic_id),
+ "partition_id",
partition_id_bytes(0), std::move(messages)));
+
+ iggy::ffi::Partition partition_before_delete{};
+ ASSERT_NO_THROW({
+ const auto topic_details =
+ client->get_topic(make_numeric_identifier(stream_id),
make_numeric_identifier(topic_id));
+ for (const auto &partition : topic_details.partitions) {
+ if (partition.id == 0) {
+ partition_before_delete = partition;
+ break;
+ }
+ }
+ });
+
+ iggy::ffi::PolledMessages polled_before_delete{};
+ ASSERT_NO_THROW({
+ polled_before_delete =
+ client->poll_messages(make_numeric_identifier(stream_id),
make_numeric_identifier(topic_id), 0, "consumer",
+ make_numeric_identifier(1005), "offset", 0,
1000, false);
+ });
+
+ ASSERT_NO_THROW(
+ client->delete_segments(make_string_identifier(stream_name),
make_string_identifier(topic_name), 0, 0));
+
+ iggy::ffi::Partition partition_after_delete{};
+ ASSERT_NO_THROW({
+ const auto topic_details =
+ client->get_topic(make_numeric_identifier(stream_id),
make_numeric_identifier(topic_id));
+ for (const auto &partition : topic_details.partitions) {
+ if (partition.id == 0) {
+ partition_after_delete = partition;
+ break;
+ }
+ }
+ });
+
+ iggy::ffi::PolledMessages polled_after_delete{};
+ ASSERT_NO_THROW({
+ polled_after_delete =
+ client->poll_messages(make_numeric_identifier(stream_id),
make_numeric_identifier(topic_id), 0, "consumer",
+ make_numeric_identifier(1006), "offset", 0,
1000, false);
+ });
+
+ EXPECT_EQ(partition_after_delete.segments_count,
partition_before_delete.segments_count);
+ EXPECT_EQ(partition_after_delete.current_offset,
partition_before_delete.current_offset);
+ EXPECT_EQ(partition_after_delete.messages_count,
partition_before_delete.messages_count);
+ EXPECT_EQ(partition_after_delete.size_bytes,
partition_before_delete.size_bytes);
+ EXPECT_EQ(polled_after_delete.count, polled_before_delete.count);
+ ASSERT_EQ(polled_after_delete.messages.size(),
polled_before_delete.messages.size());
+ for (std::size_t i = 0; i < polled_before_delete.messages.size(); ++i) {
+ EXPECT_EQ(polled_after_delete.messages[i].offset,
polled_before_delete.messages[i].offset);
+ }
+}
+
+TEST_F(LowLevelE2E_Client, DeleteSegmentsWhenOnlyActiveSegmentRemainsIsNoOp) {
+ RecordProperty("description",
+ "Keeps the partition unchanged when delete_segments is
called with only the active segment.");
+ const std::string stream_name = GetRandomName();
+ const std::string topic_name = GetRandomName();
+ iggy::ffi::Client *client = GetLoggedInClient();
+
+ ASSERT_NO_THROW(client->create_stream(stream_name));
+ TrackStream(stream_name);
+ ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", 0, "never_expire",
+ 0, "server_default"));
+
+ std::uint32_t stream_id = 0;
+ std::uint32_t topic_id = 0;
+ ASSERT_NO_THROW({
+ const auto stream_details =
client->get_stream(make_string_identifier(stream_name));
+ ASSERT_EQ(stream_details.topics.size(), 1u);
+ stream_id = stream_details.id;
+ topic_id = stream_details.topics.front().id;
+ });
+
+ rust::Vec<iggy::ffi::IggyMessageToSend> messages;
+ for (std::uint32_t i = 0; i < 5; ++i) {
+ messages.push_back(iggy::ffi::make_message(to_payload("active-only-" +
std::to_string(i)),
+
rust::Vec<iggy::ffi::HeaderEntry>()));
+ }
+ ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_id),
make_numeric_identifier(topic_id),
+ "partition_id",
partition_id_bytes(0), std::move(messages)));
+
+ iggy::ffi::Partition partition_before_delete{};
+ ASSERT_NO_THROW({
+ const auto topic_details =
+ client->get_topic(make_numeric_identifier(stream_id),
make_numeric_identifier(topic_id));
+ for (const auto &partition : topic_details.partitions) {
+ if (partition.id == 0) {
+ partition_before_delete = partition;
+ break;
+ }
+ }
+ });
+ ASSERT_EQ(partition_before_delete.segments_count, 1u);
+
+ iggy::ffi::PolledMessages polled_before_delete{};
+ ASSERT_NO_THROW({
+ polled_before_delete =
+ client->poll_messages(make_numeric_identifier(stream_id),
make_numeric_identifier(topic_id), 0, "consumer",
+ make_numeric_identifier(1007), "offset", 0,
1000, false);
+ });
+
+ ASSERT_NO_THROW(
+ client->delete_segments(make_string_identifier(stream_name),
make_string_identifier(topic_name), 0, 1));
+
+ iggy::ffi::Partition partition_after_delete{};
+ ASSERT_NO_THROW({
+ const auto topic_details =
+ client->get_topic(make_numeric_identifier(stream_id),
make_numeric_identifier(topic_id));
+ for (const auto &partition : topic_details.partitions) {
+ if (partition.id == 0) {
+ partition_after_delete = partition;
+ break;
+ }
+ }
+ });
+
+ iggy::ffi::PolledMessages polled_after_delete{};
+ ASSERT_NO_THROW({
+ polled_after_delete =
+ client->poll_messages(make_numeric_identifier(stream_id),
make_numeric_identifier(topic_id), 0, "consumer",
+ make_numeric_identifier(1008), "offset", 0,
1000, false);
+ });
+
+ EXPECT_EQ(partition_after_delete.segments_count,
partition_before_delete.segments_count);
+ EXPECT_EQ(partition_after_delete.current_offset,
partition_before_delete.current_offset);
+ EXPECT_EQ(partition_after_delete.messages_count,
partition_before_delete.messages_count);
+ EXPECT_EQ(partition_after_delete.size_bytes,
partition_before_delete.size_bytes);
+ EXPECT_EQ(polled_after_delete.count, polled_before_delete.count);
+ ASSERT_EQ(polled_after_delete.messages.size(),
polled_before_delete.messages.size());
+ for (std::size_t i = 0; i < polled_before_delete.messages.size(); ++i) {
+ EXPECT_EQ(polled_after_delete.messages[i].offset,
polled_before_delete.messages[i].offset);
+ }
+}
+
// TODO(slbotbm): add a test to create some streams, topics, partitions, and
segments, send messages, and create
// consumer groups and verify it.
TEST_F(LowLevelE2E_Client, GetStatsReturnsServerStats) {
@@ -529,12 +1127,16 @@ TEST_F(LowLevelE2E_Client,
GetStatsIsStableAcrossBackToBackCalls) {
}
TEST_F(LowLevelE2E_Client, GetMeBeforeLoginThrows) {
- RecordProperty("description", "Rejects get_me before connect, and after
connect but before login.");
+ RecordProperty("description",
+ "Rejects get_me before connect, after connect but before
login, and after disconnect.");
iggy::ffi::Client *client = GetLoggedOutClient();
ASSERT_THROW(client->get_me(), std::exception);
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->get_me(), std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->get_me(), std::exception);
}
TEST_F(LowLevelE2E_Client, GetMeReturnsCurrentClientDetails) {
@@ -699,12 +1301,16 @@ TEST_F(LowLevelE2E_Client,
GetMeReturnsValidDetailsAfterReconnect) {
}
TEST_F(LowLevelE2E_Client, GetClientBeforeLoginThrows) {
- RecordProperty("description", "Rejects get_client before connect, and
after connect but before login.");
+ RecordProperty("description",
+ "Rejects get_client before connect, after connect but
before login, and after disconnect.");
iggy::ffi::Client *client = GetLoggedOutClient();
ASSERT_THROW(client->get_client(1), std::exception);
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->get_client(1), std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->get_client(1), std::exception);
}
TEST_F(LowLevelE2E_Client, GetClientWithWrongClientIdThrows) {
@@ -786,12 +1392,16 @@ TEST_F(LowLevelE2E_Client,
GetClientIsStableAcrossBackToBackCalls) {
}
TEST_F(LowLevelE2E_Client, GetClientsBeforeLoginThrows) {
- RecordProperty("description", "Rejects get_clients before connect, and
after connect but before login.");
+ RecordProperty("description",
+ "Rejects get_clients before connect, after connect but
before login, and after disconnect.");
iggy::ffi::Client *client = GetLoggedOutClient();
ASSERT_THROW(client->get_clients(), std::exception);
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->get_clients(), std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->get_clients(), std::exception);
}
TEST_F(LowLevelE2E_Client, GetClientsReturnsActiveClientSessions) {
@@ -949,6 +1559,67 @@ TEST_F(LowLevelE2E_Client,
GetClientsReflectsAdditionalSession) {
EXPECT_TRUE(found_after);
}
+TEST_F(LowLevelE2E_Client, GetClusterMetadataBeforeLoginThrows) {
+ RecordProperty(
+ "description",
+ "Rejects get_cluster_metadata before connect, after connect but before
login, and after disconnect.");
+ iggy::ffi::Client *client = GetLoggedOutClient();
+
+ ASSERT_THROW(client->get_cluster_metadata(), std::exception);
+ ASSERT_NO_THROW(client->connect());
+ ASSERT_THROW(client->get_cluster_metadata(), std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->get_cluster_metadata(), std::exception);
+}
+
+TEST_F(LowLevelE2E_Client, GetClusterMetadataReturnsSingleNodeMetadata) {
+ RecordProperty("description",
+ "Returns the expected single-node cluster metadata shape
from the default test server.");
+ iggy::ffi::Client *client = GetLoggedInClient();
+
+ iggy::ffi::ClusterMetadata metadata{};
+ ASSERT_NO_THROW({ metadata = client->get_cluster_metadata(); });
+
+ EXPECT_EQ(static_cast<std::string>(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);
+}
+
+TEST_F(LowLevelE2E_Client, GetClusterMetadataIsStableAcrossBackToBackCalls) {
+ RecordProperty("description", "Returns stable single-node cluster metadata
across back-to-back calls.");
+ iggy::ffi::Client *client = GetLoggedInClient();
+
+ 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();
+ });
+
+ 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);
+
+ 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);
+}
+
TEST_F(LowLevelE2E_Client, PingSucceedsForNewConnection) {
RecordProperty("description", "Successfully pings the server from a fresh
unauthenticated client session.");
iggy::ffi::Client *client = GetLoggedOutClient();
@@ -980,13 +1651,17 @@ TEST_F(LowLevelE2E_Client,
HeartbeatIntervalReturnsConfiguredValueFromConnection
}
TEST_F(LowLevelE2E_Client, SnapshotBeforeLoginThrows) {
- RecordProperty("description", "Rejects snapshot before connect, and after
connect but before login.");
+ RecordProperty("description",
+ "Rejects snapshot before connect, after connect but before
login, and after disconnect.");
iggy::ffi::Client *client = GetLoggedOutClient();
ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"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"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"test"})),
std::exception);
}
TEST_F(LowLevelE2E_Client, SnapshotAllCombinedWithOtherTypeThrows) {
diff --git a/foreign/cpp/tests/common/test_helpers.hpp
b/foreign/cpp/tests/common/test_helpers.hpp
index a6f240998..de81c9c7b 100644
--- a/foreign/cpp/tests/common/test_helpers.hpp
+++ b/foreign/cpp/tests/common/test_helpers.hpp
@@ -24,6 +24,7 @@
#include <initializer_list>
#include <random>
#include <string>
+#include <utility>
#include <gtest/gtest.h>
@@ -66,6 +67,38 @@ inline rust::Vec<rust::String>
make_snapshot_types(std::initializer_list<const c
return snapshot_types;
}
+inline iggy::ffi::HeaderField make_header_field(const iggy::ffi::HeaderKind
kind, rust::Vec<std::uint8_t> value) {
+ iggy::ffi::HeaderField field;
+ field.kind = static_cast<std::uint8_t>(kind);
+ field.value = std::move(value);
+ return field;
+}
+
+inline iggy::ffi::HeaderEntry make_header_entry(iggy::ffi::HeaderField key,
iggy::ffi::HeaderField value) {
+ iggy::ffi::HeaderEntry entry;
+ entry.key = std::move(key);
+ entry.value = std::move(value);
+ return entry;
+}
+
+inline bool has_header(const rust::Vec<iggy::ffi::HeaderEntry> &headers,
+ const std::uint8_t key_kind,
+ const rust::Vec<std::uint8_t> &key_value,
+ const std::uint8_t value_kind,
+ const rust::Vec<std::uint8_t> &value_value) {
+ for (const auto &header : headers) {
+ if (header.key.kind == key_kind && header.value.kind == value_kind &&
+ header.key.value.size() == key_value.size() &&
+ std::equal(header.key.value.begin(), header.key.value.end(),
key_value.begin()) &&
+ header.value.value.size() == value_value.size() &&
+ std::equal(header.value.value.begin(), header.value.value.end(),
value_value.begin())) {
+ return true;
+ }
+ }
+
+ return false;
+}
+
struct TrackedConsumerGroup {
std::string stream_name;
std::string topic_name;
diff --git a/foreign/cpp/tests/consumer_group/low_level_e2e.cpp
b/foreign/cpp/tests/consumer_group/low_level_e2e.cpp
index f48acbd19..e729d42ff 100644
--- a/foreign/cpp/tests/consumer_group/low_level_e2e.cpp
+++ b/foreign/cpp/tests/consumer_group/low_level_e2e.cpp
@@ -150,6 +150,11 @@ TEST_F(LowLevelE2E_ConsumerGroup,
CreateConsumerGroupBeforeLoginThrows) {
ASSERT_THROW(unauthenticated_client->create_consumer_group(make_string_identifier(stream_name),
make_string_identifier(topic_name), group_name),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+
ASSERT_THROW(unauthenticated_client->create_consumer_group(make_string_identifier(stream_name),
+
make_string_identifier(topic_name), group_name),
+ std::exception);
}
TEST_F(LowLevelE2E_ConsumerGroup,
GetConsumerGroupReturnsSameInfoAsCreateConsumerGroup) {
@@ -227,6 +232,11 @@ TEST_F(LowLevelE2E_ConsumerGroup,
GetConsumerGroupsBeforeLoginThrows) {
ASSERT_THROW(unauthenticated_client->get_consumer_groups(make_string_identifier(stream_name),
make_string_identifier(topic_name)),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+
ASSERT_THROW(unauthenticated_client->get_consumer_groups(make_string_identifier(stream_name),
+
make_string_identifier(topic_name)),
+ std::exception);
}
TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupSucceeds) {
@@ -274,6 +284,12 @@ TEST_F(LowLevelE2E_ConsumerGroup,
JoinConsumerGroupBeforeLoginThrows) {
make_string_identifier(topic_name),
make_string_identifier(group_name)),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+
ASSERT_THROW(unauthenticated_client->join_consumer_group(make_string_identifier(stream_name),
+
make_string_identifier(topic_name),
+
make_string_identifier(group_name)),
+ std::exception);
}
TEST_F(LowLevelE2E_ConsumerGroup,
JoinConsumerGroupOnNonExistentResourcesThrows) {
@@ -529,6 +545,12 @@ TEST_F(LowLevelE2E_ConsumerGroup,
LeaveConsumerGroupBeforeLoginThrows) {
make_string_identifier(topic_name),
make_string_identifier(group_name)),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+
ASSERT_THROW(unauthenticated_client->leave_consumer_group(make_string_identifier(stream_name),
+
make_string_identifier(topic_name),
+
make_string_identifier(group_name)),
+ std::exception);
}
TEST_F(LowLevelE2E_ConsumerGroup,
LeaveConsumerGroupOnNonExistentResourcesThrows) {
@@ -896,6 +918,12 @@ TEST_F(LowLevelE2E_ConsumerGroup,
GetConsumerGroupBeforeLoginThrows) {
make_string_identifier(topic_name),
make_string_identifier(group_name)),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+
ASSERT_THROW(unauthenticated_client->get_consumer_group(make_string_identifier(stream_name),
+
make_string_identifier(topic_name),
+
make_string_identifier(group_name)),
+ std::exception);
}
TEST_F(LowLevelE2E_ConsumerGroup,
GetConsumerGroupOnNonExistentResourcesThrows) {
@@ -1000,6 +1028,12 @@ TEST_F(LowLevelE2E_ConsumerGroup,
DeleteConsumerGroupBeforeLoginThrows) {
make_string_identifier(topic_name),
make_string_identifier(group_name)),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+
ASSERT_THROW(unauthenticated_client->delete_consumer_group(make_string_identifier(stream_name),
+
make_string_identifier(topic_name),
+
make_string_identifier(group_name)),
+ std::exception);
}
TEST_F(LowLevelE2E_ConsumerGroup,
DeleteConsumerGroupOnNonExistentResourcesThrows) {
diff --git a/foreign/cpp/tests/message/low_level_e2e.cpp
b/foreign/cpp/tests/message/low_level_e2e.cpp
index 76c29b7a4..5fca7dc29 100644
--- a/foreign/cpp/tests/message/low_level_e2e.cpp
+++ b/foreign/cpp/tests/message/low_level_e2e.cpp
@@ -42,7 +42,8 @@ TEST_F(LowLevelE2E_Message, SendAndPollMessagesRoundTrip) {
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)));
+ 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));
}
@@ -76,7 +77,7 @@ TEST_F(LowLevelE2E_Message, PollMessagesVerifyMessageIds) {
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- auto msg = iggy::ffi::make_message(to_payload("id-test-message"));
+ 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));
@@ -117,12 +118,22 @@ TEST_F(LowLevelE2E_Message,
SendMessagesBeforeLoginThrows) {
ASSERT_NO_THROW(client->connect());
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- auto msg = iggy::ffi::make_message(to_payload("should-fail"));
+ 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) {
@@ -130,7 +141,7 @@ TEST_F(LowLevelE2E_Message,
SendMessagesWithInvalidStreamId) {
iggy::ffi::Client *client = GetLoggedInClient();
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- auto msg = iggy::ffi::make_message(to_payload("test"));
+ 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;
@@ -147,7 +158,7 @@ TEST_F(LowLevelE2E_Message,
SendMessagesToNonExistentStream) {
iggy::ffi::Client *client = GetLoggedInClient();
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- auto msg = iggy::ffi::make_message(to_payload("test"));
+ 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),
@@ -168,7 +179,7 @@ TEST_F(LowLevelE2E_Message,
SendMessagesWithInvalidPartitioningKind) {
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- auto msg = iggy::ffi::make_message(to_payload("test"));
+ 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",
@@ -189,7 +200,7 @@ TEST_F(LowLevelE2E_Message,
SendMessagesWithInvalidPartitioningValue) {
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- auto msg = iggy::ffi::make_message(to_payload("test"));
+ 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;
@@ -216,7 +227,8 @@ TEST_F(LowLevelE2E_Message,
SendMessagesToSpecificPartitionVerified) {
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)));
+ 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));
}
@@ -267,7 +279,7 @@ TEST_F(LowLevelE2E_Message,
SendMessageWithEmptyPayloadThrows) {
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
rust::Vec<std::uint8_t> empty_payload;
- auto msg = iggy::ffi::make_message(std::move(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",
@@ -296,7 +308,7 @@ TEST_F(LowLevelE2E_Message,
SendMessageWithOversizedPayloadThrows) {
}
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- auto msg = iggy::ffi::make_message(std::move(oversized_payload));
+ 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",
@@ -318,7 +330,8 @@ TEST_F(LowLevelE2E_Message, SendMessagesPreservesOrder) {
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)));
+ auto msg =
+ iggy::ffi::make_message(to_payload("order-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
@@ -351,7 +364,8 @@ TEST_F(LowLevelE2E_Message, SendMessagesWithDuplicateIds) {
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)));
+ 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));
@@ -396,16 +410,16 @@ TEST_F(LowLevelE2E_Message,
SendMessagesWithVariousPayloads) {
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- auto msg0 = iggy::ffi::make_message(to_payload("simple ascii"));
+ 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));
+ 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"));
+ 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));
+ 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",
@@ -435,11 +449,365 @@ TEST_F(LowLevelE2E_Message,
SendMessagesWithVariousPayloads) {
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);
@@ -515,7 +883,7 @@ TEST_F(LowLevelE2E_Message,
PollMessagesCountLessThanAvailable) {
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)));
+ auto msg = iggy::ffi::make_message(to_payload("msg-" +
std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
@@ -543,7 +911,7 @@ TEST_F(LowLevelE2E_Message, PollMessagesWithLargeOffset) {
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)));
+ auto msg = iggy::ffi::make_message(to_payload("msg-" +
std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
@@ -571,7 +939,7 @@ TEST_F(LowLevelE2E_Message, PollMessagesFirstStrategy) {
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)));
+ auto msg = iggy::ffi::make_message(to_payload("msg-" +
std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
@@ -606,7 +974,7 @@ TEST_F(LowLevelE2E_Message, PollMessagesLastStrategy) {
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)));
+ auto msg = iggy::ffi::make_message(to_payload("msg-" +
std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
@@ -642,7 +1010,7 @@ TEST_F(LowLevelE2E_Message,
PollMessagesNextStrategyNoAutoCommit) {
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)));
+ auto msg = iggy::ffi::make_message(to_payload("msg-" +
std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
@@ -684,7 +1052,7 @@ TEST_F(LowLevelE2E_Message,
PollMessagesNextStrategyAutoCommit) {
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)));
+ auto msg = iggy::ffi::make_message(to_payload("msg-" +
std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
@@ -732,7 +1100,7 @@ TEST_F(LowLevelE2E_Message,
PollMessagesConsumerIdIndependence) {
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)));
+ auto msg = iggy::ffi::make_message(to_payload("msg-" +
std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
@@ -766,7 +1134,8 @@ TEST_F(LowLevelE2E_Message,
PollMessagesMultipleSendsThenPollOrder) {
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)));
+ 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",
@@ -774,7 +1143,8 @@ TEST_F(LowLevelE2E_Message,
PollMessagesMultipleSendsThenPollOrder) {
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)));
+ 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",
@@ -814,7 +1184,7 @@ TEST_F(LowLevelE2E_Message, PollMessagesMultipleCustomIds)
{
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)));
+ 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));
@@ -846,7 +1216,7 @@ TEST_F(LowLevelE2E_Message,
PollMessagesAfterStreamDeletedThrows) {
"server_default");
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- auto msg = iggy::ffi::make_message(to_payload("test"));
+ 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",
@@ -910,7 +1280,7 @@ TEST_F(LowLevelE2E_Message,
PollMessagesWithoutSpecifyingPartition) {
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)));
+ 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",
@@ -946,7 +1316,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesTimestampStrategy)
{
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)));
+ 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",
@@ -956,7 +1327,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesTimestampStrategy)
{
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)));
+ 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",
@@ -1007,7 +1379,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesMonotonicOffsets)
{
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)));
+ 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",
@@ -1045,7 +1418,8 @@ TEST_F(LowLevelE2E_Message, SendMessagesLargeBatch) {
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)));
+ 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));
}
@@ -1066,7 +1440,7 @@ TEST_F(LowLevelE2E_Message,
SendMessagesWithInvalidTopicIdThrows) {
iggy::ffi::Client *client = GetLoggedInClient();
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- auto msg = iggy::ffi::make_message(to_payload("test"));
+ 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;
@@ -1139,7 +1513,8 @@ TEST_F(LowLevelE2E_Message,
ConsumerGroupCreateJoinAndPollMessages) {
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)));
+ 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",
diff --git a/foreign/cpp/tests/message/unit_tests.cpp
b/foreign/cpp/tests/message/unit_tests.cpp
index 50dd87790..f441c55e0 100644
--- a/foreign/cpp/tests/message/unit_tests.cpp
+++ b/foreign/cpp/tests/message/unit_tests.cpp
@@ -32,7 +32,7 @@ TEST(MessageTest, MakeMessageSetsPayload) {
payload.push_back(static_cast<std::uint8_t>(c));
}
- auto msg = iggy::ffi::make_message(std::move(payload));
+ auto msg = iggy::ffi::make_message(std::move(payload),
rust::Vec<iggy::ffi::HeaderEntry>());
ASSERT_EQ(msg.payload.size(), text.size());
for (std::size_t i = 0; i < text.size(); i++) {
@@ -45,7 +45,7 @@ TEST(MessageTest, MakeMessageZerosIdAndHeaders) {
rust::Vec<std::uint8_t> payload;
payload.push_back(0x42);
- auto msg = iggy::ffi::make_message(std::move(payload));
+ auto msg = iggy::ffi::make_message(std::move(payload),
rust::Vec<iggy::ffi::HeaderEntry>());
EXPECT_EQ(msg.id_lo, 0u);
EXPECT_EQ(msg.id_hi, 0u);
@@ -56,7 +56,7 @@ TEST(MessageTest, MakeMessageWithEmptyPayload) {
RecordProperty("description", "Verifies make_message accepts an empty
payload.");
rust::Vec<std::uint8_t> empty_payload;
- auto msg = iggy::ffi::make_message(std::move(empty_payload));
+ auto msg = iggy::ffi::make_message(std::move(empty_payload),
rust::Vec<iggy::ffi::HeaderEntry>());
ASSERT_EQ(msg.payload.size(), 0u);
}
@@ -66,7 +66,7 @@ TEST(MessageTest, MakeMessageWithSingleByte) {
rust::Vec<std::uint8_t> payload;
payload.push_back(0xFF);
- auto msg = iggy::ffi::make_message(std::move(payload));
+ auto msg = iggy::ffi::make_message(std::move(payload),
rust::Vec<iggy::ffi::HeaderEntry>());
ASSERT_EQ(msg.payload.size(), 1u);
EXPECT_EQ(msg.payload[0], 0xFF);
@@ -79,7 +79,7 @@ TEST(MessageTest, MakeMessageWithNullBytes) {
payload.push_back(0x01);
payload.push_back(0x00);
- auto msg = iggy::ffi::make_message(std::move(payload));
+ auto msg = iggy::ffi::make_message(std::move(payload),
rust::Vec<iggy::ffi::HeaderEntry>());
ASSERT_EQ(msg.payload.size(), 3u);
EXPECT_EQ(msg.payload[0], 0x00);
@@ -91,7 +91,7 @@ TEST(MessageTest, MakeMessageThenSetCustomId) {
RecordProperty("description", "Verifies custom ID can be set after
make_message without affecting payload.");
rust::Vec<std::uint8_t> payload;
payload.push_back(0x42);
- auto msg = iggy::ffi::make_message(std::move(payload));
+ auto msg = iggy::ffi::make_message(std::move(payload),
rust::Vec<iggy::ffi::HeaderEntry>());
msg.id_lo = 100;
msg.id_hi = 200;
@@ -109,7 +109,7 @@ TEST(MessageTest, MakeMessageWithLargePayload) {
payload.push_back(static_cast<std::uint8_t>(i % 256));
}
- auto msg = iggy::ffi::make_message(std::move(payload));
+ auto msg = iggy::ffi::make_message(std::move(payload),
rust::Vec<iggy::ffi::HeaderEntry>());
ASSERT_EQ(msg.payload.size(), 10000u);
EXPECT_EQ(msg.payload[0], 0u);
diff --git a/foreign/cpp/tests/partition/low_level_e2e.cpp
b/foreign/cpp/tests/partition/low_level_e2e.cpp
index 517c2c679..8b444eb69 100644
--- a/foreign/cpp/tests/partition/low_level_e2e.cpp
+++ b/foreign/cpp/tests/partition/low_level_e2e.cpp
@@ -62,6 +62,10 @@ TEST_F(LowLevelE2E_Partition,
CreatePartitionsBeforeLoginThrows) {
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->create_partitions(make_string_identifier(stream_name),
make_string_identifier(topic_name), 1),
std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+
ASSERT_THROW(client->create_partitions(make_string_identifier(stream_name),
make_string_identifier(topic_name), 1),
+ std::exception);
}
TEST_F(LowLevelE2E_Partition, CreatePartitionsOnNonExistentResourcesThrows) {
@@ -356,6 +360,10 @@ TEST_F(LowLevelE2E_Partition,
DeletePartitionsBeforeLoginThrows) {
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->delete_partitions(make_string_identifier(stream_name),
make_string_identifier(topic_name), 1),
std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+
ASSERT_THROW(client->delete_partitions(make_string_identifier(stream_name),
make_string_identifier(topic_name), 1),
+ std::exception);
}
TEST_F(LowLevelE2E_Partition, DeletePartitionsOnNonExistentResourcesThrows) {
diff --git a/foreign/cpp/tests/stream/low_level_e2e.cpp
b/foreign/cpp/tests/stream/low_level_e2e.cpp
index 2cdd4fc52..2c3fad218 100644
--- a/foreign/cpp/tests/stream/low_level_e2e.cpp
+++ b/foreign/cpp/tests/stream/low_level_e2e.cpp
@@ -52,6 +52,9 @@ TEST_F(LowLevelE2E_Stream, CreateStreamBeforeLoginThrows) {
ASSERT_THROW(client->create_stream(stream_name), std::exception);
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->create_stream(stream_name), std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->create_stream(stream_name), std::exception);
}
TEST_F(LowLevelE2E_Stream, CreateStreamValidatesNameConstraintsAndUniqueness) {
@@ -160,6 +163,10 @@ TEST_F(LowLevelE2E_Stream, UpdateStreamBeforeLoginThrows) {
ASSERT_NO_THROW(unauthenticated_client->connect());
ASSERT_THROW(unauthenticated_client->update_stream(make_string_identifier(stream_name),
updated_stream_name),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+
ASSERT_THROW(unauthenticated_client->update_stream(make_string_identifier(stream_name),
updated_stream_name),
+ std::exception);
}
TEST_F(LowLevelE2E_Stream, UpdateStreamWithVariousUtf8Characters) {
@@ -287,7 +294,8 @@ TEST_F(LowLevelE2E_Stream, UpdateStreamOnlyChangesName) {
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 3; ++i) {
- auto message =
iggy::ffi::make_message(to_payload("stream-update-preserve-" +
std::to_string(i)));
+ auto message =
iggy::ffi::make_message(to_payload("stream-update-preserve-" +
std::to_string(i)),
+
rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(message));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_id),
make_numeric_identifier(0),
@@ -396,6 +404,9 @@ TEST_F(LowLevelE2E_Stream, DeleteStreamBeforeLoginThrows) {
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->delete_stream(make_string_identifier(stream_name)),
std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->delete_stream(make_string_identifier(stream_name)),
std::exception);
}
TEST_F(LowLevelE2E_Stream, DeleteStreamTwiceThrows) {
@@ -476,6 +487,9 @@ TEST_F(LowLevelE2E_Stream,
GetStreamDetailsBeforeLoginThrows) {
ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)),
std::exception);
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)),
std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)),
std::exception);
}
TEST_F(LowLevelE2E_Stream, GetDeletedStreamDetailsThrows) {
@@ -564,7 +578,8 @@ TEST_F(LowLevelE2E_Stream, GetStreamsFieldsVerification) {
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 5; i++) {
- auto msg = iggy::ffi::make_message(to_payload("field-verify-message-"
+ std::to_string(i)));
+ auto msg = iggy::ffi::make_message(to_payload("field-verify-message-"
+ 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",
@@ -592,6 +607,9 @@ TEST_F(LowLevelE2E_Stream, GetStreamsBeforeLoginThrows) {
ASSERT_THROW(client->get_streams(), std::exception);
ASSERT_NO_THROW(client->connect());
ASSERT_THROW(client->get_streams(), std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(client->disconnect());
+ ASSERT_THROW(client->get_streams(), std::exception);
}
TEST_F(LowLevelE2E_Stream, GetStreamsConsistentWithGetStream) {
@@ -716,7 +734,8 @@ TEST_F(LowLevelE2E_Stream,
PurgeStreamPreservesStreamMetadata) {
ASSERT_EQ(stream_before_purge.topics.size(), 2u);
rust::Vec<iggy::ffi::IggyMessageToSend> first_topic_messages;
-
first_topic_messages.push_back(iggy::ffi::make_message(to_payload("preserve-stream-metadata")));
+ first_topic_messages.push_back(
+ iggy::ffi::make_message(to_payload("preserve-stream-metadata"),
rust::Vec<iggy::ffi::HeaderEntry>()));
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_before_purge.id),
make_string_identifier(first_topic_name), "partition_id",
partition_id_bytes(0),
std::move(first_topic_messages)));
@@ -809,7 +828,8 @@ TEST_F(LowLevelE2E_Stream,
PurgeStreamRemovesMessagesAndPreservesTopics) {
rust::Vec<iggy::ffi::IggyMessageToSend> first_topic_messages;
for (std::uint32_t i = 0; i < 3; ++i) {
-
first_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-stream-first-"
+ std::to_string(i))));
+
first_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-stream-first-"
+ std::to_string(i)),
+
rust::Vec<iggy::ffi::HeaderEntry>()));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(first_topic_id), "partition_id",
@@ -817,8 +837,8 @@ TEST_F(LowLevelE2E_Stream,
PurgeStreamRemovesMessagesAndPreservesTopics) {
rust::Vec<iggy::ffi::IggyMessageToSend> second_topic_messages;
for (std::uint32_t i = 0; i < 2; ++i) {
- second_topic_messages.push_back(
- iggy::ffi::make_message(to_payload("purge-stream-second-" +
std::to_string(i))));
+
second_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-stream-second-"
+ std::to_string(i)),
+
rust::Vec<iggy::ffi::HeaderEntry>()));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(second_topic_id), "partition_id",
@@ -913,7 +933,8 @@ TEST_F(LowLevelE2E_Stream,
PurgeStreamAcrossMultipleTopicsAndPartitionsClearsEve
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 2; ++i) {
messages.push_back(iggy::ffi::make_message(
- to_payload("purge-stream-topic-a-" +
std::to_string(partition_id) + "-" + std::to_string(i))));
+ to_payload("purge-stream-topic-a-" +
std::to_string(partition_id) + "-" + std::to_string(i)),
+ rust::Vec<iggy::ffi::HeaderEntry>()));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(first_topic_id), "partition_id",
@@ -923,7 +944,8 @@ TEST_F(LowLevelE2E_Stream,
PurgeStreamAcrossMultipleTopicsAndPartitionsClearsEve
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 2; ++i) {
messages.push_back(iggy::ffi::make_message(
- to_payload("purge-stream-topic-b-" +
std::to_string(partition_id) + "-" + std::to_string(i))));
+ to_payload("purge-stream-topic-b-" +
std::to_string(partition_id) + "-" + std::to_string(i)),
+ rust::Vec<iggy::ffi::HeaderEntry>()));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(second_topic_id), "partition_id",
@@ -961,15 +983,15 @@ TEST_F(LowLevelE2E_Stream,
PurgeStreamThenSendMessagesAgainSucceeds) {
const std::uint32_t topic_id = created_stream.topics.front().id;
rust::Vec<iggy::ffi::IggyMessageToSend> first_batch;
- first_batch.push_back(iggy::ffi::make_message(to_payload("before-purge")));
+ first_batch.push_back(iggy::ffi::make_message(to_payload("before-purge"),
rust::Vec<iggy::ffi::HeaderEntry>()));
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(topic_id),
"partition_id",
partition_id_bytes(0), std::move(first_batch)));
ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));
rust::Vec<iggy::ffi::IggyMessageToSend> second_batch;
-
second_batch.push_back(iggy::ffi::make_message(to_payload("after-purge-0")));
-
second_batch.push_back(iggy::ffi::make_message(to_payload("after-purge-1")));
+
second_batch.push_back(iggy::ffi::make_message(to_payload("after-purge-0"),
rust::Vec<iggy::ffi::HeaderEntry>()));
+
second_batch.push_back(iggy::ffi::make_message(to_payload("after-purge-1"),
rust::Vec<iggy::ffi::HeaderEntry>()));
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(topic_id),
"partition_id",
partition_id_bytes(0), std::move(second_batch)));
@@ -998,7 +1020,8 @@ TEST_F(LowLevelE2E_Stream,
PurgeStreamTwiceKeepsStreamEmptyAndTopicsIntact) {
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 3; ++i) {
-
messages.push_back(iggy::ffi::make_message(to_payload("purge-stream-twice-" +
std::to_string(i))));
+
messages.push_back(iggy::ffi::make_message(to_payload("purge-stream-twice-" +
std::to_string(i)),
+
rust::Vec<iggy::ffi::HeaderEntry>()));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(topic_id),
"partition_id",
partition_id_bytes(0), std::move(messages)));
@@ -1035,4 +1058,7 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamBeforeLoginThrows) {
ASSERT_THROW(unauthenticated_client->purge_stream(make_string_identifier(stream_name)),
std::exception);
ASSERT_NO_THROW(unauthenticated_client->connect());
ASSERT_THROW(unauthenticated_client->purge_stream(make_string_identifier(stream_name)),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+
ASSERT_THROW(unauthenticated_client->purge_stream(make_string_identifier(stream_name)),
std::exception);
}
diff --git a/foreign/cpp/tests/topic/low_level_e2e.cpp
b/foreign/cpp/tests/topic/low_level_e2e.cpp
index 8a58e065c..82f442031 100644
--- a/foreign/cpp/tests/topic/low_level_e2e.cpp
+++ b/foreign/cpp/tests/topic/low_level_e2e.cpp
@@ -295,6 +295,11 @@ TEST_F(LowLevelE2E_Topic, CreateTopicBeforeLoginThrows) {
ASSERT_THROW(unauthenticated_client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", 0,
"server_default", 0,
"server_default"),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+
ASSERT_THROW(unauthenticated_client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", 0,
+ "server_default", 0,
"server_default"),
+ std::exception);
}
TEST_F(LowLevelE2E_Topic, DeleteTopicAfterCreate) {
@@ -398,6 +403,11 @@ TEST_F(LowLevelE2E_Topic, DeleteTopicBeforeLoginThrows) {
ASSERT_THROW(
unauthenticated_client->delete_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name)),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+ ASSERT_THROW(
+
unauthenticated_client->delete_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name)),
+ std::exception);
}
TEST_F(LowLevelE2E_Topic, DeleteTopicWithInvalidStreamIdentifierThrows) {
@@ -497,6 +507,11 @@ TEST_F(LowLevelE2E_Topic, GetTopicBeforeLoginThrows) {
ASSERT_THROW(
unauthenticated_client->get_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name)),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+ ASSERT_THROW(
+ unauthenticated_client->get_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name)),
+ std::exception);
}
TEST_F(LowLevelE2E_Topic, GetTopicWithWrongStreamIdThrows) {
@@ -730,6 +745,9 @@ TEST_F(LowLevelE2E_Topic, GetTopicsBeforeLoginThrows) {
ASSERT_THROW(unauthenticated_client->get_topics(make_string_identifier(stream_name)),
std::exception);
ASSERT_NO_THROW(unauthenticated_client->connect());
ASSERT_THROW(unauthenticated_client->get_topics(make_string_identifier(stream_name)),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+
ASSERT_THROW(unauthenticated_client->get_topics(make_string_identifier(stream_name)),
std::exception);
}
TEST_F(LowLevelE2E_Topic, GetTopicsReturnsEmptyForStreamWithoutTopics) {
@@ -875,7 +893,8 @@ TEST_F(LowLevelE2E_Topic, UpdateTopicDoesNotChangeMessages)
{
const auto topic_id = created_stream.topics.front().id;
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
-
messages.push_back(iggy::ffi::make_message(to_payload("message-before-topic-update")));
+ messages.push_back(
+ iggy::ffi::make_message(to_payload("message-before-topic-update"),
rust::Vec<iggy::ffi::HeaderEntry>()));
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(topic_id),
"partition_id",
partition_id_bytes(0), std::move(messages)));
@@ -1080,6 +1099,12 @@ TEST_F(LowLevelE2E_Topic, UpdateTopicBeforeLoginThrows) {
unauthenticated_client->update_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name),
updated_topic_name, "gzip", 1,
"duration", 1000, "1GiB"),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+ ASSERT_THROW(
+
unauthenticated_client->update_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name),
+ updated_topic_name, "gzip", 1,
"duration", 1000, "1GiB"),
+ std::exception);
}
TEST_F(LowLevelE2E_Topic, UpdateTopicOnNonExistentStreamThrows) {
@@ -1243,7 +1268,8 @@ TEST_F(LowLevelE2E_Topic,
PurgeTopicPreservesTopicMetadata) {
const auto topic_before_messages = stream_before_purge.topics.front();
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
-
messages.push_back(iggy::ffi::make_message(to_payload("preserve-topic-metadata")));
+ messages.push_back(
+ iggy::ffi::make_message(to_payload("preserve-topic-metadata"),
rust::Vec<iggy::ffi::HeaderEntry>()));
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_before_purge.id),
make_numeric_identifier(topic_before_messages.id), "partition_id",
partition_id_bytes(1),
std::move(messages)));
@@ -1307,7 +1333,8 @@ TEST_F(LowLevelE2E_Topic,
PurgeTopicRemovesOnlyTargetTopicMessages) {
rust::Vec<iggy::ffi::IggyMessageToSend> first_topic_messages;
for (std::uint32_t i = 0; i < 3; ++i) {
-
first_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-topic-first-"
+ std::to_string(i))));
+
first_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-topic-first-"
+ std::to_string(i)),
+
rust::Vec<iggy::ffi::HeaderEntry>()));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(first_topic_id), "partition_id",
@@ -1315,7 +1342,8 @@ TEST_F(LowLevelE2E_Topic,
PurgeTopicRemovesOnlyTargetTopicMessages) {
rust::Vec<iggy::ffi::IggyMessageToSend> second_topic_messages;
for (std::uint32_t i = 0; i < 2; ++i) {
-
second_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-topic-second-"
+ std::to_string(i))));
+
second_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-topic-second-"
+ std::to_string(i)),
+
rust::Vec<iggy::ffi::HeaderEntry>()));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(second_topic_id), "partition_id",
@@ -1373,7 +1401,8 @@ TEST_F(LowLevelE2E_Topic,
PurgeTopicAcrossMultiplePartitionsClearsAllPartitions)
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 2; ++i) {
messages.push_back(iggy::ffi::make_message(
- to_payload("purge-topic-partition-" +
std::to_string(partition_id) + "-" + std::to_string(i))));
+ to_payload("purge-topic-partition-" +
std::to_string(partition_id) + "-" + std::to_string(i)),
+ rust::Vec<iggy::ffi::HeaderEntry>()));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(topic_id), "partition_id",
@@ -1413,15 +1442,18 @@ TEST_F(LowLevelE2E_Topic,
PurgeTopicThenSendMessagesAgainSucceeds) {
const std::uint32_t topic_id = created_stream.topics.front().id;
rust::Vec<iggy::ffi::IggyMessageToSend> first_batch;
-
first_batch.push_back(iggy::ffi::make_message(to_payload("before-topic-purge")));
+ first_batch.push_back(
+ iggy::ffi::make_message(to_payload("before-topic-purge"),
rust::Vec<iggy::ffi::HeaderEntry>()));
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(topic_id),
"partition_id",
partition_id_bytes(0), std::move(first_batch)));
ASSERT_NO_THROW(client->purge_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name)));
rust::Vec<iggy::ffi::IggyMessageToSend> second_batch;
-
second_batch.push_back(iggy::ffi::make_message(to_payload("after-topic-purge-0")));
-
second_batch.push_back(iggy::ffi::make_message(to_payload("after-topic-purge-1")));
+ second_batch.push_back(
+ iggy::ffi::make_message(to_payload("after-topic-purge-0"),
rust::Vec<iggy::ffi::HeaderEntry>()));
+ second_batch.push_back(
+ iggy::ffi::make_message(to_payload("after-topic-purge-1"),
rust::Vec<iggy::ffi::HeaderEntry>()));
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(topic_id),
"partition_id",
partition_id_bytes(0), std::move(second_batch)));
@@ -1470,8 +1502,8 @@ TEST_F(LowLevelE2E_Topic,
PurgeTopicTwiceKeepsTargetTopicEmptyAndOtherTopicsUnto
rust::Vec<iggy::ffi::IggyMessageToSend> first_topic_messages;
for (std::uint32_t i = 0; i < 3; ++i) {
- first_topic_messages.push_back(
- iggy::ffi::make_message(to_payload("purge-topic-twice-first-" +
std::to_string(i))));
+ first_topic_messages.push_back(iggy::ffi::make_message(
+ to_payload("purge-topic-twice-first-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>()));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(first_topic_id), "partition_id",
@@ -1479,8 +1511,8 @@ TEST_F(LowLevelE2E_Topic,
PurgeTopicTwiceKeepsTargetTopicEmptyAndOtherTopicsUnto
rust::Vec<iggy::ffi::IggyMessageToSend> second_topic_messages;
for (std::uint32_t i = 0; i < 2; ++i) {
- second_topic_messages.push_back(
- iggy::ffi::make_message(to_payload("purge-topic-twice-second-" +
std::to_string(i))));
+ second_topic_messages.push_back(iggy::ffi::make_message(
+ to_payload("purge-topic-twice-second-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>()));
}
ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(second_topic_id), "partition_id",
@@ -1542,4 +1574,9 @@ TEST_F(LowLevelE2E_Topic, PurgeTopicBeforeLoginThrows) {
ASSERT_THROW(
unauthenticated_client->purge_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name)),
std::exception);
+ ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client->disconnect());
+ ASSERT_THROW(
+
unauthenticated_client->purge_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name)),
+ std::exception);
}
