Publish/subscribe: one event, many consumer groups

const subscribe = (group) => queen.queue('order-placed')
  .group(group)                                   // each group gets every message
  .subscriptionMode('all')
  .each()
  .consume(async (event) => {
    console.log(group, event.data.orderId)
  })

for (const orderId of [3001, 3002, 3003]) {
  await queen.queue('order-placed').partition(`order-${orderId}`)
    .push([{ data: { orderId } }])
}
await Promise.all([subscribe('billing'), subscribe('shipping')])

// Months later, analytics subscribes too, and reads every order ever placed.
await subscribe('analytics')

def subscribe(group):
    async def handle(event):
        print(group, event["data"]["orderId"])

    return (
        queen.queue("order-placed")
        .group(group)                 # each group gets every message
        .subscription_mode("all")
        .each()
        .consume(handle)
    )

for order_id in [3001, 3002, 3003]:
    await queen.queue("order-placed").partition(f"order-{order_id}").push(
        [{"data": {"orderId": order_id}}]
    )
await asyncio.gather(subscribe("billing"), subscribe("shipping"))

# Months later, analytics subscribes too, and reads every order ever placed.
await subscribe("analytics")

subscribe := func(group string) error {
	return client.Queue("order-placed").
		Group(group). // each group gets every message
		SubscriptionMode(queen.SubscriptionModeAll).
		Each().
		Consume(ctx, func(ctx context.Context, event *queen.Message) error {
			fmt.Println(group, event.Data["orderId"])
			return nil
		}).
		Execute(ctx)
}

for _, orderID := range []int{3001, 3002, 3003} {
	_, err := client.Queue("order-placed").Partition(fmt.Sprintf("order-%d", orderID)).
		Push(map[string]interface{}{"orderId": orderID}).
		Execute(ctx)
	if err != nil {
		log.Fatal(err)
	}
}
var subscribers sync.WaitGroup
for _, group := range []string{"billing", "shipping"} {
	subscribers.Add(1)
	go func() {
		defer subscribers.Done()
		if err := subscribe(group); err != nil {
			log.Fatal(err)
		}
	}()
}
subscribers.Wait()

// Months later, analytics subscribes too, and reads every order ever placed.
if err := subscribe("analytics"); err != nil {
	log.Fatal(err)
}

let subscribe = |group: &'static str| {
    let events = queen
        .queue("order-placed")
        .group(group)                       // each group gets every message
        .subscription_mode(SubscriptionMode::All);
    async move {
        events
            .consume(move |event: Message| async move {
                println!("{group} {}", event.data["orderId"]);
                Ok::<_, queen_mq::Error>(())
            })
            .await
    }
};

for order_id in [3001, 3002, 3003] {
    queen
        .queue("order-placed")
        .partition(format!("order-{order_id}"))
        .push(json!({ "orderId": order_id }))
        .await?;
}
tokio::try_join!(subscribe("billing"), subscribe("shipping"))?;

// Months later, analytics subscribes too, and reads every order ever placed.
subscribe("analytics").await?;

$subscribe = fn (string $group) => $queen->queue('order-placed')
    ->group($group)                                 // each group gets every message
    ->subscriptionMode('all')
    ->each()
    ->consume(function (array $event) use ($group) {
        echo $group, ' ', $event['data']['orderId'], "\n";
    })
    ->execute();

foreach ([3001, 3002, 3003] as $orderId) {
    $queen->queue('order-placed')->partition("order-{$orderId}")
        ->push([['data' => ['orderId' => $orderId]]])
        ->execute();
}
$subscribe('billing');           // one after the other here: in production each group
$subscribe('shipping');          // is a worker process of its own

// Months later, analytics subscribes too, and reads every order ever placed.
$subscribe('analytics');

auto subscribe = [&](const std::string& group) {
    client.queue("order-placed")
        .group(group)                  // each group gets every message
        .subscription_mode("all")
        .each()
        .consume([&](const json& event) {
            std::cout << group + " " + event["data"]["orderId"].dump() + "\n";
        });
};

for (int order_id : {3001, 3002, 3003}) {
    client.queue("order-placed").partition("order-" + std::to_string(order_id))
        .push({{{"data", {{"orderId", order_id}}}}});
}
std::thread billing(subscribe, "billing"), shipping(subscribe, "shipping");
billing.join(); shipping.join();

// Months later, analytics subscribes too, and reads every order ever placed.
subscribe("analytics");