haubur commented on code in PR #3913:
URL: https://github.com/apache/iggy/pull/3913#discussion_r3806031068
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -100,6 +100,316 @@ pub enum AutoCommitAfter {
// 4. All `&self` methods only access Sync-safe fields
unsafe impl Sync for IggyConsumer {}
+/// Reads messages from the partitions of one topic and yields them one at a
time.
+///
+/// A topic is split into partitions, and a partition is an ordered log that
producers append to.
+/// Every message sits at an *offset*, its position in that log. Reading is
therefore always the
+/// same three decisions: which partition to read, where in it to start, and
how to keep track of
+/// how far you got so the next run can continue there.
+///
+/// `IggyConsumer` handles all three. It fetches batches of messages from the
server, keeps them in
+/// an in-memory buffer, decrypts them when the client is configured with an
encryptor, and records
+/// how far it has read. **It implements [`Stream`], so consuming is a loop
over [`StreamExt::next`].**
+///
+/// You can use a consumer as a worker draining a topic, a reader that replays
a
+/// partition from a chosen point, and a pool of consumers sharing a workload
through a consumer
+/// group.
+///
+/// # Creating a consumer
+///
+/// Easiest way is to use the [`IggyClient`] with a configured connection.
Then:
+/// - [`IggyClient::consumer()`] builds a standalone consumer, bound to the
one partition passed
+/// in.
+/// - [`IggyClient::consumer_group()`] builds a member of a consumer group.
The server gives every
+/// partition to exactly one member, so several consumers using the same
group name split the
+/// topic between them and share one set of offsets.
+///
+/// Note, building never talks to the server. [`init()`](Self::init) must be
awaited once before the
+/// first message is read.
+///
+/// # Examples
+///
+/// A standalone consumer reading partition 1 with the defaults:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer("my-consumer", "my-stream", "my-topic", 1)?
+/// .batch_length(100)
+/// .poll_interval(IggyDuration::new_from_secs(1))
+/// .build();
+/// consumer.init().await?;
+///
+/// while let Some(received) = consumer.next().await {
+/// match received {
+/// Ok(received) => println!("Offset: {}",
received.message.header.offset),
+/// Err(error) => eprintln!("Failed to read a message: {error}"),
+/// }
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
+/// A group member that commits every message right after it is handed over,
and shuts down
+/// cleanly:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn handle(message: &IggyMessage) {}
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer_group("order-workers", "my-stream", "my-topic")?
+/// .auto_commit(AutoCommit::When(AutoCommitWhen::ConsumingEachMessage))
+/// .polling_strategy(PollingStrategy::next())
+/// .build();
+/// consumer.init().await?;
+///
+/// let mut consumed = 0;
+/// while let Some(received) = consumer.next().await {
+/// handle(&received?.message).await;
+/// consumed += 1;
+/// if consumed == 100 {
+/// break;
+/// }
+/// }
+///
+/// consumer.shutdown().await?;
+/// # Ok(())
+/// # }
+/// ```
+///
+/// Committing by hand, so that a message the handler could not process comes
back:
Review Comment:
True, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
