@amqp-contract/worker

@amqp-contract/worker

Classes

MessageValidationError

Defined in: packages/core/dist/index.d.mts:83

Error thrown when message validation fails (payload or headers).

Used by both the client (publish-time payload validation) and the worker (consume-time payload and headers validation).

Param

The name of the publisher or consumer that triggered the validation

Param

The validation issues from the Standard Schema validation

Extends

Error

Constructors

Constructor

new MessageValidationError(source, issues): MessageValidationError;

Defined in: packages/core/dist/index.d.mts:86

Parameters

Parameter	Type
`source`	`string`
`issues`	`unknown`

Returns

MessageValidationError

Overrides

Error.constructor

Properties

Property	Modifier	Type	Description	Inherited from	Defined in
`cause?`	`public`	`unknown`	-	`Error.cause`	node_modules/.pnpm/typescript@5.9.3/node_modules/typescript/lib/lib.es2022.error.d.ts:26
`issues`	`readonly`	`unknown`	-	-	packages/core/dist/index.d.mts:85
`message`	`public`	`string`	-	`Error.message`	node_modules/.pnpm/typescript@5.9.3/node_modules/typescript/lib/lib.es5.d.ts:1077
`name`	`public`	`string`	-	`Error.name`	node_modules/.pnpm/typescript@5.9.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`source`	`readonly`	`string`	-	-	packages/core/dist/index.d.mts:84
`stack?`	`public`	`string`	-	`Error.stack`	node_modules/.pnpm/typescript@5.9.3/node_modules/typescript/lib/lib.es5.d.ts:1078
`stackTraceLimit`	`static`	`number`	The `Error.stackTraceLimit` property specifies the number of stack frames collected by a stack trace (whether generated by `new Error().stack` or `Error.captureStackTrace(obj)`). The default value is `10` but may be set to any valid JavaScript number. Changes will affect any stack trace captured after the value has been changed. If set to a non-number value, or set to a negative number, stack traces will not capture any frames.	`Error.stackTraceLimit`	node_modules/.pnpm/@types+node@25.3.2/node_modules/@types/node/globals.d.ts:67

Methods

captureStackTrace()

static captureStackTrace(targetObject, constructorOpt?): void;

Defined in: node_modules/.pnpm/@types+node@25.3.2/node_modules/@types/node/globals.d.ts:51

Creates a .stack property on targetObject, which when accessed returns a string representing the location in the code at which Error.captureStackTrace() was called.

const myObject = {};
Error.captureStackTrace(myObject);
myObject.stack;  // Similar to `new Error().stack`

The first line of the trace will be prefixed with ${myObject.name}: ${myObject.message}.

The optional constructorOpt argument accepts a function. If given, all frames above constructorOpt, including constructorOpt, will be omitted from the generated stack trace.

The constructorOpt argument is useful for hiding implementation details of error generation from the user. For instance:

function a() {
  b();
}

function b() {
  c();
}

function c() {
  // Create an error without stack trace to avoid calculating the stack trace twice.
  const { stackTraceLimit } = Error;
  Error.stackTraceLimit = 0;
  const error = new Error();
  Error.stackTraceLimit = stackTraceLimit;

  // Capture the stack trace above function b
  Error.captureStackTrace(error, b); // Neither function c, nor b is included in the stack trace
  throw error;
}

a();

Parameters

Parameter	Type
`targetObject`	`object`
`constructorOpt?`	`Function`

Returns

void

Inherited from

Error.captureStackTrace

prepareStackTrace()

static prepareStackTrace(err, stackTraces): any;

Defined in: node_modules/.pnpm/@types+node@25.3.2/node_modules/@types/node/globals.d.ts:55

Parameters

Parameter	Type
`err`	`Error`
`stackTraces`	`CallSite`[]

Returns

any

See

https://v8.dev/docs/stack-trace-api#customizing-stack-traces

Inherited from

Error.prepareStackTrace

NonRetryableError

Defined in: packages/worker/src/errors.ts:34

