Trait sea_streamer::Consumer

pub trait Consumer: Sized + Send + Sync {
    type Error: Error;
    type Message<'a>: Message
       where Self: 'a;
    type NextFuture<'a>: Future<Output = Result<Self::Message<'a>, StreamErr<Self::Error>>>
       where Self: 'a;
    type Stream<'a>: Stream<Item = Result<Self::Message<'a>, StreamErr<Self::Error>>>
       where Self: 'a;

    // Required methods
    fn seek(
        &mut self,
        to: OffsetDateTime
    ) -> impl Future<Output = Result<(), StreamErr<Self::Error>>> + Send;
    fn rewind(
        &mut self,
        offset: SeqPos
    ) -> impl Future<Output = Result<(), StreamErr<Self::Error>>> + Send;
    fn assign(
        &mut self,
        ss: (StreamKey, ShardId)
    ) -> Result<(), StreamErr<Self::Error>>;
    fn unassign(
        &mut self,
        ss: (StreamKey, ShardId)
    ) -> Result<(), StreamErr<Self::Error>>;
    fn next(&self) -> Self::NextFuture<'_>;
    fn stream<'a, 'b>(&'b mut self) -> Self::Stream<'a>
       where 'b: 'a;
}

Common interface of consumers, to be implemented by all backends.

Required Associated Types

type Error: Error

type Message<'a>: Message where Self: 'a

type NextFuture<'a>: Future<Output = Result<Self::Message<'a>, StreamErr<Self::Error>>> where Self: 'a

type Stream<'a>: Stream<Item = Result<Self::Message<'a>, StreamErr<Self::Error>>> where Self: 'a

Required Methods

fn seek( &mut self, to: OffsetDateTime ) -> impl Future<Output = Result<(), StreamErr<Self::Error>>> + Send

Seek all streams to an arbitrary point in time. It will start consuming from the earliest message with a timestamp later than to.

If the consumer is not already assigned, shard ZERO will be used.

fn rewind( &mut self, offset: SeqPos ) -> impl Future<Output = Result<(), StreamErr<Self::Error>>> + Send

Rewind all streams to a particular sequence number.

If the consumer is not already assigned, shard ZERO will be used.

fn assign( &mut self, ss: (StreamKey, ShardId) ) -> Result<(), StreamErr<Self::Error>>

Assign this consumer to a particular shard. Can be called multiple times to assign to multiple shards. Returns error StreamKeyNotFound if the stream is not currently subscribed.

It will only take effect on the next Consumer::seek or Consumer::rewind.

fn unassign( &mut self, ss: (StreamKey, ShardId) ) -> Result<(), StreamErr<Self::Error>>

Unassign a shard. Returns error StreamKeyNotFound if the stream is not currently subscribed. Returns error StreamKeyEmpty if all streams have been unassigned.

fn next(&self) -> Self::NextFuture<'_>

Poll and receive one message: it awaits until there are new messages. This method can be called from multiple threads.

fn stream<'a, 'b>(&'b mut self) -> Self::Stream<'a>

where 'b: 'a,

Returns an async stream. You cannot create multiple streams from the same consumer, nor perform any operation while streaming.

Object Safety

This trait is not object safe.

Implementations on Foreign Types

impl Consumer for KafkaConsumer

async fn seek( &mut self, timestamp: OffsetDateTime ) -> Result<(), StreamErr<KafkaError>>

Seek all streams to the given point in time, with all partitions assigned.

Warning

This async method is not cancel safe. You must await this future, and this Consumer will be unusable for any operations until it finishes.

async fn rewind(&mut self, offset: SeqPos) -> Result<(), StreamErr<KafkaError>>

Rewind all streams across all assigned partitions. Call Consumer::assign to assign a partition beforehand, or KafkaConsumer::reassign_partitions to assign all partitions.

fn stream<'a, 'b>(&'b mut self) -> <KafkaConsumer as Consumer>::Stream<'a>

where 'b: 'a,

Note: Kafka stream never ends.

type Error = KafkaError

type Message<'a> = KafkaMessage<'a>

type NextFuture<'a> = Map<StreamFuture<MessageStream<'a, DefaultConsumerContext>>, fn(_: (Option<Result<BorrowedMessage<'a>, KafkaError>>, MessageStream<'a, DefaultConsumerContext>)) -> Result<KafkaMessage<'a>, StreamErr<KafkaError>>>

type Stream<'a> = Map<MessageStream<'a, DefaultConsumerContext>, fn(_: Result<BorrowedMessage<'a>, KafkaError>) -> Result<KafkaMessage<'a>, StreamErr<KafkaError>>>

fn assign( &mut self, _: (StreamKey, ShardId) ) -> Result<(), StreamErr<KafkaError>>

fn unassign( &mut self, s: (StreamKey, ShardId) ) -> Result<(), StreamErr<KafkaError>>

fn next(&self) -> <KafkaConsumer as Consumer>::NextFuture<'_>

impl Consumer for StdioConsumer

fn assign(&mut self, _: (StreamKey, ShardId)) -> Result<(), StreamErr<StdioErr>>

Always succeed if the stream exists. There is only shard ZERO anyway.

fn unassign( &mut self, _: (StreamKey, ShardId) ) -> Result<(), StreamErr<StdioErr>>

Always fail. There is only shard ZERO anyway.

type Error = StdioErr

type Message<'a> = SharedMessage

type NextFuture<'a> = MapErr<RecvFut<'a, SharedMessage>, fn(_: RecvError) -> StreamErr<StdioErr>>

type Stream<'a> = Map<RecvStream<'a, SharedMessage>, fn(_: SharedMessage) -> Result<SharedMessage, StreamErr<StdioErr>>>

async fn seek(&mut self, _: OffsetDateTime) -> Result<(), StreamErr<StdioErr>>

async fn rewind(&mut self, _: SeqPos) -> Result<(), StreamErr<StdioErr>>

fn next(&self) -> <StdioConsumer as Consumer>::NextFuture<'_>

fn stream<'a, 'b>(&'b mut self) -> <StdioConsumer as Consumer>::Stream<'a>

where 'b: 'a,

Implementors

impl Consumer for SeaConsumer

type Error = BackendErr

type Message<'a> = SeaMessage<'a>

type NextFuture<'a> = NextFuture<'a>

type Stream<'a> = SeaMessageStream<'a>