Merge pull request #16 from prooph/amphp

fix amphp problems
prooph · Sep 25, 2018 · b87dd1b · b87dd1b
2 parents deee1b5 + ba30ab5
commit b87dd1b
Show file tree

Hide file tree

Showing 59 changed files with 187 additions and 1,293 deletions.
diff --git a/examples/demo-subscription5.php b/examples/demo-subscription5.php
@@ -17,7 +17,6 @@
 use Amp\Success;
 use Prooph\EventStoreClient\Exception\InvalidOperationException;
 use Prooph\EventStoreClient\Internal\AbstractEventStorePersistentSubscription;
-use Prooph\EventStoreClient\Internal\EventStorePersistentSubscription;
 use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Prooph\EventStoreClient\Messages\ClientMessages\CreatePersistentSubscription;
 use Throwable;
@@ -60,7 +59,7 @@
 
     \var_dump($result);
 
-    $subscription = $connection->connectToPersistentSubscription(
+    $connection->connectToPersistentSubscriptionAsync(
         'foo-bar',
         'test-persistent-subscription',
         new class() implements EventAppearedOnPersistentSubscription {
@@ -92,7 +91,4 @@ public function __invoke(
         true,
         new UserCredentials('admin', 'changeit')
     );
-
-    /** @var EventStorePersistentSubscription $subscription */
-    $subscription = yield $subscription->start();
 });
diff --git a/examples/demo-subscription6.php b/examples/demo-subscription6.php
diff --git a/src/ClientOperations/AbstractSubscriptionOperation.php b/src/ClientOperations/AbstractSubscriptionOperation.php
@@ -13,6 +13,7 @@
 namespace Prooph\EventStoreClient\ClientOperations;
 
 use Amp\Deferred;
+use Amp\Loop;
 use Amp\Promise;
 use Amp\Success;
 use Generator;
@@ -371,7 +372,9 @@ private function executeActionAsync(callable $action): void
             $this->dropSubscription(SubscriptionDropReason::userInitiated(), new \Exception('client buffer too big'));
         }
 
-        Promise\rethrow($this->executeActions());
+        Loop::defer(function (): Generator {
+            yield $this->executeActions();
+        });
     }
 
     /** @return Promise<void> */

diff --git a/src/EventStoreAsyncConnection.php b/src/EventStoreAsyncConnection.php
@@ -14,7 +14,6 @@
 
 use Amp\Promise;
 use Prooph\EventStoreClient\Internal\EventStoreAllCatchUpSubscription;
-use Prooph\EventStoreClient\Internal\EventStorePersistentSubscription;
 use Prooph\EventStoreClient\Internal\EventStoreStreamCatchUpSubscription;
 use Prooph\EventStoreClient\Internal\ListenerHandler;
 
@@ -197,16 +196,6 @@ public function subscribeToAllFrom(
         ?UserCredentials $userCredentials = null
     ): EventStoreAllCatchUpSubscription;
 
-    public function connectToPersistentSubscription(
-        string $stream,
-        string $groupName,
-        EventAppearedOnPersistentSubscription $eventAppeared,
-        ?PersistentSubscriptionDropped $subscriptionDropped = null,
-        int $bufferSize = 10,
-        bool $autoAck = true,
-        ?UserCredentials $userCredentials = null
-    ): EventStorePersistentSubscription;
-
     /**
      * @return Promise<AbstractEventStorePersistentSubscription>
      */

diff --git a/src/Internal/AbstractEventStorePersistentSubscription.php b/src/Internal/AbstractEventStorePersistentSubscription.php
@@ -349,11 +349,7 @@ private function enqueue(PersistentSubscriptionResolvedEvent $resolvedEvent): vo
             $this->isProcessing = true;
 
             Loop::defer(function (): Generator {
-                $promise = $this->processQueue();
-
-                Promise\rethrow($promise);
-
-                yield $promise;
+                yield $this->processQueue();
             });
         }
     }

