Struct WorkFlow 

pub struct WorkFlow<Input, Current, FlowSink, Encode, Compact>
where
    FlowSink: TaskSink<Compact>,
{ /* private fields */ }

Implementations

impl<Input, Current, FlowSink, Encode, Compact> WorkFlow<Input, Current, FlowSink, Encode, Compact>

where Current: Send + 'static, FlowSink: Send + Clone + Sync + 'static + Unpin + TaskSink<Compact>,

pub fn delay_for<CodecError>( self, duration: Duration, ) -> WorkFlow<Input, Current, FlowSink, Encode, Compact>

where Current: Send + 'static + Sync, FlowSink::Context: Send + Sync + Default + 'static + MetadataExt<WorkflowRequest>, FlowSink::Error: Into<BoxDynError> + Send + 'static, FlowSink::IdType: Send + Default, Compact: Sync + Send + 'static, Encode: Codec<Current, Compact = Compact, Error = CodecError> + Send + Sync + 'static, 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>

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

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

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>, <TaskFn<F, Current, FlowSink::Context, FnArgs> as Service<Task<Current, FlowSink::Context, FlowSink::IdType>>>::Future: Send + 'static, Output: Send + 'static + Sync, FlowSink: Sync + Clone + Send + 'static + WaitForCompletion<Output, Compact> + Unpin, FlowSink::Error: Debug + Into<BoxDynError> + Send + Sync, SvcError: Send + Sync + 'static + Into<BoxDynError>, FlowSink::IdType: Send + Default + Sync, Encode: Codec<Current, 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> + 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()
    .filter_map(|x: i32| 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>

where Current: Send + 'static, FlowSink: Send + Clone + Sync + 'static + Unpin + TaskSink<Compact>,

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

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>, FlowSink::Error: Into<BoxDynError> + Send + '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 + Default, Compact: Sync + Send + 'static, Encode: Codec<Current, Compact = Compact, Error = CodecError> + Send + Sync + Codec<O, Compact = Compact, Error = CodecError> + 'static, CodecError: Send + Sync + Error + 'static, <FlowSink::Context as MetadataExt<WorkflowRequest>>::Error: Error + Sync + Send + 'static,

impl<Input, FlowSink, Encode, Compact> WorkFlow<Input, Input, FlowSink, Encode, Compact>

where FlowSink: TaskSink<Compact>,

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

impl<Input, Current, FlowSink, Encode, Compact> WorkFlow<Input, Current, FlowSink, Encode, Compact>

where Current: Send + 'static, FlowSink: Send + Clone + Sync + 'static + Unpin + TaskSink<Compact>,

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

where Current: Send + 'static + Sync, FlowSink::Context: Send + 'static + Sync, S: Step<Current, FlowSink, Encode, Response = Res, Error = E> + Sync + Send + 'static + Clone, S::Response: Send, S::Error: Send, Res: 'static, FlowSink::IdType: Send, Encode: Codec<Current, 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,