hubcio commented on code in PR #4022:
URL: https://github.com/apache/iggy/pull/4022#discussion_r3988744906
##########
foreign/cpp/include/iggy.hpp:
##########
@@ -98,186 +876,613 @@ class SnapshotCompression final : private
detail::StringTag<SnapshotCompression>
/** @brief Returns the uncompressed storage option. */
static SnapshotCompression Stored() { return
SnapshotCompression("stored"); }
- /** @brief Returns the Deflate compression option. */
- static SnapshotCompression Deflated() { return
SnapshotCompression("deflated"); }
+ /** @brief Returns the Deflate compression option. */
+ static SnapshotCompression Deflated() { return
SnapshotCompression("deflated"); }
+
+ /** @brief Uses bzip2 for better compression with slower processing. */
+ static SnapshotCompression Bzip2() { return SnapshotCompression("bzip2"); }
+
+ /** @brief Uses Zstandard for fast compression and decompression. */
+ static SnapshotCompression Zstd() { return SnapshotCompression("zstd"); }
+
+ /** @brief Uses LZMA for high compression, especially for larger files. */
+ static SnapshotCompression Lzma() { return SnapshotCompression("lzma"); }
+
+ /** @brief Uses XZ for LZMA-like compression with faster decompression. */
+ static SnapshotCompression Xz() { return SnapshotCompression("xz"); }
+
+ /**
+ * @brief Returns the snapshot compression algorithm name.
+ * @return Snapshot compression algorithm name.
+ */
+ [[nodiscard]] std::string_view Value() const { return
detail::StringTag<SnapshotCompression>::Value(); }
+
+ private:
+ explicit SnapshotCompression(std::string snapshot_compression)
+ :
detail::StringTag<SnapshotCompression>(std::move(snapshot_compression)) {}
+};
+
+/**
+ * @brief Selects data to include in a system snapshot.
+ */
+class SystemSnapshotType final : private detail::StringTag<SystemSnapshotType>
{
+ public:
+ /** @brief Includes an overview of the file-system structure. */
+ static SystemSnapshotType FilesystemOverview() { return
SystemSnapshotType("filesystem_overview"); }
+
+ /** @brief Includes currently running processes. */
+ static SystemSnapshotType ProcessList() { return
SystemSnapshotType("process_list"); }
+
+ /** @brief Includes CPU, memory, and other resource usage statistics. */
+ static SystemSnapshotType ResourceUsage() { return
SystemSnapshotType("resource_usage"); }
+
+ /** @brief Includes the test snapshot used for development and testing. */
+ static SystemSnapshotType Test() { return SystemSnapshotType("test"); }
+
+ /** @brief Includes server logs from the configured logging directory. */
+ static SystemSnapshotType ServerLogs() { return
SystemSnapshotType("server_logs"); }
+
+ /** @brief Includes server configuration. */
+ static SystemSnapshotType ServerConfig() { return
SystemSnapshotType("server_config"); }
+
+ /** @brief Includes all available snapshot data. */
+ static SystemSnapshotType All() { return SystemSnapshotType("all"); }
+
+ /**
+ * @brief Returns the value passed to the client implementation.
+ * @return System snapshot type name.
+ */
+ [[nodiscard]] std::string_view SnapshotTypeValue() const { return Value();
}
+
+ private:
+ explicit SystemSnapshotType(std::string snapshot_type)
+ : detail::StringTag<SystemSnapshotType>(std::move(snapshot_type)) {}
+};
+
+/**
+ * @brief Maximum retained size of a topic.
+ *
+ * A topic may use the server default, have no size limit, or use an explicit
+ * byte limit.
+ *
+ * Use ServerDefault(), Unlimited(), or FromBytes() to select the retention
+ * limit.
+ */
+class MaxTopicSize final : private detail::StringTag<MaxTopicSize> {
+ public:
+ /** @brief Returns the server-default size option. */
+ static MaxTopicSize ServerDefault() { return
MaxTopicSize("server_default"); }
+
+ /** @brief Returns the unlimited size option. */
+ static MaxTopicSize Unlimited() { return MaxTopicSize("unlimited"); }
+
+ /**
+ * @brief Creates an explicit topic size limit.
+ * @param bytes Maximum topic size in bytes.
+ * @return Server-default size for zero, unlimited size for
+ * std::numeric_limits<std::uint64_t>::max(), or the requested
limit.
+ * @note The configured limit cannot be smaller than the server segment
size.
+ */
+ static MaxTopicSize FromBytes(std::uint64_t bytes) {
+ if (bytes == 0) {
+ return ServerDefault();
+ }
+ if (bytes == std::numeric_limits<std::uint64_t>::max()) {
+ return Unlimited();
+ }
+ return MaxTopicSize(std::to_string(bytes));
+ }
+
+ /**
+ * @brief Returns the value passed to the client implementation.
+ * @return Topic size option or decimal byte count.
+ */
+ [[nodiscard]] std::string_view Value() const { return
detail::StringTag<MaxTopicSize>::Value(); }
+
+ private:
+ explicit MaxTopicSize(std::string max_topic_size) :
detail::StringTag<MaxTopicSize>(std::move(max_topic_size)) {}
+};
+
+/**
+ * @brief Message retention policy for a topic.
+ *
+ * Use ServerDefault(), NeverExpire(), or Duration() to select the retention
+ * policy.
+ */
+class Expiry final {
+ public:
+ /** @brief Returns the server-default expiry policy. */
+ static Expiry ServerDefault() { return Expiry("server_default", 0); }
+
+ /**
+ * @brief Keeps messages until another operation removes them, such as
+ * topic deletion.
+ */
+ static Expiry NeverExpire() { return Expiry("never_expire",
std::numeric_limits<std::uint64_t>::max()); }
+
+ /**
+ * @brief Creates a time-based expiry policy.
+ * @param micros Message lifetime in microseconds.
+ * @return Time-based expiry policy.
+ */
+ static Expiry Duration(std::uint64_t micros) { return Expiry("duration",
micros); }
Review Comment:
warning: `Expiry::Duration(0)` is sent as 0, which means server default, so
a new topic never expires and an update keeps the old expiry. throw for `micros
== 0` here and reject `duration` with value 0 in `type_conversion.rs`.
##########
foreign/cpp/tests/unit/unit_tests.cpp:
##########
@@ -187,18 +366,6 @@ TEST(IggyExceptionTest, StoresMessage) {
EXPECT_EQ(std::string(from_string.what()), message);
}
Review Comment:
warning: `fefe92a77` deleted `MovedFromOperationsThrow` without a
replacement, so nothing tests the `Handle()` guard that every new method relies
on. restore it and cover `CreateStream`, `GetStream`, `CreateTopic` and
`CreatePartitions` too.
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -525,6 +528,167 @@ impl From<RustOptionSpec> for ffi::OptionSpec {
}
}
+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
+ )
+ },
+ )?,
+ )
+ } 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
+ ));
+ }
+ })
+ } 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
+ )
+ })?,
+ )
+ } else {
+ None
+ };
+ let durability = if options.has_durability {
Review Comment:
warning: typed durability does not win over raw entries, although the docs
say it does. `SetDurability(Replicated)` plus a raw `durability=persisted`
creates a persisted topic, so erase the matching key from `raw` when
`has_durability` or `has_consumer_offset_durability` is set.
##########
foreign/cpp/src/type_conversion.rs:
##########
@@ -525,6 +528,167 @@ impl From<RustOptionSpec> for ffi::OptionSpec {
}
}
+impl TryFrom<ffi::TopicCreateOptions> for RustTopicCreateOptions {
+ type Error = String;
+
+ fn try_from(options: ffi::TopicCreateOptions) -> Result<Self, Self::Error>
{
+ use std::str::FromStr;
Review Comment:
simplification: the compression/expiry/max-size parsing below is repeated in
the update impl (lines 639-681), along with this inline `use
std::str::FromStr;`. share one helper per field and move the import to the top
of the file.
##########
foreign/cpp/src/type_conversions.cpp:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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) {
Review Comment:
simplification: `Identifier::FromFfi` has no caller. delete it and its
declaration in `iggy.hpp`.
##########
foreign/cpp/include/iggy.hpp:
##########
@@ -366,54 +1571,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
-enum class Durability { Replicated, Persisted };
-
-constexpr std::string_view to_string(const Durability durability) {
- switch (durability) {
- case Durability::Replicated:
- return "replicated";
- case Durability::Persisted:
- return "persisted";
- }
- throw std::invalid_argument("Unknown durability");
-}
-
/**
- * @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
Review Comment:
warning: `TopicOption` is dead public API - its factories return
`iggy::HeaderEntry`, which no client method accepts, and this example does not
compile because `ResourceOptions::Explicit()` takes no arguments. delete lines
1542-1680 together with `TopicOptionTest` and its helpers in `unit_tests.cpp`.
##########
foreign/cpp/src/client.cpp:
##########
@@ -55,13 +55,251 @@ 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)); });
+ [this, &username, &password] { return
LoginInfo::FromFfi(Handle()->login_user(username, 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(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));
+ }
+ Handle()->update_stream(stream.ToFfi(), name, std::move(ffi_options));
+ });
+}
+
+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) {
+ RethrowAsIggyException([this, &stream] {
Handle()->delete_stream(stream.ToFfi()); });
+}
+
+void IggyBlockingClient::PurgeStream(const Identifier &stream) {
+ RethrowAsIggyException([this, &stream] {
Handle()->purge_stream(stream.ToFfi()); });
+}
+
+TopicDetails IggyBlockingClient::CreateTopic(const Identifier &stream,
+ std::string name,
+ const TopicCreateOptions
&options) {
+ return RethrowAsIggyException([this, &stream, &name, &options] {
+ ffi::TopicCreateOptions ffi_options;
Review Comment:
simplification: the raw-entry loop is copied three times (lines 71, 193,
242) and the compression/expiry/size block twice (lines 128, 219). move them
into helpers and drop the else branches, since cxx already initializes these
fields to zero or empty.
##########
foreign/cpp/include/iggy.hpp:
##########
@@ -98,186 +876,613 @@ class SnapshotCompression final : private
detail::StringTag<SnapshotCompression>
/** @brief Returns the uncompressed storage option. */
static SnapshotCompression Stored() { return
SnapshotCompression("stored"); }
- /** @brief Returns the Deflate compression option. */
- static SnapshotCompression Deflated() { return
SnapshotCompression("deflated"); }
+ /** @brief Returns the Deflate compression option. */
+ static SnapshotCompression Deflated() { return
SnapshotCompression("deflated"); }
+
+ /** @brief Uses bzip2 for better compression with slower processing. */
+ static SnapshotCompression Bzip2() { return SnapshotCompression("bzip2"); }
+
+ /** @brief Uses Zstandard for fast compression and decompression. */
+ static SnapshotCompression Zstd() { return SnapshotCompression("zstd"); }
+
+ /** @brief Uses LZMA for high compression, especially for larger files. */
+ static SnapshotCompression Lzma() { return SnapshotCompression("lzma"); }
+
+ /** @brief Uses XZ for LZMA-like compression with faster decompression. */
+ static SnapshotCompression Xz() { return SnapshotCompression("xz"); }
+
+ /**
+ * @brief Returns the snapshot compression algorithm name.
+ * @return Snapshot compression algorithm name.
+ */
+ [[nodiscard]] std::string_view Value() const { return
detail::StringTag<SnapshotCompression>::Value(); }
+
+ private:
+ explicit SnapshotCompression(std::string snapshot_compression)
+ :
detail::StringTag<SnapshotCompression>(std::move(snapshot_compression)) {}
+};
+
+/**
+ * @brief Selects data to include in a system snapshot.
+ */
+class SystemSnapshotType final : private detail::StringTag<SystemSnapshotType>
{
+ public:
+ /** @brief Includes an overview of the file-system structure. */
+ static SystemSnapshotType FilesystemOverview() { return
SystemSnapshotType("filesystem_overview"); }
+
+ /** @brief Includes currently running processes. */
+ static SystemSnapshotType ProcessList() { return
SystemSnapshotType("process_list"); }
+
+ /** @brief Includes CPU, memory, and other resource usage statistics. */
+ static SystemSnapshotType ResourceUsage() { return
SystemSnapshotType("resource_usage"); }
+
+ /** @brief Includes the test snapshot used for development and testing. */
+ static SystemSnapshotType Test() { return SystemSnapshotType("test"); }
+
+ /** @brief Includes server logs from the configured logging directory. */
+ static SystemSnapshotType ServerLogs() { return
SystemSnapshotType("server_logs"); }
+
+ /** @brief Includes server configuration. */
+ static SystemSnapshotType ServerConfig() { return
SystemSnapshotType("server_config"); }
+
+ /** @brief Includes all available snapshot data. */
+ static SystemSnapshotType All() { return SystemSnapshotType("all"); }
+
+ /**
+ * @brief Returns the value passed to the client implementation.
+ * @return System snapshot type name.
+ */
+ [[nodiscard]] std::string_view SnapshotTypeValue() const { return Value();
}
+
+ private:
+ explicit SystemSnapshotType(std::string snapshot_type)
+ : detail::StringTag<SystemSnapshotType>(std::move(snapshot_type)) {}
+};
+
+/**
+ * @brief Maximum retained size of a topic.
+ *
+ * A topic may use the server default, have no size limit, or use an explicit
+ * byte limit.
+ *
+ * Use ServerDefault(), Unlimited(), or FromBytes() to select the retention
+ * limit.
+ */
+class MaxTopicSize final : private detail::StringTag<MaxTopicSize> {
+ public:
+ /** @brief Returns the server-default size option. */
+ static MaxTopicSize ServerDefault() { return
MaxTopicSize("server_default"); }
+
+ /** @brief Returns the unlimited size option. */
+ static MaxTopicSize Unlimited() { return MaxTopicSize("unlimited"); }
+
+ /**
+ * @brief Creates an explicit topic size limit.
+ * @param bytes Maximum topic size in bytes.
+ * @return Server-default size for zero, unlimited size for
+ * std::numeric_limits<std::uint64_t>::max(), or the requested
limit.
+ * @note The configured limit cannot be smaller than the server segment
size.
+ */
+ static MaxTopicSize FromBytes(std::uint64_t bytes) {
+ if (bytes == 0) {
+ return ServerDefault();
+ }
+ if (bytes == std::numeric_limits<std::uint64_t>::max()) {
+ return Unlimited();
+ }
+ return MaxTopicSize(std::to_string(bytes));
+ }
+
+ /**
+ * @brief Returns the value passed to the client implementation.
+ * @return Topic size option or decimal byte count.
+ */
+ [[nodiscard]] std::string_view Value() const { return
detail::StringTag<MaxTopicSize>::Value(); }
+
+ private:
+ explicit MaxTopicSize(std::string max_topic_size) :
detail::StringTag<MaxTopicSize>(std::move(max_topic_size)) {}
+};
+
+/**
+ * @brief Message retention policy for a topic.
+ *
+ * Use ServerDefault(), NeverExpire(), or Duration() to select the retention
+ * policy.
+ */
+class Expiry final {
+ public:
+ /** @brief Returns the server-default expiry policy. */
+ static Expiry ServerDefault() { return Expiry("server_default", 0); }
+
+ /**
+ * @brief Keeps messages until another operation removes them, such as
+ * topic deletion.
+ */
+ static Expiry NeverExpire() { return Expiry("never_expire",
std::numeric_limits<std::uint64_t>::max()); }
+
+ /**
+ * @brief Creates a time-based expiry policy.
+ * @param micros Message lifetime in microseconds.
+ * @return Time-based expiry policy.
+ */
+ static Expiry Duration(std::uint64_t micros) { return Expiry("duration",
micros); }
+
+ /**
+ * @brief Returns the expiry policy kind.
+ * @return One of server_default, never_expire, or duration.
+ */
+ [[nodiscard]] std::string_view Kind() const { return expiry_kind_; }
+
+ /**
+ * @brief Returns the value associated with the expiry policy.
+ * @return Duration in microseconds for Duration(), zero for
ServerDefault(),
+ * or std::numeric_limits<std::uint64_t>::max() for NeverExpire().
+ */
+ [[nodiscard]] std::uint64_t Value() const { return expiry_value_; }
+
+ private:
+ explicit Expiry(std::string expiry_kind, std::uint64_t expiry_value)
+ : expiry_kind_(std::move(expiry_kind)), expiry_value_(expiry_value) {}
+
+ std::string expiry_kind_;
+ std::uint64_t expiry_value_;
+};
+
+enum class Durability { Replicated, Persisted };
+
+constexpr std::string_view to_string(const Durability durability) {
+ switch (durability) {
+ case Durability::Replicated:
+ return "replicated";
+ case Durability::Persisted:
+ return "persisted";
+ }
+ throw std::invalid_argument("Unknown durability");
+}
+
+/**
+ * @brief Options for creating a topic.
+ *
+ * Use the typed setters to configure supported topic settings. Leave a setting
+ * unset to use the server default. Use SetRawEntries() for supported options
+ * that do not yet have a typed setter. When both specify the same option, the
+ * typed setting takes precedence.
+ */
+class TopicCreateOptions final {
+ public:
+ TopicCreateOptions() = default;
+
+ /**
+ * @brief Returns the number of partitions to create.
+ * @return Configured partition count, or `std::nullopt` to use the server
Review Comment:
nit: an unset partition count is not a server default - the client fills in
`DEFAULT_PARTITIONS_COUNT`, which is 1. say "defaults to 1".
##########
foreign/cpp/MODULE.bazel.lock:
##########
@@ -483,109 +483,6 @@
}
}
}
Review Comment:
nit: this drops the rules_rust `cu_nr` entry that master still has, with no
`MODULE.bazel` or bazel version change - intentional? it came in with
`fefe92a77`, which also deleted `MovedFromOperationsThrow`, so check what else
that rebase dropped.
##########
foreign/cpp/tests/e2e/topic.cpp:
##########
@@ -158,171 +155,175 @@ 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", {}),
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(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,
+
iggy::TopicCreateOptions().SetPartitionsCount(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", {}),
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(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,
+
iggy::TopicCreateOptions().SetPartitionsCount(1)));
+
ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(second_stream_name),
topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(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();
const std::string invalid_expiry_topic_name = GetRandomName();
const std::string invalid_max_size_topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name),
invalid_compression_topic_name, 1,
- "invalid-compression", "server_default",
0, "server_default", {}),
+ ASSERT_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
invalid_compression_topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1).SetRawEntries(
+ {{"compression_algorithm",
"invalid-compression"}})),
std::exception);
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name),
invalid_expiry_topic_name, 1, "none",
- "invalid-expiry-kind", 0,
"server_default", {}),
+ ASSERT_THROW(client.CreateTopic(
+ iggy::Identifier::String(stream_name),
invalid_expiry_topic_name,
+
iggy::TopicCreateOptions().SetPartitionsCount(1).SetRawEntries({{"message_expiry",
"invalid"}})),
std::exception);
- ASSERT_THROW(client->create_topic(make_string_identifier(stream_name),
invalid_max_size_topic_name, 1, "none",
- "server_default", 0, "not-a-size", {}),
+ 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);
}
-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.");
const std::string stream_name = GetRandomName();
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("durability")),
-
make_header_field(iggy::ffi::HeaderKind::String, to_payload("persisted"))));
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name, 1, "none", "server_default",
- 0, "server_default",
std::move(options)));
+ iggy::TopicCreateOptions options;
+ options.SetPartitionsCount(1).SetDurability(iggy::Durability::Persisted);
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name),
topic_name, 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.
- EXPECT_TRUE(has_header(topic_details.options,
static_cast<std::uint8_t>(iggy::ffi::HeaderKind::String),
- to_payload("durability"),
static_cast<std::uint8_t>(iggy::ffi::HeaderKind::String),
- to_payload("persisted")));
-
- EXPECT_FALSE(topic_details.derived_options.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()));
- }
- EXPECT_EQ(derived_option_keys.count("max_topic_size"), 1u);
- EXPECT_EQ(derived_option_keys.count("durability"), 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)),
+ // The SDK sends both durability fields, including the independently
+ // defaulted offset policy.
+ 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'}));
+ 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_FALSE(topic_details.Options().Derived().empty());
+ const auto &derived_options = topic_details.Options().Derived();
+ EXPECT_EQ(derived_options.count("max_topic_size"), 1u);
+ EXPECT_EQ(derived_options.count("durability"), 0u);
+ EXPECT_EQ(derived_options.count("consumer_offset_durability"), 0u);
+
+ 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);
}
-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::Durability(iggy::Durability::Persisted));
-
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));
+ iggy::TopicCreateOptions options;
+ options.SetPartitionsCount(1)
+ .SetSegmentSize(segment_size_bytes)
+ .SetDurability(iggy::Durability::Persisted)
+ .SetMessagesRequiredToSave(messages_required_to_save)
+ .SetSizeOfMessagesRequiredToSave(size_of_messages_required_to_save)
+ .SetPreallocateSegments(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, options));
const auto topic_details =
- client->get_topic(make_string_identifier(stream_name),
make_string_identifier(topic_name));
-
- 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("durability"),
key_kind, to_payload("persisted")));
- 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());
- EXPECT_NE(derived_key, "segment_size") << "segment_size was set
explicitly, so it cannot be derived";
- }
+ client.GetTopic(iggy::Identifier::String(stream_name),
iggy::Identifier::String(topic_name));
+
+ const auto &explicit_options = topic_details.Options().Explicit();
+ EXPECT_EQ(explicit_options.count("segment_size"), 1u);
Review Comment:
warning: the low-level version of this test checked every option value, but
this one only checks keys and kinds. restore the value checks for
`segment_size`, `messages_required_to_save`,
`size_of_messages_required_to_save` and `preallocate_segments`, because they
now go through the new conversion code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