diff --git a/src/Internal/EventStoreAsyncNodeConnection.php b/src/Internal/EventStoreAsyncNodeConnection.php
@@ -616,7 +616,7 @@ public function subscribeToStreamFrom(
             $settings
         );
 
-        Promise\rethrow($catchUpSubscription->startAsync());
+        $catchUpSubscription->startAsync();
 
         return $catchUpSubscription;
     }
@@ -671,47 +671,11 @@ public function subscribeToAllFrom(
             $settings
         );
 
-        Promise\rethrow($catchUpSubscription->startAsync());
+        $catchUpSubscription->startAsync();
 
         return $catchUpSubscription;
     }
 
-    public function connectToPersistentSubscription(
-        string $stream,
-        string $groupName,
-        EventAppearedOnPersistentSubscription $eventAppeared,
-        ?PersistentSubscriptionDropped $subscriptionDropped = null,
-        int $bufferSize = 10,
-        bool $autoAck = true,
-        ?UserCredentials $userCredentials = null
-    ): EventStorePersistentSubscription {
-        if (empty($stream)) {
-            throw new InvalidArgumentException('Stream cannot be empty');
-        }
-
-        if (empty($groupName)) {
-            throw new InvalidArgumentException('Group cannot be empty');
-        }
-
-        $subscription = new EventStorePersistentSubscription(
-            $groupName,
-            $stream,
-            $eventAppeared,
-            $subscriptionDropped,
-            $userCredentials,
-            $this->settings->log(),
-            $this->settings->verboseLogging(),
-            $this->settings,
-            $this->handler,
-            $bufferSize,
-            $autoAck
-        );
-
-        Promise\rethrow($subscription->start());
-
-        return $subscription;
-    }
-
     /** {@inheritdoc} */
     public function connectToPersistentSubscriptionAsync(
         string $stream,

diff --git a/src/Internal/EventStoreCatchUpSubscription.php b/src/Internal/EventStoreCatchUpSubscription.php
@@ -221,10 +221,7 @@ private function onReconnect(ClientConnectionEventArgs $clientConnectionEventArg
         $this->connection->detach($this->connectListener);
 
         Loop::defer(function (): Generator {
-            $promise = $this->runSubscriptionAsync();
-            Promise\rethrow($promise);
-
-            yield $promise;
+            yield $this->runSubscriptionAsync();
         });
     }
 
@@ -462,10 +459,7 @@ private function ensureProcessingPushQueue(): void
             $this->isProcessing = true;
 
             Loop::defer(function (): Generator {
-                $promise = $this->processLiveQueueAsync();
-                Promise\rethrow($promise);
-
-                yield $promise;
+                yield $this->processLiveQueueAsync();
             });
         }
     }

diff --git a/src/Internal/EventStoreConnectionLogicHandler.php b/src/Internal/EventStoreConnectionLogicHandler.php
@@ -48,7 +48,6 @@
 use Prooph\EventStoreClient\SystemData\TcpPackage;
 use Prooph\EventStoreClient\Transport\Tcp\TcpPackageConnection;
 use Throwable;
-use function Amp\Promise\rethrow;
 
 /** @internal */
 class EventStoreConnectionLogicHandler
@@ -250,8 +249,6 @@ private function discoverEndPoint(?Deferred $deferred): void
                 $deferred->resolve(null);
             }
         });
-
-        rethrow($promise);
     }
 
     /** @throws \Exception */
