FEATURE: Subscription Engine

Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphProjectionFactory.php

@@ -8,7 +8,7 @@
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Repository\DimensionSpacePointsRepository;
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Repository\NodeFactory;
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Repository\ProjectionContentGraph;
-use Neos\ContentRepository\Core\Factory\ProjectionFactoryDependencies;
+use Neos\ContentRepository\Core\Factory\SubscriberFactoryDependencies;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionFactoryInterface;
 
 /**
@@ -24,8 +24,7 @@ public function __construct(
     }
 
     public function build(
-        ProjectionFactoryDependencies $projectionFactoryDependencies,
-        array $options,
+        SubscriberFactoryDependencies $projectionFactoryDependencies,
     ): DoctrineDbalContentGraphProjection {
         $tableNames = ContentGraphTableNames::create(
             $projectionFactoryDependencies->contentRepositoryId

Neos.ContentGraph.PostgreSQLAdapter/src/HypergraphProjectionFactory.php

@@ -7,7 +7,7 @@
 use Doctrine\DBAL\Connection;
 use Neos\ContentGraph\PostgreSQLAdapter\Domain\Projection\HypergraphProjection;
 use Neos\ContentGraph\PostgreSQLAdapter\Domain\Repository\NodeFactory;
-use Neos\ContentRepository\Core\Factory\ProjectionFactoryDependencies;
+use Neos\ContentRepository\Core\Factory\SubscriberFactoryDependencies;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionFactoryInterface;
 use Neos\ContentRepository\Core\SharedModel\ContentRepository\ContentRepositoryId;
 
@@ -28,8 +28,7 @@ public static function graphProjectionTableNamePrefix(
     }
 
     public function build(
-        ProjectionFactoryDependencies $projectionFactoryDependencies,
-        array $options,
+        SubscriberFactoryDependencies $projectionFactoryDependencies,
     ): HypergraphProjection {
         $tableNamePrefix = self::graphProjectionTableNamePrefix(
             $projectionFactoryDependencies->contentRepositoryId

Neos.ContentRepository.BehavioralTests/Classes/ProjectionRaceConditionTester/RaceTrackerCatchUpHook.php

@@ -17,7 +17,8 @@
 use Neos\ContentRepository\BehavioralTests\ProjectionRaceConditionTester\Dto\TraceEntries;
 use Neos\ContentRepository\BehavioralTests\ProjectionRaceConditionTester\Dto\TraceEntryType;
 use Neos\ContentRepository\Core\EventStore\EventInterface;
-use Neos\ContentRepository\Core\Projection\CatchUpHookInterface;
+use Neos\ContentRepository\Core\Projection\CatchUpHook\CatchUpHookInterface;
+use Neos\ContentRepository\Core\Subscription\SubscriptionStatus;
 use Neos\EventStore\Model\EventEnvelope;
 use Neos\Flow\Annotations as Flow;
 
@@ -107,7 +108,7 @@ final class RaceTrackerCatchUpHook implements CatchUpHookInterface
     protected $configuration;
     private bool $inCriticalSection = false;
 
-    public function onBeforeCatchUp(): void
+    public function onBeforeCatchUp(SubscriptionStatus $subscriptionStatus): void
     {
         RedisInterleavingLogger::connect($this->configuration['redis']['host'], $this->configuration['redis']['port']);
     }
@@ -126,16 +127,12 @@ public function onAfterEvent(EventInterface $eventInstance, EventEnvelope $event
     {
     }
 
-    public function onBeforeBatchCompleted(): void
+    public function onAfterCatchUp(): void
     {
         // we only want to track relevant lock release calls (i.e. if we were in the event processing loop before)
         if ($this->inCriticalSection) {
             $this->inCriticalSection = false;
             RedisInterleavingLogger::trace(TraceEntryType::LockWillBeReleasedIfItWasAcquiredBefore);
         }
     }
-
-    public function onAfterCatchUp(): void
-    {
-    }
 }

Neos.ContentRepository.BehavioralTests/Classes/ProjectionRaceConditionTester/RaceTrackerCatchUpHookFactory.php

@@ -14,9 +14,9 @@
 
 namespace Neos\ContentRepository\BehavioralTests\ProjectionRaceConditionTester;
 
-use Neos\ContentRepository\Core\Projection\CatchUpHookFactoryDependencies;
-use Neos\ContentRepository\Core\Projection\CatchUpHookFactoryInterface;
-use Neos\ContentRepository\Core\Projection\CatchUpHookInterface;
+use Neos\ContentRepository\Core\Projection\CatchUpHook\CatchUpHookFactoryDependencies;
+use Neos\ContentRepository\Core\Projection\CatchUpHook\CatchUpHookFactoryInterface;
+use Neos\ContentRepository\Core\Projection\CatchUpHook\CatchUpHookInterface;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
 
 /**

Neos.ContentRepository.Core/Classes/ContentRepository.php

@@ -23,9 +23,6 @@
 use Neos\ContentRepository\Core\EventStore\InitiatingEventMetadata;
 use Neos\ContentRepository\Core\Factory\ContentRepositoryFactory;
 use Neos\ContentRepository\Core\NodeType\NodeTypeManager;
-use Neos\ContentRepository\Core\Projection\CatchUp;
-use Neos\ContentRepository\Core\Projection\CatchUpHookFactoryDependencies;
-use Neos\ContentRepository\Core\Projection\CatchUpOptions;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphInterface;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
 use Neos\ContentRepository\Core\Projection\ProjectionStateInterface;
@@ -39,9 +36,6 @@
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspace;
 use Neos\ContentRepository\Core\SharedModel\Workspace\WorkspaceName;
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspaces;
-use Neos\EventStore\EventStoreInterface;
-use Neos\EventStore\Model\EventEnvelope;
-use Neos\EventStore\Model\EventStream\VirtualStreamName;
 use Psr\Clock\ClockInterface;
 
 /**

Neos.ContentRepository.Core/Classes/Factory/ContentRepositoryFactory.php

@@ -29,6 +29,10 @@
 use Neos\ContentRepository\Core\Feature\WorkspaceCommandHandler;
 use Neos\ContentRepository\Core\Infrastructure\Property\PropertyConverter;
 use Neos\ContentRepository\Core\NodeType\NodeTypeManager;
+use Neos\ContentRepository\Core\Projection\CatchUpHook\CatchUpHookFactoryDependencies;
+use Neos\ContentRepository\Core\Projection\CatchUpHook\CatchUpHookFactoryInterface;
+use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionFactoryInterface;
+use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionInterface;
 use Neos\ContentRepository\Core\Projection\ProjectionEventHandler;
 use Neos\ContentRepository\Core\Projection\ProjectionStates;
 use Neos\ContentRepository\Core\SharedModel\ContentRepository\ContentRepositoryId;
@@ -38,6 +42,7 @@
 use Neos\ContentRepository\Core\Subscription\RetryStrategy\NoRetryStrategy;
 use Neos\ContentRepository\Core\Subscription\RunMode;
 use Neos\ContentRepository\Core\Subscription\Store\SubscriptionStoreInterface;
+use Neos\ContentRepository\Core\Subscription\Subscriber\ProjectionEventHandlerInterface;
 use Neos\ContentRepository\Core\Subscription\Subscriber\Subscriber;
 use Neos\ContentRepository\Core\Subscription\Subscriber\Subscribers;
 use Neos\ContentRepository\Core\Subscription\SubscriptionGroup;
@@ -53,12 +58,15 @@
  */
 final class ContentRepositoryFactory
 {
-    private ProjectionFactoryDependencies $projectionFactoryDependencies;
+    private SubscriberFactoryDependencies $subscriberFactoryDependencies;
     private EventStoreInterface $eventStore;
-
     private SubscriptionEngine $subscriptionEngine;
 
+    private ProjectionStates $projectionStates;
+
+    private ContentGraphProjectionInterface $contentGraphProjection;
+
     public function __construct(
         private readonly ContentRepositoryId $contentRepositoryId,
         EventStoreInterface $eventStore,
         NodeTypeManager $nodeTypeManager,
@@ -67,14 +75,17 @@
         private readonly UserIdProviderInterface $userIdProvider,
         private readonly ClockInterface $clock,
         SubscriptionStoreInterface $subscriptionStore,
+        ContentGraphProjectionFactoryInterface $contentGraphProjectionFactory,
+        CatchUpHookFactoryInterface $contentGraphCatchUpHookFactory,
+        ContentRepositorySubscriberFactories $additionalSubscriberFactories,
     ) {
         $contentDimensionZookeeper = new ContentDimensionZookeeper($contentDimensionSource);
         $interDimensionalVariationGraph = new InterDimensionalVariationGraph(
             $contentDimensionSource,
             $contentDimensionZookeeper
         );
         $eventNormalizer = new EventNormalizer();
-        $this->projectionFactoryDependencies = new ProjectionFactoryDependencies(
+        $this->subscriberFactoryDependencies = new SubscriberFactoryDependencies(
             $contentRepositoryId,
             $eventNormalizer,
             $nodeTypeManager,
@@ -83,18 +94,26 @@
             $interDimensionalVariationGraph,
             new PropertyConverter($propertySerializer)
         );
-        $subscribers = [];
-        foreach ($this->projectionsAndCatchUpHooks->projections as $projection) {
-            $subscribers[] = new Subscriber(
-                SubscriptionId::fromString(substr(strrchr($projection::class, '\\'), 1)),
-                SubscriptionGroup::fromString('default'),
-                RunMode::FROM_BEGINNING,
-                new ProjectionEventHandler(
-                    $projection,
-                    $this->projectionsAndCatchUpHooks->getCatchUpHookFactoryForProjection($projection)?->build($this->getOrBuild()),
-                ),
-            );
+        $this->contentGraphProjection = $contentGraphProjectionFactory->build($this->subscriberFactoryDependencies);
+        $contentGraphSubscriber = new Subscriber(
+            SubscriptionId::fromString('contentGraph'),
+            SubscriptionGroup::fromString('default'),
+            RunMode::FROM_BEGINNING,
+            ProjectionEventHandler::createWithCatchUpHook(
+                $this->contentGraphProjection,
+                $contentGraphCatchUpHookFactory->build(new CatchUpHookFactoryDependencies($this->contentRepositoryId, $this->contentGraphProjection->getState(), $nodeTypeManager, $contentDimensionSource, $interDimensionalVariationGraph)),
+            ),
+        );
+        $subscribers = [$contentGraphSubscriber];
+        $projectionStates = [];
+        foreach ($additionalSubscriberFactories as $subscriberFactory) {
+            $subscriber = $subscriberFactory->build($this->subscriberFactoryDependencies);
+            $subscribers[] = $subscriber;
+            if ($subscriber->handler instanceof ProjectionEventHandler) {
+                $projectionStates[] = $subscriber->handler->projection->getState();
+            }
         }
+        $this->projectionStates = ProjectionStates::fromArray($projectionStates);
         $this->subscriptionEngine = new SubscriptionEngine($eventStore, $subscriptionStore, Subscribers::fromArray($subscribers), $eventNormalizer, new NoRetryStrategy());
         $this->eventStore = new RunSubscriptionEventStore($eventStore, $this->subscriptionEngine);
     }
@@ -115,59 +134,54 @@
             return $this->contentRepository;
         }
 
-        $contentGraphReadModel = $this->projectionsAndCatchUpHooks->contentGraphProjection->getState();
+        $contentGraphReadModel = $this->contentGraphProjection->getState();
         $commandHandlingDependencies = new CommandHandlingDependencies($contentGraphReadModel);
 
         // we dont need full recursion in rebase - e.g apply workspace commands - and thus we can use this set for simulation
         $commandBusForRebaseableCommands = new CommandBus(
             $commandHandlingDependencies,
             new NodeAggregateCommandHandler(
-                $this->projectionFactoryDependencies->nodeTypeManager,
-                $this->projectionFactoryDependencies->contentDimensionZookeeper,
-                $this->projectionFactoryDependencies->interDimensionalVariationGraph,
-                $this->projectionFactoryDependencies->propertyConverter,
+                $this->subscriberFactoryDependencies->nodeTypeManager,
+                $this->subscriberFactoryDependencies->contentDimensionZookeeper,
+                $this->subscriberFactoryDependencies->interDimensionalVariationGraph,
+                $this->subscriberFactoryDependencies->propertyConverter,
             ),
             new DimensionSpaceCommandHandler(
-                $this->projectionFactoryDependencies->contentDimensionZookeeper,
-                $this->projectionFactoryDependencies->interDimensionalVariationGraph,
+                $this->subscriberFactoryDependencies->contentDimensionZookeeper,
+                $this->subscriberFactoryDependencies->interDimensionalVariationGraph,
             ),
             new NodeDuplicationCommandHandler(
-                $this->projectionFactoryDependencies->nodeTypeManager,
-                $this->projectionFactoryDependencies->contentDimensionZookeeper,
-                $this->projectionFactoryDependencies->interDimensionalVariationGraph,
+                $this->subscriberFactoryDependencies->nodeTypeManager,
+                $this->subscriberFactoryDependencies->contentDimensionZookeeper,
+                $this->subscriberFactoryDependencies->interDimensionalVariationGraph,
             )
         );
 
         $commandSimulatorFactory = new CommandSimulatorFactory(
-            $this->projectionsAndCatchUpHooks->contentGraphProjection,
-            $this->projectionFactoryDependencies->eventNormalizer,
+            $this->contentGraphProjection,
+            $this->subscriberFactoryDependencies->eventNormalizer,
             $commandBusForRebaseableCommands
         );
 
         $publicCommandBus = $commandBusForRebaseableCommands->withAdditionalHandlers(
             new WorkspaceCommandHandler(
                 $commandSimulatorFactory,
                 $this->eventStore,
-                $this->projectionFactoryDependencies->eventNormalizer,
+                $this->subscriberFactoryDependencies->eventNormalizer,
             )
         );
 
-        $projectionStates = [];
-        foreach ($this->projectionsAndCatchUpHooks->projections as $projection) {
-            $projectionStates[] = $projection->getState();
-        }
-
         return $this->contentRepository = new ContentRepository(
             $this->contentRepositoryId,
             $publicCommandBus,
             $this->buildEventPersister(),
-            $this->projectionFactoryDependencies->nodeTypeManager,
-            $this->projectionFactoryDependencies->interDimensionalVariationGraph,
-            $this->projectionFactoryDependencies->contentDimensionSource,
+            $this->subscriberFactoryDependencies->nodeTypeManager,
+            $this->subscriberFactoryDependencies->interDimensionalVariationGraph,
+            $this->subscriberFactoryDependencies->contentDimensionSource,
             $this->userIdProvider,
             $this->clock,
             $contentGraphReadModel,
-            ProjectionStates::fromArray($projectionStates),
+            $this->projectionStates,
         );
     }
 
@@ -187,7 +201,7 @@
     ): ContentRepositoryServiceInterface {
 
         $serviceFactoryDependencies = ContentRepositoryServiceFactoryDependencies::create(
-            $this->projectionFactoryDependencies,
+            $this->subscriberFactoryDependencies,
             $this->eventStore,
             $this->getOrBuild(),
             $this->buildEventPersister(),
@@ -201,7 +215,7 @@
         if (!$this->eventPersister) {
             $this->eventPersister = new EventPersister(
                 $this->eventStore,
-                $this->projectionFactoryDependencies->eventNormalizer,
+                $this->subscriberFactoryDependencies->eventNormalizer,
             );
         }
         return $this->eventPersister;

Neos.ContentRepository.Core/Classes/Factory/ContentRepositoryServiceFactoryDependencies.php

@@ -54,7 +54,7 @@ private function __construct(
      * @internal
      */
     public static function create(
-        ProjectionFactoryDependencies $projectionFactoryDependencies,
+        SubscriberFactoryDependencies $projectionFactoryDependencies,
         EventStoreInterface $eventStore,
         ContentRepository $contentRepository,
         EventPersister $eventPersister,

Neos.ContentRepository.Core/Classes/Factory/ContentRepositorySubscriberFactories.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Neos\ContentRepository\Core\Factory;
+
+/**
+ * @implements \IteratorAggregate<ContentRepositorySubscriberFactoryInterface>
+ */
+final class ContentRepositorySubscriberFactories implements \IteratorAggregate
+{
+    private array $subscriberFactories;
+
+    private function __construct(ContentRepositorySubscriberFactoryInterface ...$subscriberFactories)
+    {
+        $this->subscriberFactories = $subscriberFactories;
+    }
+
+    public static function fromArray(array $subscriberFactories): self
+    {
+        return new self(...$subscriberFactories);
+    }
+
+    public static function none(): self
+    {
+        return new self();
+    }
+
+    public function isEmpty(): bool
+    {
+        return $this->subscriberFactories === [];
+    }
+
+    public function getIterator(): \Traversable
+    {
+        yield from $this->subscriberFactories;
+    }
+}

Neos.ContentRepository.Core/Classes/Factory/ContentRepositorySubscriberFactoryInterface.php

@@ -0,0 +1,25 @@
+<?php
+
+/*
+ * This file is part of the Neos.ContentRepository package.
+ *
+ * (c) Contributors of the Neos Project - www.neos.io
+ *
+ * This package is Open Source Software. For the full copyright and license
+ * information, please view the LICENSE file which was distributed with this
+ * source code.
+ */
+
+declare(strict_types=1);
+
+namespace Neos\ContentRepository\Core\Factory;
+
+use Neos\ContentRepository\Core\Subscription\Subscriber\Subscriber;
+
+/**
+ * @api for implementers of custom subscribers
+ */
+interface ContentRepositorySubscriberFactoryInterface
+{
+    public function build(SubscriberFactoryDependencies $dependencies): Subscriber;
+}