@mqueue/multicast

Mulitcast Strategy for MQueue

Broadcast a message to multiple different MQueue queue backends simultaneously, with the same interface. Publish to amqplib, azure-service-bus, rhea, and sqs with one call.

npm install --save @mqueue/queue @mqueue/multicast # + Adapter(s)...
# or use pnpm/yarn

const outgoingQueue = new MQueue.Outgoing(
  new MulticastQueue.Outgoing([
    await AmqplibOutgoingQueue.connect("amqp://rabbitmq:5271", "queue-name"),
    await AmqplibOutgoingQueue.connect("amqp://rabbitmq:5272", "queue-name2"),
  ]),
);

outgoingQueue.sendMessage({
  headers: {
    "Account-ID": "123",
  },
  body: "...",
});

// ...

const incomingQueue = new MQueue.Incoming(
  new MulticastQueue.Incoming([
    await AmqplibIncomingQueue.connect("amqp://rabbitmq:5271", "queue-name"),
    await AmqplibIncomingQueue.connect("amqp://rabbitmq:5272", "queue-name2"),
  ]),
);

Filtering & Randomisation

// Select one random adapter (for example)
const filter = (adapters) => [adapters[randomInt(adapters.length)]];

// Or filter by message detail
cosnt filter = (adapters, message) => {
  return message.headers.example === "something" ? adapter[0] : adapter;
};

const outgoingQueue = new MQueue.Outgoing(
  new MulticastQueue.Outgoing([
    await AmqplibOutgoingQueue.connect("amqp://rabbitmq:5271", "queue-name"),
    await AmqplibOutgoingQueue.connect("amqp://rabbitmq:5272", "queue-name2"),
  ]),
  { filter },
);

// ...

const incomingQueue = new MQueue.Incoming(
  new MulticastQueue.Incoming([
    await AmqplibIncomingQueue.connect("amqp://rabbitmq:5271", "queue-name"),
    await AmqplibIncomingQueue.connect("amqp://rabbitmq:5272", "queue-name2"),
  ]),
  { filter },
);