Struct Workflow 

pub struct Workflow<Input, Current, FlowSink, Encode, Compact, Context, IdType> { /* private fields */ }

Implementations

impl<Input, Current, FlowSink, Encode, Compact> Workflow<Input, Current, FlowSink, Encode, Compact, FlowSink::Context, FlowSink::IdType>

where Current: Send + 'static, FlowSink: Send + Clone + Sync + 'static + Unpin + WeakTaskSink<Current, Codec = Encode>,

pub fn delay_for<CodecError, DbError>( self, duration: Duration, ) -> Workflow<Input, Current, FlowSink, Encode, Compact, FlowSink::Context, FlowSink::IdType>

where Current: Send + 'static + Sync, FlowSink::Context: Send + Sync + Default + 'static + MetadataExt<WorkflowRequest>, FlowSink: Sink<Task<Compact, FlowSink::Context, FlowSink::IdType>, Error = DbError> + Backend<Error = DbError>, DbError: Error + Send + Sync + 'static, FlowSink::IdType: Send + GenerateId, Compact: Sync + Send + 'static, Encode: Codec<Current, Compact = Compact, Error = CodecError> + Send + Sync + 'static + Codec<GoTo<Current>, Compact = Compact, Error = CodecError>, CodecError: Send + Sync + Error + 'static, <FlowSink::Context as MetadataExt<WorkflowRequest>>::Error: Error + Sync + Send + 'static,

impl<Input, Current, FlowSink, Encode, Compact> Workflow<Input, Vec<Current>, FlowSink, Encode, Compact, FlowSink::Context, FlowSink::IdType>

where Current: Send + 'static, FlowSink: Backend, FlowSink::Context: MetadataExt<FilterContext<FlowSink::IdType>> + Send + 'static,

pub fn filter_map<F, Output, FnArgs, SvcError, MetadataError, CodecError, DbError>( self, predicate: F, ) -> Workflow<Input, Vec<Output>, FlowSink, Encode, Compact, FlowSink::Context, FlowSink::IdType>

where F: Send + 'static + Sync + Clone, TaskFn<F, Current, FlowSink::Context, FnArgs>: Service<Task<Current, FlowSink::Context, FlowSink::IdType>, Response = Option<Output>, Error = SvcError>, FnArgs: Send + 'static + Sync, Current: Send + 'static + Serialize + Sync + Debug, FlowSink::Context: Send + 'static + Sync + Default + MetadataExt<FilterContext<FlowSink::IdType>, Error = MetadataError> + MetadataExt<WorkflowRequest, Error = MetadataError> + MetadataExt<FilterState, Error = MetadataError>, <TaskFn<F, Current, FlowSink::Context, FnArgs> as Service<Task<Current, FlowSink::Context, FlowSink::IdType>>>::Future: Send + 'static, Output: Send + 'static + Sync, FlowSink: WeakTaskSink<Option<Output>, Codec = Encode, Error = DbError> + WeakTaskSink<Vec<Output>, Codec = Encode, Error = DbError> + Sink<Task<Compact, FlowSink::Context, FlowSink::IdType>, Error = DbError> + Sync + Clone + Send + 'static + WaitForCompletion<GoTo<Option<Output>>> + Unpin, DbError: Debug + Into<BoxDynError> + Send + Sync + Error + 'static, SvcError: Send + Sync + 'static + Into<BoxDynError>, FlowSink::IdType: Send + GenerateId + Sync + Display, Encode: Codec<Current, Compact = Compact, Error = CodecError> + Codec<GoTo<Option<Output>>, Compact = Compact, Error = CodecError> + Codec<Option<Output>, Compact = Compact, Error = CodecError> + Codec<Vec<Current>, Compact = Compact, Error = CodecError> + Codec<Vec<Output>, Compact = Compact, Error = CodecError> + Codec<GoTo<Vec<Output>>, Compact = Compact, Error = CodecError> + Send + Sync + 'static, Compact: Send + Sync + 'static, CodecError: Error + Sync + Send + 'static, MetadataError: Error + Sync + Send + 'static,

Adds a filter_map step to the workflow, allowing you to filter and map items in the workflow using a predicate function.

Example

