Multi-queue flow

//
// Tutorial 2 of 4: a multi-queue flow.
//
// One queue partitioned per customer, two consumer groups reading it
// independently, and a second queue downstream. This is the shape most
// applications end up with, and it shows the three things that make it work:
// a partition keeps one entity's events in order, a consumer group is a cursor
// so every group sees everything, and a queue is created by the push.
//
//   orders (partition = customer)
//     ├── group "billing"    -> charges, and pushes to the shipping queue
//     └── group "analytics"  -> counts, and pushes nothing
//   shipping
//     └── group "warehouse"  -> ships
//
// Run it:
//   QUEEN_URL=http://localhost:6632 php 02-multi-queue-flow.php

require __DIR__ . '/vendor/autoload.php';

use Queen\Queen;

$QUEEN_URL = getenv('QUEEN_URL') ?: 'http://localhost:6632';
$RUN = base_convert((string) (int) (microtime(true) * 1000), 10, 36);
$ORDERS = "tut-php-orders-{$RUN}";
$SHIPPING = "tut-php-shipping-{$RUN}";

$INPUT = [
    ['orderId' => 'A-1', 'customer' => 'acme', 'total' => 120.5],
    ['orderId' => 'A-2', 'customer' => 'acme', 'total' => 12.0],
    ['orderId' => 'B-1', 'customer' => 'globex', 'total' => 88.75],
    ['orderId' => 'C-1', 'customer' => 'initech', 'total' => 310.0],
    ['orderId' => 'A-3', 'customer' => 'acme', 'total' => 9.99],
];

$checks = 0;
$assert = function (bool $condition, string $description) use (&$checks): void {
    if (!$condition) {
        throw new RuntimeException($description);
    }
    $checks++;
    echo "  ok: {$description}\n";
};

$queen = new Queen($QUEEN_URL);
$exitCode = 0;

try {
    echo "broker {$QUEEN_URL}\n";

    // Push each order into the partition named after its customer. Everything
    // about one customer stays in order; different customers never wait for each
    // other. The partition key is the only ordering decision you make.
    echo "\npushing\n";
    foreach ($INPUT as $order) {
        $queen->queue($ORDERS)->partition($order['customer'])->push([['data' => $order]])->execute();
        echo "  {$order['orderId']} -> partition {$order['customer']}\n";
    }

    // Group one. It reads every order, charges it, and hands the paid ones to the
    // shipping queue. subscriptionMode('all') matters: a group created after the
    // messages were pushed starts at the tail by default, so without it this
    // group would see nothing.
    //
    // each() hands the closure one message at a time instead of the whole batch,
    // and consume() only describes the loop: execute() is what runs it, and it
    // blocks until limit or idleMillis stops it.
    echo "\nbilling\n";
    $billed = [];
    $queen
        ->queue($ORDERS)
        ->group('tut-php-billing')
        ->subscriptionMode('all')
        ->each()
        ->limit(count($INPUT))
        ->idleMillis(5000) // stop after 5s of silence, so a lost message fails the run instead of hanging it
        ->consume(function (array $msg) use ($queen, $SHIPPING, &$billed): void {
            $billed[] = $msg['data']['orderId'];
            echo "  charged {$msg['data']['orderId']} ({$msg['data']['total']})\n";

            // The push to the next queue creates it on first use, exactly like the
            // first queue. Partitioning it by customer as well keeps a customer's
            // shipments in the order their orders were charged.
            $queen->queue($SHIPPING)->partition($msg['data']['customer'])->push([[
                'data' => ['orderId' => $msg['data']['orderId'], 'customer' => $msg['data']['customer']],
            ]])->execute();
        })
        ->execute();

    $assert(count($billed) === count($INPUT), 'billing saw all ' . count($INPUT) . ' orders');

    // Group two reads the same stored messages through its own cursor. It was not
    // affected by billing acking them: that is what fan-out means here, and it
    // costs no extra copy of the data.
    echo "\nanalytics\n";
    $total = 0.0;
    $queen
        ->queue($ORDERS)
        ->group('tut-php-analytics')
        ->subscriptionMode('all')
        ->each()
        ->limit(count($INPUT))
        ->idleMillis(5000)
        ->consume(function (array $msg) use (&$total): void {
            $total += $msg['data']['total'];
        })
        ->execute();

    $assert(
        abs($total - array_sum(array_column($INPUT, 'total'))) < 0.001,
        'analytics summed every order, independently of billing'
    );

    // The order inside one partition is the order it was pushed in. Check the
    // customer with more than one order.
    echo "\nwarehouse\n";
    $acmeShipments = [];
    $queen
        ->queue($SHIPPING)
        ->partition('acme')
        ->group('tut-php-warehouse')
        ->subscriptionMode('all')
        ->each()
        ->limit(3)
        ->idleMillis(5000)
        ->consume(function (array $msg) use (&$acmeShipments): void {
            $acmeShipments[] = $msg['data']['orderId'];
            echo "  shipping {$msg['data']['orderId']}\n";
        })
        ->execute();

    $assert(
        $acmeShipments === ['A-1', 'A-2', 'A-3'],
        "one customer's shipments arrived in the order they were pushed"
    );

    $queen->queue($ORDERS)->delete()->execute();
    $queen->queue($SHIPPING)->delete()->execute();

    echo "\nPASS: {$checks} checks\n";
} catch (Throwable $error) {
    fwrite(STDERR, "\nFAIL: " . $error->getMessage() . "\n");
    $exitCode = 1;
} finally {
    $queen->close();
}

exit($exitCode);