You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "slbotbm (via GitHub)" <gi...@apache.org> on 2026/09/01 10:17:25 UTC
[PR] feat(cpp): add stream, topic, and partition functions to high level client [iggy]
slbotbm opened a new pull request, #4022:
URL: https://github.com/apache/iggy/pull/4022
## Which issue does this PR address?
<!--
We generally require a GitHub issue for all bug fixes and enhancements. Link it with GitHub syntax, keep the line that applies and delete the other:
- `Closes #123` auto-closes the issue when this PR merges (full fix).
- `Relates to #123` links without closing (partial or related work).
-->
Closes #
Relates to #2100
## Rationale
<!--
Why is this change needed? If the issue explains it well, a one-liner is fine.
-->
Adds function to the high level cpp client.
## What changed?
Adds the following functions to the high-level client:
- CreateStream
- GetStream
- GetStreams
- PurgeStream
- DeleteStream
- UpdateStream
- CreateTopic
- GetTopic
- GetTopics
- PurgeTopic
- DeleteTopic
- UpdateTopic
- DeletePartition
- CreatePartition
and their associated structs.
This PR also changes the tests testing these functions to utilize the high level client instead of the low level client.
Note to the reviewers: It would be better to ignore the `tests/` changes since those are just mechanical translations.
## Local Execution
- Passed
- Pre-commit hooks ran
## AI Usage
If AI tools were used, please answer:
1. Which tools? codex
2. Scope of usage? code generation
3. How did you verify the generated code works correctly? read through it. tested it.
4. Can you explain every line of the code if asked? yes.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add stream, topic, and partition functions to high level client [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add stream, topic, and partition functions to high level client [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #4022:
URL: https://github.com/apache/iggy/pull/4022#discussion_r3991304146
##########
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:
done
##########
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:
done
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add stream, topic, and partition functions to high level client [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on PR #4022:
URL: https://github.com/apache/iggy/pull/4022#issuecomment-5604053853
A note to the reviewers: since this is a pretty big PR, I would suggest ignoring that changes in `foreign/cpp/tests/*` since those are mostly mechanical translations of the tests from the low level API to the high level API, and it is still not yet complete.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add stream, topic, and partition functions to high level client [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #4022:
URL: https://github.com/apache/iggy/pull/4022#discussion_r3991294243
##########
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:
done
##########
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:
done
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add stream, topic, and partition functions to high level client [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #4022:
URL: https://github.com/apache/iggy/pull/4022#discussion_r3991292312
##########
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:
done
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add stream, topic, and partition functions to high level client [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm merged PR #4022:
URL: https://github.com/apache/iggy/pull/4022
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add stream, topic, and partition functions to high level client [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #4022:
URL: https://github.com/apache/iggy/pull/4022#discussion_r3991298358
##########
foreign/cpp/MODULE.bazel.lock:
##########
@@ -483,109 +483,6 @@
}
}
}
Review Comment:
looks like it was dropped by mistake. restored.
##########
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:
done
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add stream, topic, and partition functions to high level client [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on PR #4022:
URL: https://github.com/apache/iggy/pull/4022#issuecomment-5637776414
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add stream, topic, and partition functions to high level client [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #4022:
URL: https://github.com/apache/iggy/pull/4022#discussion_r3991301869
##########
foreign/cpp/tests/unit/unit_tests.cpp:
##########
@@ -187,18 +366,6 @@ TEST(IggyExceptionTest, StoresMessage) {
EXPECT_EQ(std::string(from_string.what()), message);
}
Review Comment:
done
##########
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:
done
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(cpp): add stream, topic, and partition functions to high level client [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #4022:
URL: https://github.com/apache/iggy/pull/4022#discussion_r3991295833
##########
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:
done
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org