This is an automated email from the ASF dual-hosted git repository. slbotbm pushed a commit to branch cpp-high-level-client-2 in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 803fec97f628441aa2fba769a5b22e48eede07f9 Author: Rimuksh Kansal <[email protected]> AuthorDate: Sat Sep 12 20:49:45 2026 +0900 make consumer and producer related functions and classes available to high level client, plus test translation --- foreign/cpp/include/iggy.hpp | 132 ++++ foreign/cpp/src/client.cpp | 74 ++ foreign/cpp/src/client.rs | 65 ++ foreign/cpp/src/consumer.cpp | 61 ++ foreign/cpp/src/consumer.rs | 17 + foreign/cpp/src/lib.rs | 28 +- foreign/cpp/src/producer.cpp | 61 ++ foreign/cpp/src/producer.rs | 16 + foreign/cpp/src/type_conversions.cpp | 21 + foreign/cpp/tests/e2e/consumer_group.cpp | 1272 +++++++++++++++--------------- 10 files changed, 1110 insertions(+), 637 deletions(-) diff --git a/foreign/cpp/include/iggy.hpp b/foreign/cpp/include/iggy.hpp index 230905fb5..397c44c32 100644 --- a/foreign/cpp/include/iggy.hpp +++ b/foreign/cpp/include/iggy.hpp @@ -42,12 +42,17 @@ namespace iggy { class IggyBlockingClient; +class IggyBlockingConsumer; +class IggyBlockingProducer; class LoginInfo; class Partition; class Topic; class TopicDetails; class Stream; class StreamDetails; +class ConsumerGroup; +class ConsumerGroupDetails; +class ConsumerGroupMember; namespace detail { /** @brief Internal base for string-backed option types. */ @@ -834,6 +839,77 @@ class Stream final { ResourceOptions options_; }; +class ConsumerGroupMember final { + public: + [[nodiscard]] std::uint32_t Id() const noexcept { return id_; } + [[nodiscard]] std::uint32_t PartitionsCount() const noexcept { return partitions_count_; } + [[nodiscard]] const std::vector<std::uint32_t> &Partitions() const noexcept { return partitions_; } + + private: + ConsumerGroupMember(std::uint32_t id, std::uint32_t partitions_count, std::vector<std::uint32_t> partitions) + : id_(id), partitions_count_(partitions_count), partitions_(std::move(partitions)) {} + + static ConsumerGroupMember FromFfi(ffi::ConsumerGroupMember member); + + friend class ConsumerGroupDetails; + + std::uint32_t id_; + std::uint32_t partitions_count_; + std::vector<std::uint32_t> partitions_; +}; + +class ConsumerGroup final { + public: + [[nodiscard]] std::uint32_t Id() const noexcept { return id_; } + [[nodiscard]] const std::string &Name() const noexcept { return name_; } + [[nodiscard]] std::uint32_t PartitionsCount() const noexcept { return partitions_count_; } + [[nodiscard]] std::uint32_t MembersCount() const noexcept { return members_count_; } + + private: + ConsumerGroup(std::uint32_t id, std::string name, std::uint32_t partitions_count, std::uint32_t members_count) + : id_(id), name_(std::move(name)), partitions_count_(partitions_count), members_count_(members_count) {} + + static ConsumerGroup FromFfi(ffi::ConsumerGroup group); + + friend class IggyBlockingClient; + + std::uint32_t id_; + std::string name_; + std::uint32_t partitions_count_; + std::uint32_t members_count_; +}; + +class ConsumerGroupDetails final { + public: + [[nodiscard]] std::uint32_t Id() const noexcept { return id_; } + [[nodiscard]] const std::string &Name() const noexcept { return name_; } + [[nodiscard]] std::uint32_t PartitionsCount() const noexcept { return partitions_count_; } + [[nodiscard]] std::uint32_t MembersCount() const noexcept { return members_count_; } + [[nodiscard]] const std::vector<ConsumerGroupMember> &Members() const noexcept { return members_; } + + private: + ConsumerGroupDetails(std::uint32_t id, + std::string name, + std::uint32_t partitions_count, + std::uint32_t members_count, + std::vector<ConsumerGroupMember> members) + : id_(id), + name_(std::move(name)), + partitions_count_(partitions_count), + members_count_(members_count), + members_(std::move(members)) {} + + static ConsumerGroupDetails FromFfi(ffi::ConsumerGroupDetails group); + + friend class IggyBlockingClient; + + std::uint32_t id_; + std::string name_; + std::uint32_t partitions_count_; + std::uint32_t members_count_; + std::vector<ConsumerGroupMember> members_; +}; + /** * @brief Compression algorithm used for topic messages. * @@ -1543,6 +1619,48 @@ class PollingStrategy final { std::uint64_t polling_strategy_value_; }; +class IggyBlockingConsumer final { + public: + IggyBlockingConsumer(const IggyBlockingConsumer &) = delete; + IggyBlockingConsumer &operator=(const IggyBlockingConsumer &) = delete; + + IggyBlockingConsumer(IggyBlockingConsumer &&other) noexcept; + IggyBlockingConsumer &operator=(IggyBlockingConsumer &&other) noexcept; + + ~IggyBlockingConsumer(); + + private: + explicit IggyBlockingConsumer(ffi::Consumer *consumer); + + [[nodiscard]] ffi::Consumer *Handle() const; + void Reset() noexcept; + + friend class IggyBlockingClient; + + ffi::Consumer *consumer_; +}; + +class IggyBlockingProducer final { + public: + IggyBlockingProducer(const IggyBlockingProducer &) = delete; + IggyBlockingProducer &operator=(const IggyBlockingProducer &) = delete; + + IggyBlockingProducer(IggyBlockingProducer &&other) noexcept; + IggyBlockingProducer &operator=(IggyBlockingProducer &&other) noexcept; + + ~IggyBlockingProducer(); + + private: + explicit IggyBlockingProducer(ffi::Producer *producer); + + [[nodiscard]] ffi::Producer *Handle() const; + void Reset() noexcept; + + friend class IggyBlockingClient; + + ffi::Producer *producer_; +}; + /** * @brief Owning client connection to an Apache Iggy server. * @@ -1985,6 +2103,20 @@ class IggyBlockingClient final { */ void DeletePartitions(const Identifier &stream, const Identifier &topic, std::uint32_t partitions_count); + ConsumerGroupDetails CreateConsumerGroup(const Identifier &stream, const Identifier &topic, std::string name); + ConsumerGroupDetails GetConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group); + std::vector<ConsumerGroup> GetConsumerGroups(const Identifier &stream, const Identifier &topic); + void DeleteConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group); + void JoinConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group); + void LeaveConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group); + + IggyBlockingConsumer CreateConsumer(std::string name, + const Identifier &stream, + const Identifier &topic, + std::uint32_t partition_id); + IggyBlockingConsumer CreateGroupConsumer(std::string name, const Identifier &stream, const Identifier &topic); + IggyBlockingProducer CreateProducer(const Identifier &stream, const Identifier &topic); + private: explicit IggyBlockingClient(ffi::Client *client); diff --git a/foreign/cpp/src/client.cpp b/foreign/cpp/src/client.cpp index 3e56a4baf..2d8d699a2 100644 --- a/foreign/cpp/src/client.cpp +++ b/foreign/cpp/src/client.cpp @@ -252,6 +252,80 @@ void IggyBlockingClient::DeletePartitions(const Identifier &stream, }); } +ConsumerGroupDetails IggyBlockingClient::CreateConsumerGroup(const Identifier &stream, + const Identifier &topic, + std::string name) { + return RethrowAsIggyException([this, &stream, &topic, &name] { + return ConsumerGroupDetails::FromFfi(Handle()->create_consumer_group(stream.ToFfi(), topic.ToFfi(), name)); + }); +} + +ConsumerGroupDetails IggyBlockingClient::GetConsumerGroup(const Identifier &stream, + const Identifier &topic, + const Identifier &group) { + return RethrowAsIggyException([this, &stream, &topic, &group] { + return ConsumerGroupDetails::FromFfi( + Handle()->get_consumer_group(stream.ToFfi(), topic.ToFfi(), group.ToFfi())); + }); +} + +std::vector<ConsumerGroup> IggyBlockingClient::GetConsumerGroups(const Identifier &stream, const Identifier &topic) { + return RethrowAsIggyException([this, &stream, &topic] { + std::vector<ConsumerGroup> groups; + auto ffi_groups = Handle()->get_consumer_groups(stream.ToFfi(), topic.ToFfi()); + groups.reserve(ffi_groups.size()); + for (auto &group : ffi_groups) { + groups.push_back(ConsumerGroup::FromFfi(std::move(group))); + } + return groups; + }); +} + +void IggyBlockingClient::DeleteConsumerGroup(const Identifier &stream, + const Identifier &topic, + const Identifier &group) { + RethrowAsIggyException([this, &stream, &topic, &group] { + Handle()->delete_consumer_group(stream.ToFfi(), topic.ToFfi(), group.ToFfi()); + }); +} + +void IggyBlockingClient::JoinConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group) { + RethrowAsIggyException([this, &stream, &topic, &group] { + Handle()->join_consumer_group(stream.ToFfi(), topic.ToFfi(), group.ToFfi()); + }); +} + +void IggyBlockingClient::LeaveConsumerGroup(const Identifier &stream, + const Identifier &topic, + const Identifier &group) { + RethrowAsIggyException([this, &stream, &topic, &group] { + Handle()->leave_consumer_group(stream.ToFfi(), topic.ToFfi(), group.ToFfi()); + }); +} + +IggyBlockingConsumer IggyBlockingClient::CreateConsumer(std::string name, + const Identifier &stream, + const Identifier &topic, + std::uint32_t partition_id) { + return RethrowAsIggyException([this, &name, &stream, &topic, partition_id] { + return IggyBlockingConsumer(Handle()->create_consumer(name, stream.ToFfi(), topic.ToFfi(), partition_id)); + }); +} + +IggyBlockingConsumer IggyBlockingClient::CreateGroupConsumer(std::string name, + const Identifier &stream, + const Identifier &topic) { + return RethrowAsIggyException([this, &name, &stream, &topic] { + return IggyBlockingConsumer(Handle()->create_group_consumer(name, stream.ToFfi(), topic.ToFfi())); + }); +} + +IggyBlockingProducer IggyBlockingClient::CreateProducer(const Identifier &stream, const Identifier &topic) { + return RethrowAsIggyException([this, &stream, &topic] { + return IggyBlockingProducer(Handle()->create_producer(stream.ToFfi(), topic.ToFfi())); + }); +} + IggyBlockingClient::IggyBlockingClient(ffi::Client *client) : client_(client) { if (client_ == nullptr) { throw IggyException("Could not create Iggy client"); diff --git a/foreign/cpp/src/client.rs b/foreign/cpp/src/client.rs index 6ea8fdc9e..22203836b 100644 --- a/foreign/cpp/src/client.rs +++ b/foreign/cpp/src/client.rs @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +use crate::consumer::Consumer as BridgeConsumer; +use crate::producer::Producer as BridgeProducer; use crate::{RUNTIME, ffi, type_conversion::ffi_options_to_raw}; use bytes::Bytes; use iggy::prelude::{ @@ -1244,6 +1246,69 @@ impl Client { Ok(Vec::from(response)) }) } + + pub fn create_consumer( + &self, + name: String, + stream_id: ffi::Identifier, + topic_id: ffi::Identifier, + partition_id: u32, + ) -> Result<*mut BridgeConsumer, String> { + let rust_stream_id = RustIdentifier::try_from(stream_id) + .map_err(|error| format!("Could not create consumer '{name}': {error}"))?; + let rust_topic_id = RustIdentifier::try_from(topic_id) + .map_err(|error| format!("Could not create consumer '{name}': {error}"))?; + let consumer = self + .inner + .consumer( + &name, + &rust_stream_id.as_string(), + &rust_topic_id.as_string(), + partition_id, + ) + .map_err(|error| format!("Could not create consumer '{name}': {error}"))? + .build(); + Ok(Box::into_raw(Box::new(BridgeConsumer { inner: consumer }))) + } + + pub fn create_group_consumer( + &self, + name: String, + stream_id: ffi::Identifier, + topic_id: ffi::Identifier, + ) -> Result<*mut BridgeConsumer, String> { + let rust_stream_id = RustIdentifier::try_from(stream_id) + .map_err(|error| format!("Could not create consumer group member '{name}': {error}"))?; + let rust_topic_id = RustIdentifier::try_from(topic_id) + .map_err(|error| format!("Could not create consumer group member '{name}': {error}"))?; + let consumer = self + .inner + .consumer_group( + &name, + &rust_stream_id.as_string(), + &rust_topic_id.as_string(), + ) + .map_err(|error| format!("Could not create consumer group member '{name}': {error}"))? + .build(); + Ok(Box::into_raw(Box::new(BridgeConsumer { inner: consumer }))) + } + + pub fn create_producer( + &self, + stream_id: ffi::Identifier, + topic_id: ffi::Identifier, + ) -> Result<*mut BridgeProducer, String> { + let rust_stream_id = RustIdentifier::try_from(stream_id) + .map_err(|error| format!("Could not create producer: {error}"))?; + let rust_topic_id = RustIdentifier::try_from(topic_id) + .map_err(|error| format!("Could not create producer: {error}"))?; + let producer = self + .inner + .producer(&rust_stream_id.as_string(), &rust_topic_id.as_string()) + .map_err(|error| format!("Could not create producer: {error}"))? + .build(); + Ok(Box::into_raw(Box::new(BridgeProducer { inner: producer }))) + } } pub unsafe fn delete_connection(client: *mut Client) { diff --git a/foreign/cpp/src/consumer.cpp b/foreign/cpp/src/consumer.cpp new file mode 100644 index 000000000..c0183495f --- /dev/null +++ b/foreign/cpp/src/consumer.cpp @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iggy.hpp" + +namespace iggy { + +IggyBlockingConsumer::IggyBlockingConsumer(IggyBlockingConsumer &&other) noexcept + : consumer_(std::exchange(other.consumer_, nullptr)) {} + +IggyBlockingConsumer &IggyBlockingConsumer::operator=(IggyBlockingConsumer &&other) noexcept { + if (this != &other) { + Reset(); + consumer_ = std::exchange(other.consumer_, nullptr); + } + return *this; +} + +IggyBlockingConsumer::~IggyBlockingConsumer() { + Reset(); +} + +IggyBlockingConsumer::IggyBlockingConsumer(ffi::Consumer *consumer) : consumer_(consumer) { + if (consumer_ == nullptr) { + throw IggyException("Could not create Iggy consumer"); + } +} + +ffi::Consumer *IggyBlockingConsumer::Handle() const { + if (consumer_ == nullptr) { + throw IggyException("Cannot use a moved-from IggyBlockingConsumer"); + } + return consumer_; +} + +void IggyBlockingConsumer::Reset() noexcept { + if (consumer_ == nullptr) { + return; + } + + ffi::Consumer *consumer{std::exchange(consumer_, nullptr)}; + ffi::delete_consumer(consumer); +} + +} // namespace iggy diff --git a/foreign/cpp/src/consumer.rs b/foreign/cpp/src/consumer.rs index 147374d4e..4250625a8 100644 --- a/foreign/cpp/src/consumer.rs +++ b/foreign/cpp/src/consumer.rs @@ -21,3 +21,20 @@ use iggy::prelude::IggyConsumer as RustIggyConsumer; pub struct Consumer { pub inner: RustIggyConsumer, } + +/// Releases a consumer previously returned by `create_consumer` or +/// `create_consumer_group`. +/// +/// # Safety +/// +/// - Passing the pointer to this function more than once is undefined +/// behaviour (double-free). +/// - Using the pointer after this function has been called is undefined +/// behaviour (use-after-free). +pub unsafe fn delete_consumer(consumer: *mut Consumer) { + if !consumer.is_null() { + unsafe { + drop(Box::from_raw(consumer)); + } + } +} diff --git a/foreign/cpp/src/lib.rs b/foreign/cpp/src/lib.rs index 3ed6d1103..d6e1b20ab 100644 --- a/foreign/cpp/src/lib.rs +++ b/foreign/cpp/src/lib.rs @@ -23,9 +23,9 @@ mod producer; mod type_conversion; use client::{Client, delete_connection as delete_client, from_connection_string, new_connection}; -use consumer::Consumer; +use consumer::{Consumer, delete_consumer}; use messages::make_message; -use producer::Producer; +use producer::{Producer, delete_producer}; use std::sync::LazyLock; static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| { @@ -685,6 +685,28 @@ mod ffi { fn set_string(self: &mut Identifier, id: String) -> Result<()>; fn set_numeric(self: &mut Identifier, id: u32) -> Result<()>; + fn create_consumer( + self: &Client, + name: String, + stream_id: Identifier, + topic_id: Identifier, + partition_id: u32, + ) -> Result<*mut Consumer>; + fn create_group_consumer( + self: &Client, + name: String, + stream_id: Identifier, + topic_id: Identifier, + ) -> Result<*mut Consumer>; + unsafe fn delete_consumer(consumer: *mut Consumer); + + fn create_producer( + self: &Client, + stream_id: Identifier, + topic_id: Identifier, + ) -> Result<*mut Producer>; + unsafe fn delete_producer(producer: *mut Producer); + // Consumer methods // fn name(self: &Consumer) -> Result<String>; // fn topic(self: &Consumer) -> Result<Identifier>; @@ -696,7 +718,6 @@ mod ffi { // fn get_last_stored_offset(self: &Consumer, partition_id: u32) -> Result<u64>; // fn init(self: &mut Consumer) -> Result<()>; // fn shutdown(self: &mut Consumer) -> Result<()>; - // unsafe fn delete_consumer(consumer: *mut Consumer) -> Result<()>; // Producer methods // fn stream(self: &Producer) -> Result<Identifier>; @@ -707,6 +728,5 @@ mod ffi { // fn send_with_partitioning(self: &Producer, partitioning_kind: String, partitioning_value: Vec<u8>, messages: Vec<IggyMessageToSend>) -> Result<()>; // fn send_to(self: &Producer, stream_id: Identifier, topic_id: Identifier, partitioning_kind: String, partitioning_value: Vec<u8>, messages: Vec<IggyMessageToSend>) -> Result<()>; // fn shutdown(self: &mut Producer) -> Result<()>; - // unsafe fn delete_producer(producer: *mut Producer) -> Result<()>; } } diff --git a/foreign/cpp/src/producer.cpp b/foreign/cpp/src/producer.cpp new file mode 100644 index 000000000..b5a823158 --- /dev/null +++ b/foreign/cpp/src/producer.cpp @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iggy.hpp" + +namespace iggy { + +IggyBlockingProducer::IggyBlockingProducer(IggyBlockingProducer &&other) noexcept + : producer_(std::exchange(other.producer_, nullptr)) {} + +IggyBlockingProducer &IggyBlockingProducer::operator=(IggyBlockingProducer &&other) noexcept { + if (this != &other) { + Reset(); + producer_ = std::exchange(other.producer_, nullptr); + } + return *this; +} + +IggyBlockingProducer::~IggyBlockingProducer() { + Reset(); +} + +IggyBlockingProducer::IggyBlockingProducer(ffi::Producer *producer) : producer_(producer) { + if (producer_ == nullptr) { + throw IggyException("Could not create Iggy producer"); + } +} + +ffi::Producer *IggyBlockingProducer::Handle() const { + if (producer_ == nullptr) { + throw IggyException("Cannot use a moved-from IggyBlockingProducer"); + } + return producer_; +} + +void IggyBlockingProducer::Reset() noexcept { + if (producer_ == nullptr) { + return; + } + + ffi::Producer *producer{std::exchange(producer_, nullptr)}; + ffi::delete_producer(producer); +} + +} // namespace iggy diff --git a/foreign/cpp/src/producer.rs b/foreign/cpp/src/producer.rs index 85b87db1c..62442924c 100644 --- a/foreign/cpp/src/producer.rs +++ b/foreign/cpp/src/producer.rs @@ -21,3 +21,19 @@ use iggy::prelude::IggyProducer as RustIggyProducer; pub struct Producer { pub inner: RustIggyProducer, } + +/// Releases a producer previously returned by `create_producer`. +/// +/// # Safety +/// +/// - Passing the pointer to this function more than once is undefined +/// behaviour (double-free). +/// - Using the pointer after this function has been called is undefined +/// behaviour (use-after-free). +pub unsafe fn delete_producer(producer: *mut Producer) { + if !producer.is_null() { + unsafe { + drop(Box::from_raw(producer)); + } + } +} diff --git a/foreign/cpp/src/type_conversions.cpp b/foreign/cpp/src/type_conversions.cpp index 31a1d33b9..ce8251050 100644 --- a/foreign/cpp/src/type_conversions.cpp +++ b/foreign/cpp/src/type_conversions.cpp @@ -115,4 +115,25 @@ Stream Stream::FromFfi(ffi::Stream stream) { ResourceOptions::FromFfi(std::move(stream.options), rust::Vec<ffi::HeaderEntry>{})); } +ConsumerGroupMember ConsumerGroupMember::FromFfi(ffi::ConsumerGroupMember member) { + return ConsumerGroupMember(member.id, member.partitions_count, + std::vector<std::uint32_t>(member.partitions.begin(), member.partitions.end())); +} + +ConsumerGroup ConsumerGroup::FromFfi(ffi::ConsumerGroup group) { + return ConsumerGroup(group.id, std::string(group.name.c_str(), group.name.size()), group.partitions_count, + group.members_count); +} + +ConsumerGroupDetails ConsumerGroupDetails::FromFfi(ffi::ConsumerGroupDetails group) { + std::vector<ConsumerGroupMember> members; + members.reserve(group.members.size()); + for (auto &member : group.members) { + members.push_back(ConsumerGroupMember::FromFfi(std::move(member))); + } + + return ConsumerGroupDetails(group.id, std::string(group.name.c_str(), group.name.size()), group.partitions_count, + group.members_count, std::move(members)); +} + } // namespace iggy diff --git a/foreign/cpp/tests/e2e/consumer_group.cpp b/foreign/cpp/tests/e2e/consumer_group.cpp index 7099e5f22..cc8d6740d 100644 --- a/foreign/cpp/tests/e2e/consumer_group.cpp +++ b/foreign/cpp/tests/e2e/consumer_group.cpp @@ -22,280 +22,280 @@ #include <gtest/gtest.h> +#include "iggy.hpp" #include "lib.rs.h" #include "tests/e2e/test_helpers.hpp" -class LowLevelE2E_ConsumerGroup : public E2ETestFixture {}; +class E2E_ConsumerGroup : public E2ETestFixture {}; -TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupSucceeds) { +TEST_F(E2E_ConsumerGroup, CreateConsumerGroupSucceeds) { RecordProperty("description", "Creates a consumer group successfully for an existing stream and topic."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); ASSERT_NO_THROW({ - const auto group = client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name); + const auto group = client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_EQ(group.name, group_name); - ASSERT_EQ(group.members_count, 0u); - ASSERT_TRUE(group.members.empty()); + ASSERT_EQ(group.Name(), group_name); + ASSERT_EQ(group.MembersCount(), 0u); + ASSERT_TRUE(group.Members().empty()); }); } -TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupOnNonExistentResourcesThrows) { +TEST_F(E2E_ConsumerGroup, CreateConsumerGroupOnNonExistentResourcesThrows) { RecordProperty("description", "Rejects creating a consumer group on streams or topics that do not exist."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string missing_stream_name = GetRandomName(); const std::string missing_topic_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - - ASSERT_THROW(client->create_consumer_group(make_string_identifier(missing_stream_name), - make_string_identifier(topic_name), GetRandomName()), - std::exception); - ASSERT_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(missing_topic_name), GetRandomName()), - std::exception); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + + ASSERT_THROW(client.CreateConsumerGroup(iggy::Identifier::String(missing_stream_name), + iggy::Identifier::String(topic_name), GetRandomName()), + iggy::IggyException); + ASSERT_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(missing_topic_name), GetRandomName()), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupTwiceOnSameInputThrows) { +TEST_F(E2E_ConsumerGroup, CreateConsumerGroupTwiceOnSameInputThrows) { RecordProperty("description", "Rejects creating the same consumer group twice for the same stream and topic."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_THROW(client->create_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - group_name), - std::exception); + ASSERT_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + group_name), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupWithInvalidNamesThrows) { +TEST_F(E2E_ConsumerGroup, CreateConsumerGroupWithInvalidNamesThrows) { RecordProperty("description", "Rejects empty and overlong consumer group names."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); const std::string invalid_names[] = {"", std::string(256, 'a')}; for (const std::string &invalid_name : invalid_names) { SCOPED_TRACE(invalid_name.size()); - ASSERT_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), invalid_name), - std::exception); + ASSERT_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), invalid_name), + iggy::IggyException); } } -TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupAfterStreamDeletionThrows) { +TEST_F(E2E_ConsumerGroup, CreateConsumerGroupAfterStreamDeletionThrows) { RecordProperty("description", "Rejects creating a consumer group after deleting the stream that owned the topic."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name))); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name))); ForgetTrackedStream(stream_name); - ASSERT_THROW(client->create_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - group_name), - std::exception); + ASSERT_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + group_name), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupBeforeLoginThrows) { +TEST_F(E2E_ConsumerGroup, CreateConsumerGroupBeforeLoginThrows) { RecordProperty("description", "Rejects creating a consumer group before connect, and after connect but before login."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *setup_client = GetLoggedInClient(); - ASSERT_NO_THROW(setup_client->create_stream(stream_name)); + auto setup_client = GetLoggedInHighLevelClient(); + ASSERT_NO_THROW(setup_client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW( - setup_client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - - iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient(); - - 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->connect()); - 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); + ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + + auto unauthenticated_client = GetLoggedOutHighLevelClient(); + + ASSERT_THROW(unauthenticated_client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Connect()); + ASSERT_THROW(unauthenticated_client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy")); + ASSERT_NO_THROW(unauthenticated_client.Disconnect()); + ASSERT_THROW(unauthenticated_client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupReturnsSameInfoAsCreateConsumerGroup) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupReturnsSameInfoAsCreateConsumerGroup) { RecordProperty("description", "Returns the same consumer group details from get_consumer_group as create_consumer_group."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); - const auto created_group = client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name); + const auto created_group = client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name); TrackConsumerGroup(stream_name, topic_name, group_name); - const auto fetched_group = client->get_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)); + const auto fetched_group = + client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)); - ASSERT_EQ(fetched_group.id, created_group.id); - ASSERT_EQ(fetched_group.name, created_group.name); - ASSERT_EQ(fetched_group.partitions_count, created_group.partitions_count); - ASSERT_EQ(fetched_group.members_count, created_group.members_count); - ASSERT_EQ(fetched_group.members.size(), created_group.members.size()); + ASSERT_EQ(fetched_group.Id(), created_group.Id()); + ASSERT_EQ(fetched_group.Name(), created_group.Name()); + ASSERT_EQ(fetched_group.PartitionsCount(), created_group.PartitionsCount()); + ASSERT_EQ(fetched_group.MembersCount(), created_group.MembersCount()); + ASSERT_EQ(fetched_group.Members().size(), created_group.Members().size()); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsReturnsCreatedGroups) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupsReturnsCreatedGroups) { RecordProperty("description", "Returns created consumer groups for an existing stream and topic."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string first_group_name = GetRandomName(); const std::string second_group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), first_group_name)); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), first_group_name)); TrackConsumerGroup(stream_name, topic_name, first_group_name); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), second_group_name)); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), second_group_name)); TrackConsumerGroup(stream_name, topic_name, second_group_name); const auto groups = - client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)); + client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)); EXPECT_EQ(groups.size(), std::size_t{2}); - EXPECT_EQ(groups[0].name, first_group_name); - EXPECT_EQ(groups[1].name, second_group_name); + EXPECT_EQ(groups[0].Name(), first_group_name); + EXPECT_EQ(groups[1].Name(), second_group_name); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsBeforeLoginThrows) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupsBeforeLoginThrows) { RecordProperty("description", "Rejects get_consumer_groups before connect, and after connect but before login."); 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)); + auto setup_client = GetLoggedInHighLevelClient(); + ASSERT_NO_THROW(setup_client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW( - setup_client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - - iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient(); - - ASSERT_THROW(unauthenticated_client->get_consumer_groups(make_string_identifier(stream_name), - make_string_identifier(topic_name)), - std::exception); - ASSERT_NO_THROW(unauthenticated_client->connect()); - 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); + ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + + auto unauthenticated_client = GetLoggedOutHighLevelClient(); + + ASSERT_THROW(unauthenticated_client.GetConsumerGroups(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name)), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Connect()); + ASSERT_THROW(unauthenticated_client.GetConsumerGroups(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name)), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy")); + ASSERT_NO_THROW(unauthenticated_client.Disconnect()); + ASSERT_THROW(unauthenticated_client.GetConsumerGroups(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupSucceeds) { +TEST_F(E2E_ConsumerGroup, JoinConsumerGroupSucceeds) { RecordProperty("description", "Joins an existing consumer group successfully."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); } -TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupBeforeLoginThrows) { +TEST_F(E2E_ConsumerGroup, JoinConsumerGroupBeforeLoginThrows) { RecordProperty("description", "Rejects join_consumer_group before connect, and after connect but before login."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *setup_client = GetLoggedInClient(); - ASSERT_NO_THROW(setup_client->create_stream(stream_name)); + auto setup_client = GetLoggedInHighLevelClient(); + ASSERT_NO_THROW(setup_client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW( - setup_client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(setup_client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(setup_client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient(); - - ASSERT_THROW(unauthenticated_client->join_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); - ASSERT_NO_THROW(unauthenticated_client->connect()); - ASSERT_THROW(unauthenticated_client->join_consumer_group(make_string_identifier(stream_name), - 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); + auto unauthenticated_client = GetLoggedOutHighLevelClient(); + + ASSERT_THROW(unauthenticated_client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Connect()); + ASSERT_THROW(unauthenticated_client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy")); + ASSERT_NO_THROW(unauthenticated_client.Disconnect()); + ASSERT_THROW(unauthenticated_client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupOnNonExistentResourcesThrows) { +TEST_F(E2E_ConsumerGroup, JoinConsumerGroupOnNonExistentResourcesThrows) { RecordProperty("description", "Rejects join_consumer_group for streams, topics, or groups that do not exist."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); @@ -303,261 +303,273 @@ TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupOnNonExistentResourcesThrows) const std::string missing_stream_name = GetRandomName(); const std::string missing_topic_name = GetRandomName(); const std::string missing_group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), created_group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), created_group_name)); TrackConsumerGroup(stream_name, topic_name, created_group_name); ASSERT_THROW( - client->join_consumer_group(make_string_identifier(missing_stream_name), make_string_identifier(topic_name), - make_string_identifier(created_group_name)), - std::exception); + client.JoinConsumerGroup(iggy::Identifier::String(missing_stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(created_group_name)), + iggy::IggyException); ASSERT_THROW( - client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(missing_topic_name), - make_string_identifier(created_group_name)), - std::exception); - ASSERT_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(missing_group_name)), - std::exception); + client.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(missing_topic_name), + iggy::Identifier::String(created_group_name)), + iggy::IggyException); + ASSERT_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(missing_group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupAfterStreamDeletionThrows) { +TEST_F(E2E_ConsumerGroup, JoinConsumerGroupAfterStreamDeletionThrows) { RecordProperty("description", "Rejects join_consumer_group after deleting the stream that owned the group."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name))); + ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name))); ForgetTrackedStream(stream_name); - ASSERT_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); + ASSERT_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupAfterTopicDeletionThrows) { +TEST_F(E2E_ConsumerGroup, JoinConsumerGroupAfterTopicDeletionThrows) { RecordProperty("description", "Rejects join_consumer_group after deleting the topic that owned the group."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->delete_topic(make_string_identifier(stream_name), make_string_identifier(topic_name))); + ASSERT_NO_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name))); ForgetTrackedConsumerGroup(stream_name, topic_name, group_name); - ASSERT_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); + ASSERT_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupReflectsInGetConsumerGroup) { +TEST_F(E2E_ConsumerGroup, JoinConsumerGroupReflectsInGetConsumerGroup) { RecordProperty("description", "Reflects a joined consumer group in get_consumer_group member details."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); - iggy::ffi::ConsumerGroupDetails created_group; - ASSERT_NO_THROW({ - created_group = client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name); - }); + const auto created_group = client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); - - const auto fetched_group = client->get_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)); - - EXPECT_EQ(fetched_group.id, created_group.id); - EXPECT_EQ(fetched_group.name, created_group.name); - EXPECT_EQ(fetched_group.partitions_count, created_group.partitions_count); - EXPECT_EQ(fetched_group.members_count, 1u); - ASSERT_EQ(fetched_group.members.size(), std::size_t{1}); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + + const auto fetched_group = + client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)); + + EXPECT_EQ(fetched_group.Id(), created_group.Id()); + EXPECT_EQ(fetched_group.Name(), created_group.Name()); + EXPECT_EQ(fetched_group.PartitionsCount(), created_group.PartitionsCount()); + EXPECT_EQ(fetched_group.MembersCount(), 1u); + ASSERT_EQ(fetched_group.Members().size(), std::size_t{1}); } -TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupTwiceKeepsSingleMember) { +TEST_F(E2E_ConsumerGroup, JoinConsumerGroupTwiceKeepsSingleMember) { RecordProperty("description", "Allows joining the same consumer group twice in a row without duplicating membership."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); - ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); - - const auto fetched_group = client->get_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)); - EXPECT_EQ(fetched_group.members_count, 1u); - ASSERT_EQ(fetched_group.members.size(), std::size_t{1}); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + + const auto fetched_group = + client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)); + EXPECT_EQ(fetched_group.MembersCount(), 1u); + ASSERT_EQ(fetched_group.Members().size(), std::size_t{1}); } -TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupFromTwoClientsIncreasesMembersCount) { +TEST_F(E2E_ConsumerGroup, JoinConsumerGroupFromTwoClientsIncreasesMembersCount) { RecordProperty("description", "Reflects two joined clients as two members in the same consumer group."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *first = GetLoggedInClient(); - iggy::ffi::Client *second = GetLoggedInClient(); + auto first = GetLoggedInHighLevelClient(); + auto second = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(first->create_stream(stream_name)); + ASSERT_NO_THROW(first.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(first->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(first->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(first.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(first.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(first->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); - ASSERT_NO_THROW(second->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); - - const auto fetched_group = first->get_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)); - EXPECT_EQ(fetched_group.members_count, 2u); - ASSERT_EQ(fetched_group.members.size(), std::size_t{2}); + ASSERT_NO_THROW(first.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + ASSERT_NO_THROW(second.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + + const auto fetched_group = + first.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)); + EXPECT_EQ(fetched_group.MembersCount(), 2u); + ASSERT_EQ(fetched_group.Members().size(), std::size_t{2}); } -TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupThenLeaveRestoresMembersCount) { +TEST_F(E2E_ConsumerGroup, JoinConsumerGroupThenLeaveRestoresMembersCount) { RecordProperty("description", "Restores the consumer group member count after a client joins and then leaves."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); - auto joined_group = client->get_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)); - EXPECT_EQ(joined_group.members_count, 1u); - ASSERT_EQ(joined_group.members.size(), std::size_t{1}); - - ASSERT_NO_THROW(client->leave_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name))); - - const auto left_group = client->get_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)); - EXPECT_EQ(left_group.members_count, 0u); - EXPECT_TRUE(left_group.members.empty()); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + const auto joined_group = + client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)); + EXPECT_EQ(joined_group.MembersCount(), 1u); + ASSERT_EQ(joined_group.Members().size(), std::size_t{1}); + + ASSERT_NO_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + + const auto left_group = + client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)); + EXPECT_EQ(left_group.MembersCount(), 0u); + EXPECT_TRUE(left_group.Members().empty()); } -TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupReducesMembersCount) { +TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupReducesMembersCount) { RecordProperty("description", "Reduces the consumer group member count after one of two joined clients leaves."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *first = GetLoggedInClient(); - iggy::ffi::Client *second = GetLoggedInClient(); + auto first = GetLoggedInHighLevelClient(); + auto second = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(first->create_stream(stream_name)); + ASSERT_NO_THROW(first.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(first->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(first->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(first.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(first.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(first->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); - ASSERT_NO_THROW(second->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); - - const auto joined_group = first->get_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)); - EXPECT_EQ(joined_group.members_count, 2u); - ASSERT_EQ(joined_group.members.size(), std::size_t{2}); - - ASSERT_NO_THROW(second->leave_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name))); - - const auto left_group = first->get_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)); - EXPECT_EQ(left_group.members_count, 1u); - ASSERT_EQ(left_group.members.size(), std::size_t{1}); + ASSERT_NO_THROW(first.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + ASSERT_NO_THROW(second.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + + const auto joined_group = + first.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)); + EXPECT_EQ(joined_group.MembersCount(), 2u); + ASSERT_EQ(joined_group.Members().size(), std::size_t{2}); + + ASSERT_NO_THROW(second.LeaveConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + + const auto left_group = + first.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)); + EXPECT_EQ(left_group.MembersCount(), 1u); + ASSERT_EQ(left_group.Members().size(), std::size_t{1}); } -TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupBeforeLoginThrows) { +TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupBeforeLoginThrows) { RecordProperty("description", "Rejects leave_consumer_group before connect, and after connect but before login."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *setup_client = GetLoggedInClient(); - ASSERT_NO_THROW(setup_client->create_stream(stream_name)); + auto setup_client = GetLoggedInHighLevelClient(); + ASSERT_NO_THROW(setup_client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW( - setup_client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(setup_client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(setup_client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(setup_client->join_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name))); - - iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient(); - - ASSERT_THROW(unauthenticated_client->leave_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); - ASSERT_NO_THROW(unauthenticated_client->connect()); - ASSERT_THROW(unauthenticated_client->leave_consumer_group(make_string_identifier(stream_name), - 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); + ASSERT_NO_THROW(setup_client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + + auto unauthenticated_client = GetLoggedOutHighLevelClient(); + + ASSERT_THROW(unauthenticated_client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Connect()); + ASSERT_THROW(unauthenticated_client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy")); + ASSERT_NO_THROW(unauthenticated_client.Disconnect()); + ASSERT_THROW(unauthenticated_client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupOnNonExistentResourcesThrows) { +TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupOnNonExistentResourcesThrows) { RecordProperty("description", "Rejects leave_consumer_group for streams, topics, or groups that do not exist."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); @@ -565,372 +577,364 @@ TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupOnNonExistentResourcesThrows const std::string missing_stream_name = GetRandomName(); const std::string missing_topic_name = GetRandomName(); const std::string missing_group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), created_group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), created_group_name)); TrackConsumerGroup(stream_name, topic_name, created_group_name); - ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(created_group_name))); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(created_group_name))); ASSERT_THROW( - client->leave_consumer_group(make_string_identifier(missing_stream_name), make_string_identifier(topic_name), - make_string_identifier(created_group_name)), - std::exception); + client.LeaveConsumerGroup(iggy::Identifier::String(missing_stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(created_group_name)), + iggy::IggyException); ASSERT_THROW( - client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(missing_topic_name), - make_string_identifier(created_group_name)), - std::exception); - ASSERT_THROW(client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(missing_group_name)), - std::exception); + client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(missing_topic_name), + iggy::Identifier::String(created_group_name)), + iggy::IggyException); + ASSERT_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(missing_group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupAfterStreamDeletionThrows) { +TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupAfterStreamDeletionThrows) { RecordProperty("description", "Rejects leave_consumer_group after deleting the stream that owned the group."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); - ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name))); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name))); ForgetTrackedStream(stream_name); - ASSERT_THROW(client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); + ASSERT_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupAfterTopicDeletionThrows) { +TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupAfterTopicDeletionThrows) { RecordProperty("description", "Rejects leave_consumer_group after deleting the topic that owned the group."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); - ASSERT_NO_THROW(client->delete_topic(make_string_identifier(stream_name), make_string_identifier(topic_name))); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + ASSERT_NO_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name))); ForgetTrackedConsumerGroup(stream_name, topic_name, group_name); - ASSERT_THROW(client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); + ASSERT_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupTwiceThrows) { +TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupTwiceThrows) { RecordProperty("description", "Rejects leaving the same consumer group twice."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name))); - ASSERT_NO_THROW(client->leave_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name))); - - ASSERT_THROW(client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + ASSERT_NO_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + + ASSERT_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupWithoutJoiningThrows) { +TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupWithoutJoiningThrows) { RecordProperty("description", "Rejects leaving a consumer group when the client is not a member."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_THROW(client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); + ASSERT_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsReflectsJoinedGroupMembersCount) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupsReflectsJoinedGroupMembersCount) { RecordProperty("description", "Reflects a joined consumer group in get_consumer_groups members_count."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string joined_group_name = GetRandomName(); const std::string other_group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); - iggy::ffi::ConsumerGroupDetails joined_group; - iggy::ffi::ConsumerGroupDetails other_group; - ASSERT_NO_THROW({ - joined_group = client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), joined_group_name); - }); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), joined_group_name)); TrackConsumerGroup(stream_name, topic_name, joined_group_name); - ASSERT_NO_THROW({ - other_group = client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), other_group_name); - }); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), other_group_name)); TrackConsumerGroup(stream_name, topic_name, other_group_name); - ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(joined_group_name))); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(joined_group_name))); const auto groups = - client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)); + client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)); ASSERT_EQ(groups.size(), std::size_t{2}); - EXPECT_EQ(groups[0].id, joined_group.id); - EXPECT_EQ(groups[0].name, joined_group.name); - EXPECT_EQ(groups[0].members_count, joined_group.members_count + 1u); + EXPECT_EQ(groups[0].Name(), joined_group_name); + EXPECT_EQ(groups[0].MembersCount(), 1u); - EXPECT_EQ(groups[1].id, other_group.id); - EXPECT_EQ(groups[1].name, other_group.name); - EXPECT_EQ(groups[1].members_count, other_group.members_count); - EXPECT_NE(groups[0].members_count, groups[1].members_count); + EXPECT_EQ(groups[1].Name(), other_group_name); + EXPECT_EQ(groups[1].MembersCount(), 0u); + EXPECT_NE(groups[0].MembersCount(), groups[1].MembersCount()); } // The VSR server rejects consumer-group reads whose parent stream or topic is // absent with the legacy typed not-found; the legacy server answered them with // an empty list. -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsOnNonExistentStreamThrows) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupsOnNonExistentStreamThrows) { RecordProperty("description", "Throws when the stream does not exist."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)), - std::exception); + ASSERT_THROW(client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsOnNonExistentTopicThrows) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupsOnNonExistentTopicThrows) { RecordProperty("description", "Throws when the topic does not exist."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)), - std::exception); + ASSERT_THROW(client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsIsStableAcrossBackToBackCalls) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupsIsStableAcrossBackToBackCalls) { RecordProperty("description", "Returns the same consumer groups across back-to-back get_consumer_groups calls."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string first_group_name = GetRandomName(); const std::string second_group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), first_group_name)); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), first_group_name)); TrackConsumerGroup(stream_name, topic_name, first_group_name); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), second_group_name)); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), second_group_name)); TrackConsumerGroup(stream_name, topic_name, second_group_name); const auto first_groups = - client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)); + client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)); const auto second_groups = - client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)); + client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)); EXPECT_EQ(second_groups.size(), first_groups.size()); EXPECT_EQ(second_groups.size(), std::size_t{2}); for (std::size_t i = 0; i < first_groups.size(); ++i) { - EXPECT_EQ(second_groups[i].id, first_groups[i].id); - EXPECT_EQ(second_groups[i].name, first_groups[i].name); - EXPECT_EQ(second_groups[i].partitions_count, first_groups[i].partitions_count); - EXPECT_EQ(second_groups[i].members_count, first_groups[i].members_count); + EXPECT_EQ(second_groups[i].Id(), first_groups[i].Id()); + EXPECT_EQ(second_groups[i].Name(), first_groups[i].Name()); + EXPECT_EQ(second_groups[i].PartitionsCount(), first_groups[i].PartitionsCount()); + EXPECT_EQ(second_groups[i].MembersCount(), first_groups[i].MembersCount()); } } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsReturnsCorrectNumberOfGroups) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupsReturnsCorrectNumberOfGroups) { RecordProperty("description", "Returns the last remaining consumer group after deleting two groups."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string deleted_group_name = GetRandomName(); const std::string other_deleted_name = GetRandomName(); const std::string remaining_group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), deleted_group_name)); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), deleted_group_name)); TrackConsumerGroup(stream_name, topic_name, deleted_group_name); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), other_deleted_name)); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), other_deleted_name)); TrackConsumerGroup(stream_name, topic_name, other_deleted_name); - iggy::ffi::ConsumerGroupDetails remaining_group; - ASSERT_NO_THROW({ - remaining_group = client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), remaining_group_name); - }); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), remaining_group_name)); TrackConsumerGroup(stream_name, topic_name, remaining_group_name); - ASSERT_NO_THROW(client->delete_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), - make_string_identifier(deleted_group_name))); + ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(deleted_group_name))); ForgetTrackedConsumerGroup(stream_name, topic_name, deleted_group_name); - ASSERT_NO_THROW(client->delete_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), - make_string_identifier(other_deleted_name))); + ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(other_deleted_name))); ForgetTrackedConsumerGroup(stream_name, topic_name, other_deleted_name); const auto groups = - client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)); + client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)); ASSERT_EQ(groups.size(), std::size_t{1}); - EXPECT_EQ(groups[0].id, remaining_group.id); - EXPECT_EQ(groups[0].name, remaining_group.name); - EXPECT_EQ(groups[0].partitions_count, remaining_group.partitions_count); - EXPECT_EQ(groups[0].members_count, remaining_group.members_count); - - ASSERT_NO_THROW(client->delete_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), - make_string_identifier(remaining_group_name))); + EXPECT_EQ(groups[0].Name(), remaining_group_name); + EXPECT_EQ(groups[0].MembersCount(), 0u); + + ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(remaining_group_name))); ForgetTrackedConsumerGroup(stream_name, topic_name, remaining_group_name); const auto groups_after_delete = - client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)); + client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)); EXPECT_TRUE(groups_after_delete.empty()); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsAfterStreamDeletionThrows) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupsAfterStreamDeletionThrows) { RecordProperty("description", "Throws after deleting the stream that owned the groups."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string first_group_name = GetRandomName(); const std::string second_group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), first_group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), first_group_name)); TrackConsumerGroup(stream_name, topic_name, first_group_name); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), second_group_name)); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), second_group_name)); TrackConsumerGroup(stream_name, topic_name, second_group_name); - ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name))); + ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name))); ForgetTrackedConsumerGroup(stream_name, topic_name, first_group_name); ForgetTrackedConsumerGroup(stream_name, topic_name, second_group_name); ForgetTrackedStream(stream_name); - ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)), - std::exception); + ASSERT_THROW(client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsAfterTopicDeletionThrows) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupsAfterTopicDeletionThrows) { RecordProperty("description", "Throws after deleting the topic that owned the groups."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string first_group_name = GetRandomName(); const std::string second_group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), first_group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), first_group_name)); TrackConsumerGroup(stream_name, topic_name, first_group_name); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), second_group_name)); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), second_group_name)); TrackConsumerGroup(stream_name, topic_name, second_group_name); - ASSERT_NO_THROW(client->delete_topic(make_string_identifier(stream_name), make_string_identifier(topic_name))); + ASSERT_NO_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name))); ForgetTrackedConsumerGroup(stream_name, topic_name, first_group_name); ForgetTrackedConsumerGroup(stream_name, topic_name, second_group_name); - ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)), - std::exception); + ASSERT_THROW(client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupBeforeLoginThrows) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupBeforeLoginThrows) { RecordProperty("description", "Rejects get_consumer_group before connect, and after connect but before login."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *setup_client = GetLoggedInClient(); - ASSERT_NO_THROW(setup_client->create_stream(stream_name)); + auto setup_client = GetLoggedInHighLevelClient(); + ASSERT_NO_THROW(setup_client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW( - setup_client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(setup_client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(setup_client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient(); - - ASSERT_THROW(unauthenticated_client->get_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); - ASSERT_NO_THROW(unauthenticated_client->connect()); - ASSERT_THROW(unauthenticated_client->get_consumer_group(make_string_identifier(stream_name), - 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); + auto unauthenticated_client = GetLoggedOutHighLevelClient(); + + ASSERT_THROW(unauthenticated_client.GetConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Connect()); + ASSERT_THROW(unauthenticated_client.GetConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy")); + ASSERT_NO_THROW(unauthenticated_client.Disconnect()); + ASSERT_THROW(unauthenticated_client.GetConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupOnNonExistentResourcesThrows) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupOnNonExistentResourcesThrows) { RecordProperty("description", "Rejects get_consumer_group for streams, topics, or groups that do not exist."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); @@ -938,110 +942,110 @@ TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupOnNonExistentResourcesThrows) const std::string missing_stream_name = GetRandomName(); const std::string missing_topic_name = GetRandomName(); const std::string missing_group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), created_group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), created_group_name)); TrackConsumerGroup(stream_name, topic_name, created_group_name); ASSERT_THROW( - client->get_consumer_group(make_string_identifier(missing_stream_name), make_string_identifier(topic_name), - make_string_identifier(created_group_name)), - std::exception); + client.GetConsumerGroup(iggy::Identifier::String(missing_stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(created_group_name)), + iggy::IggyException); ASSERT_THROW( - client->get_consumer_group(make_string_identifier(stream_name), make_string_identifier(missing_topic_name), - make_string_identifier(created_group_name)), - std::exception); - ASSERT_THROW(client->get_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(missing_group_name)), - std::exception); + client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(missing_topic_name), + iggy::Identifier::String(created_group_name)), + iggy::IggyException); + ASSERT_THROW(client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(missing_group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupAfterStreamDeletionThrows) { +TEST_F(E2E_ConsumerGroup, GetConsumerGroupAfterStreamDeletionThrows) { RecordProperty("description", "Rejects get_consumer_group after deleting the stream that owned the group."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); - ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name))); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); + ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name))); ForgetTrackedStream(stream_name); - ASSERT_THROW(client->get_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); + ASSERT_THROW(client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupSucceeds) { +TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupSucceeds) { RecordProperty("description", "Deletes an existing consumer group successfully."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->delete_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name))); + ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); ForgetTrackedConsumerGroup(stream_name, topic_name, group_name); - ASSERT_THROW(client->get_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); + ASSERT_THROW(client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupBeforeLoginThrows) { +TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupBeforeLoginThrows) { RecordProperty("description", "Rejects delete_consumer_group before connect, and after connect but before login."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *setup_client = GetLoggedInClient(); - ASSERT_NO_THROW(setup_client->create_stream(stream_name)); + auto setup_client = GetLoggedInHighLevelClient(); + ASSERT_NO_THROW(setup_client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW( - setup_client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(setup_client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(setup_client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient(); - - ASSERT_THROW(unauthenticated_client->delete_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); - ASSERT_NO_THROW(unauthenticated_client->connect()); - ASSERT_THROW(unauthenticated_client->delete_consumer_group(make_string_identifier(stream_name), - 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); + auto unauthenticated_client = GetLoggedOutHighLevelClient(); + + ASSERT_THROW(unauthenticated_client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Connect()); + ASSERT_THROW(unauthenticated_client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); + ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy")); + ASSERT_NO_THROW(unauthenticated_client.Disconnect()); + ASSERT_THROW(unauthenticated_client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupOnNonExistentResourcesThrows) { +TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupOnNonExistentResourcesThrows) { RecordProperty("description", "Rejects delete_consumer_group for streams, topics, or groups that do not exist."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); @@ -1049,102 +1053,104 @@ TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupOnNonExistentResourcesThrow const std::string missing_stream_name = GetRandomName(); const std::string missing_topic_name = GetRandomName(); const std::string missing_group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), created_group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), created_group_name)); TrackConsumerGroup(stream_name, topic_name, created_group_name); ASSERT_THROW( - client->delete_consumer_group(make_string_identifier(missing_stream_name), make_string_identifier(topic_name), - make_string_identifier(created_group_name)), - std::exception); + client.DeleteConsumerGroup(iggy::Identifier::String(missing_stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(created_group_name)), + iggy::IggyException); ASSERT_THROW( - client->delete_consumer_group(make_string_identifier(stream_name), make_string_identifier(missing_topic_name), - make_string_identifier(created_group_name)), - std::exception); - ASSERT_THROW(client->delete_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(missing_group_name)), - std::exception); + client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(missing_topic_name), + iggy::Identifier::String(created_group_name)), + iggy::IggyException); + ASSERT_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(missing_group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupTwiceThrows) { +TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupTwiceThrows) { RecordProperty("description", "Rejects deleting the same consumer group twice."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->delete_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name))); + ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); ForgetTrackedConsumerGroup(stream_name, topic_name, group_name); - ASSERT_THROW(client->delete_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); + ASSERT_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupAfterStreamDeletionThrows) { +TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupAfterStreamDeletionThrows) { RecordProperty("description", "Rejects delete_consumer_group after deleting the stream that owned the consumer group."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); TrackConsumerGroup(stream_name, topic_name, group_name); - ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name))); + ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name))); ForgetTrackedStream(stream_name); ForgetTrackedConsumerGroup(stream_name, topic_name, group_name); - ASSERT_THROW(client->delete_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name), - make_string_identifier(group_name)), - std::exception); + ASSERT_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name)), + iggy::IggyException); } -TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupAndRecreateWithSameNameSucceeds) { +TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupAndRecreateWithSameNameSucceeds) { RecordProperty("description", "Allows recreating a consumer group with the same name after the previous group is deleted."); const std::string stream_name = GetRandomName(); const std::string topic_name = GetRandomName(); const std::string group_name = GetRandomName(); - iggy::ffi::Client *client = GetLoggedInClient(); + auto client = GetLoggedInHighLevelClient(); - ASSERT_NO_THROW(client->create_stream(stream_name)); + ASSERT_NO_THROW(client.CreateStream(stream_name)); TrackStream(stream_name); - ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, - make_topic_create_options(1, "none", "server_default", 0, "server_default"))); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); - ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name)); - ASSERT_NO_THROW(client->delete_consumer_group( - make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); + ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); ASSERT_NO_THROW({ - const auto recreated_group = client->create_consumer_group(make_string_identifier(stream_name), - make_string_identifier(topic_name), group_name); + const auto recreated_group = client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name); TrackConsumerGroup(stream_name, topic_name, group_name); // The VSR server mints group ids monotonically; a recreate gets a // fresh id (the deleted group held 0), unlike the legacy server which // reused the freed slot. - ASSERT_GT(recreated_group.id, 0u); - ASSERT_EQ(recreated_group.name, group_name); - ASSERT_EQ(recreated_group.members_count, 0u); - ASSERT_TRUE(recreated_group.members.empty()); + ASSERT_GT(recreated_group.Id(), 0u); + ASSERT_EQ(recreated_group.Name(), group_name); + ASSERT_EQ(recreated_group.MembersCount(), 0u); + ASSERT_TRUE(recreated_group.Members().empty()); }); }