@@ -334,10 +331,7 @@ function (TcpPackageConnection $connection, Throwable $exception): void {
         );
 
         Loop::defer(function (): Generator {
-            $promise = $this->connection->connectAsync();
-            rethrow($promise);
-
-            yield $promise;
+            yield $this->connection->connectAsync();
 
             if (! $this->connection->isClosed()) {
                 $this->connection->startReceiving();

diff --git a/src/Internal/EventStoreSyncNodeConnection.php b/src/Internal/EventStoreSyncNodeConnection.php
@@ -20,7 +20,6 @@
 use Prooph\EventStoreClient\ClusterSettings;
 use Prooph\EventStoreClient\ConditionalWriteResult;
 use Prooph\EventStoreClient\ConnectionSettings;
-use Prooph\EventStoreClient\EventAppearedOnSubscription;
 use Prooph\EventStoreClient\EventReadResult;
 use Prooph\EventStoreClient\EventStoreAsyncConnection;
 use Prooph\EventStoreClient\EventStoreSyncConnection;
@@ -31,7 +30,6 @@
 use Prooph\EventStoreClient\StreamEventsSlice;
 use Prooph\EventStoreClient\StreamMetadata;
 use Prooph\EventStoreClient\StreamMetadataResult;
-use Prooph\EventStoreClient\SubscriptionDropped;
 use Prooph\EventStoreClient\SystemSettings;
 use Prooph\EventStoreClient\UserCredentials;
 use Prooph\EventStoreClient\WriteResult;
@@ -273,27 +271,6 @@ public function deletePersistentSubscription(
         return Promise\wait($this->asyncConnection->deletePersistentSubscriptionAsync($stream, $groupName, $userCredentials));
     }
 
-    /** @throws Throwable */
-    public function connectToPersistentSubscription(
-        string $stream,
-        string $groupName,
-        EventAppearedOnSubscription $eventAppeared,
-        ?SubscriptionDropped $subscriptionDropped = null,
-        int $bufferSize = 10,
-        bool $autoAck = true,
-        ?UserCredentials $userCredentials = null
-    ): EventStorePersistentSubscription {
-        return $this->asyncConnection->connectToPersistentSubscription(
-            $stream,
-            $groupName,
-            $eventAppeared,
-            $subscriptionDropped,
-            $bufferSize,
-            $autoAck,
-            $userCredentials
-        );
-    }
-
     public function startTransaction(
         string $stream,
         int $expectedVersion,

diff --git a/tests/SpecificationWithConnection.php b/tests/SpecificationWithConnection.php
@@ -33,7 +33,7 @@ protected function given(): Generator
     abstract protected function when();
 
     /** @throws Throwable */
-    protected function executeCallback(callable $test): void
+    protected function execute(callable $test): void
     {
         wait(call(function () use ($test) {
             $this->conn = TestConnection::createAsync();
@@ -52,8 +52,7 @@ protected function executeCallback(callable $test): void
 
     protected function end(): Generator
     {
-        // @todo is this a bug?
-        //$this->conn->close();
+        $this->conn->close();
 
         yield new Success();
     }

diff --git a/tests/a_nak_in_subscription_handler_in_autoack_mode_drops_the_subscription.php b/tests/a_nak_in_subscription_handler_in_autoack_mode_drops_the_subscription.php
@@ -77,7 +77,7 @@ protected function given(): Generator
             $this->resetEvent->resolve(true);
         };
 
-        $this->conn->connectToPersistentSubscription(
+        yield $this->conn->connectToPersistentSubscriptionAsync(
             $this->stream,
             $this->group,
             new class() implements EventAppearedOnPersistentSubscription {
@@ -129,7 +129,7 @@ protected function when(): Generator
      */
     public function the_subscription_gets_dropped(): void
     {
-        $this->executeCallback(function (): Generator {
+        $this->execute(function (): Generator {
             try {
                 $result = yield Promise\timeout($this->resetEvent->promise(), 5000);
             } catch (TimeoutException $e) {

diff --git a/tests/can_create_duplicate_persistent_subscription_group_name_on_different_streams.php b/tests/can_create_duplicate_persistent_subscription_group_name_on_different_streams.php
@@ -53,7 +53,7 @@ protected function when(): Generator
      */
     public function the_completion_succeeds(): void
     {
-        $this->executeCallback(function () {
+        $this->execute(function () {
             yield $this->conn->createPersistentSubscriptionAsync(
                 'someother' . $this->stream,
                 'group3211',