CLI

This library provides a cli to manage the event-sourcing functionalities. The commands are symfony console commands, so you can register them in any symfony console application.

You can:

  • Create and delete databases
  • Create, update and delete schemas
  • Manage subscriptions
  • Inspect your aggregates, events and subscribers
  • Migrate events from one store to another

The examples on this page use bin/console as the entry point. How to build such a cli file is shown in the CLI example.

Database commands

The database commands create and delete the database configured in your doctrine connection. They are mostly useful for local development and test environments.

Create database

The event-sourcing:database:create command creates the database.

bin/console event-sourcing:database:create

With the --if-not-exists option, the command does not fail if the database already exists.

bin/console event-sourcing:database:create --if-not-exists

Drop database

The event-sourcing:database:drop command deletes the database. Without the --force option, the command only prints which database would be dropped.

bin/console event-sourcing:database:drop --force

With the --if-exists option, the command does not fail if the database does not exist.

Dropping the database deletes all your events. They cannot be restored.

Schema commands

The schema commands manage the tables the library needs, for example the event store table and the subscription store table. They use the schema director, which knows all tables of the configured stores.

All schema commands support the --dry-run option. Instead of executing anything, the command prints the SQL queries it would run, so you can review the changes before applying them.

bin/console event-sourcing:schema:update --dry-run

Create schema

The event-sourcing:schema:create command creates all tables. Use it once when you set up a new database.

bin/console event-sourcing:schema:create

Update schema

The event-sourcing:schema:update command compares the existing tables with the expected schema and applies the differences, for example after a library update. Without the --force option, the command does nothing.

bin/console event-sourcing:schema:update --force

Drop schema

The event-sourcing:schema:drop command deletes all tables of the library. Without the --force option, the command does nothing.

bin/console event-sourcing:schema:drop --force

Dropping the schema deletes all your events. They cannot be restored.

In production, we recommend doctrine migrations instead of the schema commands, so that every schema change is versioned and reviewed.

Subscription commands

The subscription commands control the subscription engine. They move subscriptions through their lifecycle and show their current status.

Filter subscriptions

Most subscription commands can be limited to certain subscriptions. Use --id to select subscriptions by their subscriber ID and --group to select them by their group. Both options can be passed multiple times.

bin/console event-sourcing:subscription:boot --id=profile_1 --id=welcome_email
bin/console event-sourcing:subscription:run --group=projector

Ids are combined with OR, groups are combined with OR, and both options together are combined with AND. Without any filter, the command applies to all subscriptions.

Setup

The event-sourcing:subscription:setup command sets up all new subscriptions. The subscription engine calls the setup method of the subscriber, if available, for example to create the table of a projection. Afterwards the subscription is booting or, depending on the run mode, already active.

bin/console event-sourcing:subscription:setup

With --skip-booting, the subscriptions skip the booting status and become active right away. They then catch up with the event store in the run command instead of the boot command.

Boot

The event-sourcing:subscription:boot command lets all booting subscriptions catch up with the event store. After that, they are active, or finished for subscribers with RunMode::Once. The command stops as soon as all subscriptions have caught up.

bin/console event-sourcing:subscription:boot

With --setup, the command first sets up new subscriptions, so you can do both steps in one call.

bin/console event-sourcing:subscription:boot --setup

The command processes the events in chunks. The --message-limit option (default 1000) defines how many events are processed per run. If the boot is aborted by a worker limit before all subscriptions caught up, the command exits with code 1.

Run

The event-sourcing:subscription:run command is the long-running worker. It keeps the active subscriptions up to date by processing new events as they are recorded.

bin/console event-sourcing:subscription:run

By default, the command looks for new events every second (--sleep=1000) and processes up to 100 events per run (--message-limit=100). If your store supports it (DoctrineDbalStore and StreamDoctrineDbalStore with PostgreSQL), the worker does not poll but waits for a notification from the database, so new events are processed right away.

The run command only handles active subscriptions. Subscriptions that are new or booting have to be set up and booted first. After adding a new subscriber, run the setup and boot step and restart the run command.

Worker options

The boot and run commands are workers and support the following options:

