RabbitMQMatcher

RabbitMQMatcher routes Amazon MQ for RabbitMQ messages to handlers, one message at a time.

An event carries messages grouped by queue, and every message body arrives base64 encoded. The matcher decodes each one, parses it as JSON, then works out which of your routes should handle it. Your handler deals with one message and never sees the batch.

Install

bash

npm install @latcher/base @latcher/mq

@latcher/base is a peer dependency, so install it yourself. It exports Latcher, which every matcher plugs into.

The package holds ActiveMQMatcher as well, so the functions are named per matcher. You want createRabbitMQMatcher and defineRabbitMQRoute rather than the createMatcher and defineRoute a single matcher package gives you.

Create the matcher

import { createRabbitMQMatcher } from '@latcher/mq'
import { logInvocation } from './middleware/logInvocation'

const rabbitMQMatcher = createRabbitMQMatcher({
  middleware: [logInvocation],  // Optional
})

The one option can be left out, so createRabbitMQMatcher() is what you want most of the time.

Options

Option	Type	Required	Default	Description
`middleware`	`RabbitMQMiddleware[]`	No	`[]`	Runs for every message this matcher handles, before any route middleware. See Middleware

Register routes

rabbitMQMatcher.route({
  filters: {
    queue: 'orders',
    contentType: 'application/json',
  },
  bodySchema: OrderSchema,  // Optional
  middleware: [withOrderContext],  // Optional
  handler: processOrder,
})

filters and handler are the only required keys.

route() returns the matcher, so you can chain registrations.

rabbitMQMatcher.route(processOrderRoute).route(refundOrderRoute)

Routes are ranked by how specific they are. Where two overlap, the one matching only a subset of the other is tried first, whatever order you registered them in. Registration order decides the rest, so give each route filters no other route can match. See match order for how ranking works and what it cannot settle.

A message that matches no route throws, which fails the invocation and leaves the messages behind it in the batch unhandled. Register a catch-all route with empty filters if you would rather swallow unknown messages, and see nothing matched for what the other matchers do instead.

Filters

Every filter key on one route, showing each form a value can take. All of them are optional, so set the ones that pick out the messages you want and leave the rest off.

rabbitMQMatcher.route({
  filters: {
    eventSourceArn: BROKER_ARN,
    queue: ['orders', 'priority-orders'], // Or a pattern: /-retry$/
    virtualHost: '/production',
    contentType: 'application/json',
    custom: ({ record }) => {
      // Only a custom reaches the message itself
      return (record.basicProperties.priority ?? 0) >= HIGH_PRIORITY
    },
  },
  handler: processOrder,
})

Filter	Type	Description
`eventSourceArn`	`FilterStringMatcher`	Matches the ARN of the broker the event came from
`queue`	`FilterStringMatcher`	Matches the queue name, with the virtual host split off. See Queue names
`virtualHost`	`FilterStringMatcher`	Matches the virtual host the queue lives on. See Queue names
`contentType`	`FilterStringMatcher`	Matches `basicProperties.contentType`, when it is not `null`
`custom`	`(input: RabbitMQFilterInput) => boolean \| Promise<boolean>`	Anything the other keys cannot express, given the `queue`, the `virtualHost`, the `contentType`, the parsed `timestamp`, the decoded `message` and the raw `record`. Can be async

FilterStringMatcher is string | RegExp | Array<string | RegExp>. See filters for how each form matches, including the * wildcard.

A message with no content type never matches a contentType filter. Content type is an optional AMQP property, so a publisher can leave it off, and Amazon MQ then sends contentType: null. The matcher skips a contentType filter for that message rather than matching it, the same way an absent virtual host skips a virtualHost filter. So contentType: '*' picks out the messages that have a content type, not every message, and custom sees contentType as null.

custom is the only key that reaches basicProperties, so priority, headers, correlationId and the redelivered flag are filterable through it and nowhere else.

custom gets no parsed body. Schema validation runs after a route matches, so parse message.data yourself to route on the contents. The filter sees the same split as the handler: message has its data decoded to text and record is the untouched AWS message with data still base64. See custom for where it sits in the filter order.

Queue names

An event groups its messages by queue, and RabbitMQ names each group queueName::virtualHost. A group named orders::/production holds the messages from the orders queue on the /production virtual host.

The matcher splits the two apart, so request.queue is orders and request.virtualHost is /production. The queue filter matches against the name and the virtualHost filter against the host.

Use queue: 'orders', not queue: 'orders::/production'. The queue filter sees the name on its own, so a value with the virtual host in it matches nothing. Pick the host with virtualHost instead.

The same queue name in two virtual hosts routes apart once you add a virtualHost filter, and a handler reads request.virtualHost to tell which host a message came from. A key with no :: carries no virtual host, so request.virtualHost is undefined and a virtualHost filter never matches it.