Non-retryable errors - permanent failures that should not be retried Examples: invalid data, business rule violations, permanent external failures

Use this error type when retrying would not help - the message will be immediately sent to the dead letter queue (DLQ) if configured.

Extends

Error

Constructors

Constructor

new NonRetryableError(message, cause?): NonRetryableError;

Defined in: packages/worker/src/errors.ts:35

Parameters

Parameter	Type
`message`	`string`
`cause?`	`unknown`

Returns

NonRetryableError

Overrides

Error.constructor

Properties

Property	Modifier	Type	Description	Inherited from	Defined in
`cause?`	`readonly`	`unknown`	-	`Error.cause`	packages/worker/src/errors.ts:37
`message`	`public`	`string`	-	`Error.message`	node_modules/.pnpm/typescript@5.9.3/node_modules/typescript/lib/lib.es5.d.ts:1077
`name`	`public`	`string`	-	`Error.name`	node_modules/.pnpm/typescript@5.9.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`stack?`	`public`	`string`	-	`Error.stack`	node_modules/.pnpm/typescript@5.9.3/node_modules/typescript/lib/lib.es5.d.ts:1078
`stackTraceLimit`	`static`	`number`	The `Error.stackTraceLimit` property specifies the number of stack frames collected by a stack trace (whether generated by `new Error().stack` or `Error.captureStackTrace(obj)`). The default value is `10` but may be set to any valid JavaScript number. Changes will affect any stack trace captured after the value has been changed. If set to a non-number value, or set to a negative number, stack traces will not capture any frames.	`Error.stackTraceLimit`	node_modules/.pnpm/@types+node@25.3.2/node_modules/@types/node/globals.d.ts:67

Methods

captureStackTrace()

static captureStackTrace(targetObject, constructorOpt?): void;

Defined in: node_modules/.pnpm/@types+node@25.3.2/node_modules/@types/node/globals.d.ts:51

Creates a .stack property on targetObject, which when accessed returns a string representing the location in the code at which Error.captureStackTrace() was called.

const myObject = {};
Error.captureStackTrace(myObject);
myObject.stack;  // Similar to `new Error().stack`

The first line of the trace will be prefixed with ${myObject.name}: ${myObject.message}.

The optional constructorOpt argument accepts a function. If given, all frames above constructorOpt, including constructorOpt, will be omitted from the generated stack trace.

The constructorOpt argument is useful for hiding implementation details of error generation from the user. For instance:

function a() {
  b();
}

function b() {
  c();
}

function c() {
  // Create an error without stack trace to avoid calculating the stack trace twice.
  const { stackTraceLimit } = Error;
  Error.stackTraceLimit = 0;
  const error = new Error();
  Error.stackTraceLimit = stackTraceLimit;

  // Capture the stack trace above function b
  Error.captureStackTrace(error, b); // Neither function c, nor b is included in the stack trace
  throw error;
}

a();

Parameters

Parameter	Type
`targetObject`	`object`
`constructorOpt?`	`Function`

Returns

void

Inherited from

Error.captureStackTrace

prepareStackTrace()

static prepareStackTrace(err, stackTraces): any;

Defined in: node_modules/.pnpm/@types+node@25.3.2/node_modules/@types/node/globals.d.ts:55

Parameters

Parameter	Type
`err`	`Error`
`stackTraces`	`CallSite`[]

Returns

any

See

https://v8.dev/docs/stack-trace-api#customizing-stack-traces

Inherited from

Error.prepareStackTrace

RetryableError

Defined in: packages/worker/src/errors.ts:10

Retryable errors - transient failures that may succeed on retry Examples: network timeouts, rate limiting, temporary service unavailability

Use this error type when the operation might succeed if retried. The worker will apply exponential backoff and retry the message.

Extends

Error

Constructors

Constructor

new RetryableError(message, cause?): RetryableError;

Defined in: packages/worker/src/errors.ts:11

Parameters

Parameter	Type
`message`	`string`
`cause?`	`unknown`

Returns

RetryableError