use apalis_workflow::WorkFlow;
// Suppose you have a workflow of integers and want to filter even numbers and double them.
let workflow = WorkFlow::new("the-even-doubler")
    .filter_map(|x: i32| async move { if x % 2 == 0 { Some(x * 2) } else { None } });
// The resulting workflow will only contain doubled even numbers.

Returns

Returns a new WorkFlow with the filter_map step added, producing a vector of outputs.

Errors

Errors from the predicate function or encoding/decoding are propagated as boxed dynamic errors.

Notes

• The predicate function must be Send, Sync, and 'static.
• The workflow step is inserted at the end of the current steps.
• This method is intended for advanced workflow composition scenarios.

impl<Input, Current, FlowSink, Encode, Compact> Workflow<Input, Current, FlowSink, Encode, Compact, FlowSink::Context, FlowSink::IdType>

where Current: Send + 'static, FlowSink: Send + Clone + Sync + 'static + Unpin + Backend,

pub fn then<F, O, E, FnArgs, CodecError, DbError>( self, then: F, ) -> Workflow<Input, O, FlowSink, Encode, Compact, FlowSink::Context, FlowSink::IdType>

where O: Sync + Send + 'static, E: Into<BoxDynError> + Send + Sync + 'static, F: Send + 'static + Sync + Clone, TaskFn<F, Current, FlowSink::Context, FnArgs>: Service<Task<Current, FlowSink::Context, FlowSink::IdType>, Response = O, Error = E>, FnArgs: Send + 'static + Sync, Current: Send + 'static + Sync, FlowSink::Context: Send + Sync + Default + 'static + MetadataExt<WorkflowRequest>, DbError: Error + Send + Sync + 'static, <TaskFn<F, Current, FlowSink::Context, FnArgs> as Service<Task<Current, FlowSink::Context, FlowSink::IdType>>>::Future: Send + 'static, <TaskFn<F, Current, FlowSink::Context, FnArgs> as Service<Task<Current, FlowSink::Context, FlowSink::IdType>>>::Error: Into<BoxDynError>, FlowSink::IdType: Send + GenerateId, Compact: Sync + Send + 'static, Encode: Codec<Current, Compact = Compact, Error = CodecError> + Send + Sync + Codec<O, Compact = Compact, Error = CodecError> + 'static + Codec<GoTo<O>, Compact = Compact, Error = CodecError>, CodecError: Send + Sync + Error + 'static, <FlowSink::Context as MetadataExt<WorkflowRequest>>::Error: Error + Sync + Send + 'static, FlowSink: WeakTaskSink<O, Codec = Encode, Error = DbError> + Sink<Task<Compact, FlowSink::Context, FlowSink::IdType>, Error = DbError>,

impl<Input, FlowSink, Encode, Compact, Context, IdType> Workflow<Input, Input, FlowSink, Encode, Compact, Context, IdType>

pub fn new(name: &str) -> Self

impl<Input, Current, FlowSink, Encode, Compact> Workflow<Input, Current, FlowSink, Encode, Compact, FlowSink::Context, FlowSink::IdType>

where Current: Send + 'static, FlowSink: Send + Clone + Sync + 'static + Unpin + Backend,

pub fn add_step<S, Res, E, CodecError, BackendError>( self, step: S, ) -> Workflow<Input, Res, FlowSink, Encode, Compact, FlowSink::Context, FlowSink::IdType>

where FlowSink: WeakTaskSink<Res, Codec = Encode, Error = BackendError> + Sink<Task<Compact, FlowSink::Context, FlowSink::IdType>, Error = BackendError>, Current: Send + 'static + Sync, FlowSink::Context: Send + 'static + Sync + MetadataExt<WorkflowRequest>, S: Step<Current, FlowSink, Encode, Response = Res, Error = E> + Sync + Send + 'static + Clone, S::Response: Send, S::Error: Send, Res: 'static + Sync, FlowSink::IdType: Send, Encode: Codec<Current, Compact = Compact, Error = CodecError> + Codec<GoTo<Res>, Compact = Compact, Error = CodecError> + Codec<Res, Compact = Compact, Error = CodecError> + Send + Sync + 'static, Compact: Send + Sync + 'static, E: Into<BoxDynError> + Send + Sync + 'static, CodecError: Error + Send + 'static + Sync, BackendError: Error + Send + Sync + 'static,