Trait nakadion::handler::BatchHandler

pub trait BatchHandler: Send {
    fn handle<'a>(
        &'a mut self, 
        events: Bytes, 
        meta: BatchMeta<'a>
    ) -> BatchHandlerFuture<'a>;

    fn on_inactive(
        &mut self, 
        _inactive_for: Duration, 
        _last_activity: Instant
    ) -> InactivityAnswer { ... }
}

A handler that implements batch processing logic.

This trait will be called by Nakadion when a batch has to be processed. The BatchHandler only receives an EventType and a slice of bytes that contains the batch.

The events slice always contains a JSON encoded array of events.

Hint

The handle method gets called on &mut self.

Example

use futures::FutureExt;

use nakadion::handler::{BatchHandler, BatchPostAction, BatchMeta, Bytes, BatchHandlerFuture};
use nakadion::nakadi_types::subscription::EventTypeName;

// Use a struct to maintain state
struct MyHandler {
    pub count: i32,
}

// Implement the processing logic by implementing `BatchHandler`
impl BatchHandler for MyHandler {
    fn handle(&mut self, _events: Bytes, _meta: BatchMeta) -> BatchHandlerFuture {
        async move {
            self.count += 1;
            BatchPostAction::commit_no_stats()
        }.boxed()
    }
}

Required methods

fn handle<'a>(
    &'a mut self,
    events: Bytes,
    meta: BatchMeta<'a>
) -> BatchHandlerFuture<'a>

Handle a batch of bytes

Loading content...

Provided methods

fn on_inactive(
    &mut self,
    _inactive_for: Duration,
    _last_activity: Instant
) -> InactivityAnswer

Periodically called if there were no events for a given time.

This method will only be called if the parameter handler_inactivity_timeout_secs was set for the Consumer

Loading content...

Implementors

impl<F> BatchHandler for HandlerFn<F> where
    F: for<'a> FnMut(Bytes, BatchMeta<'a>) -> BatchHandlerFuture<'a> + Send, 

fn handle<'a>(
    &'a mut self,
    events: Bytes,
    meta: BatchMeta<'a>
) -> BatchHandlerFuture<'a>

impl<T> BatchHandler for T where
    T: EventsHandler + Send + 'static,
    T::Event: DeserializeOwned + Send + 'static, 

fn handle<'a>(
    &'a mut self,
    events: Bytes,
    meta: BatchMeta<'a>
) -> BatchHandlerFuture<'a>

fn on_inactive(
    &mut self,
    inactive_for: Duration,
    last_activity: Instant
) -> InactivityAnswer

Loading content...