Option Description
--run-limit Stop the worker after this number of runs.
--message-limit How many events are processed per run.
--memory-limit Stop the worker if it uses more memory than this, e.g. 250MB.
--time-limit Stop the worker after this number of seconds.
--sleep How many milliseconds the worker waits between two runs. 
bin/console event-sourcing:subscription:run --memory-limit=250MB --time-limit=3600

Use the memory and time limits together with a process manager like supervisor or systemd, which restarts the worker after it stopped. This prevents memory leaks from piling up and makes sure that a deployment is picked up.

Status

The event-sourcing:subscription:status command shows a table of all subscriptions with their group, run mode, position, status and error message. Use it to check whether your subscriptions are up to date or have an error.

bin/console event-sourcing:subscription:status
bin/console event-sourcing:subscription:status --group=projector

Pass a subscription ID to show the details of one subscription. If the subscription has an error, the command also prints the error message and the stack trace.

bin/console event-sourcing:subscription:status profile_1

Pause

The event-sourcing:subscription:pause command pauses subscriptions. The subscription engine no longer processes events for them until they are reactivated. This is useful, for example, if a third party system used by a processor is down.

bin/console event-sourcing:subscription:pause --id=welcome_email

Reactivate

The event-sourcing:subscription:reactivate command reactivates paused, finished, detached, error or failed subscriptions. The subscription then continues from its current position. A subscription with an error returns to the status it had before the error, all others become active. Use it after you fixed the cause of an error.

bin/console event-sourcing:subscription:reactivate --id=welcome_email

Only subscriptions whose subscriber exists in the code can be reactivated.

Refresh

The event-sourcing:subscription:refresh command updates the stored subscriptions if you changed the metadata of a subscriber in the code, for example the run mode or the group. The position and status of the subscriptions stay untouched.

bin/console event-sourcing:subscription:refresh

This command needs a subscription engine that supports refreshing, like the DefaultSubscriptionEngine.

Teardown

The event-sourcing:subscription:teardown command cleans up detached subscriptions. A subscription is detached if its subscriber no longer exists in the code, for example because you deleted it or created a new version with a new subscriber ID. The subscription engine calls the teardown method or runs the cleanup tasks of the subscriber, if available, and removes the subscription afterwards.

bin/console event-sourcing:subscription:teardown

Without cleanup tasks, a teardown is only possible if the code of the subscriber still exists. Otherwise, the subscription stays detached and you can remove it with the remove command.

Remove

The event-sourcing:subscription:remove command removes subscriptions regardless of their status. The subscription engine tries to call the teardown method, if available, but removes the subscription even if this fails. If the subscriber still exists, the subscription is new again and can be set up and booted from scratch. This is the way to rebuild a projection.

bin/console event-sourcing:subscription:remove --id=profile_1

Without any filter, the command asks for confirmation before it removes all subscriptions. Use --force to skip the question.

Removing a subscription deletes the data of its projection, if the subscriber has a teardown method.

Typical workflow

On every deployment, stop the old run command, set up and boot new subscriptions and start the run command again. Once the old version of your application no longer serves requests, clean up the outdated subscriptions.

bin/console event-sourcing:subscription:boot --setup
# start bin/console event-sourcing:subscription:run
# wait until the old application is gone
bin/console event-sourcing:subscription:teardown

If a subscription has an error, look at the details and reactivate it after you fixed the problem.

bin/console event-sourcing:subscription:status welcome_email
bin/console event-sourcing:subscription:reactivate --id=welcome_email

You can find out more about the subscription lifecycle.

Inspector commands

The inspector commands display the events in your store. They help you to debug your application and to understand what happened in an aggregate.

Show events

The event-sourcing:show command shows the latest events of the store, newest first. After each page, it asks whether to show the next events.

bin/console event-sourcing:show
Option Description
--limit How many events are displayed per page (default 10, 0 shows all).
--forward Start with the oldest event instead of the newest.
--stream Only show events of the given stream, wildcards like profile-* are allowed. 
bin/console event-sourcing:show --stream="profile-*" --limit=20

The --stream option needs the StreamDoctrineDbalStore.

Show aggregate

The event-sourcing:show-aggregate command shows all events of one aggregate. Pass the aggregate name and the aggregate ID as arguments. If you leave them out, the command asks for them.

