Struct KafkaEventStream 

pub struct KafkaEventStream<C = DefaultConsumerContext, R = DefaultRuntime>
where
    C: ConsumerContext,
{ /* private fields */ }

An event stream consuming from kafka.

Events are expected to be received with a json payload, and will be deserialized with the thalo::aggregate::Aggregate::Event’s serde::de::DeserializeOwned implementation.

Implementations

impl<C, R> KafkaEventStream<C, R>

where C: ConsumerContext,

pub fn new(topics: &[impl Display], consumer: StreamConsumer<C, R>) -> Self

Creates a new KafkaEventStream.

pub fn consumer(&self) -> Arc<StreamConsumer<C, R>>

Get a reference to the stream consumer.

pub fn listen_events<A>( &self, ) -> EventStreamResult<'_, Result<KafkaEventMessage<<A as Aggregate>::Event>, Error>, Error>

where A: Aggregate, <A as Aggregate>::Event: 'static + Clone + Debug + DeserializeOwned + Send, C: 'static, R: Send + Sync,

Subscribes to topics and listens to events using the consumer, returning a stream of events.

Events are expected to be JSON thalo::event::EventEnvelopes, and will be deserialized.

For more documentation, see thalo::event_stream::EventStream.

Trait Implementations

impl<C, R: Clone> Clone for KafkaEventStream<C, R>

where C: ConsumerContext + Clone,

fn clone(&self) -> KafkaEventStream<C, R>

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<A, C, R> EventStream<A> for KafkaEventStream<C, R>

where A: 'static + Aggregate, <A as Aggregate>::Event: Clone + Debug, C: 'static + ConsumerContext, R: Send + Sync,

type Item = Result<KafkaEventMessage<<A as Aggregate>::Event>, Error>

Item yielded by stream.

type Error = Error

Error before entering the stream.

fn listen_events(&mut self) -> EventStreamResult<'_, Self::Item, Self::Error>

where <A as Aggregate>::Event: 'static + DeserializeOwned + Send,

Continuously listen for events as an async stream for the given aggregate.