Middleware and app context

The application container and the request context

A worker is built once with your application's container (database pools, clients, configuration) and hands it to every job, workflow step and compensation it runs. The handler also gets the request context the producer attached (tenant_id, request_id, traceparent, ...). Work a handler starts through its context (ctx.enqueue*, ctx.start*) carries that context on.

Language	Worker option	In a handler
Python	JoobQWorker(..., app_context=app)	ctx.app, ctx.context
TypeScript	new JoobQWorker({ ..., app })	ctx.app, ctx.context
Ruby	JoobQ::Worker.new(..., app: app)	ctx.app, ctx.context
Go	joobq.NewWorker(client, id, queues, app)	ec.App, ec.Meta
Rust	Worker::new(client, id, queues, app)	ctx.app, ctx.context
Java	new JoobQWorker<>(client, id, queues, app)	ctx.app(), ctx.context()
PHP	new JoobQWorker($client, $id, $queues, $app)	$ctx->app, $ctx->context
.NET	new JoobQWorker<TApp>(client, id, queues, app)	ctx.App, ctx.Context

A Crystal application holds its container where its classes reach it: JoobQ builds jobs and activities itself, so there is no ctx.app. PHP forks its worker processes: each has its own copy of the app (appFactory builds one per process).

Crystal

struct ConfJob
  include JoobQ::Job

  @queue = QUEUE
  @retries = 3
  @max_retries = 3
  @expires = 1.day.from_now.to_unix_ms

Python

@job("conf.context", queue=QUEUE)
def context_job(payload, ctx):
    report(ctx, "context", payload, app_marker=ctx.app.marker, tenant=ctx.context.get("tenant_id"), x=payload.get("x"))

TypeScript

defineJob<App>({ name: 'conf.context', queue: QUEUE, perform: async (p, ctx) =>
  report(ctx, 'context', p, { app_marker: ctx.app.marker, tenant: ctx.context.tenant_id, x: p.x }) }),

Ruby

class ContextJob
  include JoobQ::Job
  job_name "conf.context"
  queue QUEUE
  def self.perform(p, ctx) = report(ctx, "context", p, app_marker: ctx.app.marker, tenant: ctx.context["tenant_id"], x: p["x"])
end

Go

joobq.DefineJob[*App]("conf.context", opts, func(p map[string]any, ec *EC) (any, error) {
	return report(ec, "context", p, map[string]any{"app_marker": ec.App.Marker, "tenant": ec.Meta["tenant_id"], "x": p["x"]})
}),

Rust

JobDef::new("conf.context", queue.clone(), |p: Value, ctx: &Ctx| {
    report(ctx, "context", &p, json!({"app_marker": ctx.app.marker, "tenant": ctx.context.get("tenant_id"), "x": p["x"]}))
}),

Java

JobDefinition.<App>of("conf.context", QUEUE, 3, (p, ctx) ->
    report(ctx, "context", p, map("app_marker", ctx.app().marker, "tenant", ctx.context().get("tenant_id"), "x", m(p).get("x")))),

PHP

final class ContextJob extends ConfJob
{
    public static function jobName(): string { return 'conf.context'; }
    public static function perform(array $p, ExecutionContext $ctx): mixed
    {
        return report($ctx, 'context', $p, ['app_marker' => $ctx->app->marker, 'tenant' => $ctx->context['tenant_id'] ?? null, 'x' => $p['x'] ?? null]);
    }
}

.NET

JobDefinition<App>.Define("conf.context", Queue, 3, (p, ctx) =>
    Report(ctx, "context", p, new() { ["app_marker"] = ctx.App.Marker, ["tenant"] = ctx.Context.GetValueOrDefault("tenant_id")?.GetValue<string>(), ["x"] = Num(p?["x"]) })),

Tested against a real server in every language shown.

Worker middleware

worker.use(mw) (use_middleware in Rust, where use is a keyword) adds a middleware and returns the worker, so calls chain. A middleware is mw(ctx, next):