bin/console event-sourcing:show-aggregate profile 018d6a1c-5f2b-7f3e-9c4a-2b6a1f0e8d7c

Watch events

The event-sourcing:watch command prints new events live as they are saved. It is a worker like the subscription run command and supports the same worker options, except --message-limit.

bin/console event-sourcing:watch

You can limit the output to one stream, or to an aggregate name and aggregate ID. The stream and the aggregate options cannot be combined. The --stream option needs the StreamDoctrineDbalStore, the --aggregate and --aggregate-id options need the DoctrineDbalStore.

bin/console event-sourcing:watch --stream="profile-*"
bin/console event-sourcing:watch --aggregate=profile --aggregate-id=018d6a1c-5f2b-7f3e-9c4a-2b6a1f0e8d7c

Debug command

The debug command prints everything the library knows about your application: all registered aggregates, all registered events and all subscribers with their subscribe methods. It is the fastest way to check if a class was picked up by the attribute scanning.

  • DebugCommand: event-sourcing:debug (alias debug:event-sourcing)
use Patchlevel\EventSourcing\Console\Command\DebugCommand;
use Patchlevel\EventSourcing\Metadata\AggregateRoot\AggregateRootRegistry;
use Patchlevel\EventSourcing\Metadata\Event\EventRegistry;
use Patchlevel\EventSourcing\Subscription\Subscriber\SubscriberAccessorRepository;
use Symfony\Component\Console\Application;

/**
 * @var Application $cli
 * @var AggregateRootRegistry $aggregateRootRegistry
 * @var EventRegistry $eventRegistry
 * @var SubscriberAccessorRepository $subscriberAccessorRepository
 */
$cli->add(
    new DebugCommand(
        $aggregateRootRegistry,
        $eventRegistry,
        $subscriberAccessorRepository,
    ),
);

The subscriber repository is optional. If you don't pass it, the subscriber section is skipped.

Store migration command

The store migration command copies all events from one store into another one. You need it when you switch the store implementation, for example from the DoctrineDbalStore to the StreamDoctrineDbalStore.

  • StoreMigrateCommand: event-sourcing:store:migrate
use Patchlevel\EventSourcing\Console\Command\StoreMigrateCommand;
use Patchlevel\EventSourcing\Message\Translator\AggregateToStreamHeaderTranslator;
use Patchlevel\EventSourcing\Store\Store;
use Symfony\Component\Console\Application;

/**
 * @var Application $cli
 * @var Store $oldStore
 * @var Store $newStore
 */
$cli->add(
    new StoreMigrateCommand(
        $oldStore,
        $newStore,
        [new AggregateToStreamHeaderTranslator()],
    ),
);

The third constructor argument is a list of translators that are applied to every message before it is written into the new store. The AggregateToStreamHeaderTranslator converts the AggregateHeader into the stream based headers and is what you need for a migration to the StreamDoctrineDbalStore.

Events are written in batches. You can control the batch size with the buffer option:

bin/console event-sourcing:store:migrate --buffer=5000

The command writes into the target store, it does not clean it up first. Make sure the target store is empty and create a backup before you run the migration.

The schema of the new store has to exist before you run the command. You can create it with the schema commands.

CLI example

A cli php file can look like this:

use Doctrine\DBAL\Connection;
use Patchlevel\EventSourcing\Console\Command;
use Patchlevel\EventSourcing\Console\DoctrineHelper;
use Patchlevel\EventSourcing\Message\Serializer\HeadersSerializer;
use Patchlevel\EventSourcing\Metadata\AggregateRoot\AggregateRootRegistry;
use Patchlevel\EventSourcing\Metadata\Event\EventRegistry;
use Patchlevel\EventSourcing\Schema\DoctrineSchemaDirector;
use Patchlevel\EventSourcing\Serializer\EventSerializer;
use Patchlevel\EventSourcing\Store\Store;
use Patchlevel\EventSourcing\Subscription\Engine\SubscriptionEngine;
use Patchlevel\EventSourcing\Subscription\Subscriber\SubscriberAccessorRepository;
use Symfony\Component\Console\Application;

$cli = new Application('Event-Sourcing CLI');
$cli->setCatchExceptions(true);

$doctrineHelper = new DoctrineHelper();

