Trait Processor 

pub trait Processor<T> {
    type Output;
    type Error;

    // Required methods
    fn process(&self, input: T) -> impl Future<Output = Result<(), Self::Error>>;
    fn next(&self) -> impl Future<Output = Result<Self::Output, Self::Error>>;
}

Interface for implementing data processors.

Processors take in items (input) and yield “processed” or “enriched” versions of them at some point (output). Please note that these items do not necessarily need to come out in the same order as they came in. Processors can keep items around until their internal logic decides they’re ready to go, and this can happen in any order required.

Users of processors will use the process method to insert “input” items and call next to find out if there’s ready processed “output” items to take out.

The process method might apply back-pressure, for example when the processor is busy and can’t take in more work.

Processors never terminate. The future returned by next will stay in pending state whenever there’s no work to do or the processor can’t continue because of it’s internal logic. Users can decide to drop a processor or escalate to a higher-level whenever an error occurs or the processor is no longer used.

Required Associated Types

type Output

type Error

Required Methods

fn process(&self, input: T) -> impl Future<Output = Result<(), Self::Error>>

Consumes an item for further processing.

Might apply back-pressure when being too busy.

fn next(&self) -> impl Future<Output = Result<Self::Output, Self::Error>>

Returns future with processed output whenever ready.

Please note that processors never terminate and rather return pending state (Poll::Pending) in the future when next is called and no ready items are available.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl<P1, P2, T> Processor<T> for ComposedProcessors<P1, P2>

where P1: Processor<T>, P2: Processor<P1::Output>,

type Output = <P2 as Processor<<P1 as Processor<T>>::Output>>::Output

type Error = ComposedError<<P1 as Processor<T>>::Error, <P2 as Processor<<P1 as Processor<T>>::Output>>::Error>

impl<P, T> Processor<T> for Pipeline<P>

where P: Processor<T>,

type Output = <P as Processor<T>>::Output

type Error = <P as Processor<T>>::Error

impl<S, T, L, E, TP> Processor<T> for Ingest<S, T, L, E, TP>

where S: Transaction + OperationStore<Operation<E>, Hash> + LogStore<Operation<E>, VerifyingKey, L, SeqNum, Hash> + TopicStore<TP, VerifyingKey, L>, T: Borrow<Operation<E>> + Borrow<IngestArgs<L, TP>>, L: LogId, E: Extensions,

Available on crate feature ingest only.

type Output = (T, IngestResult)

type Error = (T, IngestError)

impl<S, T, L, E> Processor<T> for LogPrune<S, T, L, E>

where S: LogStore<Operation<E>, VerifyingKey, L, SeqNum, Hash>, T: Borrow<LogPruneArgs<VerifyingKey, L, SeqNum>>, L: LogId, E: Extensions,

Available on crate feature log_prune only.

type Output = (T, LogPruneResult)

type Error = (T, LogPruneError)

impl<T, E, L, C> Processor<T> for Groups<T, E, L, C>

where T: Borrow<GroupsArgs<E>>, E: Extensions + Extension<GroupsExtensionArgs<C>> + Extension<L>, L: LogId, C: Conditions + Serialize + for<'a> Deserialize<'a>,

Available on crate feature groups only.

type Output = (T, GroupsResult)

type Error = (T, GroupsError<C>)

impl<T, ID, S> Processor<T> for Orderer<T, ID, S>

where T: Digest<ID> + Ordering<ID>, ID: OperationId, S: Transaction + OrdererStore<ID> + OperationStore<T, ID>,

Available on crate feature orderer only.

type Output = T

type Error = (Option<T>, OrdererError<T, ID, S>)

impl<T, S, F, C> Processor<T> for Spaces<T, S, F, C>

where T: Borrow<SpacesProcessorArgs<C>>, S: Clone + Transaction + SpacesStore<SpacesStoreState<C>> + SpacesMessageStore<SpacesArgs<C>> + GroupsStore<AuthMessage<C>, C> + KeyRegistryStore + KeySecretsStore, F: Forge<C>, C: Conditions,

Available on crate feature spaces only.

type Output = (T, SpacesResult<C>)

type Error = (T, ManagerError<F, C, StrongRemove<VerifyingKey, Hash, AuthMessage<C>, C>>)