Merge 69a5f98 into bc608a0

prooph · May 3, 2020 · 2263bdc · 2263bdc
2 parents bc608a0 + 69a5f98
commit 2263bdc
Show file tree

Hide file tree

Showing 19 changed files with 210 additions and 12 deletions.
diff --git a/src/Internal/EventStoreCatchUpSubscription.php b/src/Internal/EventStoreCatchUpSubscription.php
@@ -253,7 +253,7 @@ private function subscribeToStreamAsync(): Promise
                 }
 
                 $eventAppeared = new class(Closure::fromCallable([$this, 'enqueuePushedEvent'])) implements EventAppearedOnSubscription {
-                    private $callback;
+                    private Closure $callback;
 
                     public function __construct(Closure $callback)
                     {
@@ -269,7 +269,7 @@ public function __invoke(
                 };
 
                 $subscriptionDropped = new class(Closure::fromCallable([$this, 'serverSubscriptionDropped'])) implements SubscriptionDropped {
-                    private $callback;
+                    private Closure $callback;
 
                     public function __construct(Closure $callback)
                     {

diff --git a/src/Internal/EventStoreConnectionLogicHandler.php b/src/Internal/EventStoreConnectionLogicHandler.php
@@ -84,6 +84,7 @@ class EventStoreConnectionLogicHandler
     private bool $wasConnected = false;
     private int $packageNumber = 0;
     private int $lastTimeoutsTimeStamp;
+    private \SplObjectStorage $currentPromises;
 
     public function __construct(EventStoreConnection $connection, ConnectionSettings $settings)
     {
@@ -98,6 +99,7 @@ public function __construct(EventStoreConnection $connection, ConnectionSettings
         $this->stopWatch = StopWatch::startNew();
         // this allows first connection to connect quick
         $this->lastTimeoutsTimeStamp = -$this->settings->operationTimeoutCheckPeriod();
+        $this->currentPromises = new \SplObjectStorage();
 
         $this->handler->registerHandler(
             StartConnectionMessage::class,
@@ -172,9 +174,32 @@ public function enqueueMessage(Message $message): void
     {
         $this->logDebug(\sprintf('enqueuing message %s', (string) $message));
 
+        $promise = $message->promise();
+        if ($promise !== null) {
+            $this->keepLoopAlive($promise);
+        }
+
         $this->handler->handle($message);
     }
 
+    private function keepLoopAlive(Promise $promise): void
+    {
+        if ($this->connection) {
+            $this->connection->reference();
+        }
+
+        $this->currentPromises->attach($promise);
+        $promise->onResolve(
+            function () use ($promise): void {
+                $this->currentPromises->detach($promise);
+
+                if ($this->connection && $this->currentPromises->count() === 0) {
+                    $this->connection->unreference();
+                }
+            }
+        );
+    }
+
     private function startConnection(Deferred $deferred, EndPointDiscoverer $endPointDiscoverer): void
     {
         $this->logDebug('startConnection');
@@ -184,6 +209,7 @@ private function startConnection(Deferred $deferred, EndPointDiscoverer $endPoin
                 $this->timerTickWatcherId = Loop::repeat(Consts::TIMER_PERIOD, function (): void {
                     $this->timerTick();
                 });
+                Loop::unreference($this->timerTickWatcherId);
                 $this->endPointDiscoverer = $endPointDiscoverer;
                 $this->state = ConnectionState::connecting();
                 $this->connectingPhase = ConnectingPhase::reconnecting();
@@ -329,6 +355,10 @@ function (TcpPackageConnection $connection, Throwable $exception): void {
 
             yield $this->connection->connectAsync();
 
+            if ($this->currentPromises->count() === 0) {
+                $this->connection->unreference();
+            }
+
             if (null !== $this->connection && ! $this->connection->isClosed()) {
                 $this->connection->startReceiving();
             }

diff --git a/src/Internal/EventStoreNodeConnection.php b/src/Internal/EventStoreNodeConnection.php
@@ -31,6 +31,7 @@
 use Prooph\EventStore\EventData;
 use Prooph\EventStore\EventReadResult;
 use Prooph\EventStore\EventReadStatus;
+use Prooph\EventStore\EventStoreSubscription;
 use Prooph\EventStore\Exception\InvalidArgumentException;
 use Prooph\EventStore\Exception\InvalidOperationException;
 use Prooph\EventStore\Exception\MaxQueueSizeLimitReached;
@@ -45,6 +46,7 @@
 use Prooph\EventStore\StreamMetadata;
 use Prooph\EventStore\StreamMetadataResult;
 use Prooph\EventStore\SubscriptionDropped;
+use Prooph\EventStore\SubscriptionDropReason;
 use Prooph\EventStore\SystemSettings;
 use Prooph\EventStore\UserCredentials;
 use Prooph\EventStore\Util\Guid;
@@ -647,6 +649,32 @@ public function subscribeToStreamAsync(
             throw new InvalidArgumentException('Stream cannot be empty');
         }
 
+        $stopped = new Deferred();
+
+        $subscriptionDropped = new class($subscriptionDropped, $stopped) implements SubscriptionDropped {
+            private ?SubscriptionDropped $callback;
+
+            private Deferred $stopped;
+
+            public function __construct(?SubscriptionDropped $callback, Deferred $stopped)
+            {
+                $this->callback = $callback;
+                $this->stopped = $stopped;
+            }
+
+            public function __invoke(
+                EventStoreSubscription $subscription,
+                SubscriptionDropReason $reason,
+                ?Throwable $exception = null
+            ): void {
+                if ($this->callback) {
+                    ($this->callback)($subscription, $reason, $exception);
+                }
+
+                $this->stopped->resolve();
+            }
+        };
+
         $deferred = new Deferred();
 
         $this->handler->enqueueMessage(new StartSubscriptionMessage(
@@ -657,7 +685,8 @@ public function subscribeToStreamAsync(
             $eventAppeared,
             $subscriptionDropped,
             $this->settings->maxRetries(),
-            $this->settings->operationTimeout()
+            $this->settings->operationTimeout(),
+            $stopped->promise()
         ));
 
         return $deferred->promise();
@@ -705,6 +734,32 @@ public function subscribeToAllAsync(
         ?SubscriptionDropped $subscriptionDropped = null,
         ?UserCredentials $userCredentials = null
     ): Promise {
+        $stopped = new Deferred();
+
+        $subscriptionDropped = new class($subscriptionDropped, $stopped) implements SubscriptionDropped {
+            private ?SubscriptionDropped $callback;
+
+            private Deferred $stopped;
+
+            public function __construct(?SubscriptionDropped $callback, Deferred $stopped)
+            {
+                $this->callback = $callback;
+                $this->stopped = $stopped;
+            }
+
+            public function __invoke(
+                EventStoreSubscription $subscription,
+                SubscriptionDropReason $reason,
+                ?Throwable $exception = null
+            ): void {
+                if ($this->callback) {
+                    ($this->callback)($subscription, $reason, $exception);
+                }
+
+                $this->stopped->resolve();
+            }
+        };
+
         $deferred = new Deferred();
 
         $this->handler->enqueueMessage(new StartSubscriptionMessage(
@@ -715,7 +770,8 @@ public function subscribeToAllAsync(
             $eventAppeared,
             $subscriptionDropped,
             $this->settings->maxRetries(),
-            $this->settings->operationTimeout()
+            $this->settings->operationTimeout(),
+            $stopped->promise()
         ));
 
         return $deferred->promise();

diff --git a/src/Internal/EventStorePersistentSubscription.php b/src/Internal/EventStorePersistentSubscription.php
@@ -104,7 +104,8 @@ public function startSubscription(
         ?UserCredentials $userCredentials,
         Closure $onEventAppeared,
         ?Closure $onSubscriptionDropped,
-        ConnectionSettings $settings
+        ConnectionSettings $settings,
+        Promise $stopped
     ): Promise {
         $deferred = new Deferred();
 
@@ -117,7 +118,8 @@ public function startSubscription(
             $onEventAppeared,
             $onSubscriptionDropped,
             $settings->maxRetries(),
-            $settings->operationTimeout()
+            $settings->operationTimeout(),
+            $stopped
         ));
 
         return $deferred->promise();
@@ -149,7 +151,8 @@ public function start(): Promise
             $this->userCredentials,
             $eventAppeared,
             $subscriptionDropped,
-            $this->settings
+            $this->settings,
+            $this->stopped->promise()
         );
 
         $deferred = new Deferred();

diff --git a/src/Internal/Message/CloseConnectionMessage.php b/src/Internal/Message/CloseConnectionMessage.php
@@ -13,6 +13,7 @@
 
 namespace Prooph\EventStoreClient\Internal\Message;
 
+use Amp\Promise;
 use Throwable;
 
 /** @internal */
@@ -37,6 +38,11 @@ public function exception(): ?Throwable
         return $this->exception;
     }
 
+    public function promise(): ?Promise
+    {
+        return null;
+    }
+
     public function __toString(): string
     {
         return 'CloseConnectionMessage';

diff --git a/src/Internal/Message/EstablishTcpConnectionMessage.php b/src/Internal/Message/EstablishTcpConnectionMessage.php
@@ -13,6 +13,7 @@
 
 namespace Prooph\EventStoreClient\Internal\Message;
 
+use Amp\Promise;
 use Prooph\EventStoreClient\Internal\NodeEndPoints;
 
 /** @internal */
@@ -30,6 +31,11 @@ public function nodeEndPoints(): NodeEndPoints
         return $this->nodeEndPoints;
     }
 
+    public function promise(): ?Promise
+    {
+        return null;
+    }
+
     public function __toString(): string
     {
         return 'EstablishTcpConnectionMessage';

diff --git a/src/Internal/Message/HandleTcpPackageMessage.php b/src/Internal/Message/HandleTcpPackageMessage.php
@@ -13,6 +13,7 @@
 
 namespace Prooph\EventStoreClient\Internal\Message;
 
+use Amp\Promise;
 use Prooph\EventStoreClient\SystemData\TcpPackage;
 use Prooph\EventStoreClient\Transport\Tcp\TcpPackageConnection;
 
@@ -38,6 +39,11 @@ public function tcpPackage(): TcpPackage
         return $this->tcpPackage;
     }
 
+    public function promise(): ?Promise
+    {
+        return null;
+    }
+
     public function __toString(): string
     {
         return 'HandleTcpPackageMessage';

diff --git a/src/Internal/Message/Message.php b/src/Internal/Message/Message.php
@@ -13,8 +13,12 @@
 
 namespace Prooph\EventStoreClient\Internal\Message;
 
+use Amp\Promise;
+
 /** @internal */
 interface Message
 {
+    public function promise(): ?Promise;
+
     public function __toString(): string;
 }
diff --git a/src/Internal/Message/StartConnectionMessage.php b/src/Internal/Message/StartConnectionMessage.php
@@ -14,6 +14,7 @@
 namespace Prooph\EventStoreClient\Internal\Message;
 
 use Amp\Deferred;
+use Amp\Promise;
 use Prooph\EventStoreClient\Internal\EndPointDiscoverer;
 
 /** @internal */
@@ -38,6 +39,11 @@ public function endPointDiscoverer(): EndPointDiscoverer
         return $this->endPointDiscoverer;
     }
 
+    public function promise(): ?Promise
+    {
+        return $this->deferred->promise();
+    }
+
     public function __toString(): string
     {
         return 'StartConnectionMessage';

diff --git a/src/Internal/Message/StartOperationMessage.php b/src/Internal/Message/StartOperationMessage.php
@@ -13,6 +13,7 @@
 
 namespace Prooph\EventStoreClient\Internal\Message;
 
+use Amp\Promise;
 use Prooph\EventStoreClient\ClientOperations\ClientOperation;
 
 /** @internal  */
@@ -44,6 +45,11 @@ public function timeout(): int
         return $this->timeout;
     }
 
+    public function promise(): ?Promise
+    {
+        return $this->operation->promise();
+    }
+
     public function __toString(): string
     {
         return 'StartOperationMessage';

diff --git a/src/Internal/Message/StartPersistentSubscriptionMessage.php b/src/Internal/Message/StartPersistentSubscriptionMessage.php
@@ -14,6 +14,7 @@
 namespace Prooph\EventStoreClient\Internal\Message;
 
 use Amp\Deferred;
+use Amp\Promise;
 use Closure;
 use Prooph\EventStore\UserCredentials;
 
@@ -29,6 +30,7 @@ class StartPersistentSubscriptionMessage implements Message
     private ?Closure $subscriptionDropped;
     private int $maxRetries;
     private int $timeout;
+    private Promise $stopped;
 
     public function __construct(
         Deferred $deferred,
@@ -39,7 +41,8 @@ public function __construct(
         Closure $eventAppeared,
         ?Closure $subscriptionDropped,
         int $maxRetries,
-        int $timeout
+        int $timeout,
+        Promise $stopped
     ) {
         $this->deferred = $deferred;
         $this->subscriptionId = $subscriptionId;
@@ -50,6 +53,7 @@ public function __construct(
         $this->subscriptionDropped = $subscriptionDropped;
         $this->maxRetries = $maxRetries;
         $this->timeout = $timeout;
+        $this->stopped = $stopped;
     }
 
     public function deferred(): Deferred
@@ -97,6 +101,11 @@ public function timeout(): int
         return $this->timeout;
     }
 
+    public function promise(): ?Promise
+    {
+        return $this->stopped;
+    }
+
     public function __toString(): string
     {
         return 'StartPersistentSubscriptionMessage';