Idiomatic way of chaining workers

[ChristianJacobsen]

Hi,

I'm building a system where I plan on having multiple workers (with different concurrency settings) trigger each other in a chain, with each worker having multiple steps. I couldn't find an example for this, so I thought I'd ask if anyone has an idea of how to solve the queueing of tasks between workers in the most "idiomatic"/"true to the spirit of the library" way?

I know this can be solved with channels and such, but I wondered if any thought during the design of the library had gone into how these would best be shared? Would you just re-use the (Redis)Connection with multiple storages and then pass each "next storage" as a handle to the worker before it to queue up tasks? I.e.:

let conn = ...;
let storage1 = RedisStorage::new(conn.clone());
let storage2 = RedisStorage::new(conn);

let monitor = Monitor::new()
    .register(
        WorkerBuilder::new("w1")
            .data(Arc::new(Mutex::new(storage2.clone())))
            .backend(storage1.clone())
            .build_fn(logger1),
    )
    .register(
        WorkerBuilder::new("w2")
            .backend(storage2.clone())
            .build_fn(logger2),
    )
    .run()
    .await;

When trying something like this I just end up with lots and lots of duplicated task invocations.

The logger1 task here is just this:

async fn logger1(
    task: usize,
    data: Data<Arc<Mutex<RedisStorage<usize>>>>,
) -> std::result::Result<(), apalis::prelude::Error> {
    info!("L1: Task: {}", task);
    let _ = data.lock().await.push(task).await;
    Ok(())
}

For some reason this just doesn't sit right with me. I hope there's something I'm missing.

Any help or thoughts would be greatly appreciated. This library looks like it can solve the scheduling issues I'm facing, as I really need the persisted state in case of crashes.

[geofmureithi]

I think what you are looking for is piping. The concept is still in development but here is how we do it in apalis-cron.

1. Build a new type something like StorageTwoPiped:

    pub fn pipe_to_storage<S, Ctx>(self, storage: S) -> StorageTwoPiped<S>
    where
        S: Storage<Job = Req, Context = Ctx> + Clone + Send + Sync + 'static,
        S::Error: std::error::Error + Send + Sync + 'static,
    {
        let stream = self
            .into_stream()
            .then({
                let storage = storage.clone();
                move |res| {
                    let mut storage = storage.clone();
                    async move {
                        match res {
                            Ok(Some(req)) => storage
                                .push(req.args)
                                .await
                                .map(|_| ())
                                .map_err(|e| Box::new(e) as BoxDynError),
                            _ => Ok(()),
                        }
                    }
                }
            })
            .boxed();

        StorageTwoPiped {
            stream,
            inner: storage,
        }
    }

2. Implement Backend for the Piped storage:

impl<T, Ctx, Inner> Backend<Request<T, Ctx>> for StorageTwoPiped<Inner>
where
    Inner: Backend<Request<T, Ctx>>,
{
    type Stream = Inner::Stream;

    type Layer = Inner::Layer;

    type Codec = Inner::Codec;

    fn poll(mut self, worker: &Worker<Context>) -> Poller<Self::Stream, Self::Layer> {
        let pipe_heartbeat = async move { while (self.stream.next().await).is_some() {} };
        let inner = self.inner.poll(worker);
        let heartbeat = inner.heartbeat;

        Poller::new_with_layer(
            inner.stream,
            async {
                futures::join!(heartbeat, pipe_heartbeat);
            },
            inner.layer,
        )
    }
}

3. Use the new type as the backend.