add lock collection support

php-service-bus · Dec 30, 2019 · 2b6dba7 · 2b6dba7
1 parent 55c678b
commit 2b6dba7
Show file tree

Hide file tree

Showing 3 changed files with 59 additions and 47 deletions.
diff --git a/composer.lock b/composer.lock
diff --git a/src/SagaModule.php b/src/SagaModule.php
@@ -13,6 +13,8 @@
 namespace ServiceBus\Sagas\Module;
 
 use ServiceBus\AnnotationsReader\Reader;
+use ServiceBus\Mutex\InMemoryLockCollection;
+use ServiceBus\Mutex\LockCollection;
 use function ServiceBus\Common\canonicalizeFilesPath;
 use function ServiceBus\Common\extractNamespaceFromFile;
 use function ServiceBus\Common\searchFiles;
@@ -177,11 +179,12 @@ public function boot(ContainerBuilder $containerBuilder): void
     {
         $containerBuilder->setParameter('service_bus.sagas.list', $this->sagasToRegister);
 
+        $this->registerMutexCollection($containerBuilder);
         $this->registerSagaStore($containerBuilder);
         $this->registerMutexFactory($containerBuilder);
         $this->registerSagasProvider($containerBuilder);
 
-        if (null === $this->configurationLoaderServiceId)
+        if ($this->configurationLoaderServiceId === null)
         {
             $this->registerDefaultConfigurationLoader($containerBuilder);
 
@@ -191,27 +194,32 @@ public function boot(ContainerBuilder $containerBuilder): void
         $this->registerRoutesConfigurator($containerBuilder);
     }
 
-    /**
-     * @param ContainerBuilder $containerBuilder
-     */
+    private function registerMutexCollection(ContainerBuilder $containerBuilder): void
+    {
+        if ($containerBuilder->hasDefinition(LockCollection::class) === false)
+        {
+            $containerBuilder->setDefinition(LockCollection::class, new Definition(InMemoryLockCollection::class));
+        }
+    }
+
     private function registerMutexFactory(ContainerBuilder $containerBuilder): void
     {
-        if (false === $containerBuilder->hasDefinition(MutexFactory::class))
+        if ($containerBuilder->hasDefinition(MutexFactory::class) === false)
         {
             $containerBuilder->setDefinition(MutexFactory::class, new Definition(InMemoryMutexFactory::class));
         }
     }
 
     private function registerRoutesConfigurator(ContainerBuilder $containerBuilder): void
     {
-        if (false === $containerBuilder->hasDefinition(ChainRouterConfigurator::class))
+        if ($containerBuilder->hasDefinition(ChainRouterConfigurator::class) === false)
         {
             $containerBuilder->setDefinition(ChainRouterConfigurator::class, new Definition(ChainRouterConfigurator::class));
         }
 
         $routerConfiguratorDefinition = $containerBuilder->getDefinition(ChainRouterConfigurator::class);
 
-        if (false === $containerBuilder->hasDefinition(Router::class))
+        if ($containerBuilder->hasDefinition(Router::class) === false)
         {
             $containerBuilder->setDefinition(Router::class, new Definition(Router::class));
         }
@@ -251,7 +259,7 @@ private function registerSagasProvider(ContainerBuilder $containerBuilder): void
 
     private function registerSagaStore(ContainerBuilder $containerBuilder): void
     {
-        if (true === $containerBuilder->hasDefinition(SagasStore::class))
+        if ($containerBuilder->hasDefinition(SagasStore::class) === true)
         {
             return;
         }
@@ -264,12 +272,12 @@ private function registerSagaStore(ContainerBuilder $containerBuilder): void
 
     private function registerDefaultConfigurationLoader(ContainerBuilder $containerBuilder): void
     {
-        if (true === $containerBuilder->hasDefinition(SagaConfigurationLoader::class))
+        if ($containerBuilder->hasDefinition(SagaConfigurationLoader::class) === true)
         {
             return;
         }
 
-        if (true === $containerBuilder->hasDefinition(EventListenerProcessorFactory::class))
+        if ($containerBuilder->hasDefinition(EventListenerProcessorFactory::class) === true)
         {
             return;
         }

diff --git a/src/SagasProvider.php b/src/SagasProvider.php
@@ -12,6 +12,8 @@
 
 namespace ServiceBus\Sagas\Module;
 
+use ServiceBus\Mutex\InMemoryLockCollection;
+use ServiceBus\Mutex\LockCollection;
 use function Amp\call;
 use function ServiceBus\Common\datetimeInstantiator;
 use function ServiceBus\Common\invokeReflectionMethod;
@@ -50,25 +52,22 @@ final class SagasProvider
      */
     private $sagaMetaDataCollection = [];
 
-    /**
-     * @psalm-var array<string, \ServiceBus\Mutex\Lock>
-     *
-     * @var Lock[]
-     */
-    private $lockCollection = [];
-
-    public function __construct(SagasStore $sagaStore, ?MutexFactory $mutexFactory = null)
-    {
-        $this->sagaStore    = $sagaStore;
-        $this->mutexFactory = $mutexFactory ?? new InMemoryMutexFactory();
+    /** @var LockCollection */
+    private $lockCollection;
+
+    public function __construct(
+        SagasStore $sagaStore,
+        ?MutexFactory $mutexFactory = null,
+        ?LockCollection $lockCollection = null
+    ) {
+        $this->sagaStore      = $sagaStore;
+        $this->mutexFactory   = $mutexFactory ?? new InMemoryMutexFactory();
+        $this->lockCollection = $lockCollection ?? new InMemoryLockCollection();
     }
 
     public function __destruct()
     {
-        foreach ($this->lockCollection as $lock)
-        {
-            yield $lock->release();
-        }
+        unset($this->lockCollection);
     }
 
     /**
@@ -245,7 +244,10 @@ private function doStore(Saga $saga, ServiceBusContext $context, bool $isNew): \
             $promises[] = $context->delivery($message);
         }
 
-        yield $promises;
+        if (\count($promises) !== 0)
+        {
+            yield $promises;
+        }
 
         unset($promises, $commands, $events);
 
@@ -282,20 +284,23 @@ private function appendMetaData(string $sagaClass, SagaMetadata $metadata): void
     }
 
     /**
-     * Setup mutes on saga.
+     * Setup mutex on saga.
      */
     private function setupMutex(SagaId $id): \Generator
     {
         $mutexKey = createMutexKey($id);
 
-        if (\array_key_exists($mutexKey, $this->lockCollection) === false)
+        /** @var bool $hasLock */
+        $hasLock = yield $this->lockCollection->has($mutexKey);
+
+        if ($hasLock === false)
         {
             $mutex = $this->mutexFactory->create($mutexKey);
 
             /** @var \ServiceBus\Mutex\Lock $lock */
             $lock = yield $mutex->acquire();
 
-            $this->lockCollection[$mutexKey] = $lock;
+            yield $this->lockCollection->place($mutexKey, $lock);
         }
     }
 
@@ -306,12 +311,11 @@ private function releaseMutex(SagaId $id): \Generator
     {
         $mutexKey = createMutexKey($id);
 
-        $lock = $this->lockCollection[$mutexKey] ?? null;
+        /** @var Lock|null $lock */
+        $lock = yield $this->lockCollection->extract($mutexKey);
 
         if ($lock !== null)
         {
-            unset($this->lockCollection[$mutexKey]);
-
             yield $lock->release();
         }
     }