Async Middleware

The Async Middleware is built-in middleware that gives you capability to defer the handling to an consumer to achieve real asynchronousity. It’s only available for CommandBus and EventBus

How to use it

As for any Middleware, you can use it by adding it to the bus instance.

const commandBus = createCommandBus<CommandHandlerRegistry>();
commandBus.useAsyncMiddleware({
    consume: false, // KEY POINT
    produce: async (envelope) => {
        // use your favorite queue system here
        console.log('Generic Push to Queue', envelope);
    },
    async: true,// default is true
    intents: {
        createUser: {
            async: true,
            produce: async (envelope) => {
                // use your favorite queue system here
                console.log('createUser Push to Queue', envelope);
            },
        },
    },
});

Remember built-in middlewares are intent aware, therefore you can customize the behavior per intent using the key intents.

Next, you need to have a consumer that will consume it. The way to do that with Missive.js is to create another bus with this middlware with consume: true.

commandBus.useAsyncMiddleware({
    consume: true, // KEY POINT
});

The worker script that consumes the queue can dispatch the message it receives directly to the dispatch method:

// Consumer script
onMessage: async (message) => {
    const envelope = JSON.parse(message);
    await commandBus.dispatch(intent);
}

Gotchas

Generally, the dispatch receives an intent but it can also receive an envelope. In this case, the envelope is the message received from the queue.

Explanation

The flow is the following:

1. Your application (web node for instance) will have a bus on which this middleware is added with consume: false.

2. When you dispatch an intent, the middleware will push the intent to the queue system (via the produce method that you provide) instead of handling it.

3. You have another application (worker node for instance) that will have a bus on which this middleware is added with consume: true.

4. This worker will consume the intent from the queue system and handle it.

Important things to remember

• intent will pass through the bus twice, once for the produce and once for the consume, so make sure your middleware have no side effects.
• Async Middleware breaks sthe chain of middlewares and for this reason, it is usually register the last.

Gotchas

When you dispatch an envelope to the bus, the bus will save the envelope original stamps in the ReprocessedStamp stamp. This way, the bus that consumes the envelope can have access to the original stamps.

Added Stamps

The Async Middleware is going to add:

• type AsyncStamp = Stamp<undefined, 'missive:async'>;

  When the intent is pushed to the queue.

• type ReprocessedStamp = Stamp<{ stamps: Stamp[] }, 'missive:reprocessed'>;

  When the envelope is dispatched.

Going further

Look at the code of the Async Middleware