/**
 * @var Connection $connection
 * @var Store $store
 */
$schemaDirector = new DoctrineSchemaDirector($connection, $store);

/**
 * @var SubscriptionEngine $subscriptionEngine
 * @var EventSerializer $eventSerializer
 * @var HeadersSerializer $headersSerializer
 * @var AggregateRootRegistry $aggregateRootRegistry
 * @var EventRegistry $eventRegistry
 * @var SubscriberAccessorRepository $subscriberAccessorRepository
 */
$cli->addCommands([
    new Command\DatabaseCreateCommand($connection, $doctrineHelper),
    new Command\DatabaseDropCommand($connection, $doctrineHelper),
    new Command\SchemaCreateCommand($schemaDirector),
    new Command\SchemaDropCommand($schemaDirector),
    new Command\SchemaUpdateCommand($schemaDirector),
    new Command\SubscriptionSetupCommand($subscriptionEngine),
    new Command\SubscriptionBootCommand($subscriptionEngine),
    new Command\SubscriptionRunCommand($subscriptionEngine, $store),
    new Command\SubscriptionStatusCommand($subscriptionEngine),
    new Command\SubscriptionPauseCommand($subscriptionEngine),
    new Command\SubscriptionReactivateCommand($subscriptionEngine),
    new Command\SubscriptionRefreshCommand($subscriptionEngine),
    new Command\SubscriptionTeardownCommand($subscriptionEngine),
    new Command\SubscriptionRemoveCommand($subscriptionEngine),
    new Command\ShowCommand($store, $eventSerializer, $headersSerializer),
    new Command\ShowAggregateCommand($store, $eventSerializer, $headersSerializer, $aggregateRootRegistry),
    new Command\WatchCommand($store, $eventSerializer, $headersSerializer),
    new Command\DebugCommand($aggregateRootRegistry, $eventRegistry, $subscriberAccessorRepository),
]);

$cli->run();

The SubscriptionBootCommand and SubscriptionRunCommand accept a PSR-14 event dispatcher as an optional last argument. It receives the worker events, for example to reset services between two runs.

Doctrine Migrations

If you want to use doctrine migrations, you can register the commands like this:

use Doctrine\DBAL\Connection;
use Doctrine\Migrations\Configuration\Connection\ExistingConnection;
use Doctrine\Migrations\Configuration\Migration\ConfigurationLoader;
use Doctrine\Migrations\DependencyFactory;
use Doctrine\Migrations\Provider\SchemaProvider;
use Doctrine\Migrations\Tools\Console\Command;
use Patchlevel\EventSourcing\Schema\DoctrineMigrationSchemaProvider;
use Patchlevel\EventSourcing\Schema\DoctrineSchemaDirector;
use Patchlevel\EventSourcing\Store\Store;
use Symfony\Component\Console\Application;

/**
 * @var Connection $connection
 * @var Store $store
 */
$schemaDirector = new DoctrineSchemaDirector($connection, $store);

/** @var ConfigurationLoader $migrationConfig */
$dependencyFactory = DependencyFactory::fromConnection(
    $migrationConfig,
    new ExistingConnection($connection),
);


$dependencyFactory->setService(
    SchemaProvider::class,
    new DoctrineMigrationSchemaProvider($schemaDirector),
);

/** @var Application $cli */
$cli->addCommands([
    new Command\ExecuteCommand($dependencyFactory, 'event-sourcing:migrations:execute'),
    new Command\GenerateCommand($dependencyFactory, 'event-sourcing:migrations:generate'),
    new Command\LatestCommand($dependencyFactory, 'event-sourcing:migrations:latest'),
    new Command\ListCommand($dependencyFactory, 'event-sourcing:migrations:list'),
    new Command\MigrateCommand($dependencyFactory, 'event-sourcing:migrations:migrate'),
    new Command\DiffCommand($dependencyFactory, 'event-sourcing:migrations:diff'),
    new Command\StatusCommand($dependencyFactory, 'event-sourcing:migrations:status'),
    new Command\VersionCommand($dependencyFactory, 'event-sourcing:migrations:version'),
]);

Here you can find more information on how to configure doctrine migration.

Learn more

PreviousTesting
NextSupported Versions