Merge pull request #508 from patchlevel/remove-watch-server

remove watch server & use event store for watching

README.md

@@ -18,7 +18,6 @@ A lightweight but also all-inclusive event sourcing library with a focus on deve
 * Versioned and managed lifecycle of [projections](https://patchlevel.github.io/event-sourcing-docs/latest/projection/)
 * Smooth [upcasting](https://patchlevel.github.io/event-sourcing-docs/latest/upcasting/) of old events
 * Simple setup with [scheme management](https://patchlevel.github.io/event-sourcing-docs/latest/store/) and [doctrine migration](https://patchlevel.github.io/event-sourcing-docs/latest/migration/)
-* Dev [tools](https://patchlevel.github.io/event-sourcing-docs/latest/watch_server/) like realtime event watcher
 * Built in [cli commands](https://patchlevel.github.io/event-sourcing-docs/latest/cli/) with [symfony](https://symfony.com/)
 * and much more...
 

docs/mkdocs.yml

@@ -96,6 +96,4 @@ nav:
     - Split Stream: split_stream.md
     - Time / Clock: clock.md
     - Testing: testing.md
-  - Other / Tools:
     - CLI: cli.md
-    - Watch Server: watch_server.md

docs/pages/cli.md

@@ -55,6 +55,14 @@ Interacting with the outbox store is also possible via the cli.
 
     You can find out more about outbox [here](outbox.md).
 
+## Inspector commands
+
+The inspector is a tool to inspect the event streams.
+
+* ShowCommand: `event-sourcing:show`
+* ShowAggregateCommand: `event-sourcing:show-aggregate`
+* WatchCommand: `event-sourcing:watch`
+
 ## CLI example
 
 A cli php file can look like this:

docs/pages/index.md

@@ -13,7 +13,6 @@ A lightweight but also all-inclusive event sourcing library with a focus on deve
 * Versioned and managed lifecycle of [projections](projection.md)
 * Smooth [upcasting](upcasting.md) of old events
 * Simple setup with [scheme management](store.md) and [doctrine migration](migration.md)
-* Dev [tools](watch_server.md) like realtime event watcher
 * Built in [cli commands](cli.md) with [symfony](https://symfony.com/)
 * and much more...
 

phpstan-baseline.neon

@@ -39,8 +39,3 @@ parameters:
 			message: "#^Ternary operator condition is always true\\.$#"
 			count: 1
 			path: src/Store/DoctrineDbalStoreStream.php
-
-		-
-			message: "#^While loop condition is always true\\.$#"
-			count: 1
-			path: src/WatchServer/SocketWatchServer.php

src/Console/Command/WatchCommand.php

@@ -4,45 +4,102 @@
 
 namespace Patchlevel\EventSourcing\Console\Command;
 
+use Patchlevel\EventSourcing\Console\InputHelper;
 use Patchlevel\EventSourcing\Console\OutputStyle;
-use Patchlevel\EventSourcing\EventBus\Message;
 use Patchlevel\EventSourcing\Serializer\EventSerializer;
-use Patchlevel\EventSourcing\WatchServer\WatchServer;
+use Patchlevel\EventSourcing\Store\Criteria;
+use Patchlevel\EventSourcing\Store\Store;
+use Patchlevel\Worker\DefaultWorker;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 
-use function sprintf;
-
 #[AsCommand(
     'event-sourcing:watch',
     'live stream of all aggregate events',
 )]
 final class WatchCommand extends Command
 {
     public function __construct(
-        private readonly WatchServer $server,
+        private readonly Store $store,
         private readonly EventSerializer $serializer,
     ) {
         parent::__construct();
     }
 
+    protected function configure(): void
+    {
+        $this
+            ->addOption(
+                'sleep',
+                null,
+                InputOption::VALUE_REQUIRED,
+                'How much time should elapse before the next job is executed in milliseconds',
+                1000,
+            )
+            ->addOption(
+                'aggregate',
+                null,
+                InputOption::VALUE_REQUIRED,
+                'filter aggregate name',
+            )
+            ->addOption(
+                'aggregate-id',
+                null,
+                InputOption::VALUE_REQUIRED,
+                'filter aggregate id',
+            );
+    }
+
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $console = new OutputStyle($input, $output);
 
-        $this->server->start();
+        $sleep = InputHelper::positiveIntOrZero($input->getOption('sleep'));
+        $aggregate = InputHelper::nullableString($input->getOption('aggregate'));
+        $aggregateId = InputHelper::nullableString($input->getOption('aggregate-id'));
 
-        $console->success(sprintf('Server listening on %s', $this->server->host()));
-        $console->comment('Quit the server with CONTROL-C.');
+        $index = $this->currentIndex();
 
-        $this->server->listen(
-            function (Message $message) use ($console): void {
-                $console->message($this->serializer, $message);
+        $worker = DefaultWorker::create(
+            function () use ($console, &$index, $aggregate, $aggregateId): void {
+                $stream = $this->store->load(
+                    new Criteria(
+                        $aggregate,
+                        $aggregateId,
+                        $index,
+                    ),
+                );
+
+                foreach ($stream as $message) {
+                    $console->message($this->serializer, $message);
+                }
+
+                $index = $stream->index();
+
+                $stream->close();
             },
+            [],
         );
 
+        $worker->run($sleep);
+
         return 0;
     }
+
+    private function currentIndex(): int
+    {
+        $stream = $this->store->load(
+            limit: 1,
+            backwards: true,
+        );
+
+        $index = $stream->index() ?? 0;
+
+        $stream->close();
+
+        return $index;
+    }
 }