pub struct KafkaStreamConsumerAdapter(/* private fields */);Expand description
An adapter used to make easier Stream implementation of StreamConsumer.
Implementations§
Source§impl KafkaStreamConsumerAdapter
impl KafkaStreamConsumerAdapter
Sourcepub fn new(csm: StreamConsumer) -> Self
pub fn new(csm: StreamConsumer) -> Self
Creates new adapter.
Methods from Deref<Target = StreamConsumer>§
Sourcepub fn stream(&self) -> MessageStream<'_, C>
pub fn stream(&self) -> MessageStream<'_, C>
Constructs a stream that yields messages from this consumer.
It is legal to have multiple live message streams for the same consumer, and to move those message streams across threads. Note, however, that the message streams share the same underlying state. A message received by the consumer will be delivered to only one of the live message streams. If you seek the underlying consumer, all message streams created from the consumer will begin to draw messages from the new position of the consumer.
If you want multiple independent views of a Kafka topic, create multiple consumers, not multiple message streams.
Sourcepub async fn recv(&self) -> Result<BorrowedMessage<'_>, KafkaError>
pub async fn recv(&self) -> Result<BorrowedMessage<'_>, KafkaError>
Receives the next message from the stream.
This method will block until the next message is available or an error
occurs. It is legal to call recv from multiple threads simultaneously.
This method is cancellation safe.
Note that this method is exactly as efficient as constructing a single-use message stream and extracting one message from it:
use futures::stream::StreamExt;
consumer.stream().next().await.expect("MessageStream never returns None");Sourcepub fn split_partition_queue(
self: &Arc<StreamConsumer<C, R>>,
topic: &str,
partition: i32,
) -> Option<StreamPartitionQueue<C, R>>
pub fn split_partition_queue( self: &Arc<StreamConsumer<C, R>>, topic: &str, partition: i32, ) -> Option<StreamPartitionQueue<C, R>>
Splits messages for the specified partition into their own stream.
If the topic or partition is invalid, returns None.
After calling this method, newly-fetched messages for the specified
partition will be returned via StreamPartitionQueue::recv rather
than StreamConsumer::recv. Note that there may be buffered messages
for the specified partition that will continue to be returned by
StreamConsumer::recv. For best results, call split_partition_queue
before the first call to
StreamConsumer::recv.
You must periodically await StreamConsumer::recv, even if no messages
are expected, to serve events. Consider using a background task like:
tokio::spawn(async move {
let message = stream_consumer.recv().await;
panic!("main stream consumer queue unexpectedly received message: {:?}", message);
})Note that calling Consumer::assign will deactivate any existing
partition queues. You will need to call this method for every partition
that should be split after every call to assign.
Beware that this method is implemented for &Arc<Self>, not &self.
You will need to wrap your consumer in an Arc in order to call this
method. This design permits moving the partition queue to another thread
while ensuring the partition queue does not outlive the consumer.