Handler

Handlers take one argument and return nothing.

import { logger } from '@latcher/base'
import type { RabbitMQRequest, RabbitMQResponse } from '@latcher/mq'

export async function processOrder(
  request: RabbitMQRequest<Order>,
): Promise<RabbitMQResponse> {
  const { orderId, total } = request.body
  logger.info(`Processing order ${orderId} for ${total} from ${request.queue}`)
}

Request object

Field	Type	Description
`message`	`RabbitMQMessage`	The message with `data` decoded from base64 to text
`queue`	`string`	The queue name, without the virtual host
`virtualHost`	`string \| undefined`	The virtual host the queue lives on. `undefined` when the queue key carries none
`body`	`TBody`	The decoded data parsed as JSON. Data that is not JSON reaches you as the raw string
`timestamp`	`Date \| null`	The AMQP `timestamp` property as a `Date`. See Timestamp
`record`	`RabbitMQMessage`	The untouched message from AWS, so `data` is still base64
`context`	`Context`	The Lambda context

RabbitMQMessage and RabbitMQBasicProperties are declared by this package, since aws-lambda carries no RabbitMQ types. Context does come from aws-lambda.

message and record differ by one field. message.data is the decoded text and record.data is the base64 AWS sent, so reading record.data for your message body hands you base64. basicProperties and redelivered are the same on both.

Timestamp

Amazon MQ sends the AMQP timestamp property as text such as Sep 21, 2026, 2:13:20 PM. The text is UTC but carries no time zone, so new Date() on it gives the wrong instant anywhere that is not running in UTC.

The matcher parses it for you as UTC, so request.timestamp is 2026-09-21T14:13:20.000Z for that text. The original text stays on basicProperties.timestamp.

request.timestamp is null when the publisher set no timestamp, and also when the text is in a format the matcher does not recognise. The message is still handled either way.

A custom filter gets the same parsed value as timestamp on its input.

Headers

A string header arrives as bytes, not text. Amazon MQ sends orderPriority: 'urgent' as { bytes: [117, 114, 103, 101, 110, 116] }, which is the UTF-8 encoding of the string. That goes for a string inside a header array or a nested table too. Decode one with Buffer.from(value.bytes).toString('utf-8').

Numbers of every AMQP size, booleans and null arrive as they are. A decimal arrives as a plain number, so 12345 with two places reads as 123.45. A byte array arrives as signed bytes, so 255 reads as -1.

A timestamp header is the one plain string, in the same format as the timestamp property. The matcher does not parse headers, so it stays as text.

RabbitMQHeaderValue types each value, so TypeScript makes you check for bytes before you read a string header.

Response type

RabbitMQResponse is undefined. There is nothing useful to return to a broker, so handlers return Promise<RabbitMQResponse> and the matcher works out what to hand back to Lambda.

Throwing is how you signal failure. See Failures and retries for what that does to the rest of the batch.

Inferred handlers

Nothing to look up and nothing to keep in sync. defineRabbitMQRoute reads the schema and hands your handler a fully typed body, defaults and coercion included, so total below is a number without you declaring that anywhere.

import { logger } from '@latcher/base'
import { defineRabbitMQRoute } from '@latcher/mq'
import { z } from 'zod'

const OrderSchema = z.object({ orderId: z.string(), total: z.coerce.number() })

export const processOrderRoute = defineRabbitMQRoute({
  filters: { queue: 'orders', contentType: 'application/json' },
  bodySchema: OrderSchema,
}).handle(async ({ body, queue }) => {
  logger.info(`Processing order ${body.orderId} for ${body.total} from ${queue}`)
})

rabbitMQMatcher.route(processOrderRoute)

Inference pays off most in a Lambda taking several event sources, since you never have to know any of their request shapes. See inferred handlers, where the same queue is written both ways to compare.

Annotated handlers

Annotating the request yourself splits route setup from business logic, using RabbitMQRequest and your own types.

// handlers/processOrder.ts
import { logger } from '@latcher/base'
import type { RabbitMQRequest, RabbitMQResponse } from '@latcher/mq'
import { z } from 'zod'

export const OrderSchema = z.object({ orderId: z.string(), total: z.number() })
type Order = z.infer<typeof OrderSchema>

export async function processOrder(request: RabbitMQRequest<Order>): Promise<RabbitMQResponse> {
  logger.info(`Processing order ${request.body.orderId}`)
}

// rabbitmq.ts
import { createRabbitMQMatcher } from '@latcher/mq'
import { OrderSchema, processOrder } from './handlers/processOrder'

const rabbitMQMatcher = createRabbitMQMatcher()

rabbitMQMatcher.route({
  filters: { queue: 'orders', contentType: 'application/json' },
  bodySchema: OrderSchema,
  handler: processOrder,
})

Derive the type from the schema with z.infer rather than hand-writing an interface that mirrors it. Both forms hand the handler the same request here, and annotated handlers has the worked version.

Schema validation

One key takes a schema, and it is optional.

const OrderSchema = z.object({
  orderId: z.string(),
  total: z.number(),
})

rabbitMQMatcher.route({
  filters: { queue: 'orders', contentType: 'application/json' },
  bodySchema: OrderSchema,
  handler: processOrder,
})

Key	Validates
`bodySchema`	The decoded message data, parsed as JSON

Any Standard Schema library works. Validation runs after a route has matched, so a message failing its schema throws rather than falling through to the next route. See schema validation for what your handler receives after coercion.

Data that is not JSON reaches the schema as a string. A z.object() rejects that, so every plain text message on the queue throws. Pair a bodySchema with contentType: 'application/json' and give the rest of the traffic its own route.

Failures and retries

A handler that throws fails the invocation. Messages are handled one at a time, queue by queue in the order the event lists them, so a throw stops everything behind it in the batch from running.

The matcher has no way to report single messages back to Lambda, so a failure applies to all of them. None are acknowledged, the broker redelivers the whole batch, and the messages that already succeeded run a second time. Make handlers safe to run twice and read message.redelivered where a repeat costs something.

The batch size on the event source mapping decides how much gets replayed, so a smaller batch means less repeated work after a failure.

A message matching no route throws the same way, naming the queue and broker ARN it could not place.

Middleware

Matcher and route middleware are both typed RabbitMQMiddleware, and the chain runs once per message, so a batch of ten messages runs it ten times.

import { logger } from '@latcher/base'
import type { RabbitMQMiddleware } from '@latcher/mq'

export const logInvocation: RabbitMQMiddleware = async (request, next) => {
  logger.info(`Handling a ${request.message.basicProperties.contentType} message on ${request.queue}`)
  return next(request)
}

const rabbitMQMatcher = createRabbitMQMatcher({ middleware: [logInvocation] })

rabbitMQMatcher.route({
  filters: { queue: 'orders' },
  middleware: [withOrderContext],
  handler: processOrder,
})

Route middleware carries the route's body type. A route with a bodySchema needs RabbitMQMiddleware<Order>. RabbitMQMiddleware on its own does not compile there. Matcher middleware takes no type argument, because it runs for every route.

See middleware for the execution order and the three levels it attaches at.

Types

All exported from @latcher/mq.

Type	Description
`RabbitMQRequest<TBody>`	The handler argument
`RabbitMQResponse`	Handler return type, `undefined`
`RabbitMQFilters`	The `filters` object
`RabbitMQFilterInput`	What `custom` receives
`RabbitMQRouteDefinition<TBody>`	A full route passed to `route()`
`RabbitMQMatcherOptions`	Options for `createRabbitMQMatcher`
`RabbitMQMiddleware<TBody>`	Matcher and route middleware
`RabbitMQMessage`	One message, as `request.message` and `request.record`
`RabbitMQBasicProperties`	The AMQP properties on a message
`RabbitMQHeaderValue`	One value in `basicProperties.headers`. See Headers
`RabbitMQEvent`	The whole event, with its messages keyed by queue

The RabbitMQMatcher class and the createRabbitMQMatcher and defineRabbitMQRoute functions come from the same place.

Generic parameters

The types above that take a parameter all take the same one.

Parameter	Types	Default
`TBody`	`request.body`	`unknown`

Leave it off and the body is unknown, which is what RabbitMQFilterInput sees however the route is typed. You only need it for annotated handlers, since inference covers the rest.

Code example

An orders queue and a refunds queue on one broker, with anything else logged and dropped.

Open a file: index.ts | RabbitMQ matcher | handlers | schema

TSrabbitmq.ts

The first two routes match a different queue, so no message can match both and the order they are registered in makes no difference to them.

The catch-all is the one route that cares where it sits, since empty filters match anything. Register it last and it takes what the other two turned down, which here is a plain text message on orders as much as anything from a third queue. Without it those throw, and a throw takes the rest of the batch with it.

index.ts hands the matcher to Latcher, which is what AWS invokes and what every matcher in the Lambda gets registered on. See matchers for how the two levels of matching fit together.

RabbitMQMatcher ​

Install ​

Create the matcher ​

Options ​

Register routes ​

Filters ​

Queue names ​

Handler ​

Request object ​

Timestamp ​

Headers ​

Response type ​

Inferred handlers ​

Annotated handlers ​

Schema validation ​

Failures and retries ​

Middleware ​

Types ​

Generic parameters ​

Code example ​