Pipeline configuration

Overview

The bus’s pipeline is where you compose cross-cutting behaviour — auth, logging, metrics, tenant routing, redaction — without touching individual handlers. Pipeline entries are either Filters (return Continue or Stop) or Middlewares (wrap the next step with try/finally-style logic), registered against one of four named stages via bus.use(stage, ...items).

Signature

use(stage: PipelineStage, ...items: Array<FilterRegistration | MiddlewareRegistration>): this;

• stage — one of 'outgoing' | 'beforeConsuming' | 'afterConsuming' | 'onConsumedSuccessfully'.
• items — any mix of asFilter(...) and asMiddleware(...) registrations. Returns the bus for chaining.

Stages

Stage	When it runs
'outgoing'	Before a message leaves the producer — every outgoing send path: publish, send, sendToMany, route, and all request methods (sendRequest, sendRequestMulti, publishRequest).
'beforeConsuming'	Before the handler runs.
'afterConsuming'	After the handler returns OR throws.
'onConsumedSuccessfully'	After a successful handler only.

Ordering rule

Caution

Within a stage, all filters run before all middleware regardless of the order you passed them to bus.use(...). A Stop from any filter short-circuits the stage — middleware never runs for that message.

Example

import { FilterAction, asFilter, asMiddleware } from '@serviceconnect/core';

bus.use(
  'beforeConsuming',
  asFilter((envelope) => {
    return envelope.headers.tenant ? FilterAction.Continue : FilterAction.Stop;
  }),
  asMiddleware(async (context, next) => {
    const start = Date.now();
    await next();
    metrics.observe(Date.now() - start);
  }),
);

If the tenant header is missing, the filter returns Stop and the timing middleware is skipped entirely — even though both were registered in the same use(...) call.

Composing multiple middlewares

Call bus.use(stage, asMiddleware(a), asMiddleware(b)) to register more than one middleware on the same stage. They compose via next() like Express or Koa: a runs first, calls await next() to invoke b, which calls await next() to invoke the handler (or whatever follows the stage). Unwinding happens in reverse on the way back out.

See also

• Filter
• Middleware
• PipelineStage
• Filters (pattern)