Overrides

Error.constructor

Properties

Property	Modifier	Type	Description	Inherited from	Defined in
`cause?`	`readonly`	`unknown`	-	`Error.cause`	packages/worker/src/errors.ts:13
`message`	`public`	`string`	-	`Error.message`	node_modules/.pnpm/typescript@5.9.3/node_modules/typescript/lib/lib.es5.d.ts:1077
`name`	`public`	`string`	-	`Error.name`	node_modules/.pnpm/typescript@5.9.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`stack?`	`public`	`string`	-	`Error.stack`	node_modules/.pnpm/typescript@5.9.3/node_modules/typescript/lib/lib.es5.d.ts:1078
`stackTraceLimit`	`static`	`number`	The `Error.stackTraceLimit` property specifies the number of stack frames collected by a stack trace (whether generated by `new Error().stack` or `Error.captureStackTrace(obj)`). The default value is `10` but may be set to any valid JavaScript number. Changes will affect any stack trace captured after the value has been changed. If set to a non-number value, or set to a negative number, stack traces will not capture any frames.	`Error.stackTraceLimit`	node_modules/.pnpm/@types+node@25.3.2/node_modules/@types/node/globals.d.ts:67

Methods

captureStackTrace()

static captureStackTrace(targetObject, constructorOpt?): void;

Defined in: node_modules/.pnpm/@types+node@25.3.2/node_modules/@types/node/globals.d.ts:51

Creates a .stack property on targetObject, which when accessed returns a string representing the location in the code at which Error.captureStackTrace() was called.

const myObject = {};
Error.captureStackTrace(myObject);
myObject.stack;  // Similar to `new Error().stack`

The first line of the trace will be prefixed with ${myObject.name}: ${myObject.message}.

The optional constructorOpt argument accepts a function. If given, all frames above constructorOpt, including constructorOpt, will be omitted from the generated stack trace.

The constructorOpt argument is useful for hiding implementation details of error generation from the user. For instance:

function a() {
  b();
}

function b() {
  c();
}

function c() {
  // Create an error without stack trace to avoid calculating the stack trace twice.
  const { stackTraceLimit } = Error;
  Error.stackTraceLimit = 0;
  const error = new Error();
  Error.stackTraceLimit = stackTraceLimit;

  // Capture the stack trace above function b
  Error.captureStackTrace(error, b); // Neither function c, nor b is included in the stack trace
  throw error;
}

a();

Parameters

Parameter	Type
`targetObject`	`object`
`constructorOpt?`	`Function`

Returns

void

Inherited from

Error.captureStackTrace

prepareStackTrace()

static prepareStackTrace(err, stackTraces): any;

Defined in: node_modules/.pnpm/@types+node@25.3.2/node_modules/@types/node/globals.d.ts:55

Parameters

Parameter	Type
`err`	`Error`
`stackTraces`	`CallSite`[]

Returns

any

See

https://v8.dev/docs/stack-trace-api#customizing-stack-traces

Inherited from

Error.prepareStackTrace

TypedAmqpWorker

Defined in: packages/worker/src/worker.ts:158

Type-safe AMQP worker for consuming messages from RabbitMQ.

This class provides automatic message validation, connection management, and error handling for consuming messages based on a contract definition.

Example

typescript

import { TypedAmqpWorker } from '@amqp-contract/worker';
import { defineQueue, defineMessage, defineContract, defineConsumer } from '@amqp-contract/contract';
import { z } from 'zod';

const orderQueue = defineQueue('order-processing', { durable: true });
const orderMessage = defineMessage(z.object({
  orderId: z.string(),
  amount: z.number()
}));

const contract = defineContract({
  consumers: {
    processOrder: defineConsumer(orderQueue, orderMessage)
  }
});

const worker = await TypedAmqpWorker.create({
  contract,
  handlers: {
    processOrder: async (message) => {
      console.log('Processing order', message.orderId);
      // Process the order...
    }
  },
  urls: ['amqp://localhost']
}).resultToPromise();

