sql-projection.php

<?php

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

use Rx\Observable;
use Rx\Observer\CallbackObserver;

$events = Observable::fromArray([
    ['name' => 'account_was_created', 'payload' => ['id' => 1, 'owner' => 'Gildas Quéméner']],
    ['name' => 'deposit_was_performed', 'payload' => ['id' => 1, 'amount' => 100]],
    ['name' => 'withdrawal_was_performed', 'payload' => ['id' => 1, 'amount' => 100]],
    ['name' => 'deposit_was_performed', 'payload' => ['id' => 1, 'amount' => 40]],
]);

$events = $events->map(function(array $event): string {
    $payload = $event['payload'];
    switch ($event['name']) {
        case 'account_was_created':
        return sprintf('INSERT INTO `accounts` (id, owner) VALUES (%d, \'%s\');', $payload['id'], $payload['owner']);

        case 'withdrawal_was_performed':
        return sprintf('UPDATE `accounts` SET amount = amount - %d WHERE id = %d;', $payload['amount'], $payload['id']);

        case 'deposit_was_performed':
        return sprintf('UPDATE `accounts` SET amount = amount + %d WHERE id = %d;', $payload['amount'], $payload['id']);
    }

    throw new \RuntimeException(sprintf('Cannot map action "%s" to SQL.', $action['name']));
});

$events->subscribe(new CallbackObserver(
    function(string $sql) {
        printf('EXECUTE QUERY: "%s".' . PHP_EOL, $sql);
    },
    function() {
        printf('ERROR' . PHP_EOL);
    },
    function() {
        printf('COMPLETED' . PHP_EOL);
    }
));