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
The following commit(s) were added to refs/heads/cpp-high-level-client-1 by
this push:
new d698374b0 fixes
d698374b0 is described below
commit d698374b04eae05f91d4ef4fa77643b90e1e0de4
Author: Rimuksh Kansal <[email protected]>
AuthorDate: Sat Sep 12 01:24:47 2026 +0900
fixes
---
foreign/cpp/MODULE.bazel.lock | 103 +++++++++++++++++++++++
foreign/cpp/include/iggy.hpp | 152 ++--------------------------------
foreign/cpp/src/client.cpp | 142 +++++++++++--------------------
foreign/cpp/src/type_conversion.rs | 127 ++++++++++++++--------------
foreign/cpp/src/type_conversions.cpp | 21 -----
foreign/cpp/tests/e2e/partition.cpp | 112 ++++++++++++++++---------
foreign/cpp/tests/e2e/stream.cpp | 70 ++++++++--------
foreign/cpp/tests/e2e/topic.cpp | 132 ++++++++++++++++++-----------
foreign/cpp/tests/unit/unit_tests.cpp | 60 +++++++-------
9 files changed, 440 insertions(+), 479 deletions(-)
diff --git a/foreign/cpp/MODULE.bazel.lock b/foreign/cpp/MODULE.bazel.lock
index c6f3c74b7..55a016bf6 100644
--- a/foreign/cpp/MODULE.bazel.lock
+++ b/foreign/cpp/MODULE.bazel.lock
@@ -483,6 +483,109 @@
}
}
}
+ },
+ "@@rules_rust+//crate_universe/private:internal_extensions.bzl%cu_nr": {
+ "general": {
+ "bzlTransitiveDigest": "qQd7ktlamEwHDmeHLMSsEyM7+9Y2dOsIhs3Y/RXmCDY=",
+ "usagesDigest": "McimNXAztVHDzspa1Kng94T+YeyY/44zx9FkNqIMn2s=",
+ "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",
+ "REPO_MAPPING:rules_rust+,cui rules_rust++cu+cui",
+ "REPO_MAPPING:rules_rust+,rrc rules_rust++i2+rrc",
+ "REPO_MAPPING:rules_rust+,rules_cc rules_cc+",
+ "REPO_MAPPING:rules_rust+,rules_rust rules_rust+"
+ ],
+ "generatedRepoSpecs": {
+ "cargo_bazel_bootstrap": {
+ "repoRuleId":
"@@rules_rust+//cargo/private:cargo_bootstrap.bzl%cargo_bootstrap_repository",
+ "attributes": {
+ "srcs": [
+ "@@rules_rust+//crate_universe:src/api.rs",
+ "@@rules_rust+//crate_universe:src/api/lockfile.rs",
+ "@@rules_rust+//crate_universe:src/cli.rs",
+ "@@rules_rust+//crate_universe:src/cli/generate.rs",
+ "@@rules_rust+//crate_universe:src/cli/query.rs",
+ "@@rules_rust+//crate_universe:src/cli/render.rs",
+ "@@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",
+ "@@rules_rust+//crate_universe:src/lib.rs",
+ "@@rules_rust+//crate_universe:src/lockfile.rs",
+ "@@rules_rust+//crate_universe:src/main.rs",
+ "@@rules_rust+//crate_universe:src/metadata.rs",
+ "@@rules_rust+//crate_universe:src/metadata/cargo_bin.rs",
+
"@@rules_rust+//crate_universe:src/metadata/cargo_tree_resolver.rs",
+
"@@rules_rust+//crate_universe:src/metadata/cargo_tree_rustc_wrapper.bat",
+
"@@rules_rust+//crate_universe:src/metadata/cargo_tree_rustc_wrapper.sh",
+ "@@rules_rust+//crate_universe:src/metadata/dependency.rs",
+
"@@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/verbatim/alias_rules.bzl",
+ "@@rules_rust+//crate_universe:src/select.rs",
+ "@@rules_rust+//crate_universe:src/splicing.rs",
+ "@@rules_rust+//crate_universe:src/splicing/cargo_config.rs",
+
"@@rules_rust+//crate_universe:src/splicing/crate_index_lookup.rs",
+ "@@rules_rust+//crate_universe:src/splicing/splicer.rs",
+ "@@rules_rust+//crate_universe:src/test.rs",
+ "@@rules_rust+//crate_universe:src/utils.rs",
+ "@@rules_rust+//crate_universe:src/utils/starlark.rs",
+ "@@rules_rust+//crate_universe:src/utils/starlark/glob.rs",
+ "@@rules_rust+//crate_universe:src/utils/starlark/label.rs",
+ "@@rules_rust+//crate_universe:src/utils/starlark/select.rs",
+
"@@rules_rust+//crate_universe:src/utils/starlark/select_dict.rs",
+
"@@rules_rust+//crate_universe:src/utils/starlark/select_list.rs",
+
"@@rules_rust+//crate_universe:src/utils/starlark/select_scalar.rs",
+
"@@rules_rust+//crate_universe:src/utils/starlark/select_set.rs",
+
"@@rules_rust+//crate_universe:src/utils/starlark/serialize.rs",
+
"@@rules_rust+//crate_universe:src/utils/starlark/target_compatible_with.rs",
+ "@@rules_rust+//crate_universe:src/utils/symlink.rs",
+ "@@rules_rust+//crate_universe:src/utils/target_triple.rs"
+ ],
+ "binary": "cargo-bazel",
+ "cargo_lockfile": "@@rules_rust+//crate_universe:Cargo.lock",
+ "cargo_toml": "@@rules_rust+//crate_universe:Cargo.toml",
+ "version": "1.98.0",
+ "rust_toolchain_cargo_template": "@rust_host_tools//:bin/{tool}",
+ "rust_toolchain_rustc_template": "@rust_host_tools//:bin/{tool}",
+ "compressed_windows_toolchain_names": false,
+ "timeout": 900
+ }
+ }
+ },
+ "moduleExtensionMetadata": {
+ "explicitRootModuleDirectDeps": [
+ "cargo_bazel_bootstrap"
+ ],
+ "explicitRootModuleDirectDevDeps": [],
+ "useAllRepos": "NO",
+ "reproducible": false
+ }
+ }
}
},
"facts": {},
diff --git a/foreign/cpp/include/iggy.hpp b/foreign/cpp/include/iggy.hpp
index 9a1c133e0..230905fb5 100644
--- a/foreign/cpp/include/iggy.hpp
+++ b/foreign/cpp/include/iggy.hpp
@@ -167,7 +167,6 @@ class Identifier final {
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;
@@ -1004,8 +1003,14 @@ class Expiry final {
* @brief Creates a time-based expiry policy.
* @param micros Message lifetime in microseconds.
* @return Time-based expiry policy.
+ * @throws std::invalid_argument if @p micros is zero.
*/
- static Expiry Duration(std::uint64_t micros) { return Expiry("duration",
micros); }
+ static Expiry Duration(std::uint64_t micros) {
+ if (micros == 0) {
+ throw std::invalid_argument("Expiry duration must be greater than
zero");
+ }
+ return Expiry("duration", micros);
+ }
/**
* @brief Returns the expiry policy kind.
@@ -1054,8 +1059,7 @@ class TopicCreateOptions final {
/**
* @brief Returns the number of partitions to create.
- * @return Configured partition count, or `std::nullopt` to use the server
- * default.
+ * @return Configured partition count, or `std::nullopt` to default to 1.
*/
[[nodiscard]] std::optional<std::uint32_t> PartitionsCount() const
noexcept { return partitions_count_; }
@@ -1539,146 +1543,6 @@ class PollingStrategy final {
std::uint64_t polling_strategy_value_;
};
-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>
-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));
- }
-
- return 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 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));
- }
-
- return bytes;
-}
-
-/// An option key is always `String`-kinded. Only the value kind varies per
key.
-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 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.
- *
- * 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. 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:
- /**
- * @brief Set the size at which this topic's segments rotate.
- *
- * Must be a multiple of 512 bytes, at least 1 MiB, and no larger than the
- * server's segment ceiling.
- *
- * @param bytes Segment size in bytes.
- * @return Encoded topic option entry.
- */
- static HeaderEntry SegmentSize(const std::uint64_t bytes) {
- return detail::to_option_entry("segment_size", HeaderKind::Uint64,
detail::to_little_endian_bytes(bytes));
- }
-
- /**
- * @brief Choose the message completion policy.
- *
- * @param value The policy, defaulting to Replicated.
- * @return Encoded topic option entry.
- */
- static HeaderEntry Durability(const ::iggy::Durability value =
::iggy::Durability::Replicated) {
- return detail::to_option_entry("durability", HeaderKind::String,
detail::to_key_bytes(to_string(value)));
- }
-
- /**
- * @brief Choose explicit offset completion independently of message
durability.
- * @param value The policy, defaulting to Replicated.
- * @return Encoded topic option entry.
- */
- static HeaderEntry ConsumerOffsetDurability(const ::iggy::Durability value
= ::iggy::Durability::Replicated) {
- return detail::to_option_entry("consumer_offset_durability",
HeaderKind::String,
- detail::to_key_bytes(to_string(value)));
- }
-
- /**
- * @brief Flush the journal once it holds this many messages.
- *
- * Must be non-zero. Paired with
- * `SizeOfMessagesRequiredToSave(bytes)`: whichever threshold trips
- * first flushes.
- *
- * @param messages Message count at which to flush the journal.
- * @return Encoded topic option entry.
- */
- 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));
- }
-
- /**
- * @brief Flush the journal once it holds this many bytes.
- *
- * Capped at 1 GiB: a threshold above the largest a segment may be never
- * trips, and the journal does not survive a crash.
- *
- * @param bytes Byte count at which to flush the journal.
- * @return Encoded topic option entry.
- */
- 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));
- }
-
- /**
- * @brief Choose whether a segment's bytes are reserved on disk when it is
created.
- *
- * Reserves `segment_size * partitions_count` up front, which the server
- * caps at 64 GiB per topic.
- *
- * @param enabled Whether to reserve segment storage on disk.
- * @return Encoded topic option entry.
- */
- static HeaderEntry PreallocateSegments(const bool enabled) {
- return detail::to_option_entry("preallocate_segments",
HeaderKind::Bool, detail::to_bool_bytes(enabled));
- }
-
- private:
- TopicOption() = delete;
-};
-
/**
* @brief Owning client connection to an Apache Iggy server.
*
diff --git a/foreign/cpp/src/client.cpp b/foreign/cpp/src/client.cpp
index c2596a8f5..1a8546282 100644
--- a/foreign/cpp/src/client.cpp
+++ b/foreign/cpp/src/client.cpp
@@ -21,6 +21,48 @@
namespace iggy {
+namespace {
+
+rust::Vec<ffi::HeaderEntry> ToFfiRawOptions(const std::map<std::string,
std::string> &raw_options) {
+ rust::Vec<ffi::HeaderEntry> ffi_options;
+ ffi_options.reserve(raw_options.size());
+ for (const auto &[key, value] : raw_options) {
+ ffi::HeaderEntry ffi_entry;
+ ffi_entry.key.kind = static_cast<std::uint8_t>(HeaderKind::String);
+ ffi_entry.key.value.reserve(key.size());
+ for (const char character : key) {
+
ffi_entry.key.value.push_back(static_cast<std::uint8_t>(character));
+ }
+ ffi_entry.value.kind = static_cast<std::uint8_t>(HeaderKind::String);
+ ffi_entry.value.value.reserve(value.size());
+ for (const char character : value) {
+
ffi_entry.value.value.push_back(static_cast<std::uint8_t>(character));
+ }
+ ffi_options.push_back(std::move(ffi_entry));
+ }
+ return ffi_options;
+}
+
+template <typename FfiOptions, typename Options>
+void SetMutableTopicOptions(FfiOptions &ffi_options, const Options &options) {
+ if (const auto value = options.CompressionAlgorithm()) {
+ ffi_options.has_compression_algorithm = true;
+ ffi_options.compression_algorithm = std::string(value->Value());
+ }
+ if (const auto value = options.MessageExpiry()) {
+ ffi_options.has_message_expiry = true;
+ ffi_options.message_expiry_kind = std::string(value->Kind());
+ ffi_options.message_expiry_value = value->Value();
+ }
+ if (const auto value = options.MaxTopicSize()) {
+ ffi_options.has_max_topic_size = true;
+ ffi_options.max_topic_size = std::string(value->Value());
+ }
+ ffi_options.raw_options = ToFfiRawOptions(options.RawEntries());
+}
+
+} // namespace
+
IggyBlockingClient::IggyBlockingClient(IggyBlockingClient &&other) noexcept
: client_(std::exchange(other.client_, nullptr)) {}
@@ -68,22 +110,7 @@ StreamDetails IggyBlockingClient::CreateStream(std::string
name) {
void IggyBlockingClient::UpdateStream(const Identifier &stream, std::string
name, const StreamUpdateOptions &options) {
RethrowAsIggyException([this, &stream, &name, &options] {
- rust::Vec<ffi::HeaderEntry> ffi_options;
- ffi_options.reserve(options.RawEntries().size());
- for (const auto &entry : options.RawEntries()) {
- ffi::HeaderEntry ffi_entry;
- ffi_entry.key.kind = static_cast<std::uint8_t>(HeaderKind::String);
- ffi_entry.key.value.reserve(entry.first.size());
- for (char character : entry.first) {
-
ffi_entry.key.value.push_back(static_cast<std::uint8_t>(character));
- }
- ffi_entry.value.kind =
static_cast<std::uint8_t>(HeaderKind::String);
- ffi_entry.value.value.reserve(entry.second.size());
- for (char character : entry.second) {
-
ffi_entry.value.value.push_back(static_cast<std::uint8_t>(character));
- }
- ffi_options.push_back(std::move(ffi_entry));
- }
+ auto ffi_options = ToFfiRawOptions(options.RawEntries());
Handle()->update_stream(stream.ToFfi(), name, std::move(ffi_options));
});
}
@@ -117,7 +144,8 @@ TopicDetails IggyBlockingClient::CreateTopic(const
Identifier &stream,
std::string name,
const TopicCreateOptions
&options) {
return RethrowAsIggyException([this, &stream, &name, &options] {
- ffi::TopicCreateOptions ffi_options;
+ ffi::TopicCreateOptions ffi_options{};
+ SetMutableTopicOptions(ffi_options, options);
if (auto value = options.PartitionsCount()) {
ffi_options.has_partitions_count = true;
ffi_options.partitions_count = *value;
@@ -125,29 +153,6 @@ TopicDetails IggyBlockingClient::CreateTopic(const
Identifier &stream,
ffi_options.has_partitions_count = false;
ffi_options.partitions_count = 0;
}
- if (auto value = options.CompressionAlgorithm()) {
- ffi_options.has_compression_algorithm = true;
- ffi_options.compression_algorithm =
std::string(value->Value());
- } else {
- ffi_options.has_compression_algorithm = false;
- ffi_options.compression_algorithm = "";
- }
- if (auto value = options.MessageExpiry()) {
- ffi_options.has_message_expiry = true;
- ffi_options.message_expiry_kind = std::string(value->Kind());
- ffi_options.message_expiry_value = value->Value();
- } else {
- ffi_options.has_message_expiry = false;
- ffi_options.message_expiry_kind = "";
- ffi_options.message_expiry_value = 0;
- }
- if (auto value = options.MaxTopicSize()) {
- ffi_options.has_max_topic_size = true;
- ffi_options.max_topic_size = std::string(value->Value());
- } else {
- ffi_options.has_max_topic_size = false;
- ffi_options.max_topic_size = "";
- }
if (auto value = options.SegmentSize()) {
ffi_options.has_segment_size = true;
ffi_options.segment_size = *value;
@@ -190,22 +195,6 @@ TopicDetails IggyBlockingClient::CreateTopic(const
Identifier &stream,
ffi_options.has_preallocate_segments = false;
ffi_options.preallocate_segments = false;
}
- ffi_options.raw_options.reserve(options.RawEntries().size());
- for (const auto &entry : options.RawEntries()) {
- ffi::HeaderEntry ffi_entry;
- ffi_entry.key.kind = static_cast<std::uint8_t>(HeaderKind::String);
- ffi_entry.key.value.reserve(entry.first.size());
- for (char character : entry.first) {
-
ffi_entry.key.value.push_back(static_cast<std::uint8_t>(character));
- }
- ffi_entry.value.kind =
static_cast<std::uint8_t>(HeaderKind::String);
- ffi_entry.value.value.reserve(entry.second.size());
- for (char character : entry.second) {
-
ffi_entry.value.value.push_back(static_cast<std::uint8_t>(character));
- }
- ffi_options.raw_options.push_back(std::move(ffi_entry));
- }
-
return TopicDetails::FromFfi(Handle()->create_topic(stream.ToFfi(),
name, std::move(ffi_options)));
});
}
@@ -215,45 +204,8 @@ void IggyBlockingClient::UpdateTopic(const Identifier
&stream,
std::string name,
const TopicUpdateOptions &options) {
RethrowAsIggyException([this, &stream, &topic, &name, &options] {
- ffi::TopicUpdateOptions ffi_options;
- if (auto value = options.CompressionAlgorithm()) {
- ffi_options.has_compression_algorithm = true;
- ffi_options.compression_algorithm =
std::string(value->Value());
- } else {
- ffi_options.has_compression_algorithm = false;
- ffi_options.compression_algorithm = "";
- }
- if (auto value = options.MessageExpiry()) {
- ffi_options.has_message_expiry = true;
- ffi_options.message_expiry_kind = std::string(value->Kind());
- ffi_options.message_expiry_value = value->Value();
- } else {
- ffi_options.has_message_expiry = false;
- ffi_options.message_expiry_kind = "";
- ffi_options.message_expiry_value = 0;
- }
- if (auto value = options.MaxTopicSize()) {
- ffi_options.has_max_topic_size = true;
- ffi_options.max_topic_size = std::string(value->Value());
- } else {
- ffi_options.has_max_topic_size = false;
- ffi_options.max_topic_size = "";
- }
- ffi_options.raw_options.reserve(options.RawEntries().size());
- for (const auto &entry : options.RawEntries()) {
- ffi::HeaderEntry ffi_entry;
- ffi_entry.key.kind = static_cast<std::uint8_t>(HeaderKind::String);
- ffi_entry.key.value.reserve(entry.first.size());
- for (char character : entry.first) {
-
ffi_entry.key.value.push_back(static_cast<std::uint8_t>(character));
- }
- ffi_entry.value.kind =
static_cast<std::uint8_t>(HeaderKind::String);
- ffi_entry.value.value.reserve(entry.second.size());
- for (char character : entry.second) {
-
ffi_entry.value.value.push_back(static_cast<std::uint8_t>(character));
- }
- ffi_options.raw_options.push_back(std::move(ffi_entry));
- }
+ ffi::TopicUpdateOptions ffi_options{};
+ SetMutableTopicOptions(ffi_options, options);
Handle()->update_topic(stream.ToFfi(), topic.ToFfi(), name,
std::move(ffi_options));
});
diff --git a/foreign/cpp/src/type_conversion.rs
b/foreign/cpp/src/type_conversion.rs
index e1eeae7b8..0d95a692f 100644
--- a/foreign/cpp/src/type_conversion.rs
+++ b/foreign/cpp/src/type_conversion.rs
@@ -15,7 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-use crate::ffi;
+use std::{collections::BTreeMap, str::FromStr};
+
use bytes::Bytes;
use iggy::prelude::{
CompressionAlgorithm as RustCompressionAlgorithm,
@@ -42,7 +43,8 @@ use iggy_common::{
TopicPermissions as RustTopicPermissions, TransportEndpoints as
RustTransportEndpoints,
UserInfo as RustUserInfo, UserInfoDetails as RustUserInfoDetails,
UserStatus as RustUserStatus,
};
-use std::collections::BTreeMap;
+
+use crate::ffi;
impl From<RustIdentifier> for ffi::Identifier {
fn from(identifier: RustIdentifier) -> Self {
@@ -517,6 +519,30 @@ pub(crate) fn ffi_options_to_raw(
Ok(raw)
}
+fn parse_message_expiry(kind: &str, value: u64) -> Result<RustIggyExpiry,
String> {
+ match kind {
+ "server_default" | "default" => Ok(RustIggyExpiry::ServerDefault),
+ "never_expire" => Ok(RustIggyExpiry::NeverExpire),
+ "duration" if value == 0 => {
+ Err("message expiry duration must be greater than
zero".to_string())
+ }
+ "duration" => Ok(RustIggyExpiry::ExpireDuration(
+ iggy::prelude::IggyDuration::from(value),
+ )),
+ _ => Err(format!("invalid message expiry kind '{kind}'")),
+ }
+}
+
+fn parse_compression_algorithm(value: &str) ->
Result<RustCompressionAlgorithm, String> {
+ RustCompressionAlgorithm::from_str(value)
+ .map_err(|error| format!("invalid compression algorithm '{value}':
{error}"))
+}
+
+fn parse_max_topic_size(value: &str) -> Result<RustMaxTopicSize, String> {
+ RustMaxTopicSize::from_str(value)
+ .map_err(|error| format!("invalid max topic size '{value}': {error}"))
+}
+
impl From<RustOptionSpec> for ffi::OptionSpec {
fn from(spec: RustOptionSpec) -> Self {
ffi::OptionSpec {
@@ -532,47 +558,21 @@ impl TryFrom<ffi::TopicCreateOptions> for
RustTopicCreateOptions {
type Error = String;
fn try_from(options: ffi::TopicCreateOptions) -> Result<Self, Self::Error>
{
- use std::str::FromStr;
let compression_algorithm = if options.has_compression_algorithm {
- Some(
-
RustCompressionAlgorithm::from_str(&options.compression_algorithm).map_err(
- |error| {
- format!(
- "invalid compression algorithm '{}': {error}",
- options.compression_algorithm
- )
- },
- )?,
- )
+ Some(parse_compression_algorithm(&options.compression_algorithm)?)
} else {
None
};
let message_expiry = if options.has_message_expiry {
- Some(match options.message_expiry_kind.as_str() {
- "server_default" | "default" => RustIggyExpiry::ServerDefault,
- "never_expire" => RustIggyExpiry::NeverExpire,
- "duration" =>
RustIggyExpiry::ExpireDuration(iggy::prelude::IggyDuration::from(
- options.message_expiry_value,
- )),
- _ => {
- return Err(format!(
- "invalid message expiry kind '{}'",
- options.message_expiry_kind
- ));
- }
- })
+ Some(parse_message_expiry(
+ &options.message_expiry_kind,
+ options.message_expiry_value,
+ )?)
} else {
None
};
let max_topic_size = if options.has_max_topic_size {
- Some(
-
RustMaxTopicSize::from_str(&options.max_topic_size).map_err(|error| {
- format!(
- "invalid max topic size '{}': {error}",
- options.max_topic_size
- )
- })?,
- )
+ Some(parse_max_topic_size(&options.max_topic_size)?)
} else {
None
};
@@ -592,7 +592,13 @@ impl TryFrom<ffi::TopicCreateOptions> for
RustTopicCreateOptions {
} else {
RustDurability::default()
};
- let raw = ffi_options_to_raw(options.raw_options)?;
+ let mut raw = ffi_options_to_raw(options.raw_options)?;
+ if options.has_durability {
+ raw.remove("durability");
+ }
+ if options.has_consumer_offset_durability {
+ raw.remove("consumer_offset_durability");
+ }
Ok(RustTopicCreateOptions {
partitions_count: if options.has_partitions_count {
Some(options.partitions_count)
@@ -635,47 +641,21 @@ impl TryFrom<ffi::TopicUpdateOptions> for
RustTopicUpdateOptions {
type Error = String;
fn try_from(options: ffi::TopicUpdateOptions) -> Result<Self, Self::Error>
{
- use std::str::FromStr;
let compression_algorithm = if options.has_compression_algorithm {
- Some(
-
RustCompressionAlgorithm::from_str(&options.compression_algorithm).map_err(
- |error| {
- format!(
- "invalid compression algorithm '{}': {error}",
- options.compression_algorithm
- )
- },
- )?,
- )
+ Some(parse_compression_algorithm(&options.compression_algorithm)?)
} else {
None
};
let message_expiry = if options.has_message_expiry {
- Some(match options.message_expiry_kind.as_str() {
- "server_default" | "default" => RustIggyExpiry::ServerDefault,
- "never_expire" => RustIggyExpiry::NeverExpire,
- "duration" =>
RustIggyExpiry::ExpireDuration(iggy::prelude::IggyDuration::from(
- options.message_expiry_value,
- )),
- _ => {
- return Err(format!(
- "invalid message expiry kind '{}'",
- options.message_expiry_kind
- ));
- }
- })
+ Some(parse_message_expiry(
+ &options.message_expiry_kind,
+ options.message_expiry_value,
+ )?)
} else {
None
};
let max_topic_size = if options.has_max_topic_size {
- Some(
-
RustMaxTopicSize::from_str(&options.max_topic_size).map_err(|error| {
- format!(
- "invalid max topic size '{}': {error}",
- options.max_topic_size
- )
- })?,
- )
+ Some(parse_max_topic_size(&options.max_topic_size)?)
} else {
None
};
@@ -1039,3 +1019,16 @@ impl From<RustSendMessagesResponse> for
ffi::SendMessagesResponse {
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn should_reject_zero_message_expiry_duration() {
+ assert_eq!(
+ parse_message_expiry("duration", 0).err().as_deref(),
+ Some("message expiry duration must be greater than zero")
+ );
+ }
+}
diff --git a/foreign/cpp/src/type_conversions.cpp
b/foreign/cpp/src/type_conversions.cpp
index c3b08fd08..31a1d33b9 100644
--- a/foreign/cpp/src/type_conversions.cpp
+++ b/foreign/cpp/src/type_conversions.cpp
@@ -32,27 +32,6 @@ LoginInfo LoginInfo::FromFfi(ffi::LoginInfo login_info) {
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) {
diff --git a/foreign/cpp/tests/e2e/partition.cpp
b/foreign/cpp/tests/e2e/partition.cpp
index 74efc5ce9..d72442710 100644
--- a/foreign/cpp/tests/e2e/partition.cpp
+++ b/foreign/cpp/tests/e2e/partition.cpp
@@ -60,16 +60,16 @@ TEST_F(E2E_Partition, CreatePartitionsBeforeLoginThrows) {
ASSERT_THROW(
client.CreatePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name), 1),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW(client.Connect());
ASSERT_THROW(
client.CreatePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name), 1),
- std::exception);
+ iggy::IggyException);
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);
+ iggy::IggyException);
}
TEST_F(E2E_Partition, CreatePartitionsOnNonExistentResourcesThrows) {
@@ -88,33 +88,50 @@ TEST_F(E2E_Partition,
CreatePartitionsOnNonExistentResourcesThrows) {
ASSERT_THROW(
client.CreatePartitions(iggy::Identifier::String(missing_stream_name),
iggy::Identifier::String(topic_name), 1),
- std::exception);
+ iggy::IggyException);
ASSERT_THROW(
client.CreatePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(missing_topic_name), 1),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Partition, CreatePartitionsWithInvalidIdentifiersThrows) {
- RecordProperty("description", "Rejects invalid stream or topic identifiers
before creating partitions.");
+ RecordProperty("description", "Rejects create_partitions requests that use
malformed stream or topic identifiers.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- auto client = GetLoggedInHighLevelClient();
+ iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client.CreateStream(stream_name));
+ ASSERT_NO_THROW(client->create_stream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
-
iggy::TopicCreateOptions().SetPartitionsCount(1)));
-
- ASSERT_THROW(client.CreatePartitions(iggy::Identifier::String(""),
iggy::Identifier::String(topic_name), 1),
+ ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name,
+ make_topic_create_options(1, "none",
"never_expire", 0, "server_default")));
+
+ iggy::ffi::Identifier invalid_stream_kind;
+ invalid_stream_kind.kind = "invalid";
+ invalid_stream_kind.length = 4;
+ invalid_stream_kind.value = {1, 0, 0, 0};
+ ASSERT_THROW(client->create_partitions(std::move(invalid_stream_kind),
make_string_identifier(topic_name), 1),
std::exception);
-
ASSERT_THROW(client.CreatePartitions(iggy::Identifier::String(std::string(256,
'a')),
- iggy::Identifier::String(topic_name),
1),
+
+ iggy::ffi::Identifier invalid_stream_length;
+ invalid_stream_length.kind = "numeric";
+ invalid_stream_length.length = 1;
+ invalid_stream_length.value.push_back(1);
+ ASSERT_THROW(client->create_partitions(std::move(invalid_stream_length),
make_string_identifier(topic_name), 1),
std::exception);
-
ASSERT_THROW(client.CreatePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(""), 1),
+
+ iggy::ffi::Identifier invalid_topic_kind;
+ invalid_topic_kind.kind = "invalid";
+ invalid_topic_kind.length = 4;
+ invalid_topic_kind.value = {1, 0, 0, 0};
+
ASSERT_THROW(client->create_partitions(make_string_identifier(stream_name),
std::move(invalid_topic_kind), 1),
std::exception);
- ASSERT_THROW(client.CreatePartitions(iggy::Identifier::String(stream_name),
-
iggy::Identifier::String(std::string(256, 'a')), 1),
+
+ iggy::ffi::Identifier invalid_topic_length;
+ invalid_topic_length.kind = "numeric";
+ invalid_topic_length.length = 1;
+ invalid_topic_length.value.push_back(1);
+
ASSERT_THROW(client->create_partitions(make_string_identifier(stream_name),
std::move(invalid_topic_length), 1),
std::exception);
}
@@ -157,7 +174,7 @@ TEST_F(E2E_Partition,
CreatePartitionsWithBoundaryPartitionsCountValues) {
ASSERT_THROW(
client.CreatePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(test_case.topic_name), test_case.partitions_count),
- std::exception);
+ iggy::IggyException);
}
}
@@ -267,7 +284,7 @@ TEST_F(E2E_Partition,
DeleteMorePartitionsThanExistingThrows) {
ASSERT_THROW(
client.DeletePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(test_case.topic_name), test_case.partitions_count),
- std::exception);
+ iggy::IggyException);
}
}
@@ -327,7 +344,7 @@ TEST_F(E2E_Partition,
DeletePartitionsFromTopicWithZeroPartitionsThrows) {
ASSERT_THROW(
client.DeletePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name), 1),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW({
const auto stream_details =
client.GetStream(iggy::Identifier::String(stream_name));
@@ -347,16 +364,16 @@ TEST_F(E2E_Partition, DeletePartitionsBeforeLoginThrows) {
ASSERT_THROW(
client.DeletePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name), 1),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW(client.Connect());
ASSERT_THROW(
client.DeletePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name), 1),
- std::exception);
+ iggy::IggyException);
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);
+ iggy::IggyException);
}
TEST_F(E2E_Partition, DeletePartitionsOnNonExistentResourcesThrows) {
@@ -375,33 +392,50 @@ TEST_F(E2E_Partition,
DeletePartitionsOnNonExistentResourcesThrows) {
ASSERT_THROW(
client.DeletePartitions(iggy::Identifier::String(missing_stream_name),
iggy::Identifier::String(topic_name), 1),
- std::exception);
+ iggy::IggyException);
ASSERT_THROW(
client.DeletePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(missing_topic_name), 1),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Partition, DeletePartitionsWithInvalidIdentifiersThrows) {
- RecordProperty("description", "Rejects invalid stream or topic identifiers
before deleting partitions.");
+ RecordProperty("description", "Rejects delete_partitions requests that use
malformed stream or topic identifiers.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- auto client = GetLoggedInHighLevelClient();
+ iggy::ffi::Client *client = GetLoggedInClient();
- ASSERT_NO_THROW(client.CreateStream(stream_name));
+ ASSERT_NO_THROW(client->create_stream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
-
iggy::TopicCreateOptions().SetPartitionsCount(3)));
-
- ASSERT_THROW(client.DeletePartitions(iggy::Identifier::String(""),
iggy::Identifier::String(topic_name), 1),
+ ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name,
+ make_topic_create_options(3, "none",
"never_expire", 0, "server_default")));
+
+ iggy::ffi::Identifier invalid_stream_kind;
+ invalid_stream_kind.kind = "invalid";
+ invalid_stream_kind.length = 4;
+ invalid_stream_kind.value = {1, 0, 0, 0};
+ ASSERT_THROW(client->delete_partitions(std::move(invalid_stream_kind),
make_string_identifier(topic_name), 1),
std::exception);
-
ASSERT_THROW(client.DeletePartitions(iggy::Identifier::String(std::string(256,
'a')),
- iggy::Identifier::String(topic_name),
1),
+
+ iggy::ffi::Identifier invalid_stream_length;
+ invalid_stream_length.kind = "numeric";
+ invalid_stream_length.length = 1;
+ invalid_stream_length.value.push_back(1);
+ ASSERT_THROW(client->delete_partitions(std::move(invalid_stream_length),
make_string_identifier(topic_name), 1),
std::exception);
-
ASSERT_THROW(client.DeletePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(""), 1),
+
+ iggy::ffi::Identifier invalid_topic_kind;
+ invalid_topic_kind.kind = "invalid";
+ invalid_topic_kind.length = 4;
+ invalid_topic_kind.value = {1, 0, 0, 0};
+
ASSERT_THROW(client->delete_partitions(make_string_identifier(stream_name),
std::move(invalid_topic_kind), 1),
std::exception);
- ASSERT_THROW(client.DeletePartitions(iggy::Identifier::String(stream_name),
-
iggy::Identifier::String(std::string(256, 'a')), 1),
+
+ iggy::ffi::Identifier invalid_topic_length;
+ invalid_topic_length.kind = "numeric";
+ invalid_topic_length.length = 1;
+ invalid_topic_length.value.push_back(1);
+
ASSERT_THROW(client->delete_partitions(make_string_identifier(stream_name),
std::move(invalid_topic_length), 1),
std::exception);
}
@@ -449,7 +483,7 @@ TEST_F(E2E_Partition,
DeletePartitionsAfterStreamDeletionThrows) {
ASSERT_THROW(client.DeletePartitions(iggy::Identifier::Numeric(stream_details.Id()),
iggy::Identifier::Numeric(stream_details.Topics()[0].Id()), 1),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Partition, CreatePartitionsAfterTopicDeletionThrows) {
@@ -467,5 +501,5 @@ TEST_F(E2E_Partition,
CreatePartitionsAfterTopicDeletionThrows) {
ASSERT_THROW(
client.CreatePartitions(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name), 1),
- std::exception);
+ iggy::IggyException);
}
diff --git a/foreign/cpp/tests/e2e/stream.cpp b/foreign/cpp/tests/e2e/stream.cpp
index bb49e88b3..d6c633b26 100644
--- a/foreign/cpp/tests/e2e/stream.cpp
+++ b/foreign/cpp/tests/e2e/stream.cpp
@@ -41,7 +41,7 @@ TEST_F(E2E_Stream, CreateDuplicateStreamThrows) {
auto client = GetLoggedInHighLevelClient();
ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_THROW(client.CreateStream(stream_name), std::exception);
+ ASSERT_THROW(client.CreateStream(stream_name), iggy::IggyException);
}
TEST_F(E2E_Stream, CreateStreamBeforeLoginThrows) {
@@ -49,12 +49,12 @@ TEST_F(E2E_Stream, CreateStreamBeforeLoginThrows) {
const std::string stream_name = GetRandomName();
auto client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(client.CreateStream(stream_name), std::exception);
+ ASSERT_THROW(client.CreateStream(stream_name), iggy::IggyException);
ASSERT_NO_THROW(client.Connect());
- ASSERT_THROW(client.CreateStream(stream_name), std::exception);
+ ASSERT_THROW(client.CreateStream(stream_name), iggy::IggyException);
ASSERT_NO_THROW(client.Login("iggy", "iggy"));
ASSERT_NO_THROW(client.Disconnect());
- ASSERT_THROW(client.CreateStream(stream_name), std::exception);
+ ASSERT_THROW(client.CreateStream(stream_name), iggy::IggyException);
}
TEST_F(E2E_Stream, CreateStreamValidatesNameConstraintsAndUniqueness) {
@@ -67,7 +67,7 @@ TEST_F(E2E_Stream,
CreateStreamValidatesNameConstraintsAndUniqueness) {
auto client = GetLoggedInHighLevelClient();
for (const auto &stream_name : illegal_stream_names) {
SCOPED_TRACE(stream_name);
- ASSERT_THROW(client.CreateStream(stream_name), std::exception);
+ ASSERT_THROW(client.CreateStream(stream_name), iggy::IggyException);
}
const std::string max_length_name(255, 'a');
@@ -104,7 +104,7 @@ TEST_F(E2E_Stream, UpdateStreamWorksCorrectly) {
ASSERT_NO_THROW(client.UpdateStream(iggy::Identifier::String(stream_name),
updated_stream_name));
- ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
iggy::StreamDetails updated_stream_details =
client.GetStream(iggy::Identifier::Numeric(stream_id));
@@ -148,11 +148,11 @@ TEST_F(E2E_Stream,
UpdateStreamWithUnsupportedOptionsRejectsAndPreservesName) {
const auto options =
iggy::StreamUpdateOptions().SetRawEntries({{"not_a_real_option", "true"}});
ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(stream_name),
updated_stream_name, options),
- std::exception);
+ iggy::IggyException);
const auto stream =
client.GetStream(iggy::Identifier::String(stream_name));
EXPECT_EQ(stream.Name(), stream_name);
-
ASSERT_THROW(client.GetStream(iggy::Identifier::String(updated_stream_name)),
std::exception);
+
ASSERT_THROW(client.GetStream(iggy::Identifier::String(updated_stream_name)),
iggy::IggyException);
}
TEST_F(E2E_Stream, UpdateStreamBeforeLoginThrows) {
@@ -166,14 +166,14 @@ TEST_F(E2E_Stream, UpdateStreamBeforeLoginThrows) {
auto unauthenticated_client = GetLoggedOutHighLevelClient();
ASSERT_THROW(unauthenticated_client.UpdateStream(iggy::Identifier::String(stream_name),
updated_stream_name),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW(unauthenticated_client.Connect());
ASSERT_THROW(unauthenticated_client.UpdateStream(iggy::Identifier::String(stream_name),
updated_stream_name),
- std::exception);
+ iggy::IggyException);
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);
+ iggy::IggyException);
}
TEST_F(E2E_Stream, UpdateStreamWithVariousUtf8Characters) {
@@ -210,7 +210,7 @@ TEST_F(E2E_Stream, UpdateNonExistentStreamThrows) {
const std::string stream_name = GetRandomName();
const std::string updated_stream_name = GetRandomName();
auto client = GetLoggedInHighLevelClient();
- ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(stream_name),
updated_stream_name), std::exception);
+ ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(stream_name),
updated_stream_name), iggy::IggyException);
}
TEST_F(E2E_Stream, UpdateStreamWithDuplicateNameThrows) {
@@ -223,7 +223,8 @@ TEST_F(E2E_Stream, UpdateStreamWithDuplicateNameThrows) {
ASSERT_NO_THROW(client.CreateStream(second_stream_name));
TrackStream(second_stream_name);
-
ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(first_stream_name),
second_stream_name), std::exception);
+
ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(first_stream_name),
second_stream_name),
+ iggy::IggyException);
}
TEST_F(E2E_Stream, UpdateDeletedStreamThrows) {
@@ -237,7 +238,7 @@ TEST_F(E2E_Stream, UpdateDeletedStreamThrows) {
ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(stream_name),
updated_stream_name), std::exception);
+ ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(stream_name),
updated_stream_name), iggy::IggyException);
}
TEST_F(E2E_Stream, UpdateStreamOnlyChangesName) {
@@ -281,7 +282,7 @@ TEST_F(E2E_Stream, UpdateStreamOnlyChangesName) {
ASSERT_NO_THROW(client.UpdateStream(iggy::Identifier::Numeric(stream_id),
updated_stream_name));
- ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
auto stream_after_update =
client.GetStream(iggy::Identifier::Numeric(stream_id));
iggy::ffi::Stats stats_after_update{};
ASSERT_NO_THROW({ stats_after_update = ffi_client->get_stats(); });
@@ -336,7 +337,8 @@ TEST_F(E2E_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.UpdateStream(iggy::Identifier::Numeric(stream_id),
invalid_stream_name), std::exception);
+ ASSERT_THROW(client.UpdateStream(iggy::Identifier::Numeric(stream_id),
invalid_stream_name),
+ iggy::IggyException);
}
}
@@ -355,7 +357,7 @@ TEST_F(E2E_Stream, DeleteNotCreatedStreamThrows) {
RecordProperty("description", "Throws when deleting a stream that does not
exist.");
const std::string stream_name = GetRandomName();
auto client = GetLoggedInHighLevelClient();
- ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
}
TEST_F(E2E_Stream, DeleteStreamBeforeLoginThrows) {
@@ -364,14 +366,14 @@ TEST_F(E2E_Stream, DeleteStreamBeforeLoginThrows) {
auto client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
ASSERT_NO_THROW(client.Connect());
- ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
ASSERT_NO_THROW(client.Login("iggy", "iggy"));
ASSERT_NO_THROW(client.Disconnect());
- ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
}
TEST_F(E2E_Stream, DeleteStreamTwiceThrows) {
@@ -383,7 +385,7 @@ TEST_F(E2E_Stream, DeleteStreamTwiceThrows) {
ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
}
TEST_F(E2E_Stream, GetStreamByStringIdentifierReturnsStreamDetails) {
@@ -407,7 +409,7 @@ TEST_F(E2E_Stream, GetNonExistentStreamDetailsThrows) {
RecordProperty("description", "Throws when requesting details for a stream
that does not exist.");
const std::string stream_name = GetRandomName();
auto client = GetLoggedInHighLevelClient();
- ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
}
TEST_F(E2E_Stream, GetStreamDetailsBeforeLoginThrows) {
@@ -415,12 +417,12 @@ TEST_F(E2E_Stream, GetStreamDetailsBeforeLoginThrows) {
const std::string stream_name = GetRandomName();
auto client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
ASSERT_NO_THROW(client.Connect());
- ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
ASSERT_NO_THROW(client.Login("iggy", "iggy"));
ASSERT_NO_THROW(client.Disconnect());
- ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
}
TEST_F(E2E_Stream, GetDeletedStreamDetailsThrows) {
@@ -432,7 +434,7 @@ TEST_F(E2E_Stream, GetDeletedStreamDetailsThrows) {
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.GetStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
}
TEST_F(E2E_Stream, GetStreamsReturnsEmptyAfterCleanup) {
@@ -514,12 +516,12 @@ TEST_F(E2E_Stream, GetStreamsBeforeLoginThrows) {
RecordProperty("description", "Throws when get_streams is called before
authentication.");
auto client = GetLoggedOutHighLevelClient();
- ASSERT_THROW(client.GetStreams(), std::exception);
+ ASSERT_THROW(client.GetStreams(), iggy::IggyException);
ASSERT_NO_THROW(client.Connect());
- ASSERT_THROW(client.GetStreams(), std::exception);
+ ASSERT_THROW(client.GetStreams(), iggy::IggyException);
ASSERT_NO_THROW(client.Login("iggy", "iggy"));
ASSERT_NO_THROW(client.Disconnect());
- ASSERT_THROW(client.GetStreams(), std::exception);
+ ASSERT_THROW(client.GetStreams(), iggy::IggyException);
}
TEST_F(E2E_Stream, GetStreamsConsistentWithGetStream) {
@@ -590,7 +592,7 @@ TEST_F(E2E_Stream, PurgeStreamOnNonExistentStreamThrows) {
RecordProperty("description", "Throws when purging a stream that does not
exist.");
const std::string stream_name = GetRandomName();
auto client = GetLoggedInHighLevelClient();
- ASSERT_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
}
TEST_F(E2E_Stream, PurgeStreamAfterStreamDeletionThrows) {
@@ -602,7 +604,7 @@ TEST_F(E2E_Stream, PurgeStreamAfterStreamDeletionThrows) {
ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)),
std::exception);
+ ASSERT_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
}
TEST_F(E2E_Stream, PurgeStreamPreservesStreamMetadata) {
@@ -950,10 +952,10 @@ TEST_F(E2E_Stream, PurgeStreamBeforeLoginThrows) {
auto unauthenticated_client = GetLoggedOutHighLevelClient();
-
ASSERT_THROW(unauthenticated_client.PurgeStream(iggy::Identifier::String(stream_name)),
std::exception);
+
ASSERT_THROW(unauthenticated_client.PurgeStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
ASSERT_NO_THROW(unauthenticated_client.Connect());
-
ASSERT_THROW(unauthenticated_client.PurgeStream(iggy::Identifier::String(stream_name)),
std::exception);
+
ASSERT_THROW(unauthenticated_client.PurgeStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
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);
+
ASSERT_THROW(unauthenticated_client.PurgeStream(iggy::Identifier::String(stream_name)),
iggy::IggyException);
}
diff --git a/foreign/cpp/tests/e2e/topic.cpp b/foreign/cpp/tests/e2e/topic.cpp
index 7e10a7c97..027c0cde8 100644
--- a/foreign/cpp/tests/e2e/topic.cpp
+++ b/foreign/cpp/tests/e2e/topic.cpp
@@ -123,7 +123,7 @@ TEST_F(E2E_Topic,
CreateTopicWithBoundaryPartitionsCountValues) {
iggy::TopicCreateOptions().SetPartitionsCount(1000)));
ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
overflow_topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(1001)),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
zero_partitions_topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(0)));
@@ -157,7 +157,7 @@ TEST_F(E2E_Topic, CreateTopicWithInvalidNamesThrows) {
SCOPED_TRACE(topic_name);
ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(1)),
- std::exception);
+ iggy::IggyException);
}
const std::string max_length_name(255, 'a');
@@ -177,7 +177,7 @@ TEST_F(E2E_Topic, CreateDuplicateTopicThrows) {
iggy::TopicCreateOptions().SetPartitionsCount(1)));
ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(1)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, CreateSameTopicNameInDifferentStreamsSucceeds) {
@@ -213,15 +213,15 @@ TEST_F(E2E_Topic, CreateTopicWithInvalidOptionsThrows) {
ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
invalid_compression_topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(1).SetRawEntries(
{{"compression_algorithm",
"invalid-compression"}})),
- std::exception);
+ iggy::IggyException);
ASSERT_THROW(client.CreateTopic(
iggy::Identifier::String(stream_name),
invalid_expiry_topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(1).SetRawEntries({{"message_expiry",
"invalid"}})),
- std::exception);
+ iggy::IggyException);
ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
invalid_max_size_topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(1).SetRawEntries(
{{"max_topic_size", "not-a-size"}})),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic,
CreateTopicWithOptionsReturnsCanonicalKindAndDerivedRemainder) {
@@ -237,7 +237,10 @@ TEST_F(E2E_Topic,
CreateTopicWithOptionsReturnsCanonicalKindAndDerivedRemainder)
TrackStream(stream_name);
iggy::TopicCreateOptions options;
- options.SetPartitionsCount(1).SetDurability(iggy::Durability::Persisted);
+ options.SetPartitionsCount(1)
+ .SetDurability(iggy::Durability::Replicated)
+ .SetConsumerOffsetDurability(iggy::Durability::Persisted)
+ .SetRawEntries({{"durability", "persisted"},
{"consumer_offset_durability", "replicated"}});
ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name, options));
const auto topic_details =
@@ -245,18 +248,18 @@ TEST_F(E2E_Topic,
CreateTopicWithOptionsReturnsCanonicalKindAndDerivedRemainder)
// 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.
- // The SDK sends both durability fields, including the independently
- // defaulted offset policy.
+ // Typed setters take precedence over conflicting raw entries.
const auto &explicit_options = topic_details.Options().Explicit();
ASSERT_EQ(explicit_options.size(), 2u);
auto explicit_it = explicit_options.find("durability");
ASSERT_NE(explicit_it, explicit_options.end());
EXPECT_EQ(explicit_it->second.Kind(), iggy::HeaderKind::String);
- EXPECT_EQ(explicit_it->second.Value(), (std::vector<std::uint8_t>{'p',
'e', 'r', 's', 'i', 's', 't', 'e', 'd'}));
+ EXPECT_EQ(explicit_it->second.Value(),
+ (std::vector<std::uint8_t>{'r', 'e', 'p', 'l', 'i', 'c', 'a',
't', 'e', 'd'}));
auto offset_it = explicit_options.find("consumer_offset_durability");
ASSERT_NE(offset_it, explicit_options.end());
EXPECT_EQ(offset_it->second.Kind(), iggy::HeaderKind::String);
- EXPECT_EQ(offset_it->second.Value(), (std::vector<std::uint8_t>{'r', 'e',
'p', 'l', 'i', 'c', 'a', 't', 'e', 'd'}));
+ EXPECT_EQ(offset_it->second.Value(), (std::vector<std::uint8_t>{'p', 'e',
'r', 's', 'i', 's', 't', 'e', 'd'}));
EXPECT_FALSE(topic_details.Options().Derived().empty());
const auto &derived_options = topic_details.Options().Derived();
@@ -267,7 +270,7 @@ TEST_F(E2E_Topic,
CreateTopicWithOptionsReturnsCanonicalKindAndDerivedRemainder)
iggy::TopicCreateOptions unknown_options;
unknown_options.SetPartitionsCount(1).SetRawEntries({{"not_a_real_option",
"true"}});
ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
unknown_option_topic, unknown_options),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic,
CreateTopicWithTypedOptionHelpersReportsThemAsExplicitOptions) {
@@ -318,6 +321,14 @@ TEST_F(E2E_Topic,
CreateTopicWithTypedOptionHelpersReportsThemAsExplicitOptions)
check_kind("size_of_messages_required_to_save", iggy::HeaderKind::Uint64);
check_kind("preallocate_segments", iggy::HeaderKind::Bool);
+ EXPECT_EQ(explicit_options.at("segment_size").Value(),
+ (std::vector<std::uint8_t>{0x00, 0x00, 0x80, 0x00, 0x00, 0x00,
0x00, 0x00}));
+ EXPECT_EQ(explicit_options.at("messages_required_to_save").Value(),
+ (std::vector<std::uint8_t>{0x00, 0x02, 0x00, 0x00}));
+ EXPECT_EQ(explicit_options.at("size_of_messages_required_to_save").Value(),
+ (std::vector<std::uint8_t>{0x00, 0x00, 0x20, 0x00, 0x00, 0x00,
0x00, 0x00}));
+ EXPECT_EQ(explicit_options.at("preallocate_segments").Value(),
(std::vector<std::uint8_t>{0x00}));
+
EXPECT_EQ(explicit_options.size(), 6u);
EXPECT_EQ(topic_details.Options().Derived().count("segment_size"), 0u)
<< "segment_size was set explicitly, so it cannot be derived";
@@ -370,7 +381,7 @@ TEST_F(E2E_Topic,
CreateTopicWithMaxTopicSizeBelowSegmentSizeThrows) {
TrackStream(stream_name);
iggy::TopicCreateOptions options;
options.SetPartitionsCount(1).SetMaxTopicSize(iggy::MaxTopicSize::FromBytes(1024));
- ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name, options), std::exception);
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name, options), iggy::IggyException);
}
TEST_F(E2E_Topic, CreateTopicOnNonExistentStreamThrows) {
@@ -381,7 +392,7 @@ TEST_F(E2E_Topic, CreateTopicOnNonExistentStreamThrows) {
auto client = GetLoggedInHighLevelClient();
ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(1)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, CreateTopicAfterStreamDeletionThrows) {
@@ -397,7 +408,7 @@ TEST_F(E2E_Topic, CreateTopicAfterStreamDeletionThrows) {
ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(1)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, CreateTopicBeforeLoginThrows) {
@@ -413,16 +424,16 @@ TEST_F(E2E_Topic, CreateTopicBeforeLoginThrows) {
ASSERT_THROW(unauthenticated_client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(1)),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW(unauthenticated_client.Connect());
ASSERT_THROW(unauthenticated_client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(1)),
- std::exception);
+ iggy::IggyException);
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,
iggy::TopicCreateOptions().SetPartitionsCount(1)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, DeleteTopicAfterCreate) {
@@ -453,7 +464,7 @@ TEST_F(E2E_Topic, DeleteTopicOnNonExistentStreamThrows) {
auto client = GetLoggedInHighLevelClient();
ASSERT_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, DeleteTopicOnNonExistentTopicThrows) {
@@ -467,7 +478,7 @@ TEST_F(E2E_Topic, DeleteTopicOnNonExistentTopicThrows) {
TrackStream(stream_name);
ASSERT_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, DeleteTopicTwiceThrows) {
@@ -484,7 +495,7 @@ TEST_F(E2E_Topic, DeleteTopicTwiceThrows) {
ASSERT_NO_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)));
ASSERT_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, DeleteTopicAfterStreamDeletionThrows) {
@@ -502,7 +513,7 @@ TEST_F(E2E_Topic, DeleteTopicAfterStreamDeletionThrows) {
ForgetTrackedStream(stream_name);
ASSERT_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, DeleteTopicBeforeLoginThrows) {
@@ -521,16 +532,16 @@ TEST_F(E2E_Topic, DeleteTopicBeforeLoginThrows) {
ASSERT_THROW(
unauthenticated_client.DeleteTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW(unauthenticated_client.Connect());
ASSERT_THROW(
unauthenticated_client.DeleteTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
ASSERT_NO_THROW(unauthenticated_client.Disconnect());
ASSERT_THROW(
unauthenticated_client.DeleteTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, GetTopicReturnsTopicForExistingTopic) {
@@ -577,16 +588,16 @@ TEST_F(E2E_Topic, GetTopicBeforeLoginThrows) {
ASSERT_THROW(
unauthenticated_client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW(unauthenticated_client.Connect());
ASSERT_THROW(
unauthenticated_client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
ASSERT_NO_THROW(unauthenticated_client.Disconnect());
ASSERT_THROW(
unauthenticated_client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, GetTopicWithWrongStreamIdThrows) {
@@ -604,7 +615,7 @@ TEST_F(E2E_Topic, GetTopicWithWrongStreamIdThrows) {
iggy::TopicCreateOptions().SetPartitionsCount(1)));
ASSERT_THROW(client.GetTopic(iggy::Identifier::String(second_stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, GetTopicWithWrongTopicThrows) {
@@ -620,7 +631,7 @@ TEST_F(E2E_Topic, GetTopicWithWrongTopicThrows) {
iggy::TopicCreateOptions().SetPartitionsCount(1)));
ASSERT_THROW(client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(wrong_topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, GetTopicAfterStreamDeletionThrows) {
@@ -637,7 +648,7 @@ TEST_F(E2E_Topic, GetTopicAfterStreamDeletionThrows) {
ForgetTrackedStream(stream_name);
ASSERT_THROW(client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, GetTopicAfterTopicDeletionThrows) {
@@ -653,7 +664,7 @@ TEST_F(E2E_Topic, GetTopicAfterTopicDeletionThrows) {
ASSERT_NO_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)));
ASSERT_THROW(client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, GetTopicReturnsEmptyPartitionsForZeroPartitionTopic) {
@@ -828,12 +839,12 @@ TEST_F(E2E_Topic, GetTopicsBeforeLoginThrows) {
auto unauthenticated_client = GetLoggedOutHighLevelClient();
-
ASSERT_THROW(unauthenticated_client.GetTopics(iggy::Identifier::String(stream_name)),
std::exception);
+
ASSERT_THROW(unauthenticated_client.GetTopics(iggy::Identifier::String(stream_name)),
iggy::IggyException);
ASSERT_NO_THROW(unauthenticated_client.Connect());
-
ASSERT_THROW(unauthenticated_client.GetTopics(iggy::Identifier::String(stream_name)),
std::exception);
+
ASSERT_THROW(unauthenticated_client.GetTopics(iggy::Identifier::String(stream_name)),
iggy::IggyException);
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);
+
ASSERT_THROW(unauthenticated_client.GetTopics(iggy::Identifier::String(stream_name)),
iggy::IggyException);
}
TEST_F(E2E_Topic, GetTopicsReturnsEmptyForStreamWithoutTopics) {
@@ -1048,6 +1059,11 @@ TEST_F(E2E_Topic,
UpdateTopicWithAllOptionCombinationsUpdatesInputFields) {
ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
iggy::TopicCreateOptions().SetPartitionsCount(2)));
+ const auto initial_topic =
+ client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name));
+ std::uint64_t expected_message_expiry = initial_topic.MessageExpiry();
+ std::uint64_t expected_max_topic_size = initial_topic.MaxTopicSize();
+
for (const auto &compression_algorithm : compression_algorithms) {
for (const auto &expiry_option : expiry_options) {
for (const auto &max_topic_size : max_topic_sizes) {
@@ -1062,6 +1078,22 @@ TEST_F(E2E_Topic,
UpdateTopicWithAllOptionCombinationsUpdatesInputFields) {
.SetMessageExpiry(expiry_option.value)
.SetMaxTopicSize(max_topic_size.value)));
topic_name = updated_topic_name;
+
+ if (expiry_option.value.Kind() != "server_default") {
+ expected_message_expiry = expiry_option.value.Value();
+ }
+ if (max_topic_size.value.Value() == "unlimited") {
+ expected_max_topic_size =
std::numeric_limits<std::uint64_t>::max();
+ } else if (max_topic_size.value.Value() != "server_default") {
+ expected_max_topic_size = 1024ULL * 1024ULL * 1024ULL;
+ }
+
+ const auto updated_topic =
+ client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name));
+ EXPECT_EQ(updated_topic.Name(), updated_topic_name);
+ EXPECT_EQ(updated_topic.CompressionAlgorithm(),
compression_algorithm.name);
+ EXPECT_EQ(updated_topic.MessageExpiry(),
expected_message_expiry);
+ EXPECT_EQ(updated_topic.MaxTopicSize(),
expected_max_topic_size);
}
}
}
@@ -1128,7 +1160,7 @@ TEST_F(E2E_Topic,
UpdateTopicWithDuplicateTopicNameThrows) {
.SetCompressionAlgorithm(iggy::CompressionAlgorithm::Gzip())
.SetMessageExpiry(iggy::Expiry::Duration(1000))
.SetMaxTopicSize(iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL))),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, UpdateTopicWithInvalidNamesThrows) {
@@ -1157,7 +1189,7 @@ TEST_F(E2E_Topic, UpdateTopicWithInvalidNamesThrows) {
.SetCompressionAlgorithm(iggy::CompressionAlgorithm::Gzip())
.SetMessageExpiry(iggy::Expiry::Duration(1000))
.SetMaxTopicSize(iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL))),
- std::exception);
+ iggy::IggyException);
}
}
@@ -1189,7 +1221,7 @@ TEST_F(E2E_Topic,
UpdateTopicFailedValidationDoesNotMutateTopic) {
.SetMessageExpiry(iggy::Expiry::Duration(2000))
.SetMaxTopicSize(iggy::MaxTopicSize::ServerDefault())
.SetRawEntries({{"max_topic_size",
"not-a-size"}})),
- std::exception);
+ iggy::IggyException);
const auto topic_after_failed_update =
client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name));
@@ -1202,7 +1234,7 @@ TEST_F(E2E_Topic,
UpdateTopicFailedValidationDoesNotMutateTopic) {
EXPECT_EQ(topic_after_failed_update.MaxTopicSize(),
topic_before_update.MaxTopicSize());
EXPECT_THROW(client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(updated_topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, UpdateTopicBeforeLoginThrows) {
@@ -1221,16 +1253,16 @@ TEST_F(E2E_Topic, UpdateTopicBeforeLoginThrows) {
ASSERT_THROW(unauthenticated_client.UpdateTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name), updated_topic_name),
- std::exception);
+ iggy::IggyException);
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);
+ iggy::IggyException);
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);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, UpdateTopicOnNonExistentStreamThrows) {
@@ -1246,7 +1278,7 @@ TEST_F(E2E_Topic, UpdateTopicOnNonExistentStreamThrows) {
.SetCompressionAlgorithm(iggy::CompressionAlgorithm::Gzip())
.SetMessageExpiry(iggy::Expiry::Duration(1000))
.SetMaxTopicSize(iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL))),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, UpdateTopicOnNonExistentTopicThrows) {
@@ -1265,7 +1297,7 @@ TEST_F(E2E_Topic, UpdateTopicOnNonExistentTopicThrows) {
.SetCompressionAlgorithm(iggy::CompressionAlgorithm::Gzip())
.SetMessageExpiry(iggy::Expiry::Duration(1000))
.SetMaxTopicSize(iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL))),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, GetTopicsAfterStreamDeletionReturnsEmpty) {
@@ -1294,7 +1326,7 @@ TEST_F(E2E_Topic, PurgeTopicOnNonExistentStreamThrows) {
auto client = GetLoggedInHighLevelClient();
ASSERT_THROW(client.PurgeTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, PurgeTopicAfterStreamDeletionThrows) {
@@ -1311,7 +1343,7 @@ TEST_F(E2E_Topic, PurgeTopicAfterStreamDeletionThrows) {
ForgetTrackedStream(stream_name);
ASSERT_THROW(client.PurgeTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, PurgeTopicOnNonExistentTopicThrows) {
@@ -1324,7 +1356,7 @@ TEST_F(E2E_Topic, PurgeTopicOnNonExistentTopicThrows) {
TrackStream(stream_name);
ASSERT_THROW(client.PurgeTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
TEST_F(E2E_Topic, PurgeTopicPreservesTopicMetadata) {
@@ -1606,14 +1638,14 @@ TEST_F(E2E_Topic, PurgeTopicBeforeLoginThrows) {
ASSERT_THROW(
unauthenticated_client.PurgeTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW(unauthenticated_client.Connect());
ASSERT_THROW(
unauthenticated_client.PurgeTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
ASSERT_NO_THROW(unauthenticated_client.Disconnect());
ASSERT_THROW(
unauthenticated_client.PurgeTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name)),
- std::exception);
+ iggy::IggyException);
}
diff --git a/foreign/cpp/tests/unit/unit_tests.cpp
b/foreign/cpp/tests/unit/unit_tests.cpp
index 5d86cf738..840247c86 100644
--- a/foreign/cpp/tests/unit/unit_tests.cpp
+++ b/foreign/cpp/tests/unit/unit_tests.cpp
@@ -28,18 +28,6 @@
#include "iggy.hpp"
-namespace {
-
-std::string option_key(const iggy::HeaderEntry &entry) {
- return std::string(entry.Key().Value().begin(), entry.Key().Value().end());
-}
-
-const std::vector<std::uint8_t> &option_value_bytes(const iggy::HeaderEntry
&entry) {
- return entry.Value().Value();
-}
-
-} // namespace
-
TEST(CompressionAlgorithmTest, ReturnsExpectedValues) {
EXPECT_EQ(iggy::CompressionAlgorithm::None().Value(), "none");
EXPECT_EQ(iggy::CompressionAlgorithm::Gzip().Value(), "gzip");
@@ -106,6 +94,7 @@ TEST(ExpiryTest, ReturnsExpectedKindAndValue) {
const auto duration = iggy::Expiry::Duration(15);
EXPECT_EQ(duration.Kind(), "duration");
EXPECT_EQ(duration.Value(), static_cast<std::uint64_t>(15));
+ EXPECT_THROW(iggy::Expiry::Duration(0), std::invalid_argument);
}
TEST(TopicCreateOptionsTest, DefaultHasNoValues) {
@@ -123,23 +112,6 @@ TEST(TopicCreateOptionsTest, DefaultHasNoValues) {
EXPECT_TRUE(options.RawEntries().empty());
}
-TEST(TopicOptionTest, DurabilityEncodesCanonicalStrings) {
- const auto persisted =
iggy::TopicOption::Durability(iggy::Durability::Persisted);
- EXPECT_EQ(persisted.Key().Kind(), iggy::HeaderKind::String);
- EXPECT_EQ(option_key(persisted), "durability");
- EXPECT_EQ(persisted.Value().Kind(), iggy::HeaderKind::String);
- EXPECT_EQ(option_value_bytes(persisted), (std::vector<std::uint8_t>{'p',
'e', 'r', 's', 'i', 's', 't', 'e', 'd'}));
-
- const auto offset = iggy::TopicOption::ConsumerOffsetDurability();
- EXPECT_EQ(offset.Key().Kind(), iggy::HeaderKind::String);
- EXPECT_EQ(option_key(offset), "consumer_offset_durability");
- EXPECT_EQ(offset.Value().Kind(), iggy::HeaderKind::String);
- EXPECT_EQ(option_value_bytes(offset),
- (std::vector<std::uint8_t>{'r', 'e', 'p', 'l', 'i', 'c', 'a',
't', 'e', 'd'}));
-
-
EXPECT_THROW(iggy::TopicOption::Durability(static_cast<iggy::Durability>(99)),
std::invalid_argument);
-}
-
TEST(TopicCreateOptionsTest, PartitionsCountStoresValue) {
iggy::TopicCreateOptions options;
options.SetPartitionsCount(3);
@@ -366,6 +338,36 @@ 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;
+
+ const auto stream = iggy::Identifier::String("stream");
+ const auto topic = iggy::Identifier::String("topic");
+
+ 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);
+ EXPECT_THROW(client.CreateStream("stream"), iggy::IggyException);
+ EXPECT_THROW(client.UpdateStream(stream, "updated-stream"),
iggy::IggyException);
+ EXPECT_THROW(client.GetStreams(), iggy::IggyException);
+ EXPECT_THROW(client.GetStream(stream), iggy::IggyException);
+ EXPECT_THROW(client.DeleteStream(stream), iggy::IggyException);
+ EXPECT_THROW(client.PurgeStream(stream), iggy::IggyException);
+ EXPECT_THROW(client.CreateTopic(stream, "topic",
iggy::TopicCreateOptions().SetPartitionsCount(1)),
+ iggy::IggyException);
+ EXPECT_THROW(client.UpdateTopic(stream, topic, "updated-topic"),
iggy::IggyException);
+ EXPECT_THROW(client.GetTopics(stream), iggy::IggyException);
+ EXPECT_THROW(client.GetTopic(stream, topic), iggy::IggyException);
+ EXPECT_THROW(client.DeleteTopic(stream, topic), iggy::IggyException);
+ EXPECT_THROW(client.PurgeTopic(stream, topic), iggy::IggyException);
+ EXPECT_THROW(client.CreatePartitions(stream, topic, 1),
iggy::IggyException);
+ EXPECT_THROW(client.DeletePartitions(stream, topic, 1),
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);