• the first one registered is the outermost; the task's handler is the innermost call;
• it wraps one task: a job, or one workflow step (kind activity, compensation or update): the engine conducts the run and sends each step as a task of its own;
• it may run code around next, change the output next returns, or not call next at all: what it returns is acknowledged as the task's output;
• it may catch the handler's errors, and what it returns then is acknowledged as the output.

Python

def m1(ctx, nxt):
    """Outer middleware: runs first, reports what ran inside it and the output it got back."""
    if ctx.task.name not in MW_TASKS:
        return nxt()
    ctx.app.trace(ctx.task.id, "m1")
    out = nxt()
    report(ctx, "mw", ctx.task.payload, trace=ctx.app.traces[ctx.task.id], output=out)
    return out


def m2(ctx, nxt):
    """Inner middleware: wraps conf.mw's output; short-circuits conf.mw_skip without calling next."""
    if ctx.task.name not in MW_TASKS:
        return nxt()
    ctx.app.trace(ctx.task.id, "m2")
    if ctx.task.name == "conf.mw_skip":
        return {"skipped": True}
    return {"wrapped": nxt()}

TypeScript

// Outer middleware: runs first, reports what ran inside it and the output it got back.
const m1: TaskMiddleware<App> = async (ctx, next) => {
  if (!MW_TASKS.includes(ctx.task.name)) return next()
  ctx.app.trace(ctx.task.id, 'm1')
  const out = await next()
  await report(ctx, 'mw', ctx.task.payload, { trace: ctx.app.traceOf(ctx.task.id), output: out })
  return out
}
// Inner middleware: wraps conf.mw's output; short-circuits conf.mw_skip without calling next.
const m2: TaskMiddleware<App> = async (ctx, next) => {
  if (!MW_TASKS.includes(ctx.task.name)) return next()
  ctx.app.trace(ctx.task.id, 'm2')
  if (ctx.task.name === 'conf.mw_skip') return { skipped: true }
  return { wrapped: await next() }
}

Ruby

# Outer middleware: runs first, reports what ran inside it and the output it got back.
M1 = lambda do |ctx, nxt|
  next nxt.call unless MW_TASKS.include?(ctx.task["name"])

  ctx.app.trace(ctx.task["id"], "m1")
  out = nxt.call
  report(ctx, "mw", ctx.task["payload"], trace: ctx.app.traces(ctx.task["id"]), output: out)
  out
end

# Inner middleware: wraps conf.mw's output; short-circuits conf.mw_skip without calling next.
M2 = lambda do |ctx, nxt|
  next nxt.call unless MW_TASKS.include?(ctx.task["name"])

  ctx.app.trace(ctx.task["id"], "m2")
  next({ skipped: true }) if ctx.task["name"] == "conf.mw_skip"

  { wrapped: nxt.call }
end

Go

// m1 is the outer middleware: it runs first and reports what ran inside it and the output it got back.
func m1(ec *EC, next func() (any, error)) (any, error) {
	if !mwTask(ec) {
		return next()
	}
	ec.App.Trace(ec.Task.ID, "m1")
	out, err := next()
	if err != nil {
		return out, err
	}
	if _, rerr := report(ec, "mw", taskPayload(ec), map[string]any{"trace": ec.App.Traces(ec.Task.ID), "output": out}); rerr != nil {
		return nil, rerr
	}
	return out, nil
}

// m2 is the inner middleware: it wraps conf.mw's output and short-circuits conf.mw_skip without calling next.
func m2(ec *EC, next func() (any, error)) (any, error) {
	if !mwTask(ec) {
		return next()
	}
	ec.App.Trace(ec.Task.ID, "m2")
	if ec.Task.Name == "conf.mw_skip" {
		return map[string]any{"skipped": true}, nil
	}
	out, err := next()
	if err != nil {
		return out, err
	}
	return map[string]any{"wrapped": out}, nil
}

Rust

/// Outer middleware: runs first, reports what ran inside it and the output it got back.
fn m1(ctx: &Ctx, next: &dyn Fn() -> joobq::Result<Value>) -> joobq::Result<Value> {
    if !MW_TASKS.contains(&ctx.task.name.as_str()) {
        return next();
    }
    ctx.app.trace(&ctx.task.id, "m1");
    let out = next()?;
    report(ctx, "mw", &ctx.task.payload, json!({"trace": ctx.app.traced(&ctx.task.id), "output": out}))?;
    Ok(out)
}