// Close when done
await worker.close().resultToPromise();

Type Parameters

Type Parameter	Description
`TContract` extends `ContractDefinition`	The contract definition type

Methods

close()

close(): Future<Result<void, TechnicalError>>;

Defined in: packages/worker/src/worker.ts:278

Close the AMQP channel and connection.

This gracefully closes the connection to the AMQP broker, stopping all message consumption and cleaning up resources.

Returns

Future<Result<void, TechnicalError>>

A Future that resolves to a Result indicating success or failure

Example

typescript

const closeResult = await worker.close().resultToPromise();
if (closeResult.isOk()) {
  console.log('Worker closed successfully');
}

create()

static create<TContract>(options): Future<Result<TypedAmqpWorker<TContract>, TechnicalError>>;

Defined in: packages/worker/src/worker.ts:235

Create a type-safe AMQP worker from a contract.

Connection management (including automatic reconnection) is handled internally by amqp-connection-manager via the AmqpClient. The worker will set up consumers for all contract-defined handlers asynchronously in the background once the underlying connection and channels are ready.

Connections are automatically shared across clients and workers with the same URLs and connection options, following RabbitMQ best practices.

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`

Parameters

Parameter	Type	Description
`options`	`CreateWorkerOptions`<`TContract`>	Configuration options for the worker

Returns

Future<Result<TypedAmqpWorker<TContract>, TechnicalError>>

A Future that resolves to a Result containing the worker or an error

Example

typescript

const worker = await TypedAmqpWorker.create({
  contract: myContract,
  handlers: {
    processOrder: async ({ payload }) => console.log('Order:', payload.orderId)
  },
  urls: ['amqp://localhost']
}).resultToPromise();

Type Aliases

CreateWorkerOptions

type CreateWorkerOptions<TContract> = object;

Defined in: packages/worker/src/worker.ts:94

Options for creating a type-safe AMQP worker.

Example

typescript

const options: CreateWorkerOptions<typeof contract> = {
  contract: myContract,
  handlers: {
    // Simple handler
    processOrder: ({ payload }) => {
      console.log('Processing order:', payload.orderId);
      return Future.value(Result.Ok(undefined));
    },
    // Handler with prefetch configuration
    processPayment: [
      ({ payload }) => {
        console.log('Processing payment:', payload.paymentId);
        return Future.value(Result.Ok(undefined));
      },
      { prefetch: 10 }
    ]
  },
  urls: ['amqp://localhost'],
  connectionOptions: {
    heartbeatIntervalInSeconds: 30
  },
  logger: myLogger
};

Note: Retry configuration is defined at the queue level in the contract, not at the handler level. See QueueDefinition.retry for configuration options.

Type Parameters

Type Parameter	Description
`TContract` extends `ContractDefinition`	The contract definition type

Properties

Property	Type	Description	Defined in
`connectionOptions?`	`AmqpConnectionManagerOptions`	Optional connection configuration (heartbeat, reconnect settings, etc.)	packages/worker/src/worker.ts:106
`contract`	`TContract`	The AMQP contract definition specifying consumers and their message schemas	packages/worker/src/worker.ts:96
`handlers`	`WorkerInferConsumerHandlers`<`TContract`>	Handlers for each consumer defined in the contract. Handlers must return `Future<Result<void, HandlerError>>` for explicit error handling. Use defineHandler() to create handlers.	packages/worker/src/worker.ts:102
`logger?`	`Logger`	Optional logger for logging message consumption and errors	packages/worker/src/worker.ts:108
`telemetry?`	`TelemetryProvider`	Optional telemetry provider for tracing and metrics. If not provided, uses the default provider which attempts to load OpenTelemetry. OpenTelemetry instrumentation is automatically enabled if @opentelemetry/api is installed.	packages/worker/src/worker.ts:114
`urls`	`ConnectionUrl`[]	AMQP broker URL(s). Multiple URLs provide failover support	packages/worker/src/worker.ts:104

HandlerError

type HandlerError = 
  | RetryableError
  | NonRetryableError;

Defined in: packages/worker/src/errors.ts:55

Union type representing all handler errors. Use this type when defining handlers that explicitly signal error outcomes.

WorkerConsumedMessage

type WorkerConsumedMessage<TPayload, THeaders> = object;

Defined in: packages/worker/src/types.ts:104

A consumed message containing parsed payload and headers.

This type represents the first argument passed to consumer handlers. It contains the validated payload and (if defined in the message schema) the validated headers.

Example

typescript

// Handler receives the consumed message with typed payload and headers
const handler = defineHandler(contract, 'processOrder', (message, rawMessage) => {
  console.log(message.payload.orderId);  // Typed payload
  console.log(message.headers?.priority); // Typed headers (if defined)
  console.log(rawMessage.fields.deliveryTag); // Raw AMQP message
  return Future.value(Result.Ok(undefined));
});

Type Parameters

Type Parameter	Default type	Description
`TPayload`	-	The inferred payload type from the message schema
`THeaders`	`undefined`	The inferred headers type from the message schema (undefined if not defined)

Properties

Property	Type	Description	Defined in
`headers`	`THeaders` extends `undefined` ? `undefined` : `THeaders`	The validated message headers (present only when headers schema is defined)	packages/worker/src/types.ts:108
`payload`	`TPayload`	The validated message payload	packages/worker/src/types.ts:106

WorkerInferConsumedMessage

type WorkerInferConsumedMessage<TContract, TName> = WorkerConsumedMessage<WorkerInferConsumerPayload<TContract, TName>, WorkerInferConsumerHeaders<TContract, TName>>;

Defined in: packages/worker/src/types.ts:115

Infer the full consumed message type for a specific consumer. Includes both payload and headers (if defined).

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferConsumerNames`<`TContract`>

