This is an automated email from the ASF dual-hosted git repository. slbotbm pushed a commit to branch cpp-high-level-client-2 in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 7a4d58d30a9aa1d3c6cf9f171d48747c3bf6791f Author: Rimuksh Kansal <[email protected]> AuthorDate: Mon Sep 14 19:56:35 2026 +0900 remove consumer/producer code --- foreign/cpp/include/iggy.hpp | 141 +++++++++++++++++++++++++++---------------- foreign/cpp/src/client.cpp | 23 ------- foreign/cpp/src/client.rs | 65 -------------------- foreign/cpp/src/consumer.cpp | 61 ------------------- foreign/cpp/src/consumer.rs | 17 ------ foreign/cpp/src/lib.rs | 28 ++------- foreign/cpp/src/producer.cpp | 61 ------------------- foreign/cpp/src/producer.rs | 16 ----- 8 files changed, 94 insertions(+), 318 deletions(-) diff --git a/foreign/cpp/include/iggy.hpp b/foreign/cpp/include/iggy.hpp index 397c44c32..7b025a2d6 100644 --- a/foreign/cpp/include/iggy.hpp +++ b/foreign/cpp/include/iggy.hpp @@ -42,8 +42,6 @@ namespace iggy { class IggyBlockingClient; -class IggyBlockingConsumer; -class IggyBlockingProducer; class LoginInfo; class Partition; class Topic; @@ -1619,48 +1617,6 @@ class PollingStrategy final { std::uint64_t polling_strategy_value_; }; -class IggyBlockingConsumer final { - public: - IggyBlockingConsumer(const IggyBlockingConsumer &) = delete; - IggyBlockingConsumer &operator=(const IggyBlockingConsumer &) = delete; - - IggyBlockingConsumer(IggyBlockingConsumer &&other) noexcept; - IggyBlockingConsumer &operator=(IggyBlockingConsumer &&other) noexcept; - - ~IggyBlockingConsumer(); - - private: - explicit IggyBlockingConsumer(ffi::Consumer *consumer); - - [[nodiscard]] ffi::Consumer *Handle() const; - void Reset() noexcept; - - friend class IggyBlockingClient; - - ffi::Consumer *consumer_; -}; - -class IggyBlockingProducer final { - public: - IggyBlockingProducer(const IggyBlockingProducer &) = delete; - IggyBlockingProducer &operator=(const IggyBlockingProducer &) = delete; - - IggyBlockingProducer(IggyBlockingProducer &&other) noexcept; - IggyBlockingProducer &operator=(IggyBlockingProducer &&other) noexcept; - - ~IggyBlockingProducer(); - - private: - explicit IggyBlockingProducer(ffi::Producer *producer); - - [[nodiscard]] ffi::Producer *Handle() const; - void Reset() noexcept; - - friend class IggyBlockingClient; - - ffi::Producer *producer_; -}; - /** * @brief Owning client connection to an Apache Iggy server. * @@ -2103,19 +2059,102 @@ class IggyBlockingClient final { */ void DeletePartitions(const Identifier &stream, const Identifier &topic, std::uint32_t partitions_count); + /** + * @brief Creates a consumer group for a topic. + * + * The group name must be unique within the topic, non-empty, and no more + * than 255 UTF-8 bytes. The new group initially has no members. + * + * @param stream Parent stream, addressed by numeric ID or name. + * @param topic Parent topic, addressed by numeric ID or name. + * @param name Unique consumer group name within @p topic. + * @return Details of the newly created consumer group. + * @throws IggyException if the client is unavailable or unauthenticated; + * an identifier or the name is invalid; the stream or topic does + * not exist; the name is already in use; the caller lacks + * stream- or topic-management permission; or the request fails. + */ ConsumerGroupDetails CreateConsumerGroup(const Identifier &stream, const Identifier &topic, std::string name); + + /** + * @brief Retrieves one consumer group and its current members. + * + * The returned details are a snapshot. Membership and partition + * assignments can change immediately after this call returns. + * + * @param stream Parent stream, addressed by numeric ID or name. + * @param topic Parent topic, addressed by numeric ID or name. + * @param group Consumer group to retrieve, addressed by numeric ID or name. + * @return Consumer group metadata and member details. + * @throws IggyException if the client is unavailable or unauthenticated; + * an identifier is invalid; the stream, topic, or consumer group + * does not exist; the caller lacks read permission; or the + * metadata read fails. + */ ConsumerGroupDetails GetConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group); + + /** + * @brief Lists consumer group summaries for a topic. + * + * The summaries include member and partition counts but omit individual + * member details. Use GetConsumerGroup() to retrieve those details. + * + * @param stream Parent stream, addressed by numeric ID or name. + * @param topic Parent topic, addressed by numeric ID or name. + * @return Consumer group summaries for the requested topic. + * @throws IggyException if the client is unavailable or unauthenticated; + * an identifier is invalid; the stream or topic does not exist; + * the caller lacks read permission; or the metadata read fails. + */ std::vector<ConsumerGroup> GetConsumerGroups(const Identifier &stream, const Identifier &topic); + + /** + * @brief Deletes a consumer group from a topic. + * + * A failed or unknown transport outcome can leave the deletion committed. + * Query the topic's consumer groups before retrying this request. + * + * @param stream Parent stream, addressed by numeric ID or name. + * @param topic Parent topic, addressed by numeric ID or name. + * @param group Consumer group to delete, addressed by numeric ID or name. + * @throws IggyException if the client is unavailable or unauthenticated; + * an identifier is invalid; the stream, topic, or consumer group + * does not exist; the caller lacks stream- or topic-management + * permission; or the request fails. + */ void DeleteConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group); + + /** + * @brief Joins the current client to a consumer group. + * + * The server assigns topic partitions among the group's members. Joining + * the same group again does not add a second membership for this client. + * + * @param stream Parent stream, addressed by numeric ID or name. + * @param topic Parent topic, addressed by numeric ID or name. + * @param group Consumer group to join, addressed by numeric ID or name. + * @throws IggyException if the client is unavailable or unauthenticated; + * an identifier is invalid; the stream, topic, or consumer group + * does not exist; the caller lacks read permission; the transport + * does not support group membership; or the request fails. + */ void JoinConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group); - void LeaveConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group); - IggyBlockingConsumer CreateConsumer(std::string name, - const Identifier &stream, - const Identifier &topic, - std::uint32_t partition_id); - IggyBlockingConsumer CreateGroupConsumer(std::string name, const Identifier &stream, const Identifier &topic); - IggyBlockingProducer CreateProducer(const Identifier &stream, const Identifier &topic); + /** + * @brief Removes the current client from a consumer group. + * + * The server reassigns partitions among the remaining group members. + * + * @param stream Parent stream, addressed by numeric ID or name. + * @param topic Parent topic, addressed by numeric ID or name. + * @param group Consumer group to leave, addressed by numeric ID or name. + * @throws IggyException if the client is unavailable or unauthenticated; + * an identifier is invalid; the stream, topic, or consumer group + * does not exist; this client is not a member; the caller lacks + * read permission; the transport does not support group + * membership; or the request fails. + */ + void LeaveConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group); private: explicit IggyBlockingClient(ffi::Client *client); diff --git a/foreign/cpp/src/client.cpp b/foreign/cpp/src/client.cpp index 2d8d699a2..cb5041945 100644 --- a/foreign/cpp/src/client.cpp +++ b/foreign/cpp/src/client.cpp @@ -303,29 +303,6 @@ void IggyBlockingClient::LeaveConsumerGroup(const Identifier &stream, }); } -IggyBlockingConsumer IggyBlockingClient::CreateConsumer(std::string name, - const Identifier &stream, - const Identifier &topic, - std::uint32_t partition_id) { - return RethrowAsIggyException([this, &name, &stream, &topic, partition_id] { - return IggyBlockingConsumer(Handle()->create_consumer(name, stream.ToFfi(), topic.ToFfi(), partition_id)); - }); -} - -IggyBlockingConsumer IggyBlockingClient::CreateGroupConsumer(std::string name, - const Identifier &stream, - const Identifier &topic) { - return RethrowAsIggyException([this, &name, &stream, &topic] { - return IggyBlockingConsumer(Handle()->create_group_consumer(name, stream.ToFfi(), topic.ToFfi())); - }); -} - -IggyBlockingProducer IggyBlockingClient::CreateProducer(const Identifier &stream, const Identifier &topic) { - return RethrowAsIggyException([this, &stream, &topic] { - return IggyBlockingProducer(Handle()->create_producer(stream.ToFfi(), topic.ToFfi())); - }); -} - IggyBlockingClient::IggyBlockingClient(ffi::Client *client) : client_(client) { if (client_ == nullptr) { throw IggyException("Could not create Iggy client"); diff --git a/foreign/cpp/src/client.rs b/foreign/cpp/src/client.rs index 22203836b..6ea8fdc9e 100644 --- a/foreign/cpp/src/client.rs +++ b/foreign/cpp/src/client.rs @@ -15,8 +15,6 @@ // specific language governing permissions and limitations // under the License. -use crate::consumer::Consumer as BridgeConsumer; -use crate::producer::Producer as BridgeProducer; use crate::{RUNTIME, ffi, type_conversion::ffi_options_to_raw}; use bytes::Bytes; use iggy::prelude::{ @@ -1246,69 +1244,6 @@ impl Client { Ok(Vec::from(response)) }) } - - pub fn create_consumer( - &self, - name: String, - stream_id: ffi::Identifier, - topic_id: ffi::Identifier, - partition_id: u32, - ) -> Result<*mut BridgeConsumer, String> { - let rust_stream_id = RustIdentifier::try_from(stream_id) - .map_err(|error| format!("Could not create consumer '{name}': {error}"))?; - let rust_topic_id = RustIdentifier::try_from(topic_id) - .map_err(|error| format!("Could not create consumer '{name}': {error}"))?; - let consumer = self - .inner - .consumer( - &name, - &rust_stream_id.as_string(), - &rust_topic_id.as_string(), - partition_id, - ) - .map_err(|error| format!("Could not create consumer '{name}': {error}"))? - .build(); - Ok(Box::into_raw(Box::new(BridgeConsumer { inner: consumer }))) - } - - pub fn create_group_consumer( - &self, - name: String, - stream_id: ffi::Identifier, - topic_id: ffi::Identifier, - ) -> Result<*mut BridgeConsumer, String> { - let rust_stream_id = RustIdentifier::try_from(stream_id) - .map_err(|error| format!("Could not create consumer group member '{name}': {error}"))?; - let rust_topic_id = RustIdentifier::try_from(topic_id) - .map_err(|error| format!("Could not create consumer group member '{name}': {error}"))?; - let consumer = self - .inner - .consumer_group( - &name, - &rust_stream_id.as_string(), - &rust_topic_id.as_string(), - ) - .map_err(|error| format!("Could not create consumer group member '{name}': {error}"))? - .build(); - Ok(Box::into_raw(Box::new(BridgeConsumer { inner: consumer }))) - } - - pub fn create_producer( - &self, - stream_id: ffi::Identifier, - topic_id: ffi::Identifier, - ) -> Result<*mut BridgeProducer, String> { - let rust_stream_id = RustIdentifier::try_from(stream_id) - .map_err(|error| format!("Could not create producer: {error}"))?; - let rust_topic_id = RustIdentifier::try_from(topic_id) - .map_err(|error| format!("Could not create producer: {error}"))?; - let producer = self - .inner - .producer(&rust_stream_id.as_string(), &rust_topic_id.as_string()) - .map_err(|error| format!("Could not create producer: {error}"))? - .build(); - Ok(Box::into_raw(Box::new(BridgeProducer { inner: producer }))) - } } pub unsafe fn delete_connection(client: *mut Client) { diff --git a/foreign/cpp/src/consumer.cpp b/foreign/cpp/src/consumer.cpp deleted file mode 100644 index c0183495f..000000000 --- a/foreign/cpp/src/consumer.cpp +++ /dev/null @@ -1,61 +0,0 @@ -/* - * 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 { - -IggyBlockingConsumer::IggyBlockingConsumer(IggyBlockingConsumer &&other) noexcept - : consumer_(std::exchange(other.consumer_, nullptr)) {} - -IggyBlockingConsumer &IggyBlockingConsumer::operator=(IggyBlockingConsumer &&other) noexcept { - if (this != &other) { - Reset(); - consumer_ = std::exchange(other.consumer_, nullptr); - } - return *this; -} - -IggyBlockingConsumer::~IggyBlockingConsumer() { - Reset(); -} - -IggyBlockingConsumer::IggyBlockingConsumer(ffi::Consumer *consumer) : consumer_(consumer) { - if (consumer_ == nullptr) { - throw IggyException("Could not create Iggy consumer"); - } -} - -ffi::Consumer *IggyBlockingConsumer::Handle() const { - if (consumer_ == nullptr) { - throw IggyException("Cannot use a moved-from IggyBlockingConsumer"); - } - return consumer_; -} - -void IggyBlockingConsumer::Reset() noexcept { - if (consumer_ == nullptr) { - return; - } - - ffi::Consumer *consumer{std::exchange(consumer_, nullptr)}; - ffi::delete_consumer(consumer); -} - -} // namespace iggy diff --git a/foreign/cpp/src/consumer.rs b/foreign/cpp/src/consumer.rs index 4250625a8..147374d4e 100644 --- a/foreign/cpp/src/consumer.rs +++ b/foreign/cpp/src/consumer.rs @@ -21,20 +21,3 @@ use iggy::prelude::IggyConsumer as RustIggyConsumer; pub struct Consumer { pub inner: RustIggyConsumer, } - -/// Releases a consumer previously returned by `create_consumer` or -/// `create_consumer_group`. -/// -/// # Safety -/// -/// - Passing the pointer to this function more than once is undefined -/// behaviour (double-free). -/// - Using the pointer after this function has been called is undefined -/// behaviour (use-after-free). -pub unsafe fn delete_consumer(consumer: *mut Consumer) { - if !consumer.is_null() { - unsafe { - drop(Box::from_raw(consumer)); - } - } -} diff --git a/foreign/cpp/src/lib.rs b/foreign/cpp/src/lib.rs index d6e1b20ab..3ed6d1103 100644 --- a/foreign/cpp/src/lib.rs +++ b/foreign/cpp/src/lib.rs @@ -23,9 +23,9 @@ mod producer; mod type_conversion; use client::{Client, delete_connection as delete_client, from_connection_string, new_connection}; -use consumer::{Consumer, delete_consumer}; +use consumer::Consumer; use messages::make_message; -use producer::{Producer, delete_producer}; +use producer::Producer; use std::sync::LazyLock; static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| { @@ -685,28 +685,6 @@ mod ffi { fn set_string(self: &mut Identifier, id: String) -> Result<()>; fn set_numeric(self: &mut Identifier, id: u32) -> Result<()>; - fn create_consumer( - self: &Client, - name: String, - stream_id: Identifier, - topic_id: Identifier, - partition_id: u32, - ) -> Result<*mut Consumer>; - fn create_group_consumer( - self: &Client, - name: String, - stream_id: Identifier, - topic_id: Identifier, - ) -> Result<*mut Consumer>; - unsafe fn delete_consumer(consumer: *mut Consumer); - - fn create_producer( - self: &Client, - stream_id: Identifier, - topic_id: Identifier, - ) -> Result<*mut Producer>; - unsafe fn delete_producer(producer: *mut Producer); - // Consumer methods // fn name(self: &Consumer) -> Result<String>; // fn topic(self: &Consumer) -> Result<Identifier>; @@ -718,6 +696,7 @@ mod ffi { // 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<()>; // Producer methods // fn stream(self: &Producer) -> Result<Identifier>; @@ -728,5 +707,6 @@ mod ffi { // fn send_with_partitioning(self: &Producer, partitioning_kind: String, partitioning_value: Vec<u8>, messages: Vec<IggyMessageToSend>) -> Result<()>; // fn send_to(self: &Producer, stream_id: Identifier, topic_id: Identifier, partitioning_kind: String, partitioning_value: Vec<u8>, messages: Vec<IggyMessageToSend>) -> Result<()>; // fn shutdown(self: &mut Producer) -> Result<()>; + // unsafe fn delete_producer(producer: *mut Producer) -> Result<()>; } } diff --git a/foreign/cpp/src/producer.cpp b/foreign/cpp/src/producer.cpp deleted file mode 100644 index b5a823158..000000000 --- a/foreign/cpp/src/producer.cpp +++ /dev/null @@ -1,61 +0,0 @@ -/* - * 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 { - -IggyBlockingProducer::IggyBlockingProducer(IggyBlockingProducer &&other) noexcept - : producer_(std::exchange(other.producer_, nullptr)) {} - -IggyBlockingProducer &IggyBlockingProducer::operator=(IggyBlockingProducer &&other) noexcept { - if (this != &other) { - Reset(); - producer_ = std::exchange(other.producer_, nullptr); - } - return *this; -} - -IggyBlockingProducer::~IggyBlockingProducer() { - Reset(); -} - -IggyBlockingProducer::IggyBlockingProducer(ffi::Producer *producer) : producer_(producer) { - if (producer_ == nullptr) { - throw IggyException("Could not create Iggy producer"); - } -} - -ffi::Producer *IggyBlockingProducer::Handle() const { - if (producer_ == nullptr) { - throw IggyException("Cannot use a moved-from IggyBlockingProducer"); - } - return producer_; -} - -void IggyBlockingProducer::Reset() noexcept { - if (producer_ == nullptr) { - return; - } - - ffi::Producer *producer{std::exchange(producer_, nullptr)}; - ffi::delete_producer(producer); -} - -} // namespace iggy diff --git a/foreign/cpp/src/producer.rs b/foreign/cpp/src/producer.rs index 62442924c..85b87db1c 100644 --- a/foreign/cpp/src/producer.rs +++ b/foreign/cpp/src/producer.rs @@ -21,19 +21,3 @@ use iggy::prelude::IggyProducer as RustIggyProducer; pub struct Producer { pub inner: RustIggyProducer, } - -/// Releases a producer previously returned by `create_producer`. -/// -/// # Safety -/// -/// - Passing the pointer to this function more than once is undefined -/// behaviour (double-free). -/// - Using the pointer after this function has been called is undefined -/// behaviour (use-after-free). -pub unsafe fn delete_producer(producer: *mut Producer) { - if !producer.is_null() { - unsafe { - drop(Box::from_raw(producer)); - } - } -}
