Merge pull request #623 from patchlevel/cli-stream-store-support

add stream store support in cli commands
patchlevel · Jul 29, 2024 · fe23cbc · fe23cbc
2 parents c36ff32 + 677b9f4
commit fe23cbc
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 4 deletions.
diff --git a/src/Console/Command/ShowCommand.php b/src/Console/Command/ShowCommand.php
@@ -8,6 +8,8 @@
 use Patchlevel\EventSourcing\Console\OutputStyle;
 use Patchlevel\EventSourcing\Message\Serializer\HeadersSerializer;
 use Patchlevel\EventSourcing\Serializer\EventSerializer;
+use Patchlevel\EventSourcing\Store\Criteria\Criteria;
+use Patchlevel\EventSourcing\Store\Criteria\StreamCriterion;
 use Patchlevel\EventSourcing\Store\Store;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
@@ -41,6 +43,12 @@ protected function configure(): void
                 'How many messages should be displayed',
                 10,
             )
+            ->addOption(
+                'stream',
+                null,
+                InputOption::VALUE_REQUIRED,
+                'Show messages from a specific stream (e.g. "stream-*")',
+            )
             ->addOption(
                 'forward',
                 null,
@@ -53,11 +61,18 @@ protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $limit = InputHelper::positiveIntOrZero($input->getOption('limit'));
         $forward = InputHelper::bool($input->getOption('forward'));
+        $stream = InputHelper::nullableString($input->getOption('stream'));
 
         $console = new OutputStyle($input, $output);
 
-        $maxCount = $this->store->count();
-        $stream = $this->store->load(null, null, null, !$forward);
+        $criteria = null;
+
+        if ($stream !== null) {
+            $criteria = new Criteria(new StreamCriterion($stream));
+        }
+
+        $maxCount = $this->store->count($criteria);
+        $stream = $this->store->load($criteria, null, null, !$forward);
 
         $currentCount = 0;
 

diff --git a/src/Console/Command/WatchCommand.php b/src/Console/Command/WatchCommand.php
@@ -45,17 +45,23 @@ protected function configure(): void
                 'How much time should elapse before the next job is executed in milliseconds',
                 1000,
             )
+            ->addOption(
+                'stream',
+                null,
+                InputOption::VALUE_REQUIRED,
+                'Watch messages from a specific stream (e.g. "stream-*")',
+            )
             ->addOption(
                 'aggregate',
                 null,
                 InputOption::VALUE_REQUIRED,
-                'filter aggregate name',
+                'Filter aggregate name',
             )
             ->addOption(
                 'aggregate-id',
                 null,
                 InputOption::VALUE_REQUIRED,
-                'filter aggregate id',
+                'Filter aggregate id',
             );
     }
 
@@ -64,9 +70,16 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         $console = new OutputStyle($input, $output);
 
         $sleep = InputHelper::positiveIntOrZero($input->getOption('sleep'));
+        $stream = InputHelper::nullableString($input->getOption('stream'));
         $aggregate = InputHelper::nullableString($input->getOption('aggregate'));
         $aggregateId = InputHelper::nullableString($input->getOption('aggregate-id'));
 
+        if ($stream !== null && ($aggregate !== null || $aggregateId !== null)) {
+            $console->error('You can only provide stream or aggregate and aggregate-id');
+
+            return 1;
+        }
+
         $index = $this->currentIndex();
 
         if ($this->store instanceof SubscriptionStore) {
@@ -75,6 +88,10 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
         $criteriaBuilder = new CriteriaBuilder();
 
+        if ($stream !== null) {
+            $criteriaBuilder->streamName($stream);
+        }
+
         if ($this->store instanceof StreamStore) {
             if ($aggregate !== null || $aggregateId !== null) {
                 if ($aggregate === null || $aggregateId === null) {