WorkerInferConsumerHandler()

type WorkerInferConsumerHandler<TContract, TName> = (message, rawMessage) => Future<Result<void, HandlerError>>;

Defined in: packages/worker/src/types.ts:154

Consumer handler type for a specific consumer. Returns a Future<Result<void, HandlerError>> for explicit error handling.

Error handling:

RetryableError: Message will be retried with exponential backoff
NonRetryableError: Message will be immediately sent to DLQ

The WorkerInfer* naming pattern indicates type inference helpers that extract types from a contract definition at compile time.

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferConsumerNames`<`TContract`>

Parameters

Parameter	Type	Description
`message`	`WorkerInferConsumedMessage`<`TContract`, `TName`>	The parsed message containing validated payload and headers
`rawMessage`	`ConsumeMessage`	The raw AMQP message with all metadata (fields, properties, content)

Returns

Future<Result<void, HandlerError>>

Example

typescript

const handler: WorkerInferConsumerHandler<typeof contract, 'processOrder'> =
  ({ payload }, rawMessage) => {
    console.log(payload.orderId);  // Typed payload
    console.log(rawMessage.fields.deliveryTag); // Raw AMQP message
    return Future.value(Result.Ok(undefined));
  };

WorkerInferConsumerHandlerEntry

type WorkerInferConsumerHandlerEntry<TContract, TName> = 
  | WorkerInferConsumerHandler<TContract, TName>
  | readonly [WorkerInferConsumerHandler<TContract, TName>, {
  prefetch?: number;
}];

Defined in: packages/worker/src/types.ts:172

Handler entry for a consumer - either a function or a tuple of [handler, options].

Two patterns are supported:

Simple handler: ({ payload }, rawMessage) => Future.value(Result.Ok(undefined))
Handler with prefetch: [({ payload }, rawMessage) => ..., { prefetch: 10 }]

Note: Retry configuration is now defined at the queue level in the contract, not at the handler level. See QueueDefinition.retry for configuration options.

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferConsumerNames`<`TContract`>

WorkerInferConsumerHandlers

type WorkerInferConsumerHandlers<TContract> = { [K in InferConsumerNames<TContract>]: WorkerInferConsumerHandlerEntry<TContract, K> };

