Skip to main content

Message Queue

OPRA's MQ layer is a transport-agnostic framework for building message consumers on top of brokers such as Apache Kafka, RabbitMQ, Amazon SQS, and others. You define your consumers once using TypeScript decorators; the adapter package for your chosen broker handles connection management, serialization, acknowledgement, and error handling under the hood. Incoming payload and headers are automatically validated and decoded against their declared types before your handler is called, and any response payload is encoded and validated before it is published to the reply channel.

A Message Queue API in OPRA is built from two decorator types:

  • @MQController() — groups related message operations. Controllers can declare shared headers inherited by all their operations.
  • @MQOperation() — defines a single message consumer on a controller method. It declares the payload type, the channel(s) to subscribe to, and an optional response.
Message Receivedbroker → consumerChannel Matchingcontroller · operationHeadersdecode & validatePayloaddecode & validateHandlerasync methodResponseencode & publishoptionalerrornackreject · requeueerrornackreject · requeueerrordead-lettererror topic / DLQ

Controllers​

@MQController() registers a class as a message consumer controller. Use .Header() on the controller decorator to declare message headers shared by all operations — headers are matched case-insensitively and RegExp patterns are supported.

→ Controllers

Operations​

@MQOperation() defines a single message consumer. The first argument is the payload type. The channel option declares which topic or queue to subscribe to — it accepts a string, a RegExp, or an array of either. For keyed-message platforms (e.g. Kafka), declare the key type with keyType.

→ Controllers

Headers​

Use .Header() on @MQOperation() to declare headers specific to a single operation, on top of any headers inherited from the controller.

→ Controllers

Responses​

Use .Response() to declare the reply message sent after the operation processes the incoming message. Pass the response payload type and a channel option for the reply channel. Chain .Header() after .Response() to declare headers on the response message.

→ Controllers

Platform Extensions​

Platform-specific behaviour is configured via chainable extension methods added by the respective adapter packages. Import @opra/kafka to enable .Kafka() on @MQOperation(), or @opra/rabbitmq to enable .RabbitMQ().

→ Controllers