feat(cpp): let poll and offset calls take a typed consumer

bdd/cpp/features/step_definitions/messaging_steps.cpp

@@ -150,7 +150,7 @@ WHEN("^I poll messages from stream ([0-9]+), topic ([0-9]+), partition ([0-9]+)
     const auto polled           = context->client->poll_messages(
         bdd::make_numeric_identifier(static_cast<std::uint32_t>(stream_id)),
         bdd::make_numeric_identifier(static_cast<std::uint32_t>(topic_id)), static_cast<std::uint32_t>(partition_id),
-        "consumer", bdd::make_numeric_identifier(1), std::string(polling_strategy.PollingStrategyKind()),
+        iggy::Consumer::Single(1), std::string(polling_strategy.PollingStrategyKind()),
         polling_strategy.PollingStrategyValue(), 100, false);
 
     context->polled.count = polled.count;

foreign/cpp/include/iggy.hpp

@@ -485,6 +485,117 @@ class IggyException : public std::runtime_error {
     explicit IggyException(const std::string &message) : std::runtime_error(message) {}
 };
 
+/**
+ * @brief Partition value that names no partition.
+ *
+ * Polling a consumer group with it reads one of the partitions assigned to the
+ * polling member, taking the next one on every call. Polling a regular consumer
+ * with it reads partition 0, and so does `get_consumer_offset(...)`.
+ * `store_consumer_offset(...)` and `delete_consumer_offset(...)` reject it and
+ * need an explicit partition.
+ */
+inline constexpr std::uint32_t kAnyPartitionId{std::numeric_limits<std::uint32_t>::max()};
+
+namespace detail {
+
+inline iggy::ffi::Identifier to_string_identifier(const std::string &id) {
+    iggy::ffi::Identifier identifier{};
+    try {
+        identifier.set_string(id);
+    } catch (const std::exception &error) {
+        throw IggyException(error.what());
+    }
+
+    return identifier;
+}
+
+inline iggy::ffi::Identifier to_numeric_identifier(const std::uint32_t id) {
+    iggy::ffi::Identifier identifier{};
+    try {
+        identifier.set_numeric(id);
+    } catch (const std::exception &error) {
+        throw IggyException(error.what());
+    }
+
+    return identifier;
+}
+
+inline iggy::ffi::Consumer to_consumer(const iggy::ffi::ConsumerKind kind, iggy::ffi::Identifier id) {
+    iggy::ffi::Consumer consumer{};
+    consumer.kind = kind;
+    consumer.id   = std::move(id);
+
+    return consumer;
+}
+
+}  // namespace detail
+
+/**
+ * @brief Creates the consumer that polling and consumer-offset calls take.
+ *
+ * The consumer carries the kind and the identifier the server keys the stored
+ * offset on. Two callers naming the same identifier share one offset, so under
+ * `PollingStrategy::Next()` with auto-commit each of them sees only the messages
+ * the other has not read yet. Give every independent consumer its own name.
+ *
+ * @code{.cpp}
+ * const auto strategy{iggy::PollingStrategy::Next()};
+ * const auto polled{client->poll_messages(stream, topic, iggy::kAnyPartitionId,
+ *                                         iggy::Consumer::Group("my-group"),
+ *                                         std::string(strategy.PollingStrategyKind()),
+ *                                         strategy.PollingStrategyValue(), 10, true)};
+ * @endcode
+ */
+class Consumer final {
+  public:
+    /**
+     * @brief Creates a named consumer owning its offset on the polled partition.
+     * @param id Consumer name.
+     * @return Consumer accepted by the client.
+     * @throws IggyException if @p id is empty or longer than 255 bytes.
+     */
+    static iggy::ffi::Consumer Single(const std::string &id) {
+        return detail::to_consumer(iggy::ffi::ConsumerKind::Consumer, detail::to_string_identifier(id));
+    }
+
+    /**
+     * @brief Creates a numbered consumer owning its offset on the polled partition.
+     * @param id Consumer number.
+     * @return Consumer accepted by the client.
+     */
+    static iggy::ffi::Consumer Single(const std::uint32_t id) {
+        return detail::to_consumer(iggy::ffi::ConsumerKind::Consumer, detail::to_numeric_identifier(id));
+    }
+
+    /**
+     * @brief Creates a member of the named consumer group, sharing the group's offset.
+     *
+     * Join the group with `join_consumer_group(...)` before polling, and pass
+     * `kAnyPartitionId` to read the partitions assigned to this member, one per
+     * call.
+     *
+     * @param id Consumer group name.
+     * @return Consumer accepted by the client.
+     * @throws IggyException if @p id is empty or longer than 255 bytes.
+     */
+    static iggy::ffi::Consumer Group(const std::string &id) {
+        return detail::to_consumer(iggy::ffi::ConsumerKind::ConsumerGroup, detail::to_string_identifier(id));
+    }
+
+    /**
+     * @brief Creates a member of the numbered consumer group, sharing the group's offset.
+     * @param id Consumer group number.
+     * @return Consumer accepted by the client.
+     * @see Group(const std::string &)
+     */
+    static iggy::ffi::Consumer Group(const std::uint32_t id) {
+        return detail::to_consumer(iggy::ffi::ConsumerKind::ConsumerGroup, detail::to_numeric_identifier(id));
+    }
+
+  private:
+    Consumer() = delete;
+};
+
 /**
  * @brief Owning client connection to an Apache Iggy server.
  *

foreign/cpp/src/client.rs

@@ -42,14 +42,6 @@ use std::sync::Arc;
 /// reserve `u32::MAX` as the sentinel for `partition_id`.
 const ANY_PARTITION_ID: u32 = u32::MAX;
 
-fn resolve_consumer(consumer_kind: &str, consumer_id: RustIdentifier) -> Result<Consumer, String> {
-    match consumer_kind {
-        "consumer" => Ok(Consumer::new(consumer_id)),
-        "consumer_group" => Ok(Consumer::group(consumer_id)),
-        _ => Err(format!("invalid consumer kind: {consumer_kind}")),
-    }
-}
-
 fn opt_partition(partition_id: u32) -> Option<u32> {
     if partition_id == ANY_PARTITION_ID {
         None
@@ -391,8 +383,7 @@ impl Client {
         stream_id: ffi::Identifier,
         topic_id: ffi::Identifier,
         partition_id: u32,
-        consumer_kind: String,
-        consumer_id: ffi::Identifier,
+        consumer: ffi::Consumer,
         polling_strategy_kind: String,
         polling_strategy_value: u64,
         count: u32,
@@ -402,9 +393,7 @@ impl Client {
             .map_err(|error| format!("Could not poll messages: {error}"))?;
         let rust_topic_id = RustIdentifier::try_from(topic_id)
             .map_err(|error| format!("Could not poll messages: {error}"))?;
-        let rust_consumer_id = RustIdentifier::try_from(consumer_id)
-            .map_err(|error| format!("Could not poll messages: {error}"))?;
-        let consumer = resolve_consumer(&consumer_kind, rust_consumer_id)
+        let rust_consumer = Consumer::try_from(consumer)
             .map_err(|error| format!("Could not poll messages: {error}"))?;
 
         let strategy = match polling_strategy_kind.as_str() {
@@ -427,7 +416,7 @@ impl Client {
                     &rust_stream_id,
                     &rust_topic_id,
                     opt_partition(partition_id),
-                    &consumer,
+                    &rust_consumer,
                     &strategy,
                     count,
                     auto_commit,
@@ -911,23 +900,20 @@ impl Client {
         stream_id: ffi::Identifier,
         topic_id: ffi::Identifier,
         partition_id: u32,
-        consumer_kind: String,
-        consumer_id: ffi::Identifier,
+        consumer: ffi::Consumer,
         offset: u64,
     ) -> Result<(), String> {
         let rust_stream_id = RustIdentifier::try_from(stream_id)
             .map_err(|error| format!("Could not store consumer offset: {error}"))?;
         let rust_topic_id = RustIdentifier::try_from(topic_id)
             .map_err(|error| format!("Could not store consumer offset: {error}"))?;
-        let rust_consumer_id = RustIdentifier::try_from(consumer_id)
-            .map_err(|error| format!("Could not store consumer offset: {error}"))?;
-        let consumer = resolve_consumer(&consumer_kind, rust_consumer_id)
+        let rust_consumer = Consumer::try_from(consumer)
             .map_err(|error| format!("Could not store consumer offset: {error}"))?;
 
         RUNTIME.block_on(async {
             self.inner
                 .store_consumer_offset(
-                    &consumer,
+                    &rust_consumer,
                     &rust_stream_id,
                     &rust_topic_id,
                     opt_partition(partition_id),
@@ -948,23 +934,20 @@ impl Client {
         stream_id: ffi::Identifier,
         topic_id: ffi::Identifier,
         partition_id: u32,
-        consumer_kind: String,
-        consumer_id: ffi::Identifier,
+        consumer: ffi::Consumer,
     ) -> Result<ffi::ConsumerOffsetInfo, String> {
         let rust_stream_id = RustIdentifier::try_from(stream_id)
             .map_err(|error| format!("Could not get consumer offset: {error}"))?;
         let rust_topic_id = RustIdentifier::try_from(topic_id)
             .map_err(|error| format!("Could not get consumer offset: {error}"))?;
-        let rust_consumer_id = RustIdentifier::try_from(consumer_id)
-            .map_err(|error| format!("Could not get consumer offset: {error}"))?;
-        let consumer = resolve_consumer(&consumer_kind, rust_consumer_id)
+        let rust_consumer = Consumer::try_from(consumer)
             .map_err(|error| format!("Could not get consumer offset: {error}"))?;
 
         RUNTIME.block_on(async {
             let offset = self
                 .inner
                 .get_consumer_offset(
-                    &consumer,
+                    &rust_consumer,
                     &rust_stream_id,
                     &rust_topic_id,
                     opt_partition(partition_id),
@@ -988,22 +971,19 @@ impl Client {
         stream_id: ffi::Identifier,
         topic_id: ffi::Identifier,
         partition_id: u32,
-        consumer_kind: String,
-        consumer_id: ffi::Identifier,
+        consumer: ffi::Consumer,
     ) -> Result<(), String> {
         let rust_stream_id = RustIdentifier::try_from(stream_id)
             .map_err(|error| format!("Could not delete consumer offset: {error}"))?;
         let rust_topic_id = RustIdentifier::try_from(topic_id)
             .map_err(|error| format!("Could not delete consumer offset: {error}"))?;
-        let rust_consumer_id = RustIdentifier::try_from(consumer_id)
-            .map_err(|error| format!("Could not delete consumer offset: {error}"))?;
-        let consumer = resolve_consumer(&consumer_kind, rust_consumer_id)
+        let rust_consumer = Consumer::try_from(consumer)
             .map_err(|error| format!("Could not delete consumer offset: {error}"))?;
 
         RUNTIME.block_on(async {
             self.inner
                 .delete_consumer_offset(
-                    &consumer,
+                    &rust_consumer,
                     &rust_stream_id,
                     &rust_topic_id,
                     opt_partition(partition_id),

foreign/cpp/src/consumer.rs

@@ -18,6 +18,6 @@
 use iggy::prelude::IggyConsumer as RustIggyConsumer;
 
 #[allow(dead_code)]
-pub struct Consumer {
+pub struct IggyConsumer {
     pub inner: RustIggyConsumer,
 }

foreign/cpp/src/lib.rs

@@ -23,7 +23,7 @@ mod producer;
 mod type_conversion;
 
 use client::{Client, delete_connection as delete_client, from_connection_string, new_connection};
-use consumer::Consumer;
+use consumer::IggyConsumer;
 use messages::make_message;
 use producer::Producer;
 use std::sync::LazyLock;
@@ -244,6 +244,22 @@ mod ffi {
         stored_offset: u64,
     }
 
+    /// Kind of the consumer a poll or a consumer-offset call names.
+    #[repr(u8)]
+    enum ConsumerKind {
+        Consumer = 1,
+        ConsumerGroup = 2,
+    }
+
+    /// The consumer a poll or a consumer-offset call names. It carries both the
+    /// kind and the identifier the server keys the stored offset on. Two callers
+    /// sharing one identifier share one offset, so under the `next` strategy with
+    /// auto-commit each of them reads only what the other has not read yet.
+    struct Consumer {
+        kind: ConsumerKind,
+        id: Identifier,
+    }
+
     struct ClientInfo {
         client_id: u32,
         has_user_id: bool,
@@ -430,7 +446,7 @@ mod ffi {
 
     extern "Rust" {
         type Client;
-        type Consumer;
+        type IggyConsumer;
         type Producer;
 
         // Client functions
@@ -529,35 +545,35 @@ mod ffi {
             stream_id: Identifier,
             topic_id: Identifier,
             partition_id: u32,
-            consumer_kind: String,
-            consumer_id: Identifier,
+            consumer: Consumer,
             offset: u64,
         ) -> Result<()>;
         fn get_consumer_offset(
             self: &Client,
             stream_id: Identifier,
             topic_id: Identifier,
             partition_id: u32,
-            consumer_kind: String,
-            consumer_id: Identifier,
+            consumer: Consumer,
         ) -> Result<ConsumerOffsetInfo>;
         fn delete_consumer_offset(
             self: &Client,
             stream_id: Identifier,
             topic_id: Identifier,
             partition_id: u32,
-            consumer_kind: String,
-            consumer_id: Identifier,
+            consumer: Consumer,
         ) -> Result<()>;
 
+        /// Polls messages as `consumer`. A `partition_id` of `u32::MAX` names no
+        /// partition: a consumer group then reads one of the partitions assigned
+        /// to the polling member, taking the next one on every call, and a
+        /// regular consumer reads partition 0.
         #[allow(clippy::too_many_arguments)]
         fn poll_messages(
             self: &Client,
             stream_id: Identifier,
             topic_id: Identifier,
             partition_id: u32,
-            consumer_kind: String,
-            consumer_id: Identifier,
+            consumer: Consumer,
             polling_strategy_kind: String,
             polling_strategy_value: u64,
             count: u32,
@@ -657,18 +673,18 @@ mod ffi {
         fn set_string(self: &mut Identifier, id: String) -> Result<()>;
         fn set_numeric(self: &mut Identifier, id: u32) -> Result<()>;
 
-        // Consumer methods
-        // fn name(self: &Consumer) -> Result<String>;
-        // fn topic(self: &Consumer) -> Result<Identifier>;
-        // fn stream(self: &Consumer) -> Result<Identifier>;
-        // fn partition_id(self: &Consumer) -> u32;
-        // fn store_offset(self: &Consumer, offset: u64, partition_id: u32) -> Result<()>;
-        // fn delete_offset(self: &Consumer, partition_id: u32) -> Result<()>;
-        // fn get_last_consumed_offset(self: &Consumer, partition_id: u32) -> Result<u64>;
-        // fn get_last_stored_offset(self: &Consumer, partition_id: u32) -> Result<u64>;
-        // fn init(self: &mut Consumer) -> Result<()>;
-        // fn shutdown(self: &mut Consumer) -> Result<()>;
-        // unsafe fn delete_consumer(consumer: *mut Consumer) -> Result<()>;
+        // IggyConsumer methods
+        // fn name(self: &IggyConsumer) -> Result<String>;
+        // fn topic(self: &IggyConsumer) -> Result<Identifier>;
+        // fn stream(self: &IggyConsumer) -> Result<Identifier>;
+        // fn partition_id(self: &IggyConsumer) -> u32;
+        // fn store_offset(self: &IggyConsumer, offset: u64, partition_id: u32) -> Result<()>;
+        // fn delete_offset(self: &IggyConsumer, partition_id: u32) -> Result<()>;
+        // fn get_last_consumed_offset(self: &IggyConsumer, partition_id: u32) -> Result<u64>;
+        // fn get_last_stored_offset(self: &IggyConsumer, partition_id: u32) -> Result<u64>;
+        // fn init(self: &mut IggyConsumer) -> Result<()>;
+        // fn shutdown(self: &mut IggyConsumer) -> Result<()>;
+        // unsafe fn delete_consumer(consumer: *mut IggyConsumer) -> Result<()>;
 
         // Producer methods
         // fn stream(self: &Producer) -> Result<Identifier>;