Defined in: packages/worker/src/types.ts:193

Consumer handlers for a contract. All handlers return Future<Result<void, HandlerError>> for explicit error control.

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`

Example

typescript

const handlers: WorkerInferConsumerHandlers<typeof contract> = {
  processOrder: ({ payload }) =>
    Future.fromPromise(processPayment(payload))
      .mapOk(() => undefined)
      .mapError((error) => new RetryableError('Payment failed', error)),
};

WorkerInferConsumerHeaders

type WorkerInferConsumerHeaders<TContract, TName> = ConsumerInferHeadersInput<InferConsumer<TContract, TName>>;

Defined in: packages/worker/src/types.ts:79

Infer the headers type for a specific consumer Returns undefined if no headers schema is defined

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferConsumerNames`<`TContract`>

Functions

defineHandler()

Call Signature

function defineHandler<TContract, TName>(
   contract, 
   consumerName, 
   handler): WorkerInferConsumerHandlerEntry<TContract, TName>;

Defined in: packages/worker/src/handlers.ts:103

Define a type-safe handler for a specific consumer in a contract.

Recommended: This function creates handlers that return Future<Result<void, HandlerError>>, providing explicit error handling and better control over retry behavior.

Supports two patterns:

Simple handler: just the function
Handler with options: [handler, { prefetch: 10 }]

Type Parameters

Type Parameter	Description
`TContract` extends `ContractDefinition`	The contract definition type
`TName` extends `string` \| `number` \| `symbol`	The consumer name from the contract

Parameters

Parameter	Type	Description
`contract`	`TContract`	The contract definition containing the consumer
`consumerName`	`TName`	The name of the consumer from the contract
`handler`	`WorkerInferConsumerHandler`<`TContract`, `TName`>	The handler function that returns `Future<Result<void, HandlerError>>`

Returns

WorkerInferConsumerHandlerEntry<TContract, TName>

A type-safe handler that can be used with TypedAmqpWorker

Example

typescript

import { defineHandler, RetryableError, NonRetryableError } from '@amqp-contract/worker';
import { Future, Result } from '@swan-io/boxed';
import { orderContract } from './contract';

// Simple handler with explicit error handling using mapError
const processOrderHandler = defineHandler(
  orderContract,
  'processOrder',
  ({ payload }) =>
    Future.fromPromise(processPayment(payload))
      .mapOk(() => undefined)
      .mapError((error) => new RetryableError('Payment failed', error))
);

// Handler with validation (non-retryable error)
const validateOrderHandler = defineHandler(
  orderContract,
  'validateOrder',
  ({ payload }) => {
    if (payload.amount < 1) {
      // Won't be retried - goes directly to DLQ
      return Future.value(Result.Error(new NonRetryableError('Invalid order amount')));
    }
    return Future.value(Result.Ok(undefined));
  }
);

Call Signature

function defineHandler<TContract, TName>(
   contract, 
   consumerName, 
   handler, 
   options): WorkerInferConsumerHandlerEntry<TContract, TName>;

Defined in: packages/worker/src/handlers.ts:111

Define a type-safe handler for a specific consumer in a contract.

Recommended: This function creates handlers that return Future<Result<void, HandlerError>>, providing explicit error handling and better control over retry behavior.

Supports two patterns:

Simple handler: just the function
Handler with options: [handler, { prefetch: 10 }]

Type Parameters

Type Parameter	Description
`TContract` extends `ContractDefinition`	The contract definition type
`TName` extends `string` \| `number` \| `symbol`	The consumer name from the contract

Parameters

Parameter	Type	Description
`contract`	`TContract`	The contract definition containing the consumer
`consumerName`	`TName`	The name of the consumer from the contract
`handler`	`WorkerInferConsumerHandler`<`TContract`, `TName`>	The handler function that returns `Future<Result<void, HandlerError>>`
`options`	{ `prefetch?`: `number`; }	Optional consumer options (prefetch)
`options.prefetch?`	`number`	-

