You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ri...@apache.org on 2026/09/01 10:11:47 UTC
(iggy) 01/01: add stream, topic, and partition functions to high level client.
This is an automated email from the ASF dual-hosted git repository.
slbotbm pushed a commit to branch cpp-high-level-client-1
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit dca7fe6545088b1aeef7c2cc6e448f39453f7320
Author: Rimuksh Kansal <ni...@gmail.com>
AuthorDate: Tue Sep 1 16:39:22 2026 +0900
add stream, topic, and partition functions to high level client.
---
foreign/cpp/MODULE.bazel | 2 +-
foreign/cpp/MODULE.bazel.lock | 36 +-
foreign/cpp/include/iggy.hpp | 1138 +++++++++++++++++++++++++--
foreign/cpp/src/client.cpp | 114 ++-
foreign/cpp/src/type_conversion.rs | 4 +-
foreign/cpp/src/type_conversions.cpp | 165 ++++
foreign/cpp/tests/e2e/partition.cpp | 386 +++++----
foreign/cpp/tests/e2e/stream.cpp | 591 +++++++-------
foreign/cpp/tests/e2e/test_helpers.hpp | 10 +
foreign/cpp/tests/e2e/topic.cpp | 1332 +++++++++++++++-----------------
foreign/cpp/tests/unit/unit_tests.cpp | 51 +-
11 files changed, 2486 insertions(+), 1343 deletions(-)
diff --git a/foreign/cpp/MODULE.bazel b/foreign/cpp/MODULE.bazel
index f274abe1d..54329fbb5 100644
--- a/foreign/cpp/MODULE.bazel
+++ b/foreign/cpp/MODULE.bazel
@@ -29,6 +29,6 @@ bazel_dep(name = "rules_rust", version = "0.73.0")
rust_host_tools = use_extension("@rules_rust//rust:extensions.bzl", "rust_host_tools")
rust_host_tools.host_tools(
name = "rs_host_tools",
- version = "1.97.1",
+ version = "1.98.0",
)
use_repo(rust_host_tools, "rs_host_tools")
diff --git a/foreign/cpp/MODULE.bazel.lock b/foreign/cpp/MODULE.bazel.lock
index 0ac4a0878..45317aa2f 100644
--- a/foreign/cpp/MODULE.bazel.lock
+++ b/foreign/cpp/MODULE.bazel.lock
@@ -229,26 +229,6 @@
},
"selectedYankedVersions": {},
"moduleExtensions": {
- "@@pybind11_bazel+//:internal_configure.bzl%internal_configure_extension": {
- "general": {
- "bzlTransitiveDigest": "lUqTnA8vmoqQ3SKxclekEGQgFg1l4LQoUFH2ENIty5A=",
- "usagesDigest": "tVQNvLoXMWAbiK39am3yovKGpwINdftfn7RpDyN+JZc=",
- "recordedInputs": [
- "REPO_MAPPING:pybind11_bazel+,bazel_tools bazel_tools"
- ],
- "generatedRepoSpecs": {
- "pybind11": {
- "repoRuleId": "@@bazel_tools//tools/build_defs/repo:http.bzl%http_archive",
- "attributes": {
- "build_file": "@@pybind11_bazel+//:pybind11-BUILD.bazel",
- "strip_prefix": "pybind11-2.13.6",
- "url": "https://github.com/pybind/pybind11/archive/refs/tags/v2.13.6.tar.gz",
- "integrity": "sha256-4Iy4f0dz2pf6e18DXeh2OrxlbYfVdz5i9toFh9Hw7CA="
- }
- }
- }
- }
- },
"@@rules_kotlin+//src/main/starlark/core/repositories:bzlmod_setup.bzl%rules_kotlin_extensions": {
"general": {
"bzlTransitiveDigest": "+Kp6j204mBZ3mxlIDDR0gBoP45BZ4jYRhRAcB8sU0qc=",
@@ -506,17 +486,20 @@
},
"@@rules_rust+//crate_universe/private:internal_extensions.bzl%cu_nr": {
"general": {
- "bzlTransitiveDigest": "VGHJuEwR5ycRmFFD/5oLU3gxx+/rnoFRQozsFk9btzo=",
- "usagesDigest": "ZmL90WEq2B6/NJ8rtHAqdnDPn+/9xG/GWR5K4UU4tyo=",
+ "bzlTransitiveDigest": "+d1kZFExl1sIZTX4/D3oxljqMS6bRkbX/zgsuDUfEu0=",
+ "usagesDigest": "+1w4EnEfZLFpf8Ak18jhmCWsOTE75Noy2i16whyXYWs=",
"recordedInputs": [
+ "REPO_MAPPING:apple_support+,bazel_skylib bazel_skylib+",
"REPO_MAPPING:bazel_features+,bazel_features_globals bazel_features++version_extension+bazel_features_globals",
"REPO_MAPPING:bazel_features+,bazel_features_version bazel_features++version_extension+bazel_features_version",
+ "REPO_MAPPING:rules_cc+,bazel_features bazel_features+",
"REPO_MAPPING:rules_cc+,bazel_skylib bazel_skylib+",
"REPO_MAPPING:rules_cc+,bazel_tools bazel_tools",
"REPO_MAPPING:rules_cc+,cc_compatibility_proxy rules_cc++compatibility_proxy+cc_compatibility_proxy",
"REPO_MAPPING:rules_cc+,platforms platforms",
"REPO_MAPPING:rules_cc+,rules_cc rules_cc+",
"REPO_MAPPING:rules_cc++compatibility_proxy+cc_compatibility_proxy,rules_cc rules_cc+",
+ "REPO_MAPPING:rules_rust+,apple_support apple_support+",
"REPO_MAPPING:rules_rust+,bazel_features bazel_features+",
"REPO_MAPPING:rules_rust+,bazel_skylib bazel_skylib+",
"REPO_MAPPING:rules_rust+,bazel_tools bazel_tools",
@@ -539,6 +522,7 @@
"@@rules_rust+//crate_universe:src/cli/splice.rs",
"@@rules_rust+//crate_universe:src/cli/vendor.rs",
"@@rules_rust+//crate_universe:src/config.rs",
+ "@@rules_rust+//crate_universe:src/config/label_injection.rs",
"@@rules_rust+//crate_universe:src/context.rs",
"@@rules_rust+//crate_universe:src/context/crate_context.rs",
"@@rules_rust+//crate_universe:src/context/platforms.rs",
@@ -554,13 +538,13 @@
"@@rules_rust+//crate_universe:src/metadata/metadata_annotation.rs",
"@@rules_rust+//crate_universe:src/rendering.rs",
"@@rules_rust+//crate_universe:src/rendering/template_engine.rs",
+ "@@rules_rust+//crate_universe:src/rendering/templates/defs_bzl_shim.j2",
"@@rules_rust+//crate_universe:src/rendering/templates/module_bzl.j2",
"@@rules_rust+//crate_universe:src/rendering/templates/partials/header.j2",
"@@rules_rust+//crate_universe:src/rendering/templates/partials/module/aliases_map.j2",
"@@rules_rust+//crate_universe:src/rendering/templates/partials/module/deps_map.j2",
"@@rules_rust+//crate_universe:src/rendering/templates/partials/module/repo_git.j2",
"@@rules_rust+//crate_universe:src/rendering/templates/partials/module/repo_http.j2",
- "@@rules_rust+//crate_universe:src/rendering/templates/vendor_module.j2",
"@@rules_rust+//crate_universe:src/rendering/verbatim/alias_rules.bzl",
"@@rules_rust+//crate_universe:src/select.rs",
"@@rules_rust+//crate_universe:src/splicing.rs",
@@ -585,11 +569,11 @@
"binary": "cargo-bazel",
"cargo_lockfile": "@@rules_rust+//crate_universe:Cargo.lock",
"cargo_toml": "@@rules_rust+//crate_universe:Cargo.toml",
- "version": "1.95.0",
- "timeout": 900,
+ "version": "1.97.1",
"rust_toolchain_cargo_template": "@rust_host_tools//:bin/{tool}",
"rust_toolchain_rustc_template": "@rust_host_tools//:bin/{tool}",
- "compressed_windows_toolchain_names": false
+ "compressed_windows_toolchain_names": false,
+ "timeout": 900
}
}
},
diff --git a/foreign/cpp/include/iggy.hpp b/foreign/cpp/include/iggy.hpp
index 593815299..09756a134 100644
--- a/foreign/cpp/include/iggy.hpp
+++ b/foreign/cpp/include/iggy.hpp
@@ -33,15 +33,614 @@
#include <string>
#include <string_view>
#include <utility>
+#include <variant>
+#include <vector>
#include "lib.rs.h"
namespace iggy {
-using LoginInfo = ffi::LoginInfo;
+class IggyBlockingClient;
+class LoginInfo;
+class Partition;
+class Topic;
+class TopicDetails;
+class Stream;
+class StreamDetails;
-namespace detail {
+/**
+ * @brief Exception thrown when an Iggy client operation fails.
+ */
+class IggyException : public std::runtime_error {
+ public:
+ explicit IggyException(const char *message) : std::runtime_error(message) {}
+ explicit IggyException(const std::string &message) : std::runtime_error(message) {}
+};
+
+/**
+ * @brief Details returned after a successful login.
+ *
+ * The value identifies the authenticated user. HTTP logins also return the
+ * access token retained by the client for later requests. Stateful transports
+ * do not return an access token. The token is a credential: callers must not
+ * write it to logs or expose it to untrusted code.
+ */
+class LoginInfo final {
+ public:
+ /**
+ * @brief Returns the numeric ID of the authenticated user.
+ * @return Numeric user ID.
+ */
+ std::uint32_t UserId() const noexcept { return user_id_; }
+
+ /**
+ * @brief Returns the HTTP access token when the login returned one.
+ * @return Empty when the selected transport does not use an access token.
+ * A returned string view remains valid while this LoginInfo object
+ * is not modified or destroyed.
+ */
+ std::optional<std::string_view> AccessToken() const noexcept {
+ if (!access_token_) {
+ return std::nullopt;
+ }
+ return *access_token_;
+ }
+
+ /**
+ * @brief Returns the access-token expiry when a token was returned.
+ * @return Empty when no access token was returned; otherwise the
+ * server-provided expiry value.
+ */
+ std::optional<std::uint64_t> AccessTokenExpiry() const noexcept { return access_token_expiry_; }
+
+ private:
+ LoginInfo(std::uint32_t user_id,
+ std::optional<std::string> access_token,
+ std::optional<std::uint64_t> access_token_expiry)
+ : user_id_(user_id), access_token_(std::move(access_token)), access_token_expiry_(access_token_expiry) {}
+
+ static LoginInfo FromFfi(ffi::LoginInfo login_info);
+
+ friend class IggyBlockingClient;
+
+ std::uint32_t user_id_;
+ std::optional<std::string> access_token_;
+ std::optional<std::uint64_t> access_token_expiry_;
+};
+
+/**
+ * @brief Identifier for a server resource.
+ *
+ * An identifier is either a numeric server ID or a name. The factories enforce
+ * the protocol's non-empty, 255-byte name limit. Numeric zero is valid because
+ * the server can assign zero as a resource ID.
+ */
+class Identifier final {
+ public:
+ enum class Kind { Numeric, String };
+
+ /**
+ * @brief Creates a numeric identifier.
+ * @param id Numeric server ID.
+ * @return Identifier that addresses @p id.
+ */
+ static Identifier Numeric(std::uint32_t id) { return Identifier(Kind::Numeric, id); }
+
+ /**
+ * @brief Creates a name-based identifier.
+ * @param name Resource name.
+ * @return Identifier that addresses @p name.
+ * @throws IggyException if @p name is empty or exceeds 255 bytes.
+ */
+ static Identifier String(std::string name) {
+ if (name.empty() || name.size() > 255) {
+ throw IggyException("Identifier name must contain 1 to 255 bytes");
+ }
+ return Identifier(Kind::String, std::move(name));
+ }
+
+ /**
+ * @brief Returns this identifier's representation.
+ * @return Kind::Numeric or Kind::String.
+ */
+ Kind Type() const noexcept { return kind_; }
+
+ /**
+ * @brief Returns the identifier payload.
+ * @return Numeric ID for Kind::Numeric, or a view of the name for
+ * Kind::String. The string view remains valid while this
+ * Identifier object is not modified or destroyed.
+ */
+ std::variant<std::uint32_t, std::string_view> Value() const noexcept {
+ if (kind_ == Kind::Numeric) {
+ return std::get<std::uint32_t>(value_);
+ }
+ return std::string_view(std::get<std::string>(value_));
+ }
+
+ private:
+ Identifier(Kind kind, std::variant<std::uint32_t, std::string> value) : kind_(kind), value_(std::move(value)) {}
+
+ static Identifier FromFfi(ffi::Identifier identifier);
+ ffi::Identifier ToFfi() const;
+
+ friend class IggyBlockingClient;
+
+ Kind kind_;
+ std::variant<std::uint32_t, std::string> value_;
+};
+
+/**
+ * @brief Type tag for a HeaderField payload.
+ *
+ * The tag and bytes use Iggy's header codec. Numeric payloads are little
+ * endian. Create TopicOption values instead of encoding catalog options by
+ * hand; use HeaderField directly only for application-defined headers.
+ */
+enum class HeaderKind : std::uint8_t {
+ Raw = 1,
+ String = 2,
+ Bool = 3,
+ Int8 = 4,
+ Int16 = 5,
+ Int32 = 6,
+ Int64 = 7,
+ Int128 = 8,
+ Uint8 = 9,
+ Uint16 = 10,
+ Uint32 = 11,
+ Uint64 = 12,
+ Uint128 = 13,
+ Float32 = 14,
+ Float64 = 15,
+};
+
+/**
+ * @brief One typed header key or value.
+ *
+ * HeaderField maps directly to Rust's header-field representation. Create()
+ * preserves the supplied bytes and does not verify that they match @p kind.
+ * Invalid key or value encodings are rejected when a request is sent.
+ */
+class HeaderField final {
+ public:
+ /**
+ * @brief Creates a typed header field from wire-encoded bytes.
+ * @param kind Type tag for @p value.
+ * @param value Payload encoded according to @p kind.
+ * @return Header field containing the supplied type and bytes.
+ */
+ static HeaderField Create(HeaderKind kind, std::vector<std::uint8_t> value) {
+ return HeaderField(kind, std::move(value));
+ }
+
+ /**
+ * @brief Returns the wire type of Value().
+ * @return Header type tag.
+ */
+ HeaderKind Kind() const noexcept { return kind_; }
+
+ /**
+ * @brief Returns bytes owned by this field.
+ * @return Payload encoded according to Kind().
+ */
+ const std::vector<std::uint8_t> &Value() const noexcept { return value_; }
+
+ private:
+ HeaderField(HeaderKind kind, std::vector<std::uint8_t> value) : kind_(kind), value_(std::move(value)) {}
+
+ static HeaderField FromFfi(ffi::HeaderField field);
+ static ffi::HeaderField ToFfi(HeaderField field);
+
+ friend class HeaderEntry;
+
+ HeaderKind kind_;
+ std::vector<std::uint8_t> value_;
+};
+
+/**
+ * @brief One typed header key-value pair.
+ *
+ * This maps directly to Rust's HeaderEntry. Topic options cross the C++ bridge
+ * as header entries because their keys and values use the same typed codec as
+ * message user headers.
+ */
+class HeaderEntry final {
+ public:
+ /**
+ * @brief Creates a header entry from its typed key and value.
+ * @param key Typed entry key.
+ * @param value Typed entry value.
+ * @return Header entry containing @p key and @p value.
+ */
+ static HeaderEntry Create(HeaderField key, HeaderField value) {
+ return HeaderEntry(std::move(key), std::move(value));
+ }
+
+ /**
+ * @brief Returns the typed key.
+ * @return Key owned by this entry.
+ */
+ const HeaderField &Key() const noexcept { return key_; }
+
+ /**
+ * @brief Returns the typed value.
+ * @return Value owned by this entry.
+ */
+ const HeaderField &Value() const noexcept { return value_; }
+
+ private:
+ HeaderEntry(HeaderField key, HeaderField value) : key_(std::move(key)), value_(std::move(value)) {}
+
+ static HeaderEntry FromFfi(ffi::HeaderEntry entry);
+ static ffi::HeaderEntry ToFfi(HeaderEntry entry);
+
+ friend class ResourceOptions;
+
+ HeaderField key_;
+ HeaderField value_;
+};
+
+/**
+ * @brief Creation options attached to a stream or topic.
+ *
+ * This maps to Rust's ResourceOptions. Explicit() contains entries supplied by
+ * the creating client. Derived() contains values the server resolved from its
+ * configuration at admission time. Derived values describe that resource's
+ * history and may differ if the resource is recreated under another server
+ * configuration.
+ *
+ * Construct creation options with, for example,
+ * `ResourceOptions::Explicit({TopicOption::SegmentSize(1024 * 1024)})`.
+ * Only explicit entries are sent; derived entries returned by Options() are
+ * not resubmitted.
+ */
+class ResourceOptions final {
+ public:
+ /**
+ * @brief Creates an empty option collection.
+ * @return Resource options with no explicit or derived entries.
+ */
+ static ResourceOptions Empty() { return ResourceOptions({}); }
+
+ /**
+ * @brief Creates request options from entries selected by the caller.
+ * @param entries Explicit option entries.
+ * @return Resource options that will submit @p entries.
+ */
+ static ResourceOptions Explicit(std::vector<HeaderEntry> entries);
+
+ /**
+ * @brief Returns entries supplied explicitly at resource creation.
+ * @return Explicit entries owned by this option collection.
+ */
+ const std::vector<HeaderEntry> &Explicit() const noexcept { return explicit_; }
+
+ /**
+ * @brief Returns entries derived from configured defaults at admission.
+ * @return Derived entries owned by this option collection.
+ * @note Stream responses currently expose explicit entries only, so this
+ * collection is empty for Stream and StreamDetails.
+ */
+ const std::vector<HeaderEntry> &Derived() const noexcept { return derived_; }
+
+ private:
+ explicit ResourceOptions(std::vector<HeaderEntry> explicit_entries) : explicit_(std::move(explicit_entries)) {}
+ ResourceOptions(std::vector<HeaderEntry> explicit_entries, std::vector<HeaderEntry> derived_entries)
+ : explicit_(std::move(explicit_entries)), derived_(std::move(derived_entries)) {}
+
+ static ResourceOptions FromFfi(rust::Vec<ffi::HeaderEntry> explicit_entries,
+ rust::Vec<ffi::HeaderEntry> derived_entries);
+ static rust::Vec<ffi::HeaderEntry> ToFfi(ResourceOptions options);
+
+ friend class IggyBlockingClient;
+ friend class Topic;
+ friend class TopicDetails;
+ friend class Stream;
+ friend class StreamDetails;
+
+ std::vector<HeaderEntry> explicit_;
+ std::vector<HeaderEntry> derived_;
+};
+
+/**
+ * @brief Topic summary returned within StreamDetails.
+ *
+ * A Topic owns its string fields and contains no Rust bridge values. It
+ * describes server state observed during the enclosing stream read, not a live
+ * view. Its aggregate statistics and partition count can change immediately
+ * after that request completes.
+ *
+ * Options() distinguishes values selected by the creating client from values
+ * the server derived during admission. Partition details require GetTopic().
+ */
+class Topic final {
+ public:
+ /**
+ * @brief Returns the numeric topic ID assigned within its stream.
+ * @return Numeric topic ID.
+ */
+ std::uint32_t Id() const noexcept { return id_; }
+
+ /**
+ * @brief Returns the server creation timestamp.
+ * @return Timestamp in microseconds.
+ */
+ std::uint64_t CreatedAt() const noexcept { return created_at_; }
+
+ /**
+ * @brief Returns the topic name.
+ * @return Name owned by this value.
+ */
+ const std::string &Name() const noexcept { return name_; }
+
+ /**
+ * @brief Returns the aggregate retained topic size.
+ * @return Size in bytes.
+ */
+ std::uint64_t SizeBytes() const noexcept { return size_bytes_; }
+
+ /**
+ * @brief Returns the server-encoded message retention value.
+ * @return Retention value in microseconds or a protocol sentinel.
+ */
+ std::uint64_t MessageExpiry() const noexcept { return message_expiry_; }
+
+ /**
+ * @brief Returns the server-selected storage compression algorithm.
+ * @return Algorithm name owned by this value.
+ */
+ const std::string &CompressionAlgorithm() const noexcept { return compression_algorithm_; }
+
+ /**
+ * @brief Returns the configured maximum retained topic size.
+ * @return Maximum size in bytes.
+ */
+ std::uint64_t MaxTopicSize() const noexcept { return max_topic_size_; }
+
+ /**
+ * @brief Returns the aggregate number of retained messages.
+ * @return Message count.
+ */
+ std::uint64_t MessagesCount() const noexcept { return messages_count_; }
+
+ /**
+ * @brief Returns the number of partitions belonging to this topic.
+ * @return Partition count.
+ */
+ std::uint32_t PartitionsCount() const noexcept { return partitions_count_; }
+
+ /**
+ * @brief Returns topic creation options and their admission provenance.
+ * @return Options owned by this value.
+ */
+ const ResourceOptions &Options() const noexcept { return options_; }
+
+ private:
+ Topic(std::uint32_t id,
+ std::uint64_t created_at,
+ std::string name,
+ std::uint64_t size_bytes,
+ std::uint64_t message_expiry,
+ std::string compression_algorithm,
+ std::uint64_t max_topic_size,
+ std::uint64_t messages_count,
+ std::uint32_t partitions_count,
+ ResourceOptions options)
+ : id_(id),
+ created_at_(created_at),
+ name_(std::move(name)),
+ size_bytes_(size_bytes),
+ message_expiry_(message_expiry),
+ compression_algorithm_(std::move(compression_algorithm)),
+ max_topic_size_(max_topic_size),
+ messages_count_(messages_count),
+ partitions_count_(partitions_count),
+ options_(std::move(options)) {}
+
+ static Topic FromFfi(ffi::Topic topic);
+
+ friend class IggyBlockingClient;
+ friend class StreamDetails;
+
+ std::uint32_t id_;
+ std::uint64_t created_at_;
+ std::string name_;
+ std::uint64_t size_bytes_;
+ std::uint64_t message_expiry_;
+ std::string compression_algorithm_;
+ std::uint64_t max_topic_size_;
+ std::uint64_t messages_count_;
+ std::uint32_t partitions_count_;
+ ResourceOptions options_;
+};
+
+/**
+ * @brief Partition metadata returned within TopicDetails.
+ *
+ * This is an observed summary, not a live partition handle. Offsets and
+ * statistics can change immediately after GetTopic() returns.
+ */
+class Partition final {
+ public:
+ /**
+ * @brief Returns the numeric partition ID within its topic.
+ * @return Numeric partition ID.
+ */
+ std::uint32_t Id() const noexcept { return id_; }
+
+ /**
+ * @brief Returns the server creation timestamp.
+ * @return Timestamp in microseconds.
+ */
+ std::uint64_t CreatedAt() const noexcept { return created_at_; }
+
+ /**
+ * @brief Returns the number of retained storage segments.
+ * @return Segment count.
+ */
+ std::uint32_t SegmentsCount() const noexcept { return segments_count_; }
+
+ /**
+ * @brief Returns the current server-observed message offset.
+ * @return Current message offset.
+ */
+ std::uint64_t CurrentOffset() const noexcept { return current_offset_; }
+
+ /**
+ * @brief Returns the retained partition size.
+ * @return Size in bytes.
+ */
+ std::uint64_t SizeBytes() const noexcept { return size_bytes_; }
+
+ /**
+ * @brief Returns the number of retained messages.
+ * @return Message count.
+ */
+ std::uint64_t MessagesCount() const noexcept { return messages_count_; }
+
+ private:
+ Partition(std::uint32_t id,
+ std::uint64_t created_at,
+ std::uint32_t segments_count,
+ std::uint64_t current_offset,
+ std::uint64_t size_bytes,
+ std::uint64_t messages_count)
+ : id_(id),
+ created_at_(created_at),
+ segments_count_(segments_count),
+ current_offset_(current_offset),
+ size_bytes_(size_bytes),
+ messages_count_(messages_count) {}
+
+ static Partition FromFfi(ffi::Partition partition);
+
+ friend class TopicDetails;
+
+ std::uint32_t id_;
+ std::uint64_t created_at_;
+ std::uint32_t segments_count_;
+ std::uint64_t current_offset_;
+ std::uint64_t size_bytes_;
+ std::uint64_t messages_count_;
+};
+/**
+ * @brief Topic metadata and partition summaries returned by topic detail calls.
+ *
+ * Partitions() contains one observed summary per partition. It does not expose
+ * segment metadata, messages, consumer offsets, or consumer-group membership.
+ */
+class TopicDetails final {
+ public:
+ /**
+ * @brief Returns the numeric topic ID within its stream.
+ * @return Numeric topic ID.
+ */
+ std::uint32_t Id() const noexcept { return id_; }
+
+ /**
+ * @brief Returns the server creation timestamp.
+ * @return Timestamp in microseconds.
+ */
+ std::uint64_t CreatedAt() const noexcept { return created_at_; }
+
+ /**
+ * @brief Returns the topic name.
+ * @return Name owned by this value.
+ */
+ const std::string &Name() const noexcept { return name_; }
+
+ /**
+ * @brief Returns the aggregate retained topic size.
+ * @return Size in bytes.
+ */
+ std::uint64_t SizeBytes() const noexcept { return size_bytes_; }
+
+ /**
+ * @brief Returns the server-encoded message retention value.
+ * @return Retention value in microseconds or a protocol sentinel.
+ */
+ std::uint64_t MessageExpiry() const noexcept { return message_expiry_; }
+
+ /**
+ * @brief Returns the storage compression algorithm selected for this topic.
+ * @return Algorithm name owned by this value.
+ */
+ const std::string &CompressionAlgorithm() const noexcept { return compression_algorithm_; }
+
+ /**
+ * @brief Returns the maximum retained size configured for this topic.
+ * @return Maximum size in bytes.
+ */
+ std::uint64_t MaxTopicSize() const noexcept { return max_topic_size_; }
+
+ /**
+ * @brief Returns the aggregate number of retained messages.
+ * @return Message count.
+ */
+ std::uint64_t MessagesCount() const noexcept { return messages_count_; }
+
+ /**
+ * @brief Returns the number of partitions belonging to this topic.
+ * @return Partition count.
+ */
+ std::uint32_t PartitionsCount() const noexcept { return partitions_count_; }
+
+ /**
+ * @brief Returns partition summaries.
+ * @return Summaries owned by this value.
+ */
+ const std::vector<Partition> &Partitions() const noexcept { return partitions_; }
+
+ /**
+ * @brief Returns topic creation options and their admission provenance.
+ * @return Options owned by this value.
+ */
+ const ResourceOptions &Options() const noexcept { return options_; }
+
+ private:
+ TopicDetails(std::uint32_t id,
+ std::uint64_t created_at,
+ std::string name,
+ std::uint64_t size_bytes,
+ std::uint64_t message_expiry,
+ std::string compression_algorithm,
+ std::uint64_t max_topic_size,
+ std::uint64_t messages_count,
+ std::uint32_t partitions_count,
+ std::vector<Partition> partitions,
+ ResourceOptions options)
+ : id_(id),
+ created_at_(created_at),
+ name_(std::move(name)),
+ size_bytes_(size_bytes),
+ message_expiry_(message_expiry),
+ compression_algorithm_(std::move(compression_algorithm)),
+ max_topic_size_(max_topic_size),
+ messages_count_(messages_count),
+ partitions_count_(partitions_count),
+ partitions_(std::move(partitions)),
+ options_(std::move(options)) {}
+
+ static TopicDetails FromFfi(ffi::TopicDetails topic);
+
+ friend class IggyBlockingClient;
+
+ std::uint32_t id_;
+ std::uint64_t created_at_;
+ std::string name_;
+ std::uint64_t size_bytes_;
+ std::uint64_t message_expiry_;
+ std::string compression_algorithm_;
+ std::uint64_t max_topic_size_;
+ std::uint64_t messages_count_;
+ std::uint32_t partitions_count_;
+ std::vector<Partition> partitions_;
+ ResourceOptions options_;
+};
+
+namespace detail {
/** @brief Internal base for string-backed option types. */
template <typename Tag>
class StringTag {
@@ -57,6 +656,196 @@ class StringTag {
} // namespace detail
+/**
+ * @brief Snapshot of one stream's metadata and aggregate statistics.
+ *
+ * CreateStream() and GetStream() return this value after converting the Rust
+ * bridge response to ordinary C++ fields. It owns its name and topic
+ * collection, so no bridge-owned string or container leaks through the public
+ * API.
+ *
+ * The value describes the stream state observed by the server for one request.
+ * It is not a live view or an atomic snapshot of later stream, topic, or
+ * message activity. SizeBytes(), MessagesCount(), TopicsCount(), and Topics()
+ * can become stale immediately after the request completes when another client
+ * changes the stream.
+ *
+ * A newly created stream has no topics or messages, so CreateStream() returns
+ * zero for SizeBytes(), MessagesCount(), and TopicsCount(), with an empty
+ * Topics() collection. GetStream() returns the same aggregate fields and one
+ * Topic summary for each observed topic.
+ *
+ * Stream IDs identify a stream for its lifetime and remain stable when it is
+ * renamed. CreatedAt() is the server timestamp, in microseconds, recorded when
+ * the stream was created.
+ */
+class StreamDetails final {
+ public:
+ /**
+ * @brief Returns the numeric ID assigned by the server.
+ *
+ * This value can be passed to GetStream() while the stream exists. It is
+ * unchanged by a stream rename.
+ */
+ std::uint32_t Id() const noexcept { return id_; }
+
+ /**
+ * @brief Returns the server-recorded creation timestamp.
+ * @return Timestamp in microseconds.
+ */
+ std::uint64_t CreatedAt() const noexcept { return created_at_; }
+
+ /**
+ * @brief Returns the unique stream name observed by the server.
+ * @return Reference owned by this value. It remains valid until this
+ * StreamDetails object is modified or destroyed.
+ */
+ const std::string &Name() const noexcept { return name_; }
+
+ /**
+ * @brief Returns the aggregate retained size of all stream topics.
+ * @return Size in bytes observed by the server for this request.
+ */
+ std::uint64_t SizeBytes() const noexcept { return size_bytes_; }
+
+ /**
+ * @brief Returns the aggregate number of messages in all stream topics.
+ * @return Message count observed by the server for this request.
+ */
+ std::uint64_t MessagesCount() const noexcept { return messages_count_; }
+
+ /**
+ * @brief Returns the number of topics belonging to the stream.
+ * @return Topic count observed by the server for this request.
+ */
+ std::uint32_t TopicsCount() const noexcept { return topics_count_; }
+
+ /**
+ * @brief Returns the topic summaries observed by the server.
+ * @return Topic values owned by this StreamDetails object.
+ */
+ const std::vector<Topic> &Topics() const noexcept { return topics_; }
+
+ /**
+ * @brief Returns explicit stream creation options.
+ * @return Options owned by this value.
+ * @note The current bridge does not return derived stream options.
+ */
+ const ResourceOptions &Options() const noexcept { return options_; }
+
+ private:
+ StreamDetails(std::uint32_t id,
+ std::uint64_t created_at,
+ std::string name,
+ std::uint64_t size_bytes,
+ std::uint64_t messages_count,
+ std::uint32_t topics_count,
+ std::vector<Topic> topics,
+ ResourceOptions options)
+ : id_(id),
+ created_at_(created_at),
+ name_(std::move(name)),
+ size_bytes_(size_bytes),
+ messages_count_(messages_count),
+ topics_count_(topics_count),
+ topics_(std::move(topics)),
+ options_(std::move(options)) {}
+
+ static StreamDetails FromFfi(ffi::StreamDetails stream);
+
+ friend class IggyBlockingClient;
+
+ std::uint32_t id_;
+ std::uint64_t created_at_;
+ std::string name_;
+ std::uint64_t size_bytes_;
+ std::uint64_t messages_count_;
+ std::uint32_t topics_count_;
+ std::vector<Topic> topics_;
+ ResourceOptions options_;
+};
+
+/**
+ * @brief Summary of a stream returned by GetStreams().
+ *
+ * Unlike StreamDetails, this value does not include topic summaries. Its
+ * aggregate statistics describe the state observed by the server for one
+ * request and can become stale when another client modifies the stream.
+ */
+class Stream final {
+ public:
+ /**
+ * @brief Returns the numeric ID assigned by the server.
+ * @return Numeric stream ID.
+ */
+ std::uint32_t Id() const noexcept { return id_; }
+
+ /**
+ * @brief Returns the server-recorded creation timestamp.
+ * @return Timestamp in microseconds.
+ */
+ std::uint64_t CreatedAt() const noexcept { return created_at_; }
+
+ /**
+ * @brief Returns the stream name.
+ * @return Name owned by this value.
+ */
+ const std::string &Name() const noexcept { return name_; }
+
+ /**
+ * @brief Returns the aggregate retained stream size.
+ * @return Size in bytes.
+ */
+ std::uint64_t SizeBytes() const noexcept { return size_bytes_; }
+
+ /**
+ * @brief Returns the aggregate number of retained stream messages.
+ * @return Message count.
+ */
+ std::uint64_t MessagesCount() const noexcept { return messages_count_; }
+
+ /**
+ * @brief Returns the number of topics belonging to the stream.
+ * @return Topic count.
+ */
+ std::uint32_t TopicsCount() const noexcept { return topics_count_; }
+
+ /**
+ * @brief Returns explicit stream creation options.
+ * @return Options owned by this value.
+ * @note The current bridge does not return derived stream options.
+ */
+ const ResourceOptions &Options() const noexcept { return options_; }
+
+ private:
+ Stream(std::uint32_t id,
+ std::uint64_t created_at,
+ std::string name,
+ std::uint64_t size_bytes,
+ std::uint64_t messages_count,
+ std::uint32_t topics_count,
+ ResourceOptions options)
+ : id_(id),
+ created_at_(created_at),
+ name_(std::move(name)),
+ size_bytes_(size_bytes),
+ messages_count_(messages_count),
+ topics_count_(topics_count),
+ options_(std::move(options)) {}
+
+ static Stream FromFfi(ffi::Stream stream);
+
+ friend class IggyBlockingClient;
+
+ std::uint32_t id_;
+ std::uint64_t created_at_;
+ std::string name_;
+ std::uint64_t size_bytes_;
+ std::uint64_t messages_count_;
+ std::uint32_t topics_count_;
+ ResourceOptions options_;
+};
+
/**
* @brief Compression algorithm used for topic messages.
*
@@ -210,31 +999,6 @@ class MaxTopicSize final : private detail::StringTag<MaxTopicSize> {
explicit MaxTopicSize(std::string max_topic_size) : detail::StringTag<MaxTopicSize>(std::move(max_topic_size)) {}
};
-// TODO(slbotbm): Add rust bindings for Identifier that will use IdKind
-/**
- * @brief Identifies whether an identifier is numeric or text-based.
- *
- * @note Reserved for future identifier bindings and not currently passed
- * across the Rust FFI.
- */
-class IdKind final : private detail::StringTag<IdKind> {
- public:
- /** @brief Uses a numeric identifier represented as a 32-bit integer. */
- static IdKind Numeric() { return IdKind("numeric"); }
-
- /** @brief Uses a string identifier represented by its text value. */
- static IdKind String() { return IdKind("string"); }
-
- /**
- * @brief Returns the value passed to the client implementation.
- * @return Identifier kind name.
- */
- std::string_view IdKindValue() const { return Value(); }
-
- private:
- explicit IdKind(std::string id_kind) : detail::StringTag<IdKind>(std::move(id_kind)) {}
-};
-
/**
* @brief Message retention policy for a topic.
*
@@ -339,8 +1103,8 @@ namespace detail {
/// Numeric option values are little-endian on the wire. Encoded byte by byte so
/// a big-endian host produces the same block as a little-endian one.
template <typename Value>
-rust::Vec<std::uint8_t> to_little_endian_bytes(const Value value) {
- rust::Vec<std::uint8_t> bytes{};
+std::vector<std::uint8_t> to_little_endian_bytes(const Value value) {
+ std::vector<std::uint8_t> bytes{};
bytes.reserve(sizeof(Value));
for (std::size_t index{}; index < sizeof(Value); ++index) {
bytes.push_back(static_cast<std::uint8_t>((value >> (index * 8)) & 0xFF));
@@ -349,15 +1113,15 @@ rust::Vec<std::uint8_t> to_little_endian_bytes(const Value value) {
return bytes;
}
-inline rust::Vec<std::uint8_t> to_bool_bytes(const bool value) {
- rust::Vec<std::uint8_t> bytes{};
+inline std::vector<std::uint8_t> to_bool_bytes(const bool value) {
+ std::vector<std::uint8_t> bytes{};
bytes.push_back(static_cast<std::uint8_t>(value ? 1 : 0));
return bytes;
}
-inline rust::Vec<std::uint8_t> to_key_bytes(const std::string_view key) {
- rust::Vec<std::uint8_t> bytes{};
+inline std::vector<std::uint8_t> to_key_bytes(const std::string_view key) {
+ std::vector<std::uint8_t> bytes{};
bytes.reserve(key.size());
for (const char character : key) {
bytes.push_back(static_cast<std::uint8_t>(character));
@@ -366,42 +1130,32 @@ inline rust::Vec<std::uint8_t> to_key_bytes(const std::string_view key) {
return bytes;
}
-inline iggy::ffi::HeaderField to_header_field(const iggy::ffi::HeaderKind kind, rust::Vec<std::uint8_t> value) {
- iggy::ffi::HeaderField field{};
- field.kind = static_cast<std::uint8_t>(kind);
- field.value = std::move(value);
-
- return field;
-}
-
/// An option key is always `String`-kinded. Only the value kind varies per key.
-inline iggy::ffi::HeaderEntry to_option_entry(const std::string_view key,
- const iggy::ffi::HeaderKind value_kind,
- rust::Vec<std::uint8_t> value) {
- iggy::ffi::HeaderEntry entry{};
- entry.key = to_header_field(iggy::ffi::HeaderKind::String, to_key_bytes(key));
- entry.value = to_header_field(value_kind, std::move(value));
-
- return entry;
+inline HeaderEntry to_option_entry(const std::string_view key,
+ const HeaderKind value_kind,
+ std::vector<std::uint8_t> value) {
+ return HeaderEntry::Create(HeaderField::Create(HeaderKind::String, to_key_bytes(key)),
+ HeaderField::Create(value_kind, std::move(value)));
}
} // namespace detail
/**
- * @brief Creates topic option entries for `Client::create_topic(...)`.
+ * @brief Creates catalog entries for ResourceOptions::Explicit().
*
* Each factory encodes one key from the server's topic option catalog using
* that key's required value kind. The server rejects unknown keys and values
* encoded with a different kind.
*
- * Use `Client::describe_options("topic")` to retrieve the supported keys,
- * value kinds, and defaults. Binary transport errors do not identify a rejected
- * key, so the catalog is the discovery mechanism for those transports.
+ * For example, pass
+ * `ResourceOptions::Explicit({TopicOption::SegmentSize(1024 * 1024)})` to
+ * CreateTopic(). The server rejects unknown keys and values encoded with a
+ * different kind.
*
- * @note These options are accepted only during topic creation.
- * `Client::update_topic(...)` rejects them because they define how
- * partition storage is created. Changing them later could leave existing
- * and new segments with different storage settings.
+ * @note These options are accepted only during topic creation. UpdateTopic()
+ * rejects them because they define how partition storage is created.
+ * Changing them later could leave existing and new segments with
+ * different storage settings.
*/
class TopicOption final {
public:
@@ -414,9 +1168,8 @@ class TopicOption final {
* @param bytes Segment size in bytes.
* @return Encoded topic option entry.
*/
- static iggy::ffi::HeaderEntry SegmentSize(const std::uint64_t bytes) {
- return detail::to_option_entry("segment_size", iggy::ffi::HeaderKind::Uint64,
- detail::to_little_endian_bytes(bytes));
+ static HeaderEntry SegmentSize(const std::uint64_t bytes) {
+ return detail::to_option_entry("segment_size", HeaderKind::Uint64, detail::to_little_endian_bytes(bytes));
}
/**
@@ -425,8 +1178,8 @@ class TopicOption final {
* @param enabled Whether partition writes are fsynced.
* @return Encoded topic option entry.
*/
- static iggy::ffi::HeaderEntry EnforceFsync(const bool enabled) {
- return detail::to_option_entry("enforce_fsync", iggy::ffi::HeaderKind::Bool, detail::to_bool_bytes(enabled));
+ static HeaderEntry EnforceFsync(const bool enabled) {
+ return detail::to_option_entry("enforce_fsync", HeaderKind::Bool, detail::to_bool_bytes(enabled));
}
/**
@@ -439,8 +1192,8 @@ class TopicOption final {
* @param messages Message count at which to flush the journal.
* @return Encoded topic option entry.
*/
- static iggy::ffi::HeaderEntry MessagesRequiredToSave(const std::uint32_t messages) {
- return detail::to_option_entry("messages_required_to_save", iggy::ffi::HeaderKind::Uint32,
+ static HeaderEntry MessagesRequiredToSave(const std::uint32_t messages) {
+ return detail::to_option_entry("messages_required_to_save", HeaderKind::Uint32,
detail::to_little_endian_bytes(messages));
}
@@ -453,8 +1206,8 @@ class TopicOption final {
* @param bytes Byte count at which to flush the journal.
* @return Encoded topic option entry.
*/
- static iggy::ffi::HeaderEntry SizeOfMessagesRequiredToSave(const std::uint64_t bytes) {
- return detail::to_option_entry("size_of_messages_required_to_save", iggy::ffi::HeaderKind::Uint64,
+ static HeaderEntry SizeOfMessagesRequiredToSave(const std::uint64_t bytes) {
+ return detail::to_option_entry("size_of_messages_required_to_save", HeaderKind::Uint64,
detail::to_little_endian_bytes(bytes));
}
@@ -467,24 +1220,14 @@ class TopicOption final {
* @param enabled Whether to reserve segment storage on disk.
* @return Encoded topic option entry.
*/
- static iggy::ffi::HeaderEntry PreallocateSegments(const bool enabled) {
- return detail::to_option_entry("preallocate_segments", iggy::ffi::HeaderKind::Bool,
- detail::to_bool_bytes(enabled));
+ static HeaderEntry PreallocateSegments(const bool enabled) {
+ return detail::to_option_entry("preallocate_segments", HeaderKind::Bool, detail::to_bool_bytes(enabled));
}
private:
TopicOption() = delete;
};
-/**
- * @brief Exception thrown when an Iggy client operation fails.
- */
-class IggyException : public std::runtime_error {
- public:
- explicit IggyException(const char *message) : std::runtime_error(message) {}
- explicit IggyException(const std::string &message) : std::runtime_error(message) {}
-};
-
/**
* @brief Owning client connection to an Apache Iggy server.
*
@@ -709,6 +1452,245 @@ class IggyBlockingClient final {
*/
void Logout();
+ /**
+ * @brief Creates a top-level stream in the cluster metadata.
+ *
+ * A stream is the top-level namespace for topics. This creates no topics,
+ * partitions, or messages. Its name must be unique, non-empty, and no more
+ * than 255 UTF-8 bytes.
+ *
+ * A transport failure after submission can leave the stream created. Look
+ * it up by name before retrying or choosing another name.
+ *
+ * @param name Unique stream name.
+ * @return Details of the newly created, topic-less stream.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * the name is invalid or already in use; the caller lacks
+ * stream-management permission; or the request fails.
+ */
+ StreamDetails CreateStream(std::string name);
+
+ /**
+ * @brief Renames a stream.
+ *
+ * @param stream Stream to rename, addressed by numeric ID or name.
+ * @param name New unique stream name.
+ * @throws IggyException if the client is unavailable, the caller lacks
+ * stream-management permission, either value is invalid, the stream
+ * does not exist, the name is already taken, or the request fails.
+ */
+ void UpdateStream(const Identifier &stream, std::string name);
+
+ /**
+ * @brief Lists stream summaries visible to the authenticated user.
+ *
+ * The summaries exclude per-topic details. Use GetStream() for those.
+ * @return Stream summaries visible to the authenticated user.
+ * @throws IggyException if the client is unavailable, the caller lacks
+ * permission to read streams, or the request fails.
+ */
+ std::vector<Stream> GetStreams();
+
+ /**
+ * @brief Retrieves one stream by numeric ID or name.
+ *
+ * The result includes observed aggregate statistics and topic summaries.
+ * It does not include partition details or messages, and its statistics can
+ * become stale immediately after the request completes.
+ *
+ * @param stream Stream to retrieve. Numeric IDs remain stable if a stream
+ * is renamed.
+ * @return Details for the requested stream.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * the stream does not exist; the caller lacks read permission; or
+ * the metadata read fails.
+ */
+ StreamDetails GetStream(const Identifier &stream);
+
+ /**
+ * @brief Deletes a stream and all of its topics, partitions, and messages.
+ *
+ * This is irreversible. A transport failure after submission can leave the
+ * deletion committed, so query the stream before retrying this request.
+ *
+ * @param stream Stream to delete, addressed by numeric ID or name.
+ * @throws IggyException if the client is unavailable, the caller lacks
+ * stream-management permission, the stream does not exist, or the
+ * request fails.
+ */
+ void DeleteStream(const Identifier &stream);
+
+ /**
+ * @brief Removes all messages from every topic in a stream.
+ *
+ * The stream, its topics, and topic configuration remain available. A
+ * transport failure after submission can still leave the purge committed.
+ * @param stream Stream to purge, addressed by numeric ID or name.
+ * @throws IggyException if the client is unavailable, the caller lacks
+ * stream-management permission, the stream does not exist, or the
+ * request fails.
+ */
+ void PurgeStream(const Identifier &stream);
+
+ /**
+ * @brief Creates a topic and its initial partitions in a stream.
+ *
+ * A topic is the message namespace within a stream. The server creates
+ * exactly @p partitions_count partitions and returns their initial
+ * summaries. Its name is unique only within the selected stream.
+ *
+ * @p options must contain only entries selected by the caller. Use
+ * ResourceOptions::Explicit() with TopicOption factories for catalog
+ * options. The server resolves omitted keys from its configuration and
+ * returns those choices in TopicDetails::Options().Derived().
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param name Unique topic name within @p stream.
+ * @param partitions_count Number of initial partitions.
+ * @param compression_algorithm Storage compression setting.
+ * @param message_expiry Message retention setting.
+ * @param max_topic_size Maximum retained topic size.
+ * @param options Explicit catalog options for the new topic.
+ * @return Metadata and initial partition summaries for the created topic.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier, name, partition count, or option is invalid; the
+ * stream does not exist; the caller lacks topic-management
+ * permission; or the server rejects or cannot commit the write.
+ */
+ TopicDetails CreateTopic(const Identifier &stream,
+ std::string name,
+ std::uint32_t partitions_count,
+ CompressionAlgorithm compression_algorithm = CompressionAlgorithm::None(),
+ Expiry message_expiry = Expiry::ServerDefault(),
+ MaxTopicSize max_topic_size = MaxTopicSize::ServerDefault(),
+ ResourceOptions options = ResourceOptions::Empty());
+
+ /**
+ * @brief Renames a topic and updates its mutable configuration.
+ *
+ * The partition count and creation-only options, including segment size,
+ * fsync policy, flush thresholds, and segment preallocation, cannot be
+ * changed here. Default arguments leave compression, expiry, and maximum
+ * size unchanged. The current update catalog exposes those three mutable
+ * settings only through their named parameters, so leave @p options empty.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Topic to update, addressed by numeric ID or name.
+ * @param name New unique topic name within @p stream.
+ * @param compression_algorithm New compression setting, or the default to
+ * preserve the existing setting.
+ * @param message_expiry New retention setting, or the default to preserve
+ * the existing setting.
+ * @param max_topic_size New maximum size, or the default to preserve the
+ * existing setting.
+ * @param options Explicit mutable catalog options.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier, name, setting, or option is invalid; the stream or
+ * topic does not exist; the caller lacks permission; or the server
+ * rejects or cannot commit the write.
+ */
+ void UpdateTopic(const Identifier &stream,
+ const Identifier &topic,
+ std::string name,
+ CompressionAlgorithm compression_algorithm = CompressionAlgorithm::None(),
+ Expiry message_expiry = Expiry::ServerDefault(),
+ MaxTopicSize max_topic_size = MaxTopicSize::ServerDefault(),
+ ResourceOptions options = ResourceOptions::Empty());
+
+ /**
+ * @brief Lists topic summaries in a stream.
+ *
+ * The returned summaries do not include partition details. Use GetTopic()
+ * when partition offsets, sizes, and segment counts are needed.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @return Topic summaries visible to the authenticated user.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * the stream does not exist; the caller lacks read permission; or
+ * the metadata read fails.
+ */
+ std::vector<Topic> GetTopics(const Identifier &stream);
+
+ /**
+ * @brief Retrieves one topic and its partition summaries.
+ *
+ * The result is an observed metadata read. Partition offsets and retained
+ * statistics can change immediately after this call returns.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Topic to retrieve, addressed by numeric ID or name.
+ * @return Topic metadata and one summary per partition.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * the stream or topic does not exist; the caller lacks read
+ * permission; or the metadata read fails.
+ */
+ TopicDetails GetTopic(const Identifier &stream, const Identifier &topic);
+
+ /**
+ * @brief Deletes a topic, its partitions, and retained messages.
+ *
+ * A failed or unknown transport outcome can leave the deletion committed.
+ * Query the topic before retrying a destructive request.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Topic to delete, addressed by numeric ID or name.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * the stream or topic does not exist; the caller lacks
+ * topic-management permission; or the server rejects or cannot
+ * commit the write.
+ */
+ void DeleteTopic(const Identifier &stream, const Identifier &topic);
+
+ /**
+ * @brief Removes retained messages from every partition of a topic.
+ *
+ * The topic, its partitions, names, and configuration remain. New messages
+ * can be sent after a purge. A failed or unknown transport outcome can
+ * still leave the purge committed.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Topic to purge, addressed by numeric ID or name.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * the stream or topic does not exist; the caller lacks
+ * topic-management permission; or the server rejects or cannot
+ * commit the write.
+ */
+ void PurgeTopic(const Identifier &stream, const Identifier &topic);
+
+ /**
+ * @brief Adds partitions to a topic.
+ *
+ * New partitions receive IDs after the topic's existing partitions. The
+ * requested count must be between 1 and 1000. A transport failure after
+ * submission can still leave the partitions created.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Topic to extend, addressed by numeric ID or name.
+ * @param partitions_count Number of partitions to add.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier or count is invalid; the stream or topic does not
+ * exist; the caller lacks topic-management permission; or the
+ * request fails.
+ */
+ void CreatePartitions(const Identifier &stream, const Identifier &topic, std::uint32_t partitions_count);
+
+ /**
+ * @brief Deletes the highest-numbered partitions from a topic.
+ *
+ * The deleted partitions and their retained messages are removed. The
+ * requested count must be between 1 and 1000. A transport failure after
+ * submission can still leave the deletion committed.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Topic to shrink, addressed by numeric ID or name.
+ * @param partitions_count Number of partitions to delete.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier or count is invalid; the stream or topic does not
+ * exist; the caller lacks topic-management permission; or the
+ * request fails.
+ */
+ void DeletePartitions(const Identifier &stream, const Identifier &topic, std::uint32_t partitions_count);
+
private:
explicit IggyBlockingClient(ffi::Client *client);
diff --git a/foreign/cpp/src/client.cpp b/foreign/cpp/src/client.cpp
index f81e7956c..8b083fbd7 100644
--- a/foreign/cpp/src/client.cpp
+++ b/foreign/cpp/src/client.cpp
@@ -54,14 +54,124 @@ void IggyBlockingClient::Shutdown() {
}
LoginInfo IggyBlockingClient::Login(std::string username, std::string password) {
- return RethrowAsIggyException(
- [this, &username, &password] { return Handle()->login_user(std::move(username), std::move(password)); });
+ return RethrowAsIggyException([this, &username, &password] {
+ return LoginInfo::FromFfi(Handle()->login_user(std::move(username), std::move(password)));
+ });
}
void IggyBlockingClient::Logout() {
RethrowAsIggyException([this] { Handle()->logout_user(); });
}
+StreamDetails IggyBlockingClient::CreateStream(std::string name) {
+ return RethrowAsIggyException(
+ [this, &name] { return StreamDetails::FromFfi(Handle()->create_stream(std::move(name))); });
+}
+
+void IggyBlockingClient::UpdateStream(const Identifier &stream, std::string name) {
+ return RethrowAsIggyException([this, &stream, &name] { Handle()->update_stream(stream.ToFfi(), std::move(name)); });
+}
+
+std::vector<Stream> IggyBlockingClient::GetStreams() {
+ return RethrowAsIggyException([this] {
+ std::vector<Stream> streams;
+ auto ffi_streams = Handle()->get_streams();
+ streams.reserve(ffi_streams.size());
+ for (auto &stream : ffi_streams) {
+ streams.push_back(Stream::FromFfi(std::move(stream)));
+ }
+ return streams;
+ });
+}
+
+StreamDetails IggyBlockingClient::GetStream(const Identifier &stream) {
+ return RethrowAsIggyException(
+ [this, &stream] { return StreamDetails::FromFfi(Handle()->get_stream(stream.ToFfi())); });
+}
+
+void IggyBlockingClient::DeleteStream(const Identifier &stream) {
+ return RethrowAsIggyException([this, &stream] { Handle()->delete_stream(stream.ToFfi()); });
+}
+
+void IggyBlockingClient::PurgeStream(const Identifier &stream) {
+ return RethrowAsIggyException([this, &stream] { Handle()->purge_stream(stream.ToFfi()); });
+}
+
+TopicDetails IggyBlockingClient::CreateTopic(const Identifier &stream,
+ std::string name,
+ const std::uint32_t partitions_count,
+ const CompressionAlgorithm compression_algorithm,
+ const Expiry message_expiry,
+ const MaxTopicSize max_topic_size,
+ ResourceOptions options) {
+ return RethrowAsIggyException(
+ [this, &stream, &name, partitions_count, &compression_algorithm, &message_expiry, &max_topic_size, &options] {
+ return TopicDetails::FromFfi(Handle()->create_topic(
+ stream.ToFfi(), std::move(name), partitions_count,
+ std::string(compression_algorithm.CompressionAlgorithmValue()),
+ std::string(message_expiry.ExpiryKind()), message_expiry.ExpiryValue(),
+ std::string(max_topic_size.MaxTopicSizeValue()), ResourceOptions::ToFfi(std::move(options))));
+ });
+}
+
+void IggyBlockingClient::UpdateTopic(const Identifier &stream,
+ const Identifier &topic,
+ std::string name,
+ const CompressionAlgorithm compression_algorithm,
+ const Expiry message_expiry,
+ const MaxTopicSize max_topic_size,
+ ResourceOptions options) {
+ return RethrowAsIggyException(
+ [this, &stream, &topic, &name, &compression_algorithm, &message_expiry, &max_topic_size, &options] {
+ Handle()->update_topic(stream.ToFfi(), topic.ToFfi(), std::move(name),
+ std::string(compression_algorithm.CompressionAlgorithmValue()),
+ std::string(message_expiry.ExpiryKind()), message_expiry.ExpiryValue(),
+ std::string(max_topic_size.MaxTopicSizeValue()),
+ ResourceOptions::ToFfi(std::move(options)));
+ });
+}
+
+std::vector<Topic> IggyBlockingClient::GetTopics(const Identifier &stream) {
+ return RethrowAsIggyException([this, &stream] {
+ std::vector<Topic> topics;
+ auto ffi_topics = Handle()->get_topics(stream.ToFfi());
+ topics.reserve(ffi_topics.size());
+ for (auto &topic : ffi_topics) {
+ topics.push_back(Topic::FromFfi(std::move(topic)));
+ }
+ return topics;
+ });
+}
+
+TopicDetails IggyBlockingClient::GetTopic(const Identifier &stream, const Identifier &topic) {
+ return RethrowAsIggyException(
+ [this, &stream, &topic] { return TopicDetails::FromFfi(Handle()->get_topic(stream.ToFfi(), topic.ToFfi())); });
+}
+
+void IggyBlockingClient::DeleteTopic(const Identifier &stream, const Identifier &topic) {
+ return RethrowAsIggyException([this, &stream, &topic] { Handle()->delete_topic(stream.ToFfi(), topic.ToFfi()); });
+}
+
+void IggyBlockingClient::PurgeTopic(const Identifier &stream, const Identifier &topic) {
+ return RethrowAsIggyException([this, &stream, &topic] { Handle()->purge_topic(stream.ToFfi(), topic.ToFfi()); });
+}
+
+void IggyBlockingClient::CreatePartitions(const Identifier &stream,
+ const Identifier &topic,
+ const std::uint32_t partitions_count) {
+ return RethrowAsIggyException([this, &stream, &topic, partitions_count] {
+ Handle()->create_partitions(stream.ToFfi(), topic.ToFfi(), partitions_count);
+ });
+}
+
+void IggyBlockingClient::DeletePartitions(const Identifier &stream,
+ const Identifier &topic,
+ const std::uint32_t partitions_count) {
+ return RethrowAsIggyException([this, &stream, &topic, partitions_count] {
+ Handle()->delete_partitions(stream.ToFfi(), topic.ToFfi(), partitions_count);
+ });
+}
+
IggyBlockingClient::IggyBlockingClient(ffi::Client *client) : client_(client) {
if (client_ == nullptr) {
throw IggyException("Could not create Iggy client");
diff --git a/foreign/cpp/src/type_conversion.rs b/foreign/cpp/src/type_conversion.rs
index 83d4c52b6..e643b7b4e 100644
--- a/foreign/cpp/src/type_conversion.rs
+++ b/foreign/cpp/src/type_conversion.rs
@@ -499,8 +499,8 @@ fn resource_options_to_ffi(
/// value. A key this build cannot read at all is dropped rather than guessed.
pub(crate) fn ffi_options_to_raw(
options: Vec<ffi::HeaderEntry>,
-) -> Result<std::collections::BTreeMap<String, String>, String> {
- let mut raw = std::collections::BTreeMap::new();
+) -> Result<BTreeMap<String, String>, String> {
+ let mut raw = BTreeMap::new();
for entry in options {
let RustHeaderEntry { key, value } = RustHeaderEntry::try_from(entry)?;
let key = key
diff --git a/foreign/cpp/src/type_conversions.cpp b/foreign/cpp/src/type_conversions.cpp
new file mode 100644
index 000000000..81e17c0cd
--- /dev/null
+++ b/foreign/cpp/src/type_conversions.cpp
@@ -0,0 +1,165 @@
+/*
+ * 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 {
+
+LoginInfo LoginInfo::FromFfi(ffi::LoginInfo login_info) {
+ std::optional<std::string> access_token;
+ std::optional<std::uint64_t> access_token_expiry;
+ if (login_info.has_access_token) {
+ access_token.emplace(login_info.access_token.c_str(), login_info.access_token.size());
+ access_token_expiry = login_info.access_token_expiry;
+ }
+
+ return LoginInfo(login_info.user_id, std::move(access_token), access_token_expiry);
+}
+
+Identifier Identifier::FromFfi(ffi::Identifier identifier) {
+ const std::string kind(identifier.kind.c_str(), identifier.kind.size());
+ if (kind == "numeric") {
+ if (identifier.length != sizeof(std::uint32_t) || identifier.value.size() != sizeof(std::uint32_t)) {
+ throw IggyException("Invalid numeric identifier returned by Rust");
+ }
+ const auto id = static_cast<std::uint32_t>(identifier.value[0]) |
+ (static_cast<std::uint32_t>(identifier.value[1]) << 8U) |
+ (static_cast<std::uint32_t>(identifier.value[2]) << 16U) |
+ (static_cast<std::uint32_t>(identifier.value[3]) << 24U);
+ return Numeric(id);
+ }
+ if (kind == "string") {
+ if (identifier.length != identifier.value.size()) {
+ throw IggyException("Invalid string identifier returned by Rust");
+ }
+ return String(std::string(identifier.value.begin(), identifier.value.end()));
+ }
+ throw IggyException("Invalid identifier kind returned by Rust");
+}
+
+ffi::Identifier Identifier::ToFfi() const {
+ ffi::Identifier identifier{};
+ if (kind_ == Kind::Numeric) {
+ identifier.set_numeric(std::get<std::uint32_t>(value_));
+ } else {
+ identifier.set_string(std::get<std::string>(value_));
+ }
+ return identifier;
+}
+
+HeaderField HeaderField::FromFfi(ffi::HeaderField field) {
+ return HeaderField(static_cast<HeaderKind>(field.kind),
+ std::vector<std::uint8_t>(field.value.begin(), field.value.end()));
+}
+
+ffi::HeaderField HeaderField::ToFfi(HeaderField field) {
+ ffi::HeaderField ffi_field{};
+ ffi_field.kind = static_cast<std::uint8_t>(field.kind_);
+ ffi_field.value.reserve(field.value_.size());
+ for (const auto byte : field.value_) {
+ ffi_field.value.push_back(byte);
+ }
+ return ffi_field;
+}
+
+HeaderEntry HeaderEntry::FromFfi(ffi::HeaderEntry entry) {
+ return HeaderEntry(HeaderField::FromFfi(std::move(entry.key)), HeaderField::FromFfi(std::move(entry.value)));
+}
+
+ffi::HeaderEntry HeaderEntry::ToFfi(HeaderEntry entry) {
+ ffi::HeaderEntry ffi_entry{};
+ ffi_entry.key = HeaderField::ToFfi(std::move(entry.key_));
+ ffi_entry.value = HeaderField::ToFfi(std::move(entry.value_));
+ return ffi_entry;
+}
+
+ResourceOptions ResourceOptions::Explicit(std::vector<HeaderEntry> entries) {
+ return ResourceOptions(std::move(entries));
+}
+
+ResourceOptions ResourceOptions::FromFfi(rust::Vec<ffi::HeaderEntry> explicit_entries,
+ rust::Vec<ffi::HeaderEntry> derived_entries) {
+ std::vector<HeaderEntry> explicit_options;
+ explicit_options.reserve(explicit_entries.size());
+ for (auto &entry : explicit_entries) {
+ explicit_options.push_back(HeaderEntry::FromFfi(std::move(entry)));
+ }
+ std::vector<HeaderEntry> derived_options;
+ derived_options.reserve(derived_entries.size());
+ for (auto &entry : derived_entries) {
+ derived_options.push_back(HeaderEntry::FromFfi(std::move(entry)));
+ }
+ return ResourceOptions(std::move(explicit_options), std::move(derived_options));
+}
+
+rust::Vec<ffi::HeaderEntry> ResourceOptions::ToFfi(ResourceOptions options) {
+ rust::Vec<ffi::HeaderEntry> headers{};
+ headers.reserve(options.explicit_.size());
+ for (auto &entry : options.explicit_) {
+ headers.push_back(HeaderEntry::ToFfi(std::move(entry)));
+ }
+ return headers;
+}
+
+Topic Topic::FromFfi(ffi::Topic topic) {
+ return Topic(topic.id, topic.created_at, std::string(topic.name.c_str(), topic.name.size()), topic.size_bytes,
+ topic.message_expiry,
+ std::string(topic.compression_algorithm.c_str(), topic.compression_algorithm.size()),
+ topic.max_topic_size, topic.messages_count, topic.partitions_count,
+ ResourceOptions::FromFfi(std::move(topic.options), std::move(topic.derived_options)));
+}
+
+Partition Partition::FromFfi(ffi::Partition partition) {
+ return Partition(partition.id, partition.created_at, partition.segments_count, partition.current_offset,
+ partition.size_bytes, partition.messages_count);
+}
+
+TopicDetails TopicDetails::FromFfi(ffi::TopicDetails topic) {
+ std::vector<Partition> partitions;
+ partitions.reserve(topic.partitions.size());
+ for (auto &partition : topic.partitions) {
+ partitions.push_back(Partition::FromFfi(std::move(partition)));
+ }
+
+ return TopicDetails(topic.id, topic.created_at, std::string(topic.name.c_str(), topic.name.size()),
+ topic.size_bytes, topic.message_expiry,
+ std::string(topic.compression_algorithm.c_str(), topic.compression_algorithm.size()),
+ topic.max_topic_size, topic.messages_count, topic.partitions_count, std::move(partitions),
+ ResourceOptions::FromFfi(std::move(topic.options), std::move(topic.derived_options)));
+}
+
+StreamDetails StreamDetails::FromFfi(ffi::StreamDetails stream) {
+ std::vector<Topic> topics;
+ topics.reserve(stream.topics.size());
+ for (auto &topic : stream.topics) {
+ topics.push_back(Topic::FromFfi(std::move(topic)));
+ }
+
+ return StreamDetails(stream.id, stream.created_at, std::string(stream.name.c_str(), stream.name.size()),
+ stream.size_bytes, stream.messages_count, stream.topics_count, std::move(topics),
+ ResourceOptions::FromFfi(std::move(stream.options), rust::Vec<ffi::HeaderEntry>{}));
+}
+
+Stream Stream::FromFfi(ffi::Stream stream) {
+ return Stream(stream.id, stream.created_at, std::string(stream.name.c_str(), stream.name.size()), stream.size_bytes,
+ stream.messages_count, stream.topics_count,
+ ResourceOptions::FromFfi(std::move(stream.options), rust::Vec<ffi::HeaderEntry>{}));
+}
+
+} // namespace iggy
diff --git a/foreign/cpp/tests/e2e/partition.cpp b/foreign/cpp/tests/e2e/partition.cpp
index 8f305db97..030bed90e 100644
--- a/foreign/cpp/tests/e2e/partition.cpp
+++ b/foreign/cpp/tests/e2e/partition.cpp
@@ -19,126 +19,110 @@
#include <cstdint>
#include <string>
+#include <vector>
#include <gtest/gtest.h>
-#include "lib.rs.h"
+#include "iggy.hpp"
#include "tests/e2e/test_helpers.hpp"
-class LowLevelE2E_Partition : public E2ETestFixture {};
+class E2E_Partition : public E2ETestFixture {};
-TEST_F(LowLevelE2E_Partition, CreatePartitionsSucceeds) {
+TEST_F(E2E_Partition, CreatePartitionsSucceeds) {
RecordProperty("description", "Creates partitions for an existing topic and verifies the resulting count.");
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, 1, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
ASSERT_NO_THROW(
- client->create_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 43));
+ client.CreatePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 43));
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_details.topics.size(), 1u);
- EXPECT_EQ(stream_details.topics[0].name, topic_name);
- EXPECT_EQ(stream_details.topics[0].partitions_count, 44u);
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_details.Topics().size(), 1u);
+ EXPECT_EQ(stream_details.Topics()[0].Name(), topic_name);
+ EXPECT_EQ(stream_details.Topics()[0].PartitionsCount(), 44u);
});
}
-TEST_F(LowLevelE2E_Partition, CreatePartitionsBeforeLoginThrows) {
+TEST_F(E2E_Partition, CreatePartitionsBeforeLoginThrows) {
RecordProperty("description",
"Throws when create_partitions is called before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedOutClient();
+ auto client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(client->create_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 1),
- std::exception);
- ASSERT_NO_THROW(client->connect());
- ASSERT_THROW(client->create_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 1),
- std::exception);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(client->disconnect());
- ASSERT_THROW(client->create_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 1),
- std::exception);
+ ASSERT_THROW(
+ client.CreatePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 1),
+ std::exception);
+ ASSERT_NO_THROW(client.Connect());
+ ASSERT_THROW(
+ client.CreatePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 1),
+ std::exception);
+ ASSERT_NO_THROW(client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(client.Disconnect());
+ ASSERT_THROW(
+ client.CreatePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 1),
+ std::exception);
}
-TEST_F(LowLevelE2E_Partition, CreatePartitionsOnNonExistentResourcesThrows) {
+TEST_F(E2E_Partition, CreatePartitionsOnNonExistentResourcesThrows) {
RecordProperty("description", "Throws when create_partitions is called for a stream or topic that does 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, 1, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
ASSERT_THROW(
- client->create_partitions(make_string_identifier(missing_stream_name), make_string_identifier(topic_name), 1),
+ client.CreatePartitions(iggy::Identifier::String(missing_stream_name), iggy::Identifier::String(topic_name), 1),
std::exception);
ASSERT_THROW(
- client->create_partitions(make_string_identifier(stream_name), make_string_identifier(missing_topic_name), 1),
+ client.CreatePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(missing_topic_name), 1),
std::exception);
}
-TEST_F(LowLevelE2E_Partition, CreatePartitionsWithInvalidIdentifiersThrows) {
- RecordProperty("description", "Rejects create_partitions requests that use invalid stream or topic identifiers.");
+TEST_F(E2E_Partition, CreatePartitionsWithInvalidIdentifiersThrows) {
+ RecordProperty("description", "Rejects invalid stream or topic identifiers before creating partitions.");
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, 1, "none", "server_default",
- 0, "server_default", {}));
-
- iggy::ffi::Identifier invalid_stream_kind_id;
- invalid_stream_kind_id.kind = "invalid";
- invalid_stream_kind_id.length = 4;
- invalid_stream_kind_id.value = {1, 0, 0, 0};
- ASSERT_THROW(client->create_partitions(std::move(invalid_stream_kind_id), make_string_identifier(topic_name), 1),
- std::exception);
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
- iggy::ffi::Identifier invalid_stream_numeric_id;
- invalid_stream_numeric_id.kind = "numeric";
- invalid_stream_numeric_id.length = 1;
- invalid_stream_numeric_id.value.push_back(1);
- ASSERT_THROW(client->create_partitions(std::move(invalid_stream_numeric_id), make_string_identifier(topic_name), 1),
+ ASSERT_THROW(client.CreatePartitions(iggy::Identifier::String(""), iggy::Identifier::String(topic_name), 1),
std::exception);
-
- iggy::ffi::Identifier invalid_topic_kind_id;
- invalid_topic_kind_id.kind = "invalid";
- invalid_topic_kind_id.length = 4;
- invalid_topic_kind_id.value = {1, 0, 0, 0};
- ASSERT_THROW(client->create_partitions(make_string_identifier(stream_name), std::move(invalid_topic_kind_id), 1),
+ ASSERT_THROW(client.CreatePartitions(iggy::Identifier::String(std::string(256, 'a')),
+ iggy::Identifier::String(topic_name), 1),
std::exception);
-
- iggy::ffi::Identifier invalid_topic_numeric_id;
- invalid_topic_numeric_id.kind = "numeric";
- invalid_topic_numeric_id.length = 1;
- invalid_topic_numeric_id.value.push_back(1);
- ASSERT_THROW(client->create_partitions(make_string_identifier(stream_name), std::move(invalid_topic_numeric_id), 1),
+ ASSERT_THROW(client.CreatePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(""), 1),
+ std::exception);
+ ASSERT_THROW(client.CreatePartitions(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(std::string(256, 'a')), 1),
std::exception);
}
-TEST_F(LowLevelE2E_Partition, CreatePartitionsWithBoundaryPartitionsCountValues) {
+TEST_F(E2E_Partition, CreatePartitionsWithBoundaryPartitionsCountValues) {
RecordProperty("description",
"Accepts supported create_partitions counts and rejects values outside the allowed range.");
const std::string stream_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);
struct TestCase {
@@ -159,29 +143,28 @@ TEST_F(LowLevelE2E_Partition, CreatePartitionsWithBoundaryPartitionsCountValues)
for (const auto &test_case : test_cases) {
SCOPED_TRACE(test_case.topic_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), test_case.topic_name, 1, "none",
- "server_default", 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), test_case.topic_name, 1));
if (test_case.should_succeed) {
- ASSERT_NO_THROW(client->create_partitions(make_string_identifier(stream_name),
- make_string_identifier(test_case.topic_name),
- test_case.partitions_count));
+ ASSERT_NO_THROW(client.CreatePartitions(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(test_case.topic_name),
+ test_case.partitions_count));
} else {
ASSERT_THROW(
- client->create_partitions(make_string_identifier(stream_name),
- make_string_identifier(test_case.topic_name), test_case.partitions_count),
+ client.CreatePartitions(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(test_case.topic_name), test_case.partitions_count),
std::exception);
}
}
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_details.topics.size(), test_cases.size());
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_details.Topics().size(), test_cases.size());
for (const auto &test_case : test_cases) {
bool found = false;
- for (const auto &topic : stream_details.topics) {
- if (topic.name == test_case.topic_name) {
- EXPECT_EQ(topic.partitions_count, test_case.expected_total_partitions);
+ for (const auto &topic : stream_details.Topics()) {
+ if (topic.Name() == test_case.topic_name) {
+ EXPECT_EQ(topic.PartitionsCount(), test_case.expected_total_partitions);
found = true;
break;
}
@@ -191,64 +174,62 @@ TEST_F(LowLevelE2E_Partition, CreatePartitionsWithBoundaryPartitionsCountValues)
});
}
-TEST_F(LowLevelE2E_Partition, CreatePartitionsWithNumericIdentifiersSucceeds) {
+TEST_F(E2E_Partition, CreatePartitionsWithNumericIdentifiersSucceeds) {
RecordProperty("description",
"Creates partitions successfully when valid numeric stream and topic identifiers are used.");
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, 1, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_details.topics.size(), 1u);
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_details.Topics().size(), 1u);
- ASSERT_NO_THROW(client->create_partitions(make_numeric_identifier(stream_details.id),
- make_numeric_identifier(stream_details.topics[0].id), 43));
+ ASSERT_NO_THROW(client.CreatePartitions(iggy::Identifier::Numeric(stream_details.Id()),
+ iggy::Identifier::Numeric(stream_details.Topics()[0].Id()), 43));
ASSERT_NO_THROW({
- const auto updated_stream_details = client->get_stream(make_numeric_identifier(stream_details.id));
- ASSERT_EQ(updated_stream_details.topics.size(), 1u);
- EXPECT_EQ(updated_stream_details.topics[0].id, stream_details.topics[0].id);
- EXPECT_EQ(updated_stream_details.topics[0].name, topic_name);
- EXPECT_EQ(updated_stream_details.topics[0].partitions_count, 44u);
+ const auto updated_stream_details = client.GetStream(iggy::Identifier::Numeric(stream_details.Id()));
+ ASSERT_EQ(updated_stream_details.Topics().size(), 1u);
+ EXPECT_EQ(updated_stream_details.Topics()[0].Id(), stream_details.Topics()[0].Id());
+ EXPECT_EQ(updated_stream_details.Topics()[0].Name(), topic_name);
+ EXPECT_EQ(updated_stream_details.Topics()[0].PartitionsCount(), 44u);
});
}
-TEST_F(LowLevelE2E_Partition, DeletePartitionsSucceeds) {
+TEST_F(E2E_Partition, DeletePartitionsSucceeds) {
RecordProperty("description", "Deletes partitions from an existing topic and verifies the resulting count.");
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, 44, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 44));
ASSERT_NO_THROW(
- client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 43));
+ client.DeletePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 43));
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_details.topics.size(), 1u);
- EXPECT_EQ(stream_details.topics[0].name, topic_name);
- EXPECT_EQ(stream_details.topics[0].partitions_count, 1u);
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_details.Topics().size(), 1u);
+ EXPECT_EQ(stream_details.Topics()[0].Name(), topic_name);
+ EXPECT_EQ(stream_details.Topics()[0].PartitionsCount(), 1u);
});
}
-TEST_F(LowLevelE2E_Partition, DeleteMorePartitionsThanExistingThrows) {
+TEST_F(E2E_Partition, DeleteMorePartitionsThanExistingThrows) {
RecordProperty("description",
"Rejects delete_partitions counts outside the allowed range and counts greater than existing.");
const std::string stream_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);
struct TestCase {
@@ -268,30 +249,29 @@ TEST_F(LowLevelE2E_Partition, DeleteMorePartitionsThanExistingThrows) {
for (const auto &test_case : test_cases) {
SCOPED_TRACE(test_case.topic_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), test_case.topic_name,
- test_case.initial_partitions, "none", "server_default", 0,
- "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), test_case.topic_name,
+ test_case.initial_partitions));
if (test_case.should_succeed) {
- ASSERT_NO_THROW(client->delete_partitions(make_string_identifier(stream_name),
- make_string_identifier(test_case.topic_name),
- test_case.partitions_count));
+ ASSERT_NO_THROW(client.DeletePartitions(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(test_case.topic_name),
+ test_case.partitions_count));
} else {
ASSERT_THROW(
- client->delete_partitions(make_string_identifier(stream_name),
- make_string_identifier(test_case.topic_name), test_case.partitions_count),
+ client.DeletePartitions(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(test_case.topic_name), test_case.partitions_count),
std::exception);
}
}
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_details.topics.size(), test_cases.size());
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_details.Topics().size(), test_cases.size());
for (const auto &test_case : test_cases) {
bool found = false;
- for (const auto &topic : stream_details.topics) {
- if (topic.name == test_case.topic_name) {
- EXPECT_EQ(topic.partitions_count, test_case.expected_total_partitions);
+ for (const auto &topic : stream_details.Topics()) {
+ if (topic.Name() == test_case.topic_name) {
+ EXPECT_EQ(topic.PartitionsCount(), test_case.expected_total_partitions);
found = true;
break;
}
@@ -301,179 +281,175 @@ TEST_F(LowLevelE2E_Partition, DeleteMorePartitionsThanExistingThrows) {
});
}
-TEST_F(LowLevelE2E_Partition, DeletePartitionsBeforeCreatingAdditionalPartitionsSucceeds) {
+TEST_F(E2E_Partition, DeletePartitionsBeforeCreatingAdditionalPartitionsSucceeds) {
RecordProperty("description",
"Deletes partitions from the initial topic allocation without calling create_partitions first.");
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, 3, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 3));
ASSERT_NO_THROW(
- client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 1));
+ client.DeletePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 1));
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_details.topics.size(), 1u);
- EXPECT_EQ(stream_details.topics[0].partitions_count, 2u);
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_details.Topics().size(), 1u);
+ EXPECT_EQ(stream_details.Topics()[0].PartitionsCount(), 2u);
});
}
-TEST_F(LowLevelE2E_Partition, DeletePartitionsFromTopicWithZeroPartitionsThrows) {
+TEST_F(E2E_Partition, DeletePartitionsFromTopicWithZeroPartitionsThrows) {
RecordProperty("description",
"Throws when delete_partitions is called with count 1 for a topic that currently has 0 partitions.");
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, 0, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 0));
- ASSERT_THROW(client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 1),
- std::exception);
+ ASSERT_THROW(
+ client.DeletePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 1),
+ std::exception);
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_details.topics.size(), 1u);
- EXPECT_EQ(stream_details.topics[0].name, topic_name);
- EXPECT_EQ(stream_details.topics[0].partitions_count, 0u);
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_details.Topics().size(), 1u);
+ EXPECT_EQ(stream_details.Topics()[0].Name(), topic_name);
+ EXPECT_EQ(stream_details.Topics()[0].PartitionsCount(), 0u);
});
}
-TEST_F(LowLevelE2E_Partition, DeletePartitionsBeforeLoginThrows) {
+TEST_F(E2E_Partition, DeletePartitionsBeforeLoginThrows) {
RecordProperty("description",
"Throws when delete_partitions is called before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedOutClient();
+ auto client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 1),
- std::exception);
- ASSERT_NO_THROW(client->connect());
- ASSERT_THROW(client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 1),
- std::exception);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(client->disconnect());
- ASSERT_THROW(client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 1),
- std::exception);
+ ASSERT_THROW(
+ client.DeletePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 1),
+ std::exception);
+ ASSERT_NO_THROW(client.Connect());
+ ASSERT_THROW(
+ client.DeletePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 1),
+ std::exception);
+ ASSERT_NO_THROW(client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(client.Disconnect());
+ ASSERT_THROW(
+ client.DeletePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 1),
+ std::exception);
}
-TEST_F(LowLevelE2E_Partition, DeletePartitionsOnNonExistentResourcesThrows) {
+TEST_F(E2E_Partition, DeletePartitionsOnNonExistentResourcesThrows) {
RecordProperty("description", "Throws when delete_partitions is called for a stream or topic that does 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, 3, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 3));
ASSERT_THROW(
- client->delete_partitions(make_string_identifier(missing_stream_name), make_string_identifier(topic_name), 1),
+ client.DeletePartitions(iggy::Identifier::String(missing_stream_name), iggy::Identifier::String(topic_name), 1),
std::exception);
ASSERT_THROW(
- client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(missing_topic_name), 1),
+ client.DeletePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(missing_topic_name), 1),
std::exception);
}
-TEST_F(LowLevelE2E_Partition, DeletePartitionsWithInvalidIdentifiersThrows) {
- RecordProperty("description", "Rejects delete_partitions requests that use invalid stream or topic identifiers.");
+TEST_F(E2E_Partition, DeletePartitionsWithInvalidIdentifiersThrows) {
+ RecordProperty("description", "Rejects invalid stream or topic identifiers before deleting partitions.");
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, 3, "none", "server_default",
- 0, "server_default", {}));
-
- iggy::ffi::Identifier invalid_stream_kind_id;
- invalid_stream_kind_id.kind = "invalid";
- invalid_stream_kind_id.length = 4;
- invalid_stream_kind_id.value = {1, 0, 0, 0};
- ASSERT_THROW(client->delete_partitions(std::move(invalid_stream_kind_id), make_string_identifier(topic_name), 1),
- std::exception);
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 3));
- iggy::ffi::Identifier invalid_stream_numeric_id;
- invalid_stream_numeric_id.kind = "numeric";
- invalid_stream_numeric_id.length = 1;
- invalid_stream_numeric_id.value.push_back(1);
- ASSERT_THROW(client->delete_partitions(std::move(invalid_stream_numeric_id), make_string_identifier(topic_name), 1),
+ ASSERT_THROW(client.DeletePartitions(iggy::Identifier::String(""), iggy::Identifier::String(topic_name), 1),
std::exception);
-
- iggy::ffi::Identifier invalid_topic_kind_id;
- invalid_topic_kind_id.kind = "invalid";
- invalid_topic_kind_id.length = 4;
- invalid_topic_kind_id.value = {1, 0, 0, 0};
- ASSERT_THROW(client->delete_partitions(make_string_identifier(stream_name), std::move(invalid_topic_kind_id), 1),
+ ASSERT_THROW(client.DeletePartitions(iggy::Identifier::String(std::string(256, 'a')),
+ iggy::Identifier::String(topic_name), 1),
std::exception);
-
- iggy::ffi::Identifier invalid_topic_numeric_id;
- invalid_topic_numeric_id.kind = "numeric";
- invalid_topic_numeric_id.length = 1;
- invalid_topic_numeric_id.value.push_back(1);
- ASSERT_THROW(client->delete_partitions(make_string_identifier(stream_name), std::move(invalid_topic_numeric_id), 1),
+ ASSERT_THROW(client.DeletePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(""), 1),
+ std::exception);
+ ASSERT_THROW(client.DeletePartitions(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(std::string(256, 'a')), 1),
std::exception);
}
-TEST_F(LowLevelE2E_Partition, DeletePartitionsTwiceForSameTopicSucceeds) {
+TEST_F(E2E_Partition, DeletePartitionsTwiceForSameTopicSucceeds) {
RecordProperty("description", "Allows delete_partitions to be called twice for the same stream and topic.");
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, 45, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 45));
ASSERT_NO_THROW(
- client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 20));
+ client.DeletePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 20));
ASSERT_NO_THROW(
- client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 20));
+ client.DeletePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 20));
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_details.topics.size(), 1u);
- EXPECT_EQ(stream_details.topics[0].name, topic_name);
- EXPECT_EQ(stream_details.topics[0].partitions_count, 5u);
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_details.Topics().size(), 1u);
+ EXPECT_EQ(stream_details.Topics()[0].Name(), topic_name);
+ EXPECT_EQ(stream_details.Topics()[0].PartitionsCount(), 5u);
});
}
-TEST_F(LowLevelE2E_Partition, DeletePartitionsAfterStreamDeletionThrows) {
+TEST_F(E2E_Partition, DeletePartitionsAfterStreamDeletionThrows) {
RecordProperty("description", "Throws when delete_partitions is called after the stream has been deleted.");
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, 3, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 3));
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_details.topics.size(), 1u);
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_details.Topics().size(), 1u);
- ASSERT_NO_THROW(client->delete_stream(make_numeric_identifier(stream_details.id)));
+ ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::Numeric(stream_details.Id())));
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client->delete_partitions(make_numeric_identifier(stream_details.id),
- make_numeric_identifier(stream_details.topics[0].id), 1),
+ ASSERT_THROW(client.DeletePartitions(iggy::Identifier::Numeric(stream_details.Id()),
+ iggy::Identifier::Numeric(stream_details.Topics()[0].Id()), 1),
std::exception);
}
-// TODO(slbotbm): Add CreatePartitionsAfterTopicDeletionThrows test case after delete_topic function is added.
+TEST_F(E2E_Partition, CreatePartitionsAfterTopicDeletionThrows) {
+ RecordProperty("description", "Throws when creating partitions after the topic has been deleted.");
+ const std::string stream_name = GetRandomName();
+ const std::string topic_name = GetRandomName();
+
+ auto client = GetLoggedInHighLevelClient();
+
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
+ TrackStream(stream_name);
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
+ ASSERT_NO_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)));
+
+ ASSERT_THROW(
+ client.CreatePartitions(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 1),
+ std::exception);
+}
diff --git a/foreign/cpp/tests/e2e/stream.cpp b/foreign/cpp/tests/e2e/stream.cpp
index c4f33e873..052e60d92 100644
--- a/foreign/cpp/tests/e2e/stream.cpp
+++ b/foreign/cpp/tests/e2e/stream.cpp
@@ -25,66 +25,66 @@
#include "lib.rs.h"
#include "tests/e2e/test_helpers.hpp"
-class LowLevelE2E_Stream : public E2ETestFixture {};
+class E2E_Stream : public E2ETestFixture {};
-TEST_F(LowLevelE2E_Stream, CreateStreamAfterLogin) {
+TEST_F(E2E_Stream, CreateStreamAfterLogin) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Creates a stream successfully after authenticating.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
}
-TEST_F(LowLevelE2E_Stream, CreateDuplicateStreamThrows) {
+TEST_F(E2E_Stream, CreateDuplicateStreamThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Rejects creating the same stream twice.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_THROW(client->create_stream(stream_name), std::exception);
+ ASSERT_THROW(client.CreateStream(stream_name), std::exception);
}
-TEST_F(LowLevelE2E_Stream, CreateStreamBeforeLoginThrows) {
+TEST_F(E2E_Stream, CreateStreamBeforeLoginThrows) {
RecordProperty("description", "Throws when stream creation is attempted before authentication.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedOutClient();
-
- ASSERT_THROW(client->create_stream(stream_name), std::exception);
- ASSERT_NO_THROW(client->connect());
- ASSERT_THROW(client->create_stream(stream_name), std::exception);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(client->disconnect());
- ASSERT_THROW(client->create_stream(stream_name), std::exception);
+ auto client = GetLoggedOutHighLevelClient();
+
+ ASSERT_THROW(client.CreateStream(stream_name), std::exception);
+ ASSERT_NO_THROW(client.Connect());
+ ASSERT_THROW(client.CreateStream(stream_name), std::exception);
+ ASSERT_NO_THROW(client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(client.Disconnect());
+ ASSERT_THROW(client.CreateStream(stream_name), std::exception);
}
-TEST_F(LowLevelE2E_Stream, CreateStreamValidatesNameConstraintsAndUniqueness) {
+TEST_F(E2E_Stream, CreateStreamValidatesNameConstraintsAndUniqueness) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description",
"Validates stream name length constraints and accepts the maximum allowed name length.");
- iggy::ffi::Client *client = GetLoggedInClient();
-
const std::string illegal_stream_names[] = {
"",
std::string(256, 'b'),
};
for (const auto &stream_name : illegal_stream_names) {
SCOPED_TRACE(stream_name);
- ASSERT_THROW(client->create_stream(stream_name), std::exception);
+ ASSERT_THROW(client.CreateStream(stream_name), std::exception);
}
const std::string max_length_name(255, 'a');
- ASSERT_NO_THROW(client->create_stream(max_length_name));
+ ASSERT_NO_THROW(client.CreateStream(max_length_name));
TrackStream(max_length_name);
}
-TEST_F(LowLevelE2E_Stream, CreateStreamWithEmojiName) {
+TEST_F(E2E_Stream, CreateStreamWithEmojiName) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Creates a stream with a UTF-8 emoji name.");
const std::string stream_name = "🚀🚀🚀🚀Apache Iggy🚀🚀🚀🚀";
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_details = ffi_client->get_stream(make_string_identifier(stream_name));
const std::string returned_stream_name = static_cast<std::string>(stream_details.name);
EXPECT_EQ(returned_stream_name, stream_name);
EXPECT_EQ(stream_details.topics_count, 0u);
@@ -92,27 +92,28 @@ TEST_F(LowLevelE2E_Stream, CreateStreamWithEmojiName) {
});
}
-TEST_F(LowLevelE2E_Stream, UpdateStreamWorksCorrectly) {
+TEST_F(E2E_Stream, UpdateStreamWorksCorrectly) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Updates an existing stream name while preserving the stream identity.");
const std::string stream_name = GetRandomName();
const std::string updated_stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
iggy::ffi::StreamDetails original_stream_details{};
- ASSERT_NO_THROW({ original_stream_details = client->get_stream(make_string_identifier(stream_name)); });
+ ASSERT_NO_THROW({ original_stream_details = ffi_client->get_stream(make_string_identifier(stream_name)); });
const std::uint32_t stream_id = original_stream_details.id;
ForgetTrackedStream(stream_name);
TrackStream(stream_id);
- ASSERT_NO_THROW(client->update_stream(make_string_identifier(stream_name), updated_stream_name));
+ ASSERT_NO_THROW(
+ client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name));
- ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(ffi_client->get_stream(make_string_identifier(stream_name)), std::exception);
iggy::ffi::StreamDetails updated_stream_details{};
- ASSERT_NO_THROW({ updated_stream_details = client->get_stream(make_numeric_identifier(stream_id)); });
+ ASSERT_NO_THROW({ updated_stream_details = ffi_client->get_stream(make_numeric_identifier(stream_id)); });
EXPECT_EQ(updated_stream_details.id, original_stream_details.id);
EXPECT_EQ(updated_stream_details.created_at, original_stream_details.created_at);
@@ -123,20 +124,21 @@ TEST_F(LowLevelE2E_Stream, UpdateStreamWorksCorrectly) {
EXPECT_EQ(updated_stream_details.topics.size(), original_stream_details.topics.size());
}
-TEST_F(LowLevelE2E_Stream, UpdateStreamWithSameNameIsIdempotent) {
+TEST_F(E2E_Stream, UpdateStreamWithSameNameIsIdempotent) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description",
"Calling update_stream with the current name succeeds without changing stream details.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
iggy::ffi::StreamDetails first_read{};
iggy::ffi::StreamDetails second_read{};
- ASSERT_NO_THROW({ first_read = client->get_stream(make_string_identifier(stream_name)); });
- ASSERT_NO_THROW(client->update_stream(make_numeric_identifier(first_read.id), stream_name));
- ASSERT_NO_THROW({ second_read = client->get_stream(make_numeric_identifier(first_read.id)); });
+ ASSERT_NO_THROW({ first_read = ffi_client->get_stream(make_string_identifier(stream_name)); });
+ ASSERT_NO_THROW(client.UpdateStream(iggy::Identifier::Numeric(first_read.id), stream_name));
+ ASSERT_NO_THROW({ second_read = ffi_client->get_stream(make_numeric_identifier(first_read.id)); });
EXPECT_EQ(second_read.id, first_read.id);
EXPECT_EQ(second_read.created_at, first_read.created_at);
@@ -147,39 +149,39 @@ TEST_F(LowLevelE2E_Stream, UpdateStreamWithSameNameIsIdempotent) {
EXPECT_EQ(second_read.topics.size(), first_read.topics.size());
}
-TEST_F(LowLevelE2E_Stream, UpdateStreamBeforeLoginThrows) {
+TEST_F(E2E_Stream, UpdateStreamBeforeLoginThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Rejects update_stream before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string updated_stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(unauthenticated_client->update_stream(make_string_identifier(stream_name), updated_stream_name),
+ ASSERT_THROW(unauthenticated_client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name),
std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
- ASSERT_THROW(unauthenticated_client->update_stream(make_string_identifier(stream_name), updated_stream_name),
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
+ ASSERT_THROW(unauthenticated_client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name),
std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
- ASSERT_THROW(unauthenticated_client->update_stream(make_string_identifier(stream_name), updated_stream_name),
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
+ ASSERT_THROW(unauthenticated_client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name),
std::exception);
}
-TEST_F(LowLevelE2E_Stream, UpdateStreamWithVariousUtf8Characters) {
+TEST_F(E2E_Stream, UpdateStreamWithVariousUtf8Characters) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Updates a stream name with various UTF-8 values.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
std::uint32_t stream_id = 0;
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_details = ffi_client->get_stream(make_string_identifier(stream_name));
stream_id = stream_details.id;
});
ForgetTrackedStream(stream_name);
@@ -191,72 +193,76 @@ TEST_F(LowLevelE2E_Stream, UpdateStreamWithVariousUtf8Characters) {
for (const auto &updated_stream_name : updated_stream_names) {
SCOPED_TRACE(updated_stream_name);
- ASSERT_NO_THROW(client->update_stream(make_numeric_identifier(stream_id), updated_stream_name));
+ ASSERT_NO_THROW(
+ client.UpdateStream(iggy::Identifier::Numeric(stream_id), updated_stream_name));
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_numeric_identifier(stream_id));
+ const auto stream_details = ffi_client->get_stream(make_numeric_identifier(stream_id));
EXPECT_EQ(stream_details.name, updated_stream_name);
});
}
}
-TEST_F(LowLevelE2E_Stream, UpdateNonExistentStreamThrows) {
+TEST_F(E2E_Stream, UpdateNonExistentStreamThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Throws when updating a stream that does not exist.");
const std::string stream_name = GetRandomName();
const std::string updated_stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_THROW(client->update_stream(make_string_identifier(stream_name), updated_stream_name), std::exception);
+ ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name),
+ std::exception);
}
-TEST_F(LowLevelE2E_Stream, UpdateStreamWithDuplicateNameThrows) {
+TEST_F(E2E_Stream, UpdateStreamWithDuplicateNameThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Rejects renaming a stream to another stream's existing name.");
const std::string first_stream_name = GetRandomName();
const std::string second_stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(first_stream_name));
+ ASSERT_NO_THROW(client.CreateStream(first_stream_name));
TrackStream(first_stream_name);
- ASSERT_NO_THROW(client->create_stream(second_stream_name));
+ ASSERT_NO_THROW(client.CreateStream(second_stream_name));
TrackStream(second_stream_name);
- ASSERT_THROW(client->update_stream(make_string_identifier(first_stream_name), second_stream_name), std::exception);
+ ASSERT_THROW(
+ client.UpdateStream(iggy::Identifier::String(first_stream_name), second_stream_name),
+ std::exception);
}
-TEST_F(LowLevelE2E_Stream, UpdateDeletedStreamThrows) {
+TEST_F(E2E_Stream, UpdateDeletedStreamThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Throws when updating a stream after it has been deleted.");
const std::string stream_name = GetRandomName();
const std::string updated_stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_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->update_stream(make_string_identifier(stream_name), updated_stream_name), std::exception);
+ ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name),
+ std::exception);
}
-TEST_F(LowLevelE2E_Stream, UpdateStreamFailedValidationDoesNotMutateStream) {
+TEST_F(E2E_Stream, UpdateStreamFailedValidationDoesNotMutateStream) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Keeps the stream unchanged when update_stream fails wrapper validation.");
const std::string stream_name = GetRandomName();
const std::string updated_stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
iggy::ffi::StreamDetails stream_before_failed_update{};
- ASSERT_NO_THROW({ stream_before_failed_update = client->get_stream(make_string_identifier(stream_name)); });
+ ASSERT_NO_THROW({ stream_before_failed_update = ffi_client->get_stream(make_string_identifier(stream_name)); });
iggy::ffi::Identifier invalid_numeric_id;
invalid_numeric_id.kind = "numeric";
invalid_numeric_id.length = 1;
invalid_numeric_id.value.push_back(1);
- ASSERT_THROW(client->update_stream(std::move(invalid_numeric_id), updated_stream_name), std::exception);
+ ASSERT_THROW(ffi_client->update_stream(std::move(invalid_numeric_id), updated_stream_name), std::exception);
iggy::ffi::StreamDetails stream_after_failed_update{};
- ASSERT_NO_THROW({ stream_after_failed_update = client->get_stream(make_string_identifier(stream_name)); });
+ ASSERT_NO_THROW({ stream_after_failed_update = ffi_client->get_stream(make_string_identifier(stream_name)); });
EXPECT_EQ(stream_after_failed_update.id, stream_before_failed_update.id);
EXPECT_EQ(stream_after_failed_update.created_at, stream_before_failed_update.created_at);
@@ -266,30 +272,31 @@ TEST_F(LowLevelE2E_Stream, UpdateStreamFailedValidationDoesNotMutateStream) {
EXPECT_EQ(stream_after_failed_update.topics_count, stream_before_failed_update.topics_count);
EXPECT_EQ(stream_after_failed_update.topics.size(), stream_before_failed_update.topics.size());
- ASSERT_THROW(client->get_stream(make_string_identifier(updated_stream_name)), std::exception);
+ ASSERT_THROW(ffi_client->get_stream(make_string_identifier(updated_stream_name)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, UpdateStreamOnlyChangesName) {
+TEST_F(E2E_Stream, UpdateStreamOnlyChangesName) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty(
"description",
"Changes only the stream name and leaves stream, topic, message, partition, and segment data intact.");
const std::string stream_name = GetRandomName();
const std::string updated_stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
std::uint32_t stream_id = 0;
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_details = ffi_client->get_stream(make_string_identifier(stream_name));
stream_id = stream_details.id;
});
ForgetTrackedStream(stream_name);
TrackStream(stream_id);
- ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream_id), topic_name, 2, "none", "never_expire", 0,
+ ASSERT_NO_THROW(ffi_client->create_topic(make_numeric_identifier(stream_id), topic_name, 2, "none", "never_expire", 0,
"server_default", {}));
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
@@ -298,7 +305,7 @@ TEST_F(LowLevelE2E_Stream, UpdateStreamOnlyChangesName) {
rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(message));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_id), make_numeric_identifier(0),
+ ASSERT_NO_THROW(ffi_client->send_messages(make_numeric_identifier(stream_id), make_numeric_identifier(0),
"partition_id", partition_id_bytes(0), std::move(messages)));
iggy::ffi::StreamDetails stream_before_update{};
@@ -306,16 +313,17 @@ TEST_F(LowLevelE2E_Stream, UpdateStreamOnlyChangesName) {
iggy::ffi::Stats stats_before_update{};
iggy::ffi::Stats stats_after_update{};
ASSERT_NO_THROW({
- stream_before_update = client->get_stream(make_numeric_identifier(stream_id));
- stats_before_update = client->get_stats();
+ stream_before_update = ffi_client->get_stream(make_numeric_identifier(stream_id));
+ stats_before_update = ffi_client->get_stats();
});
- ASSERT_NO_THROW(client->update_stream(make_numeric_identifier(stream_id), updated_stream_name));
+ ASSERT_NO_THROW(
+ client.UpdateStream(iggy::Identifier::Numeric(stream_id), updated_stream_name));
- ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(ffi_client->get_stream(make_string_identifier(stream_name)), std::exception);
ASSERT_NO_THROW({
- stream_after_update = client->get_stream(make_numeric_identifier(stream_id));
- stats_after_update = client->get_stats();
+ stream_after_update = ffi_client->get_stream(make_numeric_identifier(stream_id));
+ stats_after_update = ffi_client->get_stats();
});
EXPECT_EQ(stream_after_update.id, stream_before_update.id);
@@ -346,18 +354,19 @@ TEST_F(LowLevelE2E_Stream, UpdateStreamOnlyChangesName) {
EXPECT_EQ(stats_after_update.messages_count, stats_before_update.messages_count);
}
-TEST_F(LowLevelE2E_Stream, UpdateStreamValidatesNameBounds) {
+TEST_F(E2E_Stream, UpdateStreamValidatesNameBounds) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description",
"Rejects invalid stream name lengths during update and accepts the maximum allowed name length.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
std::uint32_t stream_id = 0;
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_details = ffi_client->get_stream(make_string_identifier(stream_name));
stream_id = stream_details.id;
});
ForgetTrackedStream(stream_name);
@@ -369,58 +378,59 @@ TEST_F(LowLevelE2E_Stream, UpdateStreamValidatesNameBounds) {
};
for (const auto &invalid_stream_name : invalid_stream_names) {
SCOPED_TRACE("invalid_stream_name_length=" + std::to_string(invalid_stream_name.size()));
- ASSERT_THROW(client->update_stream(make_numeric_identifier(stream_id), invalid_stream_name), std::exception);
+ ASSERT_THROW(
+ client.UpdateStream(iggy::Identifier::Numeric(stream_id), invalid_stream_name),
+ std::exception);
}
}
-TEST_F(LowLevelE2E_Stream, StreamCreatedAndDeletedSuccessfully) {
+TEST_F(E2E_Stream, StreamCreatedAndDeletedSuccessfully) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Creates a stream and deletes it successfully by string identifier.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_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);
}
-TEST_F(LowLevelE2E_Stream, DeleteNotCreatedStreamThrows) {
+TEST_F(E2E_Stream, DeleteNotCreatedStreamThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Throws when deleting a stream that does not exist.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_THROW(client->delete_stream(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, DeleteStreamBeforeLoginThrows) {
+TEST_F(E2E_Stream, DeleteStreamBeforeLoginThrows) {
RecordProperty("description", "Throws when stream deletion is attempted before authentication.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedOutClient();
+ auto client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(client->delete_stream(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)), std::exception);
- ASSERT_NO_THROW(client->connect());
+ ASSERT_NO_THROW(client.Connect());
- ASSERT_THROW(client->delete_stream(make_string_identifier(stream_name)), std::exception);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(client->disconnect());
- ASSERT_THROW(client->delete_stream(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)), std::exception);
+ ASSERT_NO_THROW(client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(client.Disconnect());
+ ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, DeleteStreamTwiceThrows) {
+TEST_F(E2E_Stream, DeleteStreamTwiceThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Throws when deleting the same stream a second time.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_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->delete_stream(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, DeleteStreamWithInvalidIdentifierThrows) {
+TEST_F(E2E_Stream, DeleteStreamWithInvalidIdentifierThrows) {
RecordProperty("description", "Rejects stream deletion requests that use invalid identifier formats.");
iggy::ffi::Client *client = GetLoggedInClient();
@@ -437,7 +447,7 @@ TEST_F(LowLevelE2E_Stream, DeleteStreamWithInvalidIdentifierThrows) {
ASSERT_THROW(client->delete_stream(std::move(invalid_numeric_id)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, GetStreamDetailsWithInvalidIdentifierThrows) {
+TEST_F(E2E_Stream, GetStreamDetailsWithInvalidIdentifierThrows) {
RecordProperty("description", "Rejects stream detail lookups that use invalid identifier formats.");
iggy::ffi::Client *client = GetLoggedInClient();
@@ -454,126 +464,126 @@ TEST_F(LowLevelE2E_Stream, GetStreamDetailsWithInvalidIdentifierThrows) {
ASSERT_THROW(client->get_stream(std::move(invalid_numeric_id)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, GetStreamByStringIdentifierReturnsStreamDetails) {
+TEST_F(E2E_Stream, GetStreamByStringIdentifierReturnsStreamDetails) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Returns expected stream details when looked up by string identifier.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- EXPECT_EQ(stream_details.name, stream_name);
- EXPECT_EQ(stream_details.topics_count, 0u);
- EXPECT_EQ(stream_details.topics.size(), 0u);
- EXPECT_EQ(stream_details.messages_count, 0u);
- EXPECT_EQ(stream_details.size_bytes, 0u);
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ EXPECT_EQ(stream_details.Name(), stream_name);
+ EXPECT_EQ(stream_details.TopicsCount(), 0u);
+ EXPECT_EQ(stream_details.Topics().size(), 0u);
+ EXPECT_EQ(stream_details.MessagesCount(), 0u);
+ EXPECT_EQ(stream_details.SizeBytes(), 0u);
});
}
-TEST_F(LowLevelE2E_Stream, GetNonExistentStreamDetailsThrows) {
+TEST_F(E2E_Stream, GetNonExistentStreamDetailsThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Throws when requesting details for a stream that does not exist.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, GetStreamDetailsBeforeLoginThrows) {
+TEST_F(E2E_Stream, GetStreamDetailsBeforeLoginThrows) {
RecordProperty("description", "Throws when stream details are requested before authentication.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedOutClient();
-
- ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
- ASSERT_NO_THROW(client->connect());
- ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(client->disconnect());
- ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
+ auto client = GetLoggedOutHighLevelClient();
+
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);
+ ASSERT_NO_THROW(client.Connect());
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);
+ ASSERT_NO_THROW(client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(client.Disconnect());
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, GetDeletedStreamDetailsThrows) {
+TEST_F(E2E_Stream, GetDeletedStreamDetailsThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Throws when requesting details for a stream after it has been deleted.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->get_stream(make_string_identifier(stream_name)));
- ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name)));
+ ASSERT_NO_THROW(client.GetStream(iggy::Identifier::String(stream_name)));
+ ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, GetStreamByNumericIdentifierReturnsStreamDetails) {
+TEST_F(E2E_Stream, GetStreamByNumericIdentifierReturnsStreamDetails) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Returns expected stream details when looked up by numeric identifier.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
std::uint32_t stream_id = 0;
ASSERT_NO_THROW({
- const auto by_string = client->get_stream(make_string_identifier(stream_name));
- stream_id = by_string.id;
+ const auto by_string = client.GetStream(iggy::Identifier::String(stream_name));
+ stream_id = by_string.Id();
});
ASSERT_NO_THROW({
- const auto by_numeric = client->get_stream(make_numeric_identifier(stream_id));
- EXPECT_EQ(by_numeric.id, stream_id);
- EXPECT_EQ(by_numeric.name, stream_name);
- EXPECT_EQ(by_numeric.topics_count, 0u);
- EXPECT_EQ(by_numeric.topics.size(), 0u);
+ const auto by_numeric = client.GetStream(iggy::Identifier::Numeric(stream_id));
+ EXPECT_EQ(by_numeric.Id(), stream_id);
+ EXPECT_EQ(by_numeric.Name(), stream_name);
+ EXPECT_EQ(by_numeric.TopicsCount(), 0u);
+ EXPECT_EQ(by_numeric.Topics().size(), 0u);
});
}
-TEST_F(LowLevelE2E_Stream, GetStreamsReturnsEmptyAfterCleanup) {
+TEST_F(E2E_Stream, GetStreamsReturnsEmptyAfterCleanup) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Verifies get_streams returns empty vector after cleaning up all streams.");
- iggy::ffi::Client *client = GetLoggedInClient();
-
- auto streams = client->get_streams();
+ auto streams = client.GetStreams();
for (const auto &s : streams) {
- client->delete_stream(make_numeric_identifier(s.id));
+ client.DeleteStream(iggy::Identifier::Numeric(s.Id()));
}
- streams = client->get_streams();
+ streams = client.GetStreams();
ASSERT_EQ(streams.size(), 0u);
}
-TEST_F(LowLevelE2E_Stream, GetStreamsReturnsStreamAfterCreation) {
+TEST_F(E2E_Stream, GetStreamsReturnsStreamAfterCreation) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Verifies created stream appears in get_streams result.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- client->create_stream(stream_name);
+ client.CreateStream(stream_name);
TrackStream(stream_name);
- auto streams = client->get_streams();
+ auto streams = client.GetStreams();
ASSERT_GE(streams.size(), 1u);
bool found = false;
for (const auto &s : streams) {
- if (std::string(s.name) == stream_name) {
+ if (s.Name() == stream_name) {
found = true;
- EXPECT_GT(s.created_at, static_cast<std::uint64_t>(0));
- EXPECT_EQ(s.size_bytes, static_cast<std::uint64_t>(0));
- EXPECT_EQ(s.messages_count, static_cast<std::uint64_t>(0));
- EXPECT_EQ(s.topics_count, 0u);
+ EXPECT_GT(s.CreatedAt(), 0u);
+ EXPECT_EQ(s.SizeBytes(), 0u);
+ EXPECT_EQ(s.MessagesCount(), 0u);
+ EXPECT_EQ(s.TopicsCount(), 0u);
break;
}
}
ASSERT_TRUE(found) << "Stream '" << stream_name << "' not found in get_streams result";
}
-TEST_F(LowLevelE2E_Stream, GetStreamsFieldsVerification) {
+TEST_F(E2E_Stream, GetStreamsFieldsVerification) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description",
"Verifies get_streams returns correct field values after creating stream with topic and messages.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- client->create_stream(stream_name);
+ client.CreateStream(stream_name);
TrackStream(stream_name);
- auto stream = client->get_stream(make_string_identifier(stream_name));
+ auto stream = client.GetStream(iggy::Identifier::String(stream_name));
const std::string topic_name = GetRandomName();
- client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default",
- {});
+ ffi_client->create_topic(make_numeric_identifier(stream.Id()), topic_name, 1, "none", "never_expire", 0,
+ "server_default", {});
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
for (std::uint32_t i = 0; i < 5; i++) {
@@ -581,42 +591,42 @@ TEST_F(LowLevelE2E_Stream, GetStreamsFieldsVerification) {
rust::Vec<iggy::ffi::HeaderEntry>());
messages.push_back(std::move(msg));
}
- client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
+ ffi_client->send_messages(make_numeric_identifier(stream.Id()), make_numeric_identifier(0), "partition_id",
partition_id_bytes(0), std::move(messages));
- auto streams = client->get_streams();
+ auto streams = client.GetStreams();
ASSERT_GE(streams.size(), 1u);
bool found = false;
for (const auto &s : streams) {
- if (std::string(s.name) == stream_name) {
+ if (s.Name() == stream_name) {
found = true;
- EXPECT_EQ(s.topics_count, 1u);
- EXPECT_EQ(s.messages_count, 5u);
+ EXPECT_EQ(s.TopicsCount(), 1u);
+ EXPECT_EQ(s.MessagesCount(), 5u);
break;
}
}
ASSERT_TRUE(found) << "Stream '" << stream_name << "' not found in get_streams result";
}
-TEST_F(LowLevelE2E_Stream, GetStreamsBeforeLoginThrows) {
+TEST_F(E2E_Stream, GetStreamsBeforeLoginThrows) {
RecordProperty("description", "Throws when get_streams is called before authentication.");
- iggy::ffi::Client *client = GetLoggedOutClient();
-
- ASSERT_THROW(client->get_streams(), std::exception);
- ASSERT_NO_THROW(client->connect());
- ASSERT_THROW(client->get_streams(), std::exception);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(client->disconnect());
- ASSERT_THROW(client->get_streams(), std::exception);
+ auto client = GetLoggedOutHighLevelClient();
+
+ ASSERT_THROW(client.GetStreams(), std::exception);
+ ASSERT_NO_THROW(client.Connect());
+ ASSERT_THROW(client.GetStreams(), std::exception);
+ ASSERT_NO_THROW(client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(client.Disconnect());
+ ASSERT_THROW(client.GetStreams(), std::exception);
}
-TEST_F(LowLevelE2E_Stream, GetStreamsConsistentWithGetStream) {
+TEST_F(E2E_Stream, GetStreamsConsistentWithGetStream) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Verifies get_streams result is consistent with get_stream for the same stream.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- client->create_stream(stream_name);
+ client.CreateStream(stream_name);
TrackStream(stream_name);
std::string list_name;
@@ -624,48 +634,48 @@ TEST_F(LowLevelE2E_Stream, GetStreamsConsistentWithGetStream) {
std::uint32_t list_topics_count = 0;
std::uint64_t list_created_at = 0;
std::uint64_t list_size_bytes = 0;
- auto streams = client->get_streams();
+ auto streams = client.GetStreams();
for (const auto &s : streams) {
- if (std::string(s.name) == stream_name) {
- list_name = std::string(s.name);
- list_id = s.id;
- list_topics_count = s.topics_count;
- list_created_at = s.created_at;
- list_size_bytes = s.size_bytes;
+ if (s.Name() == stream_name) {
+ list_name = s.Name();
+ list_id = s.Id();
+ list_topics_count = s.TopicsCount();
+ list_created_at = s.CreatedAt();
+ list_size_bytes = s.SizeBytes();
break;
}
}
ASSERT_FALSE(list_name.empty()) << "Stream '" << stream_name << "' not found in get_streams result";
- auto single = client->get_stream(make_string_identifier(stream_name));
- auto single_name = std::string(single.name);
- auto single_topics = single.topics_count;
+ auto single = client.GetStream(iggy::Identifier::String(stream_name));
+ auto single_name = single.Name();
+ auto single_topics = single.TopicsCount();
EXPECT_EQ(list_name, single_name);
- EXPECT_EQ(list_id, single.id);
+ EXPECT_EQ(list_id, single.Id());
EXPECT_EQ(list_topics_count, single_topics);
- EXPECT_EQ(list_created_at, single.created_at);
- EXPECT_EQ(list_size_bytes, single.size_bytes);
+ EXPECT_EQ(list_created_at, single.CreatedAt());
+ EXPECT_EQ(list_size_bytes, single.SizeBytes());
}
-TEST_F(LowLevelE2E_Stream, GetStreamsRepeatedCallsReturnSameResult) {
+TEST_F(E2E_Stream, GetStreamsRepeatedCallsReturnSameResult) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Verifies repeated get_streams calls return consistent results.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- client->create_stream(stream_name);
+ client.CreateStream(stream_name);
TrackStream(stream_name);
- auto streams1 = client->get_streams();
- auto streams2 = client->get_streams();
- auto streams3 = client->get_streams();
+ auto streams1 = client.GetStreams();
+ auto streams2 = client.GetStreams();
+ auto streams3 = client.GetStreams();
ASSERT_EQ(streams1.size(), streams2.size());
ASSERT_EQ(streams2.size(), streams3.size());
- auto contains_stream = [&](const rust::Vec<iggy::ffi::Stream> &vec) {
+ auto contains_stream = [&](const std::vector<iggy::Stream> &vec) {
for (const auto &s : vec) {
- if (std::string(s.name) == stream_name) {
+ if (s.Name() == stream_name) {
return true;
}
}
@@ -677,28 +687,26 @@ TEST_F(LowLevelE2E_Stream, GetStreamsRepeatedCallsReturnSameResult) {
ASSERT_TRUE(contains_stream(streams3)) << "Stream not found in third call";
}
-TEST_F(LowLevelE2E_Stream, PurgeStreamOnNonExistentStreamThrows) {
+TEST_F(E2E_Stream, PurgeStreamOnNonExistentStreamThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Throws when purging a stream that does not exist.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_THROW(client->purge_stream(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, PurgeStreamAfterStreamDeletionThrows) {
+TEST_F(E2E_Stream, PurgeStreamAfterStreamDeletionThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Throws when purging a stream after it has been deleted.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_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->purge_stream(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, PurgeStreamWithInvalidIdentifierThrows) {
+TEST_F(E2E_Stream, PurgeStreamWithInvalidIdentifierThrows) {
RecordProperty("description", "Rejects stream purge requests that use invalid identifier formats.");
iggy::ffi::Client *client = GetLoggedInClient();
@@ -715,31 +723,32 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamWithInvalidIdentifierThrows) {
ASSERT_THROW(client->purge_stream(std::move(invalid_numeric_id)), std::exception);
}
-TEST_F(LowLevelE2E_Stream, PurgeStreamPreservesStreamMetadata) {
+TEST_F(E2E_Stream, PurgeStreamPreservesStreamMetadata) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Preserves stream identity and topic metadata after purging stream messages.");
const std::string stream_name = GetRandomName();
const std::string first_topic_name = GetRandomName();
const std::string second_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- 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), first_topic_name, 2, "gzip", "duration",
+ ASSERT_NO_THROW(ffi_client->create_topic(make_string_identifier(stream_name), first_topic_name, 2, "gzip", "duration",
1000, "1GiB", {}));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), second_topic_name, 3, "none",
+ ASSERT_NO_THROW(ffi_client->create_topic(make_string_identifier(stream_name), second_topic_name, 3, "none",
"never_expire", 0, "server_default", {}));
- const auto stream_before_purge = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_before_purge = ffi_client->get_stream(make_string_identifier(stream_name));
ASSERT_EQ(stream_before_purge.topics.size(), 2u);
rust::Vec<iggy::ffi::IggyMessageToSend> first_topic_messages;
first_topic_messages.push_back(
iggy::ffi::make_message(to_payload("preserve-stream-metadata"), rust::Vec<iggy::ffi::HeaderEntry>()));
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_before_purge.id),
+ ASSERT_NO_THROW(ffi_client->send_messages(make_numeric_identifier(stream_before_purge.id),
make_string_identifier(first_topic_name), "partition_id",
partition_id_bytes(0), std::move(first_topic_messages)));
- const auto stream_with_messages = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_with_messages = ffi_client->get_stream(make_string_identifier(stream_name));
EXPECT_GT(stream_with_messages.messages_count, 0u);
EXPECT_GT(stream_with_messages.size_bytes, 0u);
@@ -764,9 +773,9 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamPreservesStreamMetadata) {
topic.partitions_count};
}
- ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));
+ ASSERT_NO_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)));
- const auto stream_after_purge = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_after_purge = ffi_client->get_stream(make_string_identifier(stream_name));
EXPECT_EQ(stream_after_purge.id, stream_with_messages.id);
EXPECT_EQ(stream_after_purge.created_at, stream_with_messages.created_at);
EXPECT_EQ(stream_after_purge.name, stream_with_messages.name);
@@ -788,21 +797,22 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamPreservesStreamMetadata) {
}
}
-TEST_F(LowLevelE2E_Stream, PurgeStreamRemovesMessagesAndPreservesTopics) {
+TEST_F(E2E_Stream, PurgeStreamRemovesMessagesAndPreservesTopics) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Purges all stream messages while keeping the stream and topics intact.");
const std::string stream_name = GetRandomName();
const std::string first_topic_name = GetRandomName();
const std::string second_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- 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), first_topic_name, 1, "none",
+ ASSERT_NO_THROW(ffi_client->create_topic(make_string_identifier(stream_name), first_topic_name, 1, "none",
"server_default", 0, "server_default", {}));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), second_topic_name, 1, "none",
+ ASSERT_NO_THROW(ffi_client->create_topic(make_string_identifier(stream_name), second_topic_name, 1, "none",
"server_default", 0, "server_default", {}));
- const auto created_stream = client->get_stream(make_string_identifier(stream_name));
+ const auto created_stream = ffi_client->get_stream(make_string_identifier(stream_name));
ASSERT_EQ(created_stream.topics.size(), 2u);
std::uint32_t first_topic_id = 0;
@@ -827,7 +837,7 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamRemovesMessagesAndPreservesTopics) {
first_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-stream-first-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>()));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
+ ASSERT_NO_THROW(ffi_client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(first_topic_id), "partition_id",
partition_id_bytes(0), std::move(first_topic_messages)));
@@ -836,11 +846,11 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamRemovesMessagesAndPreservesTopics) {
second_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-stream-second-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>()));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
+ ASSERT_NO_THROW(ffi_client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(second_topic_id), "partition_id",
partition_id_bytes(0), std::move(second_topic_messages)));
- const auto stream_before_purge = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_before_purge = ffi_client->get_stream(make_string_identifier(stream_name));
EXPECT_EQ(stream_before_purge.topics_count, 2u);
EXPECT_EQ(stream_before_purge.messages_count, 5u);
EXPECT_GT(stream_before_purge.size_bytes, 0u);
@@ -852,7 +862,7 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamRemovesMessagesAndPreservesTopics) {
EXPECT_EQ(messages_before_purge[first_topic_name], 3u);
EXPECT_EQ(messages_before_purge[second_topic_name], 2u);
- const auto streams_before_purge = client->get_streams();
+ const auto streams_before_purge = ffi_client->get_streams();
bool found_stream_before_purge = false;
for (const auto &stream : streams_before_purge) {
if (static_cast<std::string>(stream.name) == stream_name) {
@@ -865,9 +875,9 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamRemovesMessagesAndPreservesTopics) {
}
ASSERT_TRUE(found_stream_before_purge);
- ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));
+ ASSERT_NO_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)));
- const auto stream_after_purge = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_after_purge = ffi_client->get_stream(make_string_identifier(stream_name));
EXPECT_EQ(stream_after_purge.topics_count, 2u);
EXPECT_EQ(stream_after_purge.messages_count, 0u);
EXPECT_EQ(stream_after_purge.size_bytes, 0u);
@@ -877,7 +887,7 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamRemovesMessagesAndPreservesTopics) {
EXPECT_EQ(topic.size_bytes, 0u);
}
- const auto streams_after_purge = client->get_streams();
+ const auto streams_after_purge = ffi_client->get_streams();
bool found_stream_after_purge = false;
for (const auto &stream : streams_after_purge) {
if (static_cast<std::string>(stream.name) == stream_name) {
@@ -891,21 +901,22 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamRemovesMessagesAndPreservesTopics) {
ASSERT_TRUE(found_stream_after_purge);
}
-TEST_F(LowLevelE2E_Stream, PurgeStreamAcrossMultipleTopicsAndPartitionsClearsEverything) {
+TEST_F(E2E_Stream, PurgeStreamAcrossMultipleTopicsAndPartitionsClearsEverything) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Purges all messages across multiple topics and partitions in the stream.");
const std::string stream_name = GetRandomName();
const std::string first_topic_name = GetRandomName();
const std::string second_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- 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), first_topic_name, 2, "none",
+ ASSERT_NO_THROW(ffi_client->create_topic(make_string_identifier(stream_name), first_topic_name, 2, "none",
"server_default", 0, "server_default", {}));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), second_topic_name, 3, "none",
+ ASSERT_NO_THROW(ffi_client->create_topic(make_string_identifier(stream_name), second_topic_name, 3, "none",
"server_default", 0, "server_default", {}));
- const auto created_stream = client->get_stream(make_string_identifier(stream_name));
+ const auto created_stream = ffi_client->get_stream(make_string_identifier(stream_name));
ASSERT_EQ(created_stream.topics.size(), 2u);
std::uint32_t first_topic_id = 0;
@@ -932,7 +943,7 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamAcrossMultipleTopicsAndPartitionsClearsEve
to_payload("purge-stream-topic-a-" + std::to_string(partition_id) + "-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>()));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
+ ASSERT_NO_THROW(ffi_client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(first_topic_id), "partition_id",
partition_id_bytes(partition_id), std::move(messages)));
}
@@ -943,17 +954,17 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamAcrossMultipleTopicsAndPartitionsClearsEve
to_payload("purge-stream-topic-b-" + std::to_string(partition_id) + "-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>()));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
+ ASSERT_NO_THROW(ffi_client->send_messages(make_numeric_identifier(created_stream.id),
make_numeric_identifier(second_topic_id), "partition_id",
partition_id_bytes(partition_id), std::move(messages)));
}
- const auto stream_before_purge = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_before_purge = ffi_client->get_stream(make_string_identifier(stream_name));
EXPECT_EQ(stream_before_purge.messages_count, 10u);
- ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));
+ ASSERT_NO_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)));
- const auto stream_after_purge = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_after_purge = ffi_client->get_stream(make_string_identifier(stream_name));
EXPECT_EQ(stream_after_purge.topics_count, 2u);
EXPECT_EQ(stream_after_purge.messages_count, 0u);
EXPECT_EQ(stream_after_purge.size_bytes, 0u);
@@ -963,35 +974,36 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamAcrossMultipleTopicsAndPartitionsClearsEve
}
}
-TEST_F(LowLevelE2E_Stream, PurgeStreamThenSendMessagesAgainSucceeds) {
+TEST_F(E2E_Stream, PurgeStreamThenSendMessagesAgainSucceeds) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Allows sending fresh messages to a topic after purging its parent stream.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- 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, 1, "none", "server_default",
+ ASSERT_NO_THROW(ffi_client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
0, "server_default", {}));
- const auto created_stream = client->get_stream(make_string_identifier(stream_name));
+ const auto created_stream = ffi_client->get_stream(make_string_identifier(stream_name));
ASSERT_EQ(created_stream.topics.size(), 1u);
const std::uint32_t topic_id = created_stream.topics.front().id;
rust::Vec<iggy::ffi::IggyMessageToSend> first_batch;
first_batch.push_back(iggy::ffi::make_message(to_payload("before-purge"), rust::Vec<iggy::ffi::HeaderEntry>()));
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
+ ASSERT_NO_THROW(ffi_client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
"partition_id", partition_id_bytes(0), std::move(first_batch)));
- ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));
+ ASSERT_NO_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)));
rust::Vec<iggy::ffi::IggyMessageToSend> second_batch;
second_batch.push_back(iggy::ffi::make_message(to_payload("after-purge-0"), rust::Vec<iggy::ffi::HeaderEntry>()));
second_batch.push_back(iggy::ffi::make_message(to_payload("after-purge-1"), rust::Vec<iggy::ffi::HeaderEntry>()));
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
+ ASSERT_NO_THROW(ffi_client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
"partition_id", partition_id_bytes(0), std::move(second_batch)));
- const auto stream_after_resend = client->get_stream(make_string_identifier(stream_name));
+ const auto stream_after_resend = ffi_client->get_stream(make_string_identifier(stream_name));
EXPECT_EQ(stream_after_resend.topics_count, 1u);
EXPECT_EQ(stream_after_resend.messages_count, 2u);
EXPECT_GT(stream_after_resend.size_bytes, 0u);
@@ -999,18 +1011,19 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamThenSendMessagesAgainSucceeds) {
EXPECT_EQ(stream_after_resend.topics.front().messages_count, 2u);
}
-TEST_F(LowLevelE2E_Stream, PurgeStreamTwiceKeepsStreamEmptyAndTopicsIntact) {
+TEST_F(E2E_Stream, PurgeStreamTwiceKeepsStreamEmptyAndTopicsIntact) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Allows purging the same stream twice and keeps the stream empty after both calls.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *ffi_client = GetLoggedInClient();
- 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, 1, "none", "server_default",
+ ASSERT_NO_THROW(ffi_client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
0, "server_default", {}));
- const auto created_stream = client->get_stream(make_string_identifier(stream_name));
+ const auto created_stream = ffi_client->get_stream(make_string_identifier(stream_name));
ASSERT_EQ(created_stream.topics.size(), 1u);
const std::uint32_t topic_id = created_stream.topics.front().id;
@@ -1019,11 +1032,11 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamTwiceKeepsStreamEmptyAndTopicsIntact) {
messages.push_back(iggy::ffi::make_message(to_payload("purge-stream-twice-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>()));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
+ ASSERT_NO_THROW(ffi_client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
"partition_id", partition_id_bytes(0), std::move(messages)));
- ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));
- const auto stream_after_first_purge = client->get_stream(make_string_identifier(stream_name));
+ ASSERT_NO_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)));
+ const auto stream_after_first_purge = ffi_client->get_stream(make_string_identifier(stream_name));
EXPECT_EQ(stream_after_first_purge.topics_count, 1u);
EXPECT_EQ(stream_after_first_purge.messages_count, 0u);
EXPECT_EQ(stream_after_first_purge.size_bytes, 0u);
@@ -1031,8 +1044,8 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamTwiceKeepsStreamEmptyAndTopicsIntact) {
EXPECT_EQ(stream_after_first_purge.topics.front().messages_count, 0u);
EXPECT_EQ(stream_after_first_purge.topics.front().size_bytes, 0u);
- ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));
- const auto stream_after_second_purge = client->get_stream(make_string_identifier(stream_name));
+ ASSERT_NO_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)));
+ const auto stream_after_second_purge = ffi_client->get_stream(make_string_identifier(stream_name));
EXPECT_EQ(stream_after_second_purge.topics_count, 1u);
EXPECT_EQ(stream_after_second_purge.messages_count, 0u);
EXPECT_EQ(stream_after_second_purge.size_bytes, 0u);
@@ -1041,20 +1054,20 @@ TEST_F(LowLevelE2E_Stream, PurgeStreamTwiceKeepsStreamEmptyAndTopicsIntact) {
EXPECT_EQ(stream_after_second_purge.topics.front().size_bytes, 0u);
}
-TEST_F(LowLevelE2E_Stream, PurgeStreamBeforeLoginThrows) {
+TEST_F(E2E_Stream, PurgeStreamBeforeLoginThrows) {
+ auto client = GetLoggedInHighLevelClient();
RecordProperty("description", "Throws when stream purge is attempted before authentication.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(unauthenticated_client->purge_stream(make_string_identifier(stream_name)), std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
- ASSERT_THROW(unauthenticated_client->purge_stream(make_string_identifier(stream_name)), std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
- ASSERT_THROW(unauthenticated_client->purge_stream(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(unauthenticated_client.PurgeStream(iggy::Identifier::String(stream_name)), std::exception);
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
+ ASSERT_THROW(unauthenticated_client.PurgeStream(iggy::Identifier::String(stream_name)), std::exception);
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
+ ASSERT_THROW(unauthenticated_client.PurgeStream(iggy::Identifier::String(stream_name)), std::exception);
}
diff --git a/foreign/cpp/tests/e2e/test_helpers.hpp b/foreign/cpp/tests/e2e/test_helpers.hpp
index 22a4b142a..918a64db7 100644
--- a/foreign/cpp/tests/e2e/test_helpers.hpp
+++ b/foreign/cpp/tests/e2e/test_helpers.hpp
@@ -29,6 +29,7 @@
#include <gtest/gtest.h>
+#include "iggy.hpp"
#include "lib.rs.h"
inline iggy::ffi::Identifier make_string_identifier(const std::string &value) {
@@ -141,6 +142,15 @@ class E2ETestFixture : public ::testing::Test {
return client;
}
+ iggy::IggyBlockingClient GetLoggedOutHighLevelClient() { return iggy::IggyBlockingClient::Builder().Build(); }
+
+ iggy::IggyBlockingClient GetLoggedInHighLevelClient(std::string username = "iggy", std::string password = "iggy") {
+ auto client = GetLoggedOutHighLevelClient();
+ client.Connect();
+ client.Login(std::move(username), std::move(password));
+ return client;
+ }
+
std::string GetRandomName(const std::size_t max_length = 255) {
if (max_length == 0) {
return {};
diff --git a/foreign/cpp/tests/e2e/topic.cpp b/foreign/cpp/tests/e2e/topic.cpp
index 5bc3bdaa4..6bab34830 100644
--- a/foreign/cpp/tests/e2e/topic.cpp
+++ b/foreign/cpp/tests/e2e/topic.cpp
@@ -17,10 +17,12 @@
* under the License.
*/
+#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <string>
+#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>
@@ -31,50 +33,43 @@
#include "lib.rs.h"
#include "tests/e2e/test_helpers.hpp"
-namespace {
+class E2E_Topic : public E2ETestFixture {};
-rust::Vec<std::uint8_t> little_endian_bytes(const std::uint64_t value, const std::size_t width) {
- rust::Vec<std::uint8_t> bytes;
- bytes.reserve(width);
- for (std::size_t index = 0; index < width; ++index) {
- bytes.push_back(static_cast<std::uint8_t>((value >> (index * 8)) & 0xFF));
- }
-
- return bytes;
-}
-
-rust::Vec<std::uint8_t> bool_bytes(const bool value) {
- rust::Vec<std::uint8_t> bytes;
- bytes.push_back(static_cast<std::uint8_t>(value ? 1 : 0));
-
- return bytes;
-}
-
-} // namespace
-
-class LowLevelE2E_Topic : public E2ETestFixture {};
-
-TEST_F(LowLevelE2E_Topic, CreateTopicWithAllOptionCombinations) {
+TEST_F(E2E_Topic, CreateTopicWithAllOptionCombinations) {
RecordProperty("description",
"Creates topics across supported option combinations and verifies they are all returned.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- const std::vector<std::string> compression_algorithms = {"none", "gzip"};
+ struct CompressionOption {
+ std::string name;
+ iggy::CompressionAlgorithm value;
+ };
+ const std::vector<CompressionOption> compression_algorithms = {
+ {"none", iggy::CompressionAlgorithm::None()},
+ {"gzip", iggy::CompressionAlgorithm::Gzip()},
+ };
struct ExpiryOption {
- std::string kind;
- std::uint64_t value;
+ std::string name;
+ iggy::Expiry value;
};
const std::vector<ExpiryOption> expiry_options = {
- {"server_default", 0},
- {"never_expire", 0},
- {"duration", 1000},
+ {"server_default", iggy::Expiry::ServerDefault()},
+ {"never_expire", iggy::Expiry::NeverExpire()},
+ {"duration", iggy::Expiry::Duration(1000)},
+ };
+ struct MaxTopicSizeOption {
+ std::string name;
+ iggy::MaxTopicSize value;
+ };
+ const std::vector<MaxTopicSizeOption> max_topic_sizes = {
+ {"server_default", iggy::MaxTopicSize::ServerDefault()},
+ {"unlimited", iggy::MaxTopicSize::Unlimited()},
+ {"1GiB", iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)},
};
- const std::vector<std::string> max_topic_sizes = {"server_default", "unlimited", "1GiB"};
std::size_t expected_topics_count = 0;
std::unordered_set<std::string> expected_topic_names;
@@ -82,13 +77,12 @@ TEST_F(LowLevelE2E_Topic, CreateTopicWithAllOptionCombinations) {
for (const auto &expiry_option : expiry_options) {
for (const auto &max_topic_size : max_topic_sizes) {
const std::string topic_name = GetRandomName();
- SCOPED_TRACE("compression=" + compression_algorithm + ", expiry_kind=" + expiry_option.kind +
- ", expiry_value=" + std::to_string(expiry_option.value) +
- ", max_topic_size=" + max_topic_size);
+ SCOPED_TRACE("compression=" + compression_algorithm.name + ", expiry_kind=" + expiry_option.name +
+ ", max_topic_size=" + max_topic_size.name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1,
- compression_algorithm, expiry_option.kind, expiry_option.value,
- max_topic_size, {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1,
+ compression_algorithm.value, expiry_option.value,
+ max_topic_size.value));
++expected_topics_count;
expected_topic_names.insert(topic_name);
}
@@ -96,46 +90,39 @@ TEST_F(LowLevelE2E_Topic, CreateTopicWithAllOptionCombinations) {
}
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- EXPECT_EQ(stream_details.name, stream_name);
- EXPECT_EQ(stream_details.topics_count, expected_topics_count);
- ASSERT_EQ(stream_details.topics.size(), expected_topics_count);
- for (const auto &topic : stream_details.topics) {
- const std::string topic_name = static_cast<std::string>(topic.name);
- const auto erased = expected_topic_names.erase(topic_name);
- EXPECT_EQ(erased, 1u) << "Unexpected topic name returned: " << topic_name;
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ EXPECT_EQ(stream_details.Name(), stream_name);
+ EXPECT_EQ(stream_details.TopicsCount(), expected_topics_count);
+ ASSERT_EQ(stream_details.Topics().size(), expected_topics_count);
+ for (const auto &topic : stream_details.Topics()) {
+ const auto erased = expected_topic_names.erase(topic.Name());
+ EXPECT_EQ(erased, 1u) << "Unexpected topic name returned: " << topic.Name();
}
EXPECT_TRUE(expected_topic_names.empty());
});
}
-TEST_F(LowLevelE2E_Topic, CreateTopicWithBoundaryPartitionsCountValues) {
+TEST_F(E2E_Topic, CreateTopicWithBoundaryPartitionsCountValues) {
RecordProperty("description", "Accepts boundary partition counts and rejects values above the supported maximum.");
const std::string stream_name = GetRandomName();
const std::string zero_partitions_topic_name = GetRandomName();
const std::string max_partitions_topic_name = GetRandomName();
const std::string overflow_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), zero_partitions_topic_name, 0, "none",
- "server_default", 0, "server_default", {}));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), max_partitions_topic_name, 1000, "none",
- "server_default", 0, "server_default", {}));
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name), overflow_topic_name, 1001, "none",
- "server_default", 0, "server_default", {}),
- std::exception);
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), zero_partitions_topic_name, 0));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), max_partitions_topic_name, 1000));
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), overflow_topic_name, 1001), std::exception);
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- EXPECT_EQ(stream_details.topics_count, 2u);
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ EXPECT_EQ(stream_details.TopicsCount(), 2u);
std::unordered_map<std::string, std::uint32_t> topic_partitions;
- for (const auto &topic : stream_details.topics) {
- const std::string topic_name = static_cast<std::string>(topic.name);
- topic_partitions[topic_name] = topic.partitions_count;
+ for (const auto &topic : stream_details.Topics()) {
+ topic_partitions[topic.Name()] = topic.PartitionsCount();
}
EXPECT_EQ(topic_partitions.size(), 2u);
@@ -143,13 +130,12 @@ TEST_F(LowLevelE2E_Topic, CreateTopicWithBoundaryPartitionsCountValues) {
EXPECT_EQ(topic_partitions[max_partitions_topic_name], 1000u);
}
-TEST_F(LowLevelE2E_Topic, CreateTopicWithInvalidNamesThrows) {
+TEST_F(E2E_Topic, CreateTopicWithInvalidNamesThrows) {
RecordProperty("description", "Rejects invalid topic names and accepts the maximum allowed name length.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
const std::string illegal_topic_names[] = {
@@ -158,52 +144,42 @@ TEST_F(LowLevelE2E_Topic, CreateTopicWithInvalidNamesThrows) {
};
for (const auto &topic_name : illegal_topic_names) {
SCOPED_TRACE(topic_name);
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
- 0, "server_default", {}),
- std::exception);
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1), std::exception);
}
const std::string max_length_name(255, 'a');
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), max_length_name, 1, "none",
- "server_default", 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), max_length_name, 1));
}
-TEST_F(LowLevelE2E_Topic, CreateDuplicateTopicThrows) {
+TEST_F(E2E_Topic, CreateDuplicateTopicThrows) {
RecordProperty("description", "Rejects creating a duplicate topic within the same stream.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
- 0, "server_default", {}));
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default", 0,
- "server_default", {}),
- std::exception);
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1), std::exception);
}
-TEST_F(LowLevelE2E_Topic, CreateSameTopicNameInDifferentStreamsSucceeds) {
+TEST_F(E2E_Topic, CreateSameTopicNameInDifferentStreamsSucceeds) {
RecordProperty("description", "Allows the same topic name to be created in different streams.");
const std::string first_stream_name = GetRandomName();
const std::string second_stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(first_stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(first_stream_name));
TrackStream(first_stream_name);
- ASSERT_NO_THROW(client->create_stream(second_stream_name));
+ ASSERT_NO_THROW(client.CreateStream(second_stream_name));
TrackStream(second_stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(first_stream_name), topic_name, 1, "none",
- "server_default", 0, "server_default", {}));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(second_stream_name), topic_name, 1, "none",
- "server_default", 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(first_stream_name), topic_name, 1));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(second_stream_name), topic_name, 1));
}
-TEST_F(LowLevelE2E_Topic, CreateTopicWithInvalidOptionsThrows) {
+TEST_F(E2E_Topic, CreateTopicWithInvalidOptionsThrows) {
RecordProperty("description", "Rejects topic creation requests that use invalid option values.");
const std::string stream_name = GetRandomName();
const std::string invalid_compression_topic_name = GetRandomName();
@@ -226,7 +202,7 @@ TEST_F(LowLevelE2E_Topic, CreateTopicWithInvalidOptionsThrows) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, CreateTopicWithOptionsReturnsCanonicalKindAndDerivedRemainder) {
+TEST_F(E2E_Topic, CreateTopicWithOptionsReturnsCanonicalKindAndDerivedRemainder) {
RecordProperty("description",
"Returns an explicitly set option in its canonical kind, derives the keys left unset, and rejects "
"an option key outside the server catalog.");
@@ -234,96 +210,104 @@ TEST_F(LowLevelE2E_Topic, CreateTopicWithOptionsReturnsCanonicalKindAndDerivedRe
const std::string topic_name = GetRandomName();
const std::string unknown_option_topic = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- rust::Vec<iggy::ffi::HeaderEntry> options;
- options.push_back(make_header_entry(make_header_field(iggy::ffi::HeaderKind::String, to_payload("enforce_fsync")),
- make_header_field(iggy::ffi::HeaderKind::String, to_payload("true"))));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
- 0, "server_default", std::move(options)));
+ const auto options = iggy::ResourceOptions::Explicit({iggy::HeaderEntry::Create(
+ iggy::HeaderField::Create(iggy::HeaderKind::String,
+ {'e', 'n', 'f', 'o', 'r', 'c', 'e', '_', 'f', 's', 'y', 'n', 'c'}),
+ iggy::HeaderField::Create(iggy::HeaderKind::String, {'t', 'r', 'u', 'e'}))});
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1,
+ iggy::CompressionAlgorithm::None(), iggy::Expiry::ServerDefault(),
+ iggy::MaxTopicSize::ServerDefault(), options));
const auto topic_details =
- client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name));
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
// Admission re-encodes the block from its own parse, so a value comes back
// in its key's catalog kind rather than in the kind that was sent.
- rust::Vec<std::uint8_t> enforce_fsync_enabled;
- enforce_fsync_enabled.push_back(1);
- EXPECT_TRUE(has_header(topic_details.options, static_cast<std::uint8_t>(iggy::ffi::HeaderKind::String),
- to_payload("enforce_fsync"), static_cast<std::uint8_t>(iggy::ffi::HeaderKind::Bool),
- enforce_fsync_enabled));
-
- EXPECT_FALSE(topic_details.derived_options.empty());
+ const auto &explicit_options = topic_details.Options().Explicit();
+ ASSERT_EQ(explicit_options.size(), 1u);
+ EXPECT_EQ(explicit_options.front().Key().Kind(), iggy::HeaderKind::String);
+ EXPECT_EQ(explicit_options.front().Key().Value(),
+ (std::vector<std::uint8_t>{'e', 'n', 'f', 'o', 'r', 'c', 'e', '_', 'f', 's', 'y', 'n', 'c'}));
+ EXPECT_EQ(explicit_options.front().Value().Kind(), iggy::HeaderKind::Bool);
+ EXPECT_EQ(explicit_options.front().Value().Value(), (std::vector<std::uint8_t>{1}));
+
+ EXPECT_FALSE(topic_details.Options().Derived().empty());
std::unordered_set<std::string> derived_option_keys;
- for (const auto &derived_option : topic_details.derived_options) {
- derived_option_keys.insert(std::string(derived_option.key.value.begin(), derived_option.key.value.end()));
+ for (const auto &derived_option : topic_details.Options().Derived()) {
+ const auto &key = derived_option.Key().Value();
+ derived_option_keys.emplace(key.begin(), key.end());
}
EXPECT_EQ(derived_option_keys.count("max_topic_size"), 1u);
EXPECT_EQ(derived_option_keys.count("enforce_fsync"), 0u);
- rust::Vec<iggy::ffi::HeaderEntry> unknown_options;
- unknown_options.push_back(
- make_header_entry(make_header_field(iggy::ffi::HeaderKind::String, to_payload("not_a_real_option")),
- make_header_field(iggy::ffi::HeaderKind::String, to_payload("true"))));
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name), unknown_option_topic, 1, "none",
- "server_default", 0, "server_default", std::move(unknown_options)),
+ const auto unknown_options = iggy::ResourceOptions::Explicit({iggy::HeaderEntry::Create(
+ iggy::HeaderField::Create(iggy::HeaderKind::String, {'n', 'o', 't', '_', 'a', '_', 'r', 'e', 'a', 'l', '_', 'o',
+ 'p', 't', 'i', 'o', 'n'}),
+ iggy::HeaderField::Create(iggy::HeaderKind::String, {'t', 'r', 'u', 'e'}))});
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), unknown_option_topic, 1,
+ iggy::CompressionAlgorithm::None(), iggy::Expiry::ServerDefault(),
+ iggy::MaxTopicSize::ServerDefault(), unknown_options),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, CreateTopicWithTypedOptionHelpersReportsThemAsExplicitOptions) {
+TEST_F(E2E_Topic, CreateTopicWithTypedOptionHelpersReportsThemAsExplicitOptions) {
RecordProperty("description",
"Creates a topic with every typed option helper and verifies each key comes back as an explicit "
"option in its catalog kind.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
constexpr std::uint64_t segment_size_bytes = 8ULL * 1024ULL * 1024ULL;
constexpr std::uint32_t messages_required_to_save = 512;
constexpr std::uint64_t size_of_messages_required_to_save = 2ULL * 1024ULL * 1024ULL;
- rust::Vec<iggy::ffi::HeaderEntry> options;
- options.push_back(iggy::TopicOption::SegmentSize(segment_size_bytes));
- options.push_back(iggy::TopicOption::EnforceFsync(true));
- options.push_back(iggy::TopicOption::MessagesRequiredToSave(messages_required_to_save));
- options.push_back(iggy::TopicOption::SizeOfMessagesRequiredToSave(size_of_messages_required_to_save));
- options.push_back(iggy::TopicOption::PreallocateSegments(false));
+ const auto options = iggy::ResourceOptions::Explicit({
+ iggy::TopicOption::SegmentSize(segment_size_bytes),
+ iggy::TopicOption::EnforceFsync(true),
+ iggy::TopicOption::MessagesRequiredToSave(messages_required_to_save),
+ iggy::TopicOption::SizeOfMessagesRequiredToSave(size_of_messages_required_to_save),
+ iggy::TopicOption::PreallocateSegments(false),
+ });
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
- 0, "server_default", std::move(options)));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1,
+ iggy::CompressionAlgorithm::None(), iggy::Expiry::ServerDefault(),
+ iggy::MaxTopicSize::ServerDefault(), options));
const auto topic_details =
- client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name));
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
+
+ auto expected_options = options.Explicit();
+ auto explicit_options = topic_details.Options().Explicit();
+ const auto compare_option_keys = [](const auto &left, const auto &right) {
+ return left.Key().Value() < right.Key().Value();
+ };
+ std::sort(expected_options.begin(), expected_options.end(), compare_option_keys);
+ std::sort(explicit_options.begin(), explicit_options.end(), compare_option_keys);
+
+ ASSERT_EQ(explicit_options.size(), expected_options.size());
+ for (std::size_t index = 0; index < explicit_options.size(); ++index) {
+ EXPECT_EQ(explicit_options[index].Key().Kind(), expected_options[index].Key().Kind());
+ EXPECT_EQ(explicit_options[index].Key().Value(), expected_options[index].Key().Value());
+ EXPECT_EQ(explicit_options[index].Value().Kind(), expected_options[index].Value().Kind());
+ EXPECT_EQ(explicit_options[index].Value().Value(), expected_options[index].Value().Value());
+ }
- const auto key_kind = static_cast<std::uint8_t>(iggy::ffi::HeaderKind::String);
- const auto bool_kind = static_cast<std::uint8_t>(iggy::ffi::HeaderKind::Bool);
- const auto uint32_kind = static_cast<std::uint8_t>(iggy::ffi::HeaderKind::Uint32);
- const auto uint64_kind = static_cast<std::uint8_t>(iggy::ffi::HeaderKind::Uint64);
-
- EXPECT_TRUE(has_header(topic_details.options, key_kind, to_payload("segment_size"), uint64_kind,
- little_endian_bytes(segment_size_bytes, 8)));
- EXPECT_TRUE(has_header(topic_details.options, key_kind, to_payload("enforce_fsync"), bool_kind, bool_bytes(true)));
- EXPECT_TRUE(has_header(topic_details.options, key_kind, to_payload("messages_required_to_save"), uint32_kind,
- little_endian_bytes(messages_required_to_save, 4)));
- EXPECT_TRUE(has_header(topic_details.options, key_kind, to_payload("size_of_messages_required_to_save"),
- uint64_kind, little_endian_bytes(size_of_messages_required_to_save, 8)));
- EXPECT_TRUE(
- has_header(topic_details.options, key_kind, to_payload("preallocate_segments"), bool_kind, bool_bytes(false)));
-
- for (const auto &derived_option : topic_details.derived_options) {
- const std::string derived_key(derived_option.key.value.begin(), derived_option.key.value.end());
+ for (const auto &derived_option : topic_details.Options().Derived()) {
+ const auto &key = derived_option.Key().Value();
+ const std::string derived_key(key.begin(), key.end());
EXPECT_NE(derived_key, "segment_size") << "segment_size was set explicitly, so it cannot be derived";
}
}
-TEST_F(LowLevelE2E_Topic, DescribeOptionsServesTopicCatalogAndRejectsUnknownScope) {
+TEST_F(E2E_Topic, DescribeOptionsServesTopicCatalogAndRejectsUnknownScope) {
RecordProperty("description",
"Serves the topic option catalog with each key's kind, default and description, returns an empty "
"catalog for the stream scope, and rejects an unknown scope name.");
@@ -359,51 +343,45 @@ TEST_F(LowLevelE2E_Topic, DescribeOptionsServesTopicCatalogAndRejectsUnknownScop
ASSERT_THROW(client->describe_options("not_a_scope"), std::exception);
}
-TEST_F(LowLevelE2E_Topic, CreateTopicWithMaxTopicSizeBelowSegmentSizeThrows) {
+TEST_F(E2E_Topic, CreateTopicWithMaxTopicSizeBelowSegmentSizeThrows) {
RecordProperty("description",
"Rejects topic creation when the maximum topic size is smaller than the segment size.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default", 0,
- "1024", {}),
- std::exception);
+ ASSERT_THROW(
+ client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1, iggy::CompressionAlgorithm::None(),
+ iggy::Expiry::ServerDefault(), iggy::MaxTopicSize::FromBytes(1024)),
+ std::exception);
}
-TEST_F(LowLevelE2E_Topic, CreateTopicOnNonExistentStreamThrows) {
+TEST_F(E2E_Topic, CreateTopicOnNonExistentStreamThrows) {
RecordProperty("description", "Throws when creating a topic on a stream that does not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default", 0,
- "server_default", {}),
- std::exception);
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1), std::exception);
}
-TEST_F(LowLevelE2E_Topic, CreateTopicAfterStreamDeletionThrows) {
+TEST_F(E2E_Topic, CreateTopicAfterStreamDeletionThrows) {
RecordProperty("description", "Throws when creating a topic after its stream has been deleted.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_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->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default", 0,
- "server_default", {}),
- std::exception);
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1), std::exception);
}
-TEST_F(LowLevelE2E_Topic, CreateTopicWithInvalidStreamIdentifierThrows) {
+TEST_F(E2E_Topic, CreateTopicWithInvalidStreamIdentifierThrows) {
RecordProperty("description", "Rejects topic creation requests that use invalid stream identifier formats.");
const std::string stream_name = GetRandomName();
const std::string first_topic_name = GetRandomName();
@@ -431,30 +409,29 @@ TEST_F(LowLevelE2E_Topic, CreateTopicWithInvalidStreamIdentifierThrows) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, CreateTopicBeforeLoginThrows) {
+TEST_F(E2E_Topic, CreateTopicBeforeLoginThrows) {
RecordProperty("description", "Throws when topic creation is attempted from an unauthenticated client.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
- ASSERT_NO_THROW(unauthenticated_client->connect());
- ASSERT_THROW(unauthenticated_client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none",
- "server_default", 0, "server_default", {}),
+ ASSERT_THROW(unauthenticated_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1),
std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
- ASSERT_THROW(unauthenticated_client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none",
- "server_default", 0, "server_default", {}),
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
+ ASSERT_THROW(unauthenticated_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1),
+ std::exception);
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
+ ASSERT_THROW(unauthenticated_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, DeleteTopicAfterCreate) {
+TEST_F(E2E_Topic, DeleteTopicAfterCreate) {
RecordProperty("description", "Deletes an existing topic after creating it.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -474,7 +451,7 @@ TEST_F(LowLevelE2E_Topic, DeleteTopicAfterCreate) {
});
}
-TEST_F(LowLevelE2E_Topic, DeleteTopicOnNonExistentStreamThrows) {
+TEST_F(E2E_Topic, DeleteTopicOnNonExistentStreamThrows) {
RecordProperty("description", "Throws when deleting a topic from a stream that does not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -485,7 +462,7 @@ TEST_F(LowLevelE2E_Topic, DeleteTopicOnNonExistentStreamThrows) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, DeleteTopicOnNonExistentTopicThrows) {
+TEST_F(E2E_Topic, DeleteTopicOnNonExistentTopicThrows) {
RecordProperty("description", "Throws when deleting a topic that does not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -499,7 +476,7 @@ TEST_F(LowLevelE2E_Topic, DeleteTopicOnNonExistentTopicThrows) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, DeleteTopicTwiceThrows) {
+TEST_F(E2E_Topic, DeleteTopicTwiceThrows) {
RecordProperty("description", "Throws when deleting the same topic a second time.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -516,7 +493,7 @@ TEST_F(LowLevelE2E_Topic, DeleteTopicTwiceThrows) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, DeleteTopicAfterStreamDeletionThrows) {
+TEST_F(E2E_Topic, DeleteTopicAfterStreamDeletionThrows) {
RecordProperty("description", "Throws when deleting a topic after its stream has been deleted.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -534,7 +511,7 @@ TEST_F(LowLevelE2E_Topic, DeleteTopicAfterStreamDeletionThrows) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, DeleteTopicBeforeLoginThrows) {
+TEST_F(E2E_Topic, DeleteTopicBeforeLoginThrows) {
RecordProperty("description", "Rejects delete_topic before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -562,7 +539,7 @@ TEST_F(LowLevelE2E_Topic, DeleteTopicBeforeLoginThrows) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, DeleteTopicWithInvalidStreamIdentifierThrows) {
+TEST_F(E2E_Topic, DeleteTopicWithInvalidStreamIdentifierThrows) {
RecordProperty("description", "Rejects topic deletion requests that use invalid stream identifier formats.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -588,7 +565,7 @@ TEST_F(LowLevelE2E_Topic, DeleteTopicWithInvalidStreamIdentifierThrows) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, DeleteTopicWithInvalidTopicIdentifierThrows) {
+TEST_F(E2E_Topic, DeleteTopicWithInvalidTopicIdentifierThrows) {
RecordProperty("description", "Rejects topic deletion requests that use invalid topic identifier formats.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -614,221 +591,215 @@ TEST_F(LowLevelE2E_Topic, DeleteTopicWithInvalidTopicIdentifierThrows) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, GetTopicReturnsTopicForExistingTopic) {
+TEST_F(E2E_Topic, GetTopicReturnsTopicForExistingTopic) {
RecordProperty("description", "Returns topic details for an existing topic.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(
- client->create_topic(make_string_identifier(stream_name), topic_name, 3, "gzip", "duration", 1000, "1GiB", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 3,
+ iggy::CompressionAlgorithm::Gzip(), iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)));
ASSERT_NO_THROW({
const auto topic_details =
- client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name));
- EXPECT_EQ(topic_details.name, topic_name);
- EXPECT_EQ(topic_details.partitions_count, 3u);
- EXPECT_EQ(topic_details.partitions.size(), 3u);
- EXPECT_EQ(topic_details.compression_algorithm, "gzip");
- EXPECT_EQ(topic_details.message_expiry, 1000u);
- EXPECT_EQ(topic_details.max_topic_size, 1024ULL * 1024ULL * 1024ULL);
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
+ EXPECT_EQ(topic_details.Name(), topic_name);
+ EXPECT_EQ(topic_details.PartitionsCount(), 3u);
+ EXPECT_EQ(topic_details.Partitions().size(), 3u);
+ EXPECT_EQ(topic_details.CompressionAlgorithm(), "gzip");
+ EXPECT_EQ(topic_details.MessageExpiry(), 1000u);
+ EXPECT_EQ(topic_details.MaxTopicSize(), 1024ULL * 1024ULL * 1024ULL);
});
}
-TEST_F(LowLevelE2E_Topic, GetTopicBeforeLoginThrows) {
+TEST_F(E2E_Topic, GetTopicBeforeLoginThrows) {
RecordProperty("description", "Rejects get_topic before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
ASSERT_THROW(
- unauthenticated_client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)),
+ unauthenticated_client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
ASSERT_THROW(
- unauthenticated_client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)),
+ unauthenticated_client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
ASSERT_THROW(
- unauthenticated_client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)),
+ unauthenticated_client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, GetTopicWithWrongStreamIdThrows) {
+TEST_F(E2E_Topic, GetTopicWithWrongStreamIdThrows) {
RecordProperty("description", "Rejects get_topic when the topic belongs to a different stream.");
const std::string first_stream_name = GetRandomName();
const std::string second_stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(first_stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(first_stream_name));
TrackStream(first_stream_name);
- ASSERT_NO_THROW(client->create_stream(second_stream_name));
+ ASSERT_NO_THROW(client.CreateStream(second_stream_name));
TrackStream(second_stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(first_stream_name), topic_name, 1, "none",
- "server_default", 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(first_stream_name), topic_name, 1));
- ASSERT_THROW(client->get_topic(make_string_identifier(second_stream_name), make_string_identifier(topic_name)),
+ ASSERT_THROW(client.GetTopic(iggy::Identifier::String(second_stream_name), iggy::Identifier::String(topic_name)),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, GetTopicWithWrongTopicThrows) {
+TEST_F(E2E_Topic, GetTopicWithWrongTopicThrows) {
RecordProperty("description", "Rejects get_topic when the topic does not exist in the stream.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string wrong_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
- ASSERT_THROW(client->get_topic(make_string_identifier(stream_name), make_string_identifier(wrong_topic_name)),
+ ASSERT_THROW(client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(wrong_topic_name)),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, GetTopicAfterStreamDeletionThrows) {
+TEST_F(E2E_Topic, GetTopicAfterStreamDeletionThrows) {
RecordProperty("description", "Rejects get_topic after the stream has been deleted.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 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, 1));
+ ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)),
+ ASSERT_THROW(client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, GetTopicAfterTopicDeletionThrows) {
+TEST_F(E2E_Topic, GetTopicAfterTopicDeletionThrows) {
RecordProperty("description", "Rejects get_topic after the topic has been deleted.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
- 0, "server_default", {}));
- ASSERT_NO_THROW(client->delete_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
+ ASSERT_NO_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)));
- ASSERT_THROW(client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)),
+ ASSERT_THROW(client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, GetTopicReturnsEmptyPartitionsForZeroPartitionTopic) {
+TEST_F(E2E_Topic, GetTopicReturnsEmptyPartitionsForZeroPartitionTopic) {
RecordProperty("description", "Returns an empty partitions vector for a topic created with zero partitions.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 0, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 0));
ASSERT_NO_THROW({
const auto topic_details =
- client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name));
- EXPECT_EQ(topic_details.name, topic_name);
- EXPECT_EQ(topic_details.partitions_count, 0u);
- EXPECT_TRUE(topic_details.partitions.empty());
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
+ EXPECT_EQ(topic_details.Name(), topic_name);
+ EXPECT_EQ(topic_details.PartitionsCount(), 0u);
+ EXPECT_TRUE(topic_details.Partitions().empty());
});
}
-TEST_F(LowLevelE2E_Topic, GetTopicReturnsMaxBoundaryPartitionCount) {
+TEST_F(E2E_Topic, GetTopicReturnsMaxBoundaryPartitionCount) {
RecordProperty("description", "Returns the maximum boundary partition count for a topic created with it.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1000, "none",
- "server_default", 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1000));
ASSERT_NO_THROW({
const auto topic_details =
- client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name));
- EXPECT_EQ(topic_details.name, topic_name);
- EXPECT_EQ(topic_details.partitions_count, 1000u);
- EXPECT_EQ(topic_details.partitions.size(), 1000u);
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
+ EXPECT_EQ(topic_details.Name(), topic_name);
+ EXPECT_EQ(topic_details.PartitionsCount(), 1000u);
+ EXPECT_EQ(topic_details.Partitions().size(), 1000u);
});
}
-TEST_F(LowLevelE2E_Topic, GetTopicIsStableAcrossBackToBackCalls) {
+TEST_F(E2E_Topic, GetTopicIsStableAcrossBackToBackCalls) {
RecordProperty("description", "Returns stable topic details across back-to-back get_topic calls.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(
- client->create_topic(make_string_identifier(stream_name), topic_name, 3, "gzip", "duration", 1000, "1GiB", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 3,
+ iggy::CompressionAlgorithm::Gzip(), iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)));
- iggy::ffi::TopicDetails first_topic{};
- iggy::ffi::TopicDetails second_topic{};
ASSERT_NO_THROW({
- first_topic = client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name));
- second_topic = client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name));
+ const auto first_topic =
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
+ const auto second_topic =
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
+
+ EXPECT_EQ(second_topic.Name(), first_topic.Name());
+ EXPECT_EQ(second_topic.MessageExpiry(), first_topic.MessageExpiry());
+ EXPECT_EQ(second_topic.CompressionAlgorithm(), first_topic.CompressionAlgorithm());
+ EXPECT_EQ(second_topic.MaxTopicSize(), first_topic.MaxTopicSize());
+ EXPECT_EQ(second_topic.PartitionsCount(), first_topic.PartitionsCount());
+ EXPECT_EQ(second_topic.Partitions().size(), first_topic.Partitions().size());
});
-
- EXPECT_EQ(static_cast<std::string>(second_topic.name), static_cast<std::string>(first_topic.name));
- EXPECT_EQ(second_topic.message_expiry, first_topic.message_expiry);
- EXPECT_EQ(static_cast<std::string>(second_topic.compression_algorithm),
- static_cast<std::string>(first_topic.compression_algorithm));
- EXPECT_EQ(second_topic.max_topic_size, first_topic.max_topic_size);
- EXPECT_EQ(second_topic.partitions_count, first_topic.partitions_count);
- EXPECT_EQ(second_topic.partitions.size(), first_topic.partitions.size());
}
-TEST_F(LowLevelE2E_Topic, GetTopicAgreesWithGetStreamTopicSummary) {
+TEST_F(E2E_Topic, GetTopicAgreesWithGetStreamTopicSummary) {
RecordProperty("description", "Returns topic details that agree with get_stream topic summary fields.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(
- client->create_topic(make_string_identifier(stream_name), topic_name, 3, "gzip", "duration", 1000, "1GiB", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 3,
+ iggy::CompressionAlgorithm::Gzip(), iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)));
ASSERT_NO_THROW({
- const auto stream_details = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_details.topics.size(), 1u);
+ const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_details.Topics().size(), 1u);
- const auto topic_summary = stream_details.topics.front();
+ const auto &topic_summary = stream_details.Topics().front();
const auto topic_details =
- client->get_topic(make_string_identifier(stream_name), make_string_identifier(topic_name));
-
- EXPECT_EQ(static_cast<std::string>(topic_details.name), static_cast<std::string>(topic_summary.name));
- EXPECT_EQ(topic_details.message_expiry, topic_summary.message_expiry);
- EXPECT_EQ(static_cast<std::string>(topic_details.compression_algorithm),
- static_cast<std::string>(topic_summary.compression_algorithm));
- EXPECT_EQ(topic_details.max_topic_size, topic_summary.max_topic_size);
- EXPECT_EQ(topic_details.partitions_count, topic_summary.partitions_count);
- EXPECT_EQ(topic_details.partitions.size(), topic_summary.partitions_count);
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
+
+ EXPECT_EQ(topic_details.Name(), topic_summary.Name());
+ EXPECT_EQ(topic_details.MessageExpiry(), topic_summary.MessageExpiry());
+ EXPECT_EQ(topic_details.CompressionAlgorithm(), topic_summary.CompressionAlgorithm());
+ EXPECT_EQ(topic_details.MaxTopicSize(), topic_summary.MaxTopicSize());
+ EXPECT_EQ(topic_details.PartitionsCount(), topic_summary.PartitionsCount());
+ EXPECT_EQ(topic_details.Partitions().size(), topic_summary.PartitionsCount());
});
}
-TEST_F(LowLevelE2E_Topic, GetTopicsReturnsCreatedTopicInputFields) {
+TEST_F(E2E_Topic, GetTopicsReturnsCreatedTopicInputFields) {
RecordProperty("description", "Creates topics in a stream, gets them, and verifies user-provided topic fields.");
const std::string stream_name = GetRandomName();
const std::string first_topic_name = GetRandomName();
@@ -847,209 +818,208 @@ TEST_F(LowLevelE2E_Topic, GetTopicsReturnsCreatedTopicInputFields) {
{0, "none", std::numeric_limits<std::uint64_t>::max(), std::numeric_limits<std::uint64_t>::max()}},
};
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), first_topic_name, 2, "gzip", "duration",
- 1000, "1GiB", {}));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), second_topic_name, 0, "none",
- "never_expire", 0, "unlimited", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), first_topic_name, 2,
+ iggy::CompressionAlgorithm::Gzip(), iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), second_topic_name, 0,
+ iggy::CompressionAlgorithm::None(), iggy::Expiry::NeverExpire(),
+ iggy::MaxTopicSize::Unlimited()));
ASSERT_NO_THROW({
- const auto topics = client->get_topics(make_string_identifier(stream_name));
+ const auto topics = client.GetTopics(iggy::Identifier::String(stream_name));
ASSERT_EQ(topics.size(), expected_topics.size());
- EXPECT_EQ(topics[0].name, first_topic_name);
- EXPECT_EQ(topics[1].name, second_topic_name);
+ EXPECT_EQ(topics[0].Name(), first_topic_name);
+ EXPECT_EQ(topics[1].Name(), second_topic_name);
std::unordered_set<std::string> found_topic_names;
for (const auto &topic : topics) {
- const std::string topic_name = static_cast<std::string>(topic.name);
- const auto expected = expected_topics.find(topic_name);
- ASSERT_NE(expected, expected_topics.end()) << "Unexpected topic name returned: " << topic_name;
-
- EXPECT_EQ(topic.name, expected->first);
- EXPECT_EQ(topic.partitions_count, expected->second.partitions_count);
- EXPECT_EQ(topic.compression_algorithm, expected->second.compression_algorithm);
- EXPECT_EQ(topic.message_expiry, expected->second.message_expiry);
- EXPECT_EQ(topic.max_topic_size, expected->second.max_topic_size);
- found_topic_names.insert(topic_name);
+ const auto expected = expected_topics.find(topic.Name());
+ ASSERT_NE(expected, expected_topics.end()) << "Unexpected topic name returned: " << topic.Name();
+
+ EXPECT_EQ(topic.Name(), expected->first);
+ EXPECT_EQ(topic.PartitionsCount(), expected->second.partitions_count);
+ EXPECT_EQ(topic.CompressionAlgorithm(), expected->second.compression_algorithm);
+ EXPECT_EQ(topic.MessageExpiry(), expected->second.message_expiry);
+ EXPECT_EQ(topic.MaxTopicSize(), expected->second.max_topic_size);
+ found_topic_names.insert(topic.Name());
}
EXPECT_EQ(found_topic_names.size(), expected_topics.size());
});
}
-TEST_F(LowLevelE2E_Topic, GetTopicsBeforeLoginThrows) {
+TEST_F(E2E_Topic, GetTopicsBeforeLoginThrows) {
RecordProperty("description", "Rejects get_topics before connect, and after connect but before login.");
const std::string stream_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);
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(unauthenticated_client->get_topics(make_string_identifier(stream_name)), std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
- ASSERT_THROW(unauthenticated_client->get_topics(make_string_identifier(stream_name)), std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
- ASSERT_THROW(unauthenticated_client->get_topics(make_string_identifier(stream_name)), std::exception);
+ ASSERT_THROW(unauthenticated_client.GetTopics(iggy::Identifier::String(stream_name)), std::exception);
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
+ ASSERT_THROW(unauthenticated_client.GetTopics(iggy::Identifier::String(stream_name)), std::exception);
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
+ ASSERT_THROW(unauthenticated_client.GetTopics(iggy::Identifier::String(stream_name)), std::exception);
}
-TEST_F(LowLevelE2E_Topic, GetTopicsReturnsEmptyForStreamWithoutTopics) {
+TEST_F(E2E_Topic, GetTopicsReturnsEmptyForStreamWithoutTopics) {
RecordProperty("description", "Returns an empty topic list for an existing stream that has no topics.");
const std::string stream_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
ASSERT_NO_THROW({
- const auto topics = client->get_topics(make_string_identifier(stream_name));
+ const auto topics = client.GetTopics(iggy::Identifier::String(stream_name));
EXPECT_TRUE(topics.empty());
});
}
-TEST_F(LowLevelE2E_Topic, GetTopicsAfterTopicDeletionReturnsRemainingTopics) {
+TEST_F(E2E_Topic, GetTopicsAfterTopicDeletionReturnsRemainingTopics) {
RecordProperty("description", "Returns only non-deleted topics after a topic is deleted from the stream.");
const std::string stream_name = GetRandomName();
const std::string deleted_topic = GetRandomName();
const std::string remaining_topic = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), deleted_topic, 1, "none",
- "server_default", 0, "server_default", {}));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), remaining_topic, 1, "none",
- "server_default", 0, "server_default", {}));
- ASSERT_NO_THROW(client->delete_topic(make_string_identifier(stream_name), make_string_identifier(deleted_topic)));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), deleted_topic, 1));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), remaining_topic, 1));
+ ASSERT_NO_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(deleted_topic)));
ASSERT_NO_THROW({
- const auto topics = client->get_topics(make_string_identifier(stream_name));
+ const auto topics = client.GetTopics(iggy::Identifier::String(stream_name));
ASSERT_EQ(topics.size(), 1u);
- EXPECT_EQ(topics.front().name, remaining_topic);
+ EXPECT_EQ(topics.front().Name(), remaining_topic);
});
}
-TEST_F(LowLevelE2E_Topic, GetTopicsAfterTopicUpdateReturnsUpdatedInputFields) {
+TEST_F(E2E_Topic, GetTopicsAfterTopicUpdateReturnsUpdatedInputFields) {
RecordProperty("description", "Returns updated user-provided topic fields after a topic update.");
const std::string stream_name = GetRandomName();
const std::string original_topic = GetRandomName();
const std::string updated_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), original_topic, 2, "none",
- "server_default", 0, "server_default", {}));
- ASSERT_NO_THROW(client->update_topic(make_string_identifier(stream_name), make_string_identifier(original_topic),
- updated_topic_name, "gzip", "duration", 1000, "1GiB", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), original_topic, 2));
+ ASSERT_NO_THROW(client.UpdateTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(original_topic),
+ updated_topic_name, iggy::CompressionAlgorithm::Gzip(),
+ iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)));
ASSERT_NO_THROW({
- const auto topics = client->get_topics(make_string_identifier(stream_name));
+ const auto topics = client.GetTopics(iggy::Identifier::String(stream_name));
ASSERT_EQ(topics.size(), 1u);
- EXPECT_EQ(topics.front().name, updated_topic_name);
- EXPECT_EQ(topics.front().partitions_count, 2u);
- EXPECT_EQ(topics.front().compression_algorithm, "gzip");
- EXPECT_EQ(topics.front().message_expiry, 1000u);
- EXPECT_EQ(topics.front().max_topic_size, 1024ULL * 1024ULL * 1024ULL);
+ EXPECT_EQ(topics.front().Name(), updated_topic_name);
+ EXPECT_EQ(topics.front().PartitionsCount(), 2u);
+ EXPECT_EQ(topics.front().CompressionAlgorithm(), "gzip");
+ EXPECT_EQ(topics.front().MessageExpiry(), 1000u);
+ EXPECT_EQ(topics.front().MaxTopicSize(), 1024ULL * 1024ULL * 1024ULL);
});
}
-TEST_F(LowLevelE2E_Topic, UpdateTopicWorksCorrectly) {
+TEST_F(E2E_Topic, UpdateTopicWorksCorrectly) {
RecordProperty("description", "Returns a topic summary that matches topic details after updating a topic.");
const std::string stream_name = GetRandomName();
const std::string original_topic = GetRandomName();
const std::string updated_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), original_topic, 2, "none",
- "server_default", 0, "server_default", {}));
- ASSERT_NO_THROW(client->update_topic(make_string_identifier(stream_name), make_string_identifier(original_topic),
- updated_topic_name, "gzip", "duration", 1000, "1GiB", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), original_topic, 2));
+ ASSERT_NO_THROW(client.UpdateTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(original_topic),
+ updated_topic_name, iggy::CompressionAlgorithm::Gzip(),
+ iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)));
ASSERT_NO_THROW({
const auto topic_details =
- client->get_topic(make_string_identifier(stream_name), make_string_identifier(updated_topic_name));
- const auto topics = client->get_topics(make_string_identifier(stream_name));
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(updated_topic_name));
+ const auto topics = client.GetTopics(iggy::Identifier::String(stream_name));
ASSERT_EQ(topics.size(), 1u);
const auto &topic_summary = topics.front();
- EXPECT_EQ(topic_summary.id, topic_details.id);
- EXPECT_EQ(topic_summary.created_at, topic_details.created_at);
- EXPECT_EQ(topic_summary.name, topic_details.name);
- EXPECT_EQ(topic_summary.size_bytes, topic_details.size_bytes);
- EXPECT_EQ(topic_summary.message_expiry, topic_details.message_expiry);
- EXPECT_EQ(topic_summary.compression_algorithm, topic_details.compression_algorithm);
- EXPECT_EQ(topic_summary.max_topic_size, topic_details.max_topic_size);
- EXPECT_EQ(topic_summary.messages_count, topic_details.messages_count);
- EXPECT_EQ(topic_summary.partitions_count, topic_details.partitions_count);
+ EXPECT_EQ(topic_summary.Id(), topic_details.Id());
+ EXPECT_EQ(topic_summary.CreatedAt(), topic_details.CreatedAt());
+ EXPECT_EQ(topic_summary.Name(), topic_details.Name());
+ EXPECT_EQ(topic_summary.SizeBytes(), topic_details.SizeBytes());
+ EXPECT_EQ(topic_summary.MessageExpiry(), topic_details.MessageExpiry());
+ EXPECT_EQ(topic_summary.CompressionAlgorithm(), topic_details.CompressionAlgorithm());
+ EXPECT_EQ(topic_summary.MaxTopicSize(), topic_details.MaxTopicSize());
+ EXPECT_EQ(topic_summary.MessagesCount(), topic_details.MessagesCount());
+ EXPECT_EQ(topic_summary.PartitionsCount(), topic_details.PartitionsCount());
});
}
-TEST_F(LowLevelE2E_Topic, UpdateTopicDoesNotChangePartitionsCount) {
+TEST_F(E2E_Topic, UpdateTopicDoesNotChangePartitionsCount) {
RecordProperty("description", "Preserves the topic partition count after updating topic metadata.");
const std::string stream_name = GetRandomName();
const std::string original_topic = GetRandomName();
const std::string updated_topic_name = GetRandomName();
constexpr std::uint32_t partitions_count = 3;
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), original_topic, partitions_count, "none",
- "server_default", 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), original_topic, partitions_count));
- ASSERT_NO_THROW(client->update_topic(make_string_identifier(stream_name), make_string_identifier(original_topic),
- updated_topic_name, "gzip", "duration", 1000, "1GiB", {}));
+ ASSERT_NO_THROW(client.UpdateTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(original_topic),
+ updated_topic_name, iggy::CompressionAlgorithm::Gzip(),
+ iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)));
ASSERT_NO_THROW({
const auto topic_details =
- client->get_topic(make_string_identifier(stream_name), make_string_identifier(updated_topic_name));
- EXPECT_EQ(topic_details.partitions_count, partitions_count);
- EXPECT_EQ(topic_details.partitions.size(), partitions_count);
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(updated_topic_name));
+ EXPECT_EQ(topic_details.PartitionsCount(), partitions_count);
+ EXPECT_EQ(topic_details.Partitions().size(), partitions_count);
});
}
-TEST_F(LowLevelE2E_Topic, UpdateTopicDoesNotChangeMessages) {
+TEST_F(E2E_Topic, UpdateTopicDoesNotChangeMessages) {
RecordProperty("description", "Keeps existing messages readable after updating topic metadata.");
const std::string stream_name = GetRandomName();
const std::string original_topic = GetRandomName();
const std::string updated_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *message_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), original_topic, 1, "none",
- "server_default", 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), original_topic, 1));
- const auto created_stream = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(created_stream.topics.size(), 1u);
- const auto topic_id = created_stream.topics.front().id;
+ const auto created_stream = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(created_stream.Topics().size(), 1u);
+ const auto topic_id = created_stream.Topics().front().Id();
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
messages.push_back(
iggy::ffi::make_message(to_payload("message-before-topic-update"), rust::Vec<iggy::ffi::HeaderEntry>()));
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
- "partition_id", partition_id_bytes(0), std::move(messages)));
+ ASSERT_NO_THROW(message_client->send_messages(make_numeric_identifier(created_stream.Id()),
+ make_numeric_identifier(topic_id), "partition_id",
+ partition_id_bytes(0), std::move(messages)));
- ASSERT_NO_THROW(client->update_topic(make_string_identifier(stream_name), make_string_identifier(original_topic),
- updated_topic_name, "gzip", "duration", 1000, "1GiB", {}));
+ ASSERT_NO_THROW(client.UpdateTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(original_topic),
+ updated_topic_name, iggy::CompressionAlgorithm::Gzip(),
+ iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)));
ASSERT_NO_THROW({
- const auto polled = client->poll_messages(make_numeric_identifier(created_stream.id),
- make_string_identifier(updated_topic_name), 0, "consumer",
- make_numeric_identifier(1), "offset", 0, 10, false);
+ const auto polled = message_client->poll_messages(make_numeric_identifier(created_stream.Id()),
+ make_string_identifier(updated_topic_name), 0, "consumer",
+ make_numeric_identifier(1), "offset", 0, 10, false);
ASSERT_EQ(polled.count, 1u);
ASSERT_EQ(polled.messages.size(), 1u);
const std::string actual(polled.messages[0].payload.begin(), polled.messages[0].payload.end());
@@ -1057,113 +1027,123 @@ TEST_F(LowLevelE2E_Topic, UpdateTopicDoesNotChangeMessages) {
});
}
-TEST_F(LowLevelE2E_Topic, UpdateTopicWithAllOptionCombinationsUpdatesInputFields) {
+TEST_F(E2E_Topic, UpdateTopicWithAllOptionCombinationsUpdatesInputFields) {
RecordProperty("description",
"Updates a topic across supported option combinations and verifies deterministic updated fields.");
const std::string stream_name = GetRandomName();
std::string topic_name = GetRandomName();
- const std::vector<std::string> compression_algorithms = {"none", "gzip"};
+ struct CompressionOption {
+ std::string name;
+ iggy::CompressionAlgorithm value;
+ };
+ const std::vector<CompressionOption> compression_algorithms = {
+ {"none", iggy::CompressionAlgorithm::None()},
+ {"gzip", iggy::CompressionAlgorithm::Gzip()},
+ };
struct ExpiryOption {
- std::string kind;
- std::uint64_t value;
+ std::string name;
+ iggy::Expiry value;
};
const std::vector<ExpiryOption> expiry_options = {
- {"server_default", 0},
- {"never_expire", 0},
- {"duration", 1000},
+ {"server_default", iggy::Expiry::ServerDefault()},
+ {"never_expire", iggy::Expiry::NeverExpire()},
+ {"duration", iggy::Expiry::Duration(1000)},
+ };
+ struct MaxTopicSizeOption {
+ std::string name;
+ iggy::MaxTopicSize value;
+ };
+ const std::vector<MaxTopicSizeOption> max_topic_sizes = {
+ {"server_default", iggy::MaxTopicSize::ServerDefault()},
+ {"unlimited", iggy::MaxTopicSize::Unlimited()},
+ {"1GiB", iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)},
};
- const std::vector<std::string> max_topic_sizes = {"server_default", "unlimited", "1GiB"};
-
- iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 2, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 2));
for (const auto &compression_algorithm : compression_algorithms) {
for (const auto &expiry_option : expiry_options) {
for (const auto &max_topic_size : max_topic_sizes) {
const std::string updated_topic_name = GetRandomName();
- SCOPED_TRACE("compression=" + compression_algorithm + ", expiry_kind=" + expiry_option.kind +
- ", expiry_value=" + std::to_string(expiry_option.value) +
- ", max_topic_size=" + max_topic_size);
+ SCOPED_TRACE("compression=" + compression_algorithm.name + ", expiry_kind=" + expiry_option.name +
+ ", max_topic_size=" + max_topic_size.name);
- ASSERT_NO_THROW(client->update_topic(
- make_string_identifier(stream_name), make_string_identifier(topic_name), updated_topic_name,
- compression_algorithm, expiry_option.kind, expiry_option.value, max_topic_size, {}));
+ ASSERT_NO_THROW(client.UpdateTopic(
+ iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), updated_topic_name,
+ compression_algorithm.value, expiry_option.value, max_topic_size.value));
topic_name = updated_topic_name;
}
}
}
}
-TEST_F(LowLevelE2E_Topic, UpdateTopicWithSameOptionsIsIdempotent) {
+TEST_F(E2E_Topic, UpdateTopicWithSameOptionsIsIdempotent) {
RecordProperty("description", "Calling update_topic twice with the same options returns the same topic details.");
const std::string stream_name = GetRandomName();
const std::string original_topic = GetRandomName();
const std::string updated_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), original_topic, 2, "none",
- "server_default", 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), original_topic, 2));
const auto created_topic =
- client->get_topic(make_string_identifier(stream_name), make_string_identifier(original_topic));
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(original_topic));
- ASSERT_NO_THROW(client->update_topic(make_string_identifier(stream_name), make_numeric_identifier(created_topic.id),
- updated_topic_name, "gzip", "duration", 1000, "1GiB", {}));
+ ASSERT_NO_THROW(client.UpdateTopic(iggy::Identifier::String(stream_name),
+ iggy::Identifier::Numeric(created_topic.Id()), updated_topic_name,
+ iggy::CompressionAlgorithm::Gzip(), iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)));
const auto first_update =
- client->get_topic(make_string_identifier(stream_name), make_numeric_identifier(created_topic.id));
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::Numeric(created_topic.Id()));
- ASSERT_NO_THROW(client->update_topic(make_string_identifier(stream_name), make_numeric_identifier(created_topic.id),
- updated_topic_name, "gzip", "duration", 1000, "1GiB", {}));
+ ASSERT_NO_THROW(client.UpdateTopic(iggy::Identifier::String(stream_name),
+ iggy::Identifier::Numeric(created_topic.Id()), updated_topic_name,
+ iggy::CompressionAlgorithm::Gzip(), iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)));
const auto second_update =
- client->get_topic(make_string_identifier(stream_name), make_numeric_identifier(created_topic.id));
-
- EXPECT_EQ(second_update.id, first_update.id);
- EXPECT_EQ(second_update.name, first_update.name);
- EXPECT_EQ(second_update.partitions_count, first_update.partitions_count);
- EXPECT_EQ(second_update.compression_algorithm, first_update.compression_algorithm);
- EXPECT_EQ(second_update.message_expiry, first_update.message_expiry);
- EXPECT_EQ(second_update.max_topic_size, first_update.max_topic_size);
+ client.GetTopic(iggy::Identifier::String(stream_name), iggy::Identifier::Numeric(created_topic.Id()));
+
+ EXPECT_EQ(second_update.Id(), first_update.Id());
+ EXPECT_EQ(second_update.Name(), first_update.Name());
+ EXPECT_EQ(second_update.PartitionsCount(), first_update.PartitionsCount());
+ EXPECT_EQ(second_update.CompressionAlgorithm(), first_update.CompressionAlgorithm());
+ EXPECT_EQ(second_update.MessageExpiry(), first_update.MessageExpiry());
+ EXPECT_EQ(second_update.MaxTopicSize(), first_update.MaxTopicSize());
}
-TEST_F(LowLevelE2E_Topic, UpdateTopicWithDuplicateTopicNameThrows) {
+TEST_F(E2E_Topic, UpdateTopicWithDuplicateTopicNameThrows) {
RecordProperty("description", "Rejects renaming a topic to another topic's existing name.");
const std::string stream_name = GetRandomName();
const std::string first_topic_name = GetRandomName();
const std::string second_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), first_topic_name, 1, "none",
- "server_default", 0, "server_default", {}));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), second_topic_name, 1, "none",
- "server_default", 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), first_topic_name, 1));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), second_topic_name, 1));
- ASSERT_THROW(client->update_topic(make_string_identifier(stream_name), make_string_identifier(first_topic_name),
- second_topic_name, "gzip", "duration", 1000, "1GiB", {}),
+ ASSERT_THROW(client.UpdateTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(first_topic_name),
+ second_topic_name, iggy::CompressionAlgorithm::Gzip(), iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, UpdateTopicWithInvalidNamesThrows) {
+TEST_F(E2E_Topic, UpdateTopicWithInvalidNamesThrows) {
RecordProperty("description", "Rejects invalid topic names when updating a topic.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
const std::vector<std::string> invalid_topic_names = {
"",
@@ -1173,13 +1153,15 @@ TEST_F(LowLevelE2E_Topic, UpdateTopicWithInvalidNamesThrows) {
for (const auto &invalid_topic_name : invalid_topic_names) {
SCOPED_TRACE("invalid_topic_name_length=" + std::to_string(invalid_topic_name.size()));
- ASSERT_THROW(client->update_topic(make_string_identifier(stream_name), make_string_identifier(topic_name),
- invalid_topic_name, "gzip", "duration", 1000, "1GiB", {}),
- std::exception);
+ ASSERT_THROW(
+ client.UpdateTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ invalid_topic_name, iggy::CompressionAlgorithm::Gzip(), iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)),
+ std::exception);
}
}
-TEST_F(LowLevelE2E_Topic, UpdateTopicFailedValidationDoesNotMutateTopic) {
+TEST_F(E2E_Topic, UpdateTopicFailedValidationDoesNotMutateTopic) {
RecordProperty("description", "Keeps the topic unchanged when update_topic fails wrapper validation.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -1213,131 +1195,122 @@ TEST_F(LowLevelE2E_Topic, UpdateTopicFailedValidationDoesNotMutateTopic) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, UpdateTopicBeforeLoginThrows) {
+TEST_F(E2E_Topic, UpdateTopicBeforeLoginThrows) {
RecordProperty("description", "Rejects update_topic before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string updated_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(
- unauthenticated_client->update_topic(make_string_identifier(stream_name), make_string_identifier(topic_name),
- updated_topic_name, "gzip", "duration", 1000, "1GiB", {}),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
- ASSERT_THROW(
- unauthenticated_client->update_topic(make_string_identifier(stream_name), make_string_identifier(topic_name),
- updated_topic_name, "gzip", "duration", 1000, "1GiB", {}),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
- ASSERT_THROW(
- unauthenticated_client->update_topic(make_string_identifier(stream_name), make_string_identifier(topic_name),
- updated_topic_name, "gzip", "duration", 1000, "1GiB", {}),
- std::exception);
+ ASSERT_THROW(unauthenticated_client.UpdateTopic(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), updated_topic_name),
+ std::exception);
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
+ ASSERT_THROW(unauthenticated_client.UpdateTopic(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), updated_topic_name),
+ std::exception);
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
+ ASSERT_THROW(unauthenticated_client.UpdateTopic(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), updated_topic_name),
+ std::exception);
}
-TEST_F(LowLevelE2E_Topic, UpdateTopicOnNonExistentStreamThrows) {
+TEST_F(E2E_Topic, UpdateTopicOnNonExistentStreamThrows) {
RecordProperty("description", "Throws when updating a topic on a stream that does not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string updated_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_THROW(client->update_topic(make_string_identifier(stream_name), make_string_identifier(topic_name),
- updated_topic_name, "gzip", "duration", 1000, "1GiB", {}),
- std::exception);
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_THROW(
+ client.UpdateTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ updated_topic_name, iggy::CompressionAlgorithm::Gzip(), iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)),
+ std::exception);
}
-TEST_F(LowLevelE2E_Topic, UpdateTopicOnNonExistentTopicThrows) {
+TEST_F(E2E_Topic, UpdateTopicOnNonExistentTopicThrows) {
RecordProperty("description", "Throws when updating a topic that does not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string updated_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_THROW(client->update_topic(make_string_identifier(stream_name), make_string_identifier(topic_name),
- updated_topic_name, "gzip", "duration", 1000, "1GiB", {}),
- std::exception);
+ ASSERT_THROW(
+ client.UpdateTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ updated_topic_name, iggy::CompressionAlgorithm::Gzip(), iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)),
+ std::exception);
}
-TEST_F(LowLevelE2E_Topic, GetTopicsAfterStreamDeletionReturnsEmpty) {
+TEST_F(E2E_Topic, GetTopicsAfterStreamDeletionReturnsEmpty) {
RecordProperty("description", "Returns an empty topic list after deleting the stream that owned the topic.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 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, 1));
+ ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
ASSERT_NO_THROW({
- const auto topics = client->get_topics(make_string_identifier(stream_name));
+ const auto topics = client.GetTopics(iggy::Identifier::String(stream_name));
EXPECT_TRUE(topics.empty());
});
}
-TEST_F(LowLevelE2E_Topic, PurgeTopicOnNonExistentStreamThrows) {
+TEST_F(E2E_Topic, PurgeTopicOnNonExistentStreamThrows) {
RecordProperty("description", "Throws when purging a topic on a stream that does not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_THROW(client->purge_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)),
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_THROW(client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, PurgeTopicAfterStreamDeletionThrows) {
+TEST_F(E2E_Topic, PurgeTopicAfterStreamDeletionThrows) {
RecordProperty("description", "Throws when purging a topic after its stream has been deleted.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 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, 1));
+ ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client->purge_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)),
+ ASSERT_THROW(client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, PurgeTopicOnNonExistentTopicThrows) {
+TEST_F(E2E_Topic, PurgeTopicOnNonExistentTopicThrows) {
RecordProperty("description", "Throws when purging a topic that does not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_THROW(client->purge_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)),
+ ASSERT_THROW(client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
std::exception);
}
-TEST_F(LowLevelE2E_Topic, PurgeTopicWithInvalidStreamIdentifierThrows) {
+TEST_F(E2E_Topic, PurgeTopicWithInvalidStreamIdentifierThrows) {
RecordProperty("description", "Rejects topic purge requests that use invalid stream identifier formats.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -1363,7 +1336,7 @@ TEST_F(LowLevelE2E_Topic, PurgeTopicWithInvalidStreamIdentifierThrows) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, PurgeTopicWithInvalidTopicIdentifierThrows) {
+TEST_F(E2E_Topic, PurgeTopicWithInvalidTopicIdentifierThrows) {
RecordProperty("description", "Rejects topic purge requests that use invalid topic identifier formats.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -1389,128 +1362,108 @@ TEST_F(LowLevelE2E_Topic, PurgeTopicWithInvalidTopicIdentifierThrows) {
std::exception);
}
-TEST_F(LowLevelE2E_Topic, PurgeTopicPreservesTopicMetadata) {
+TEST_F(E2E_Topic, PurgeTopicPreservesTopicMetadata) {
RecordProperty("description", "Preserves topic metadata after purging its messages.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *message_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(
- client->create_topic(make_string_identifier(stream_name), topic_name, 3, "gzip", "duration", 1000, "1GiB", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 3,
+ iggy::CompressionAlgorithm::Gzip(), iggy::Expiry::Duration(1000),
+ iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL)));
- auto stream_before_purge = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_before_purge.topics.size(), 1u);
- const auto topic_before_messages = stream_before_purge.topics.front();
+ auto stream_before_purge = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_before_purge.Topics().size(), 1u);
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
messages.push_back(
iggy::ffi::make_message(to_payload("preserve-topic-metadata"), rust::Vec<iggy::ffi::HeaderEntry>()));
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_before_purge.id),
- make_numeric_identifier(topic_before_messages.id), "partition_id",
- partition_id_bytes(1), std::move(messages)));
-
- stream_before_purge = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_before_purge.topics.size(), 1u);
- const auto topic_with_messages = stream_before_purge.topics.front();
- EXPECT_GT(topic_with_messages.messages_count, 0u);
- EXPECT_GT(topic_with_messages.size_bytes, 0u);
-
- ASSERT_NO_THROW(client->purge_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)));
-
- const auto stream_after_purge = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(stream_after_purge.topics.size(), 1u);
- const auto topic_after_purge = stream_after_purge.topics.front();
-
- EXPECT_EQ(topic_after_purge.id, topic_with_messages.id);
- EXPECT_EQ(topic_after_purge.created_at, topic_with_messages.created_at);
- EXPECT_EQ(topic_after_purge.name, topic_with_messages.name);
- EXPECT_EQ(topic_after_purge.message_expiry, topic_with_messages.message_expiry);
- EXPECT_EQ(topic_after_purge.compression_algorithm, topic_with_messages.compression_algorithm);
- EXPECT_EQ(topic_after_purge.max_topic_size, topic_with_messages.max_topic_size);
- EXPECT_EQ(topic_after_purge.partitions_count, topic_with_messages.partitions_count);
+ ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name),
+ make_string_identifier(topic_name), "partition_id",
+ partition_id_bytes(1), std::move(messages)));
+
+ stream_before_purge = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_before_purge.Topics().size(), 1u);
+ const auto &topic_with_messages = stream_before_purge.Topics().front();
+ EXPECT_GT(topic_with_messages.MessagesCount(), 0u);
+ EXPECT_GT(topic_with_messages.SizeBytes(), 0u);
+
+ ASSERT_NO_THROW(client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)));
+
+ const auto stream_after_purge = client.GetStream(iggy::Identifier::String(stream_name));
+ ASSERT_EQ(stream_after_purge.Topics().size(), 1u);
+ const auto &topic_after_purge = stream_after_purge.Topics().front();
+
+ EXPECT_EQ(topic_after_purge.Id(), topic_with_messages.Id());
+ EXPECT_EQ(topic_after_purge.CreatedAt(), topic_with_messages.CreatedAt());
+ EXPECT_EQ(topic_after_purge.Name(), topic_with_messages.Name());
+ EXPECT_EQ(topic_after_purge.MessageExpiry(), topic_with_messages.MessageExpiry());
+ EXPECT_EQ(topic_after_purge.CompressionAlgorithm(), topic_with_messages.CompressionAlgorithm());
+ EXPECT_EQ(topic_after_purge.MaxTopicSize(), topic_with_messages.MaxTopicSize());
+ EXPECT_EQ(topic_after_purge.PartitionsCount(), topic_with_messages.PartitionsCount());
}
-TEST_F(LowLevelE2E_Topic, PurgeTopicRemovesOnlyTargetTopicMessages) {
+TEST_F(E2E_Topic, PurgeTopicRemovesOnlyTargetTopicMessages) {
RecordProperty("description", "Purges one topic's messages without affecting the other topics in the stream.");
const std::string stream_name = GetRandomName();
const std::string first_topic_name = GetRandomName();
const std::string second_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *message_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), first_topic_name, 1, "none",
- "server_default", 0, "server_default", {}));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), second_topic_name, 1, "none",
- "server_default", 0, "server_default", {}));
-
- const auto created_stream = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(created_stream.topics.size(), 2u);
-
- std::uint32_t first_topic_id = 0;
- std::uint32_t second_topic_id = 0;
- bool first_topic_found = false;
- bool second_topic_found = false;
- for (const auto &topic : created_stream.topics) {
- const std::string topic_name = static_cast<std::string>(topic.name);
- if (topic_name == first_topic_name) {
- first_topic_id = topic.id;
- first_topic_found = true;
- } else if (topic_name == second_topic_name) {
- second_topic_id = topic.id;
- second_topic_found = true;
- }
- }
- ASSERT_TRUE(first_topic_found);
- ASSERT_TRUE(second_topic_found);
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), first_topic_name, 1));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), second_topic_name, 1));
rust::Vec<iggy::ffi::IggyMessageToSend> first_topic_messages;
for (std::uint32_t i = 0; i < 3; ++i) {
first_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-topic-first-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>()));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
- make_numeric_identifier(first_topic_id), "partition_id",
- partition_id_bytes(0), std::move(first_topic_messages)));
+ ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name),
+ make_string_identifier(first_topic_name), "partition_id",
+ partition_id_bytes(0), std::move(first_topic_messages)));
rust::Vec<iggy::ffi::IggyMessageToSend> second_topic_messages;
for (std::uint32_t i = 0; i < 2; ++i) {
second_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-topic-second-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>()));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
- make_numeric_identifier(second_topic_id), "partition_id",
- partition_id_bytes(0), std::move(second_topic_messages)));
+ ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name),
+ make_string_identifier(second_topic_name), "partition_id",
+ partition_id_bytes(0), std::move(second_topic_messages)));
- const auto stream_before_purge = client->get_stream(make_string_identifier(stream_name));
- EXPECT_EQ(stream_before_purge.messages_count, 5u);
- EXPECT_GT(stream_before_purge.size_bytes, 0u);
+ const auto stream_before_purge = client.GetStream(iggy::Identifier::String(stream_name));
+ EXPECT_EQ(stream_before_purge.MessagesCount(), 5u);
+ EXPECT_GT(stream_before_purge.SizeBytes(), 0u);
std::unordered_map<std::string, std::uint64_t> messages_before_purge;
- for (const auto &topic : stream_before_purge.topics) {
- messages_before_purge[static_cast<std::string>(topic.name)] = topic.messages_count;
+ for (const auto &topic : stream_before_purge.Topics()) {
+ messages_before_purge[topic.Name()] = topic.MessagesCount();
}
EXPECT_EQ(messages_before_purge[first_topic_name], 3u);
EXPECT_EQ(messages_before_purge[second_topic_name], 2u);
- ASSERT_NO_THROW(client->purge_topic(make_string_identifier(stream_name), make_string_identifier(first_topic_name)));
+ ASSERT_NO_THROW(
+ client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(first_topic_name)));
- const auto stream_after_purge = client->get_stream(make_string_identifier(stream_name));
- EXPECT_EQ(stream_after_purge.topics_count, 2u);
- EXPECT_EQ(stream_after_purge.messages_count, 2u);
- EXPECT_GT(stream_after_purge.size_bytes, 0u);
- EXPECT_LT(stream_after_purge.size_bytes, stream_before_purge.size_bytes);
+ const auto stream_after_purge = client.GetStream(iggy::Identifier::String(stream_name));
+ EXPECT_EQ(stream_after_purge.TopicsCount(), 2u);
+ EXPECT_EQ(stream_after_purge.MessagesCount(), 2u);
+ EXPECT_GT(stream_after_purge.SizeBytes(), 0u);
+ EXPECT_LT(stream_after_purge.SizeBytes(), stream_before_purge.SizeBytes());
std::unordered_map<std::string, std::uint64_t> messages_after_purge;
std::unordered_map<std::string, std::uint64_t> sizes_after_purge;
- for (const auto &topic : stream_after_purge.topics) {
- const std::string topic_name = static_cast<std::string>(topic.name);
- messages_after_purge[topic_name] = topic.messages_count;
- sizes_after_purge[topic_name] = topic.size_bytes;
+ for (const auto &topic : stream_after_purge.Topics()) {
+ messages_after_purge[topic.Name()] = topic.MessagesCount();
+ sizes_after_purge[topic.Name()] = topic.SizeBytes();
}
EXPECT_EQ(messages_after_purge[first_topic_name], 0u);
EXPECT_EQ(sizes_after_purge[first_topic_name], 0u);
@@ -1518,21 +1471,17 @@ TEST_F(LowLevelE2E_Topic, PurgeTopicRemovesOnlyTargetTopicMessages) {
EXPECT_GT(sizes_after_purge[second_topic_name], 0u);
}
-TEST_F(LowLevelE2E_Topic, PurgeTopicAcrossMultiplePartitionsClearsAllPartitions) {
+TEST_F(E2E_Topic, PurgeTopicAcrossMultiplePartitionsClearsAllPartitions) {
RecordProperty("description", "Purges all messages from every partition in the topic.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *message_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 3, "none", "server_default",
- 0, "server_default", {}));
-
- const auto created_stream = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(created_stream.topics.size(), 1u);
- const std::uint32_t topic_id = created_stream.topics.front().id;
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 3));
for (std::uint32_t partition_id = 0; partition_id < 3; ++partition_id) {
rust::Vec<iggy::ffi::IggyMessageToSend> messages;
@@ -1541,148 +1490,125 @@ TEST_F(LowLevelE2E_Topic, PurgeTopicAcrossMultiplePartitionsClearsAllPartitions)
to_payload("purge-topic-partition-" + std::to_string(partition_id) + "-" + std::to_string(i)),
rust::Vec<iggy::ffi::HeaderEntry>()));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
- make_numeric_identifier(topic_id), "partition_id",
- partition_id_bytes(partition_id), std::move(messages)));
+ ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name),
+ make_string_identifier(topic_name), "partition_id",
+ partition_id_bytes(partition_id), std::move(messages)));
}
- const auto stream_before_purge = client->get_stream(make_string_identifier(stream_name));
- EXPECT_EQ(stream_before_purge.messages_count, 6u);
- ASSERT_EQ(stream_before_purge.topics.size(), 1u);
- EXPECT_EQ(stream_before_purge.topics.front().partitions_count, 3u);
- EXPECT_EQ(stream_before_purge.topics.front().messages_count, 6u);
+ const auto stream_before_purge = client.GetStream(iggy::Identifier::String(stream_name));
+ EXPECT_EQ(stream_before_purge.MessagesCount(), 6u);
+ ASSERT_EQ(stream_before_purge.Topics().size(), 1u);
+ EXPECT_EQ(stream_before_purge.Topics().front().PartitionsCount(), 3u);
+ EXPECT_EQ(stream_before_purge.Topics().front().MessagesCount(), 6u);
- ASSERT_NO_THROW(client->purge_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)));
+ ASSERT_NO_THROW(client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)));
- const auto stream_after_purge = client->get_stream(make_string_identifier(stream_name));
- EXPECT_EQ(stream_after_purge.messages_count, 0u);
- ASSERT_EQ(stream_after_purge.topics.size(), 1u);
- EXPECT_EQ(stream_after_purge.topics.front().partitions_count, 3u);
- EXPECT_EQ(stream_after_purge.topics.front().messages_count, 0u);
- EXPECT_EQ(stream_after_purge.topics.front().size_bytes, 0u);
+ const auto stream_after_purge = client.GetStream(iggy::Identifier::String(stream_name));
+ EXPECT_EQ(stream_after_purge.MessagesCount(), 0u);
+ ASSERT_EQ(stream_after_purge.Topics().size(), 1u);
+ EXPECT_EQ(stream_after_purge.Topics().front().PartitionsCount(), 3u);
+ EXPECT_EQ(stream_after_purge.Topics().front().MessagesCount(), 0u);
+ EXPECT_EQ(stream_after_purge.Topics().front().SizeBytes(), 0u);
}
-TEST_F(LowLevelE2E_Topic, PurgeTopicThenSendMessagesAgainSucceeds) {
+TEST_F(E2E_Topic, PurgeTopicThenSendMessagesAgainSucceeds) {
RecordProperty("description", "Allows sending fresh messages again after purging the topic.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *message_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
- 0, "server_default", {}));
-
- const auto created_stream = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(created_stream.topics.size(), 1u);
- const std::uint32_t topic_id = created_stream.topics.front().id;
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
rust::Vec<iggy::ffi::IggyMessageToSend> first_batch;
first_batch.push_back(
iggy::ffi::make_message(to_payload("before-topic-purge"), rust::Vec<iggy::ffi::HeaderEntry>()));
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
- "partition_id", partition_id_bytes(0), std::move(first_batch)));
+ ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name),
+ make_string_identifier(topic_name), "partition_id",
+ partition_id_bytes(0), std::move(first_batch)));
- ASSERT_NO_THROW(client->purge_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)));
+ ASSERT_NO_THROW(client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)));
rust::Vec<iggy::ffi::IggyMessageToSend> second_batch;
second_batch.push_back(
iggy::ffi::make_message(to_payload("after-topic-purge-0"), rust::Vec<iggy::ffi::HeaderEntry>()));
second_batch.push_back(
iggy::ffi::make_message(to_payload("after-topic-purge-1"), rust::Vec<iggy::ffi::HeaderEntry>()));
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
- "partition_id", partition_id_bytes(0), std::move(second_batch)));
-
- const auto stream_after_resend = client->get_stream(make_string_identifier(stream_name));
- EXPECT_EQ(stream_after_resend.messages_count, 2u);
- ASSERT_EQ(stream_after_resend.topics.size(), 1u);
- EXPECT_EQ(stream_after_resend.topics.front().messages_count, 2u);
- EXPECT_GT(stream_after_resend.topics.front().size_bytes, 0u);
+ ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name),
+ make_string_identifier(topic_name), "partition_id",
+ partition_id_bytes(0), std::move(second_batch)));
+
+ const auto stream_after_resend = client.GetStream(iggy::Identifier::String(stream_name));
+ EXPECT_EQ(stream_after_resend.MessagesCount(), 2u);
+ ASSERT_EQ(stream_after_resend.Topics().size(), 1u);
+ EXPECT_EQ(stream_after_resend.Topics().front().MessagesCount(), 2u);
+ EXPECT_GT(stream_after_resend.Topics().front().SizeBytes(), 0u);
}
-TEST_F(LowLevelE2E_Topic, PurgeTopicTwiceKeepsTargetTopicEmptyAndOtherTopicsUntouched) {
+TEST_F(E2E_Topic, PurgeTopicTwiceKeepsTargetTopicEmptyAndOtherTopicsUntouched) {
RecordProperty("description",
"Allows purging the same topic twice and keeps the target topic empty without affecting siblings.");
const std::string stream_name = GetRandomName();
const std::string first_topic_name = GetRandomName();
const std::string second_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ iggy::ffi::Client *message_client = GetLoggedInClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), first_topic_name, 1, "none",
- "server_default", 0, "server_default", {}));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), second_topic_name, 1, "none",
- "server_default", 0, "server_default", {}));
-
- const auto created_stream = client->get_stream(make_string_identifier(stream_name));
- ASSERT_EQ(created_stream.topics.size(), 2u);
-
- std::uint32_t first_topic_id = 0;
- std::uint32_t second_topic_id = 0;
- bool first_topic_found = false;
- bool second_topic_found = false;
- for (const auto &topic : created_stream.topics) {
- const std::string topic_name = static_cast<std::string>(topic.name);
- if (topic_name == first_topic_name) {
- first_topic_id = topic.id;
- first_topic_found = true;
- } else if (topic_name == second_topic_name) {
- second_topic_id = topic.id;
- second_topic_found = true;
- }
- }
- ASSERT_TRUE(first_topic_found);
- ASSERT_TRUE(second_topic_found);
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), first_topic_name, 1));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), second_topic_name, 1));
rust::Vec<iggy::ffi::IggyMessageToSend> first_topic_messages;
for (std::uint32_t i = 0; i < 3; ++i) {
first_topic_messages.push_back(iggy::ffi::make_message(
to_payload("purge-topic-twice-first-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>()));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
- make_numeric_identifier(first_topic_id), "partition_id",
- partition_id_bytes(0), std::move(first_topic_messages)));
+ ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name),
+ make_string_identifier(first_topic_name), "partition_id",
+ partition_id_bytes(0), std::move(first_topic_messages)));
rust::Vec<iggy::ffi::IggyMessageToSend> second_topic_messages;
for (std::uint32_t i = 0; i < 2; ++i) {
second_topic_messages.push_back(iggy::ffi::make_message(
to_payload("purge-topic-twice-second-" + std::to_string(i)), rust::Vec<iggy::ffi::HeaderEntry>()));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
- make_numeric_identifier(second_topic_id), "partition_id",
- partition_id_bytes(0), std::move(second_topic_messages)));
+ ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name),
+ make_string_identifier(second_topic_name), "partition_id",
+ partition_id_bytes(0), std::move(second_topic_messages)));
- ASSERT_NO_THROW(client->purge_topic(make_string_identifier(stream_name), make_string_identifier(first_topic_name)));
- const auto stream_after_first_purge = client->get_stream(make_string_identifier(stream_name));
- EXPECT_EQ(stream_after_first_purge.topics_count, 2u);
- EXPECT_EQ(stream_after_first_purge.messages_count, 2u);
+ ASSERT_NO_THROW(
+ client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(first_topic_name)));
+ const auto stream_after_first_purge = client.GetStream(iggy::Identifier::String(stream_name));
+ EXPECT_EQ(stream_after_first_purge.TopicsCount(), 2u);
+ EXPECT_EQ(stream_after_first_purge.MessagesCount(), 2u);
std::unordered_map<std::string, std::uint64_t> messages_after_first_purge;
std::unordered_map<std::string, std::uint64_t> sizes_after_first_purge;
- for (const auto &topic : stream_after_first_purge.topics) {
- const std::string topic_name = static_cast<std::string>(topic.name);
- messages_after_first_purge[topic_name] = topic.messages_count;
- sizes_after_first_purge[topic_name] = topic.size_bytes;
+ for (const auto &topic : stream_after_first_purge.Topics()) {
+ messages_after_first_purge[topic.Name()] = topic.MessagesCount();
+ sizes_after_first_purge[topic.Name()] = topic.SizeBytes();
}
EXPECT_EQ(messages_after_first_purge[first_topic_name], 0u);
EXPECT_EQ(sizes_after_first_purge[first_topic_name], 0u);
EXPECT_EQ(messages_after_first_purge[second_topic_name], 2u);
EXPECT_GT(sizes_after_first_purge[second_topic_name], 0u);
- ASSERT_NO_THROW(client->purge_topic(make_string_identifier(stream_name), make_string_identifier(first_topic_name)));
- const auto stream_after_second_purge = client->get_stream(make_string_identifier(stream_name));
- EXPECT_EQ(stream_after_second_purge.topics_count, 2u);
- EXPECT_EQ(stream_after_second_purge.messages_count, 2u);
+ ASSERT_NO_THROW(
+ client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(first_topic_name)));
+ const auto stream_after_second_purge = client.GetStream(iggy::Identifier::String(stream_name));
+ EXPECT_EQ(stream_after_second_purge.TopicsCount(), 2u);
+ EXPECT_EQ(stream_after_second_purge.MessagesCount(), 2u);
std::unordered_map<std::string, std::uint64_t> messages_after_second_purge;
std::unordered_map<std::string, std::uint64_t> sizes_after_second_purge;
- for (const auto &topic : stream_after_second_purge.topics) {
- const std::string topic_name = static_cast<std::string>(topic.name);
- messages_after_second_purge[topic_name] = topic.messages_count;
- sizes_after_second_purge[topic_name] = topic.size_bytes;
+ for (const auto &topic : stream_after_second_purge.Topics()) {
+ messages_after_second_purge[topic.Name()] = topic.MessagesCount();
+ sizes_after_second_purge[topic.Name()] = topic.SizeBytes();
}
EXPECT_EQ(messages_after_second_purge[first_topic_name], 0u);
EXPECT_EQ(sizes_after_second_purge[first_topic_name], 0u);
@@ -1690,30 +1616,28 @@ TEST_F(LowLevelE2E_Topic, PurgeTopicTwiceKeepsTargetTopicEmptyAndOtherTopicsUnto
EXPECT_GT(sizes_after_second_purge[second_topic_name], 0u);
}
-TEST_F(LowLevelE2E_Topic, PurgeTopicBeforeLoginThrows) {
+TEST_F(E2E_Topic, PurgeTopicBeforeLoginThrows) {
RecordProperty("description", "Throws when topic purge is attempted before authentication.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ auto client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", "server_default",
- 0, "server_default", {}));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, 1));
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
ASSERT_THROW(
- unauthenticated_client->purge_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)),
+ unauthenticated_client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
ASSERT_THROW(
- unauthenticated_client->purge_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)),
+ unauthenticated_client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
ASSERT_THROW(
- unauthenticated_client->purge_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)),
+ unauthenticated_client.PurgeTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
std::exception);
}
diff --git a/foreign/cpp/tests/unit/unit_tests.cpp b/foreign/cpp/tests/unit/unit_tests.cpp
index 18db58f65..4e9179a0a 100644
--- a/foreign/cpp/tests/unit/unit_tests.cpp
+++ b/foreign/cpp/tests/unit/unit_tests.cpp
@@ -28,16 +28,12 @@
namespace {
-std::string option_key(const iggy::ffi::HeaderEntry &entry) {
- return std::string(entry.key.value.begin(), entry.key.value.end());
+std::string option_key(const iggy::HeaderEntry &entry) {
+ return std::string(entry.Key().Value().begin(), entry.Key().Value().end());
}
-std::vector<std::uint8_t> option_value_bytes(const iggy::ffi::HeaderEntry &entry) {
- return std::vector<std::uint8_t>(entry.value.value.begin(), entry.value.value.end());
-}
-
-constexpr std::uint8_t kind_code(const iggy::ffi::HeaderKind kind) {
- return static_cast<std::uint8_t>(kind);
+const std::vector<std::uint8_t> &option_value_bytes(const iggy::HeaderEntry &entry) {
+ return entry.Value().Value();
}
} // namespace
@@ -66,11 +62,6 @@ TEST(SystemSnapshotTypeTest, ReturnsExpectedValues) {
EXPECT_EQ(iggy::SystemSnapshotType::All().SnapshotTypeValue(), "all");
}
-TEST(IdKindTest, ReturnsExpectedValues) {
- EXPECT_EQ(iggy::IdKind::Numeric().IdKindValue(), "numeric");
- EXPECT_EQ(iggy::IdKind::String().IdKindValue(), "string");
-}
-
TEST(MaxTopicSizeTest, ReturnsExpectedValues) {
EXPECT_EQ(iggy::MaxTopicSize::ServerDefault().MaxTopicSizeValue(), "server_default");
EXPECT_EQ(iggy::MaxTopicSize::Unlimited().MaxTopicSizeValue(), "unlimited");
@@ -119,51 +110,51 @@ TEST(ExpiryTest, ReturnsExpectedKindAndValue) {
TEST(TopicOptionTest, SegmentSizeEncodesLittleEndianUint64) {
const auto option = iggy::TopicOption::SegmentSize(0x0102030405060708ULL);
- EXPECT_EQ(option.key.kind, kind_code(iggy::ffi::HeaderKind::String));
+ EXPECT_EQ(option.Key().Kind(), iggy::HeaderKind::String);
EXPECT_EQ(option_key(option), "segment_size");
- EXPECT_EQ(option.value.kind, kind_code(iggy::ffi::HeaderKind::Uint64));
+ EXPECT_EQ(option.Value().Kind(), iggy::HeaderKind::Uint64);
EXPECT_EQ(option_value_bytes(option), (std::vector<std::uint8_t>{0x08, 0x07, 0x06, 0x05, 0x04, 0x03, 0x02, 0x01}));
}
TEST(TopicOptionTest, EnforceFsyncEncodesSingleBoolByte) {
const auto enabled = iggy::TopicOption::EnforceFsync(true);
- EXPECT_EQ(enabled.key.kind, kind_code(iggy::ffi::HeaderKind::String));
+ EXPECT_EQ(enabled.Key().Kind(), iggy::HeaderKind::String);
EXPECT_EQ(option_key(enabled), "enforce_fsync");
- EXPECT_EQ(enabled.value.kind, kind_code(iggy::ffi::HeaderKind::Bool));
+ EXPECT_EQ(enabled.Value().Kind(), iggy::HeaderKind::Bool);
EXPECT_EQ(option_value_bytes(enabled), (std::vector<std::uint8_t>{1}));
const auto disabled = iggy::TopicOption::EnforceFsync(false);
EXPECT_EQ(option_key(disabled), "enforce_fsync");
- EXPECT_EQ(disabled.value.kind, kind_code(iggy::ffi::HeaderKind::Bool));
+ EXPECT_EQ(disabled.Value().Kind(), iggy::HeaderKind::Bool);
EXPECT_EQ(option_value_bytes(disabled), (std::vector<std::uint8_t>{0}));
}
TEST(TopicOptionTest, MessagesRequiredToSaveEncodesLittleEndianUint32) {
const auto option = iggy::TopicOption::MessagesRequiredToSave(0x01020304U);
- EXPECT_EQ(option.key.kind, kind_code(iggy::ffi::HeaderKind::String));
+ EXPECT_EQ(option.Key().Kind(), iggy::HeaderKind::String);
EXPECT_EQ(option_key(option), "messages_required_to_save");
- EXPECT_EQ(option.value.kind, kind_code(iggy::ffi::HeaderKind::Uint32));
+ EXPECT_EQ(option.Value().Kind(), iggy::HeaderKind::Uint32);
EXPECT_EQ(option_value_bytes(option), (std::vector<std::uint8_t>{0x04, 0x03, 0x02, 0x01}));
}
TEST(TopicOptionTest, SizeOfMessagesRequiredToSaveEncodesLittleEndianUint64) {
const auto option = iggy::TopicOption::SizeOfMessagesRequiredToSave(1024ULL * 1024ULL);
- EXPECT_EQ(option.key.kind, kind_code(iggy::ffi::HeaderKind::String));
+ EXPECT_EQ(option.Key().Kind(), iggy::HeaderKind::String);
EXPECT_EQ(option_key(option), "size_of_messages_required_to_save");
- EXPECT_EQ(option.value.kind, kind_code(iggy::ffi::HeaderKind::Uint64));
+ EXPECT_EQ(option.Value().Kind(), iggy::HeaderKind::Uint64);
EXPECT_EQ(option_value_bytes(option), (std::vector<std::uint8_t>{0x00, 0x00, 0x10, 0x00, 0x00, 0x00, 0x00, 0x00}));
}
TEST(TopicOptionTest, PreallocateSegmentsEncodesSingleBoolByte) {
const auto enabled = iggy::TopicOption::PreallocateSegments(true);
- EXPECT_EQ(enabled.key.kind, kind_code(iggy::ffi::HeaderKind::String));
+ EXPECT_EQ(enabled.Key().Kind(), iggy::HeaderKind::String);
EXPECT_EQ(option_key(enabled), "preallocate_segments");
- EXPECT_EQ(enabled.value.kind, kind_code(iggy::ffi::HeaderKind::Bool));
+ EXPECT_EQ(enabled.Value().Kind(), iggy::HeaderKind::Bool);
EXPECT_EQ(option_value_bytes(enabled), (std::vector<std::uint8_t>{1}));
const auto disabled = iggy::TopicOption::PreallocateSegments(false);
@@ -190,18 +181,6 @@ TEST(IggyExceptionTest, StoresMessage) {
EXPECT_EQ(std::string(from_string.what()), message);
}
-TEST(IggyBlockingClientTest, MovedFromOperationsThrow) {
- auto client = iggy::IggyBlockingClient::Builder().Build();
- auto moved_to = std::move(client);
- (void)moved_to;
-
- EXPECT_THROW(client.Connect(), iggy::IggyException);
- EXPECT_THROW(client.Disconnect(), iggy::IggyException);
- EXPECT_THROW(client.Shutdown(), iggy::IggyException);
- EXPECT_THROW(client.Login("iggy", "iggy"), iggy::IggyException);
- EXPECT_THROW(client.Logout(), iggy::IggyException);
-}
-
TEST(AutoLoginKindTest, HasStableDiscriminantsAndZeroInitializedDefault) {
EXPECT_EQ(static_cast<std::uint8_t>(iggy::ffi::AutoLoginKind::Disabled), 0u);
EXPECT_EQ(static_cast<std::uint8_t>(iggy::ffi::AutoLoginKind::UsernamePassword), 1u);