/// Inner middleware: wraps conf.mw's output; short-circuits conf.mw_skip without calling next.
fn m2(ctx: &Ctx, next: &dyn Fn() -> joobq::Result<Value>) -> joobq::Result<Value> {
    if !MW_TASKS.contains(&ctx.task.name.as_str()) {
        return next();
    }
    ctx.app.trace(&ctx.task.id, "m2");
    if ctx.task.name == "conf.mw_skip" {
        return Ok(json!({"skipped": true}));
    }
    Ok(json!({"wrapped": next()?}))
}

Java

/** Outer middleware: runs first, reports what ran inside it and the output it got back. */
static final Middleware<App> M1 = (ctx, next) -> {
    if (!MW_TASKS.contains(ctx.getTask().name())) return next.call();
    String id = ctx.getTask().id();
    ctx.app().trace(id, "m1");
    Object out = next.call();
    report(ctx, "mw", ctx.getTask().payload(), map("trace", ctx.app().traceOf(id), "output", out));
    return out;
};

/** Inner middleware: wraps conf.mw's output; short-circuits conf.mw_skip without calling next. */
static final Middleware<App> M2 = (ctx, next) -> {
    if (!MW_TASKS.contains(ctx.getTask().name())) return next.call();
    ctx.app().trace(ctx.getTask().id(), "m2");
    if (ctx.getTask().name().equals("conf.mw_skip")) return map("skipped", true);
    return map("wrapped", next.call());
};

PHP

/** Outer middleware: runs first, reports what ran inside it and the output it got back. */
function m1(ExecutionContext $ctx, callable $next): mixed
{
    if (!in_array($ctx->task['name'] ?? '', MW_TASKS, true)) {
        return $next();
    }
    $id = (string)$ctx->task['id'];
    $ctx->app->trace($id, 'm1');
    $out = $next();
    report($ctx, 'mw', (array)($ctx->task['payload'] ?? []), ['trace' => $ctx->app->traces[$id], 'output' => (object)$out]);
    return $out;
}

/** Inner middleware: wraps conf.mw's output; short-circuits conf.mw_skip without calling next. */
function m2(ExecutionContext $ctx, callable $next): mixed
{
    if (!in_array($ctx->task['name'] ?? '', MW_TASKS, true)) {
        return $next();
    }
    $ctx->app->trace((string)$ctx->task['id'], 'm2');
    if ($ctx->task['name'] === 'conf.mw_skip') {
        return ['skipped' => true];
    }
    return ['wrapped' => $next()];
}

.NET

/// <summary>Outer middleware: runs first, reports what ran inside it and the output it got back.</summary>
static async Task<object?> M1(ExecutionContext<App> ctx, Func<Task<object?>> next)
{
    if (!MwTasks.Contains(ctx.Task.Name)) return await next();
    ctx.App.Trace(ctx.Task.Id, "m1");
    var output = await next();
    await Report(ctx, "mw", ctx.Task.Payload, new() { ["trace"] = ctx.App.Traces(ctx.Task.Id), ["output"] = output });
    return output;
}

/// <summary>Inner middleware: wraps conf.mw's output; short-circuits conf.mw_skip without calling next.</summary>
static async Task<object?> M2(ExecutionContext<App> ctx, Func<Task<object?>> next)
{
    if (!MwTasks.Contains(ctx.Task.Name)) return await next();
    ctx.App.Trace(ctx.Task.Id, "m2");
    if (ctx.Task.Name == "conf.mw_skip") return new Dictionary<string, object?> { ["skipped"] = true };
    return new Dictionary<string, object?> { ["wrapped"] = await next() };
}

Not in every language

• Crystal: SDK worker middleware: JoobQ Core's middleware is a different contract (it wraps each job and returns no output).

Tested against a real server in every language shown.

Every SDK is tested for the order (m1, m2, handler), the output wrapping and the short-circuit. JoobQ Core's own middleware for Crystal jobs (JoobQ::Middleware: matches? and call) is a different contract: it wraps each job and returns no output.