in-memory-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
    ->scan(function(array $state, array $event): array {
        switch ($event['name']) {
            case 'account_was_created':
                return array_merge($state, $event['payload'], ['amount' => 0]);

            case 'withdrawal_was_performed':
                return array_merge($state, ['amount' => $state['amount'] - $event['payload']['amount']]);

            case 'deposit_was_performed':
                return array_merge($state, ['amount' => $state['amount'] + $event['payload']['amount']]);
        }

        return $state;
    }, [])
    ->subscribe(new CallbackObserver(
        function(array $state) {
            echo json_encode($state) . PHP_EOL;
        },
        function() {
            printf('ERROR' . PHP_EOL);
        },
        function() {
            printf('COMPLETED' . PHP_EOL);
        }
    ));