Returns

WorkerInferConsumerHandlerEntry<TContract, TName>

A type-safe handler that can be used with TypedAmqpWorker

Example

typescript

import { defineHandler, RetryableError, NonRetryableError } from '@amqp-contract/worker';
import { Future, Result } from '@swan-io/boxed';
import { orderContract } from './contract';

// Simple handler with explicit error handling using mapError
const processOrderHandler = defineHandler(
  orderContract,
  'processOrder',
  ({ payload }) =>
    Future.fromPromise(processPayment(payload))
      .mapOk(() => undefined)
      .mapError((error) => new RetryableError('Payment failed', error))
);

// Handler with validation (non-retryable error)
const validateOrderHandler = defineHandler(
  orderContract,
  'validateOrder',
  ({ payload }) => {
    if (payload.amount < 1) {
      // Won't be retried - goes directly to DLQ
      return Future.value(Result.Error(new NonRetryableError('Invalid order amount')));
    }
    return Future.value(Result.Ok(undefined));
  }
);

defineHandlers()

function defineHandlers<TContract>(contract, handlers): WorkerInferConsumerHandlers<TContract>;

Defined in: packages/worker/src/handlers.ts:166

Define multiple type-safe handlers for consumers in a contract.

Recommended: This function creates handlers that return Future<Result<void, HandlerError>>, providing explicit error handling and better control over retry behavior.

Type Parameters

Type Parameter	Description
`TContract` extends `ContractDefinition`	The contract definition type

Parameters

Parameter	Type	Description
`contract`	`TContract`	The contract definition containing the consumers
`handlers`	`WorkerInferConsumerHandlers`<`TContract`>	An object with handler functions for each consumer

Returns

WorkerInferConsumerHandlers<TContract>

A type-safe handlers object that can be used with TypedAmqpWorker

Example

typescript

import { defineHandlers, RetryableError } from '@amqp-contract/worker';
import { Future } from '@swan-io/boxed';
import { orderContract } from './contract';

const handlers = defineHandlers(orderContract, {
  processOrder: ({ payload }) =>
    Future.fromPromise(processPayment(payload))
      .mapOk(() => undefined)
      .mapError((error) => new RetryableError('Payment failed', error)),
  notifyOrder: ({ payload }) =>
    Future.fromPromise(sendNotification(payload))
      .mapOk(() => undefined)
      .mapError((error) => new RetryableError('Notification failed', error)),
});

isHandlerError()

function isHandlerError(error): error is HandlerError;

Defined in: packages/worker/src/errors.ts:131

Type guard to check if an error is any HandlerError (RetryableError or NonRetryableError).

Parameters

Parameter	Type	Description
`error`	`unknown`	The error to check

Returns

error is HandlerError

True if the error is a HandlerError

Example

typescript

import { isHandlerError } from '@amqp-contract/worker';

function handleError(error: unknown) {
  if (isHandlerError(error)) {
    // error is RetryableError | NonRetryableError
    console.log('Handler error:', error.name, error.message);
  }
}

isNonRetryableError()

function isNonRetryableError(error): error is NonRetryableError;

Defined in: packages/worker/src/errors.ts:109

Type guard to check if an error is a NonRetryableError.

Use this to check error types in catch blocks or error handlers.

Parameters

Parameter	Type	Description
`error`	`unknown`	The error to check

Returns

error is NonRetryableError

True if the error is a NonRetryableError

Example

typescript

import { isNonRetryableError } from '@amqp-contract/worker';

try {
  await processMessage();
} catch (error) {
  if (isNonRetryableError(error)) {
    console.log('Will not retry:', error.message);
  }
}

isRetryableError()

function isRetryableError(error): error is RetryableError;

Defined in: packages/worker/src/errors.ts:84

Type guard to check if an error is a RetryableError.

Use this to check error types in catch blocks or error handlers.

Parameters

Parameter	Type	Description
`error`	`unknown`	The error to check

Returns

error is RetryableError

True if the error is a RetryableError

Example

typescript

import { isRetryableError } from '@amqp-contract/worker';

try {
  await processMessage();
} catch (error) {
  if (isRetryableError(error)) {
    console.log('Will retry:', error.message);
  } else {
    console.log('Permanent failure:', error);
  }
}

nonRetryable()

function nonRetryable(message, cause?): NonRetryableError;

Defined in: packages/worker/src/errors.ts:193

Create a NonRetryableError with less verbosity.

This is a shorthand factory function for creating NonRetryableError instances. Use it for cleaner error creation in handlers.

Parameters

Parameter	Type	Description
`message`	`string`	Error message describing the failure
`cause?`	`unknown`	Optional underlying error that caused this failure

Returns

NonRetryableError

A new NonRetryableError instance

Example

typescript

import { nonRetryable } from '@amqp-contract/worker';
import { Future, Result } from '@swan-io/boxed';

const handler = ({ payload }) => {
  if (!isValidPayload(payload)) {
    return Future.value(Result.Error(nonRetryable('Invalid payload format')));
  }
  return Future.value(Result.Ok(undefined));
};

// Equivalent to:
// return Future.value(Result.Error(new NonRetryableError('Invalid payload format')));

retryable()

function retryable(message, cause?): RetryableError;

Defined in: packages/worker/src/errors.ts:163

Create a RetryableError with less verbosity.

This is a shorthand factory function for creating RetryableError instances. Use it for cleaner error creation in handlers.

Parameters

Parameter	Type	Description
`message`	`string`	Error message describing the failure
`cause?`	`unknown`	Optional underlying error that caused this failure

Returns

RetryableError

A new RetryableError instance

Example

typescript

import { retryable } from '@amqp-contract/worker';
import { Future, Result } from '@swan-io/boxed';

const handler = ({ payload }) =>
  Future.fromPromise(processPayment(payload))
    .mapOk(() => undefined)
    .mapError((e) => retryable('Payment service unavailable', e));

// Equivalent to:
// .mapError((e) => new RetryableError('Payment service unavailable', e));

@amqp-contract/worker ​

Classes ​

MessageValidationError ​

Param ​

Param ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

Methods ​

captureStackTrace() ​

Parameters ​

Returns ​

Inherited from ​

prepareStackTrace() ​

Parameters ​

Returns ​

See ​

Inherited from ​

NonRetryableError ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

Methods ​

captureStackTrace() ​

Parameters ​

Returns ​

Inherited from ​

prepareStackTrace() ​

Parameters ​

Returns ​

See ​

Inherited from ​

RetryableError ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

Methods ​

captureStackTrace() ​

Parameters ​

Returns ​

Inherited from ​

prepareStackTrace() ​

Parameters ​

Returns ​

See ​

Inherited from ​

TypedAmqpWorker ​

Example ​

Type Parameters ​

Methods ​

close() ​

Returns ​

Example ​

create() ​

Type Parameters ​

Parameters ​

Returns ​

Example ​

Type Aliases ​

CreateWorkerOptions ​

Example ​

Type Parameters ​

Properties ​

HandlerError ​

WorkerConsumedMessage ​

Example ​

Type Parameters ​

Properties ​

@amqp-contract/worker

Classes

MessageValidationError

Param

Param

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

Methods

captureStackTrace()

Parameters

Returns

Inherited from

prepareStackTrace()

Parameters

Returns

See

Inherited from

NonRetryableError

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

Methods

captureStackTrace()

Parameters

Returns

Inherited from

prepareStackTrace()

Parameters

Returns

See

Inherited from

RetryableError

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

Methods

captureStackTrace()

Parameters

Returns

Inherited from

prepareStackTrace()

Parameters

Returns

See

Inherited from

TypedAmqpWorker

Example

Type Parameters

Methods

close()

Returns

Example

create()

Type Parameters

Parameters

Returns

Example

Type Aliases

CreateWorkerOptions

Example

Type Parameters

Properties

HandlerError

WorkerConsumedMessage

Example

Type Parameters

Properties