Broker

Source: src/AWS/MQ/Broker.ts

An Amazon MQ broker — a managed message broker running Apache ActiveMQ or RabbitMQ. Amazon MQ handles provisioning, patching, and (for multi-AZ deployments) failover, exposing standard wire protocols (OpenWire, AMQP, MQTT, STOMP, WSS) so existing clients connect unchanged.

Broker creation and deletion are asynchronous and take several minutes; the provider waits (bounded) for the broker to reach RUNNING before returning, and waits for it to disappear on delete.

Creating a Broker

Single-instance ActiveMQ (cheapest)

const broker = yield* MQ.Broker("Orders", {
  engineType: "ACTIVEMQ",
  engineVersion: "5.18",
  hostInstanceType: "mq.t3.micro",
  deploymentMode: "SINGLE_INSTANCE",
  publiclyAccessible: true,
  users: [{ username: "admin", password: Redacted.make("SuperSecretPassw0rd") }],
});
// broker.endpoints -> ["ssl://b-xxxx-1.mq.us-west-2.amazonaws.com:61617", ...]

Single-instance RabbitMQ

const broker = yield* MQ.Broker("Events", {
  engineType: "RABBITMQ",
  engineVersion: "3.13",
  hostInstanceType: "mq.t3.micro",
  publiclyAccessible: true,
  users: [{ username: "admin", password: Redacted.make("SuperSecretPassw0rd") }],
});

Networking and Encryption

const broker = yield* MQ.Broker("Orders", {
  engineType: "ACTIVEMQ",
  engineVersion: "5.18",
  hostInstanceType: "mq.m5.large",
  deploymentMode: "ACTIVE_STANDBY_MULTI_AZ",
  publiclyAccessible: false,
  subnetIds: [subnetA.subnetId, subnetB.subnetId],
  securityGroups: [group.groupId],
  encryptionOptions: { kmsKeyId: key.keyArn },
  users: [{ username: "admin", password: Redacted.make("SuperSecretPassw0rd") }],
});

Logging and Maintenance

const broker = yield* MQ.Broker("Orders", {
  engineType: "ACTIVEMQ",
  engineVersion: "5.18",
  hostInstanceType: "mq.t3.micro",
  users: [{ username: "admin", password: Redacted.make("SuperSecretPassw0rd") }],
  logs: { general: true, audit: true },
  maintenanceWindow: {
    dayOfWeek: "SUNDAY",
    timeOfDay: "03:00",
    timeZone: "UTC",
  },
});

Consuming Messages

Subscribe a Lambda function to broker queues from the init phase via consumeBrokerMessages. The event-source mapping, IAM grants, and runtime dispatch are created automatically (provide Lambda.BrokerEventSource on the function).

// init
yield* MQ.consumeBrokerMessages(
  broker,
  {
    queues: ["orders"],
    credentialsSecretArn: secret.secretArn,
  },
  (messages) =>
    messages.pipe(
      Stream.runForEach((message) =>
        Effect.log(`received: ${message.data}`),
      ),
    ),
);