refactor event appeared callbacks

examples/demo-persistent-subscription-with-dns-cluster.php

@@ -19,7 +19,6 @@
 use Prooph\EventStoreClient\Exception\InvalidOperationException;
 use Prooph\EventStoreClient\Internal\EventStorePersistentSubscription;
 use Prooph\EventStoreClient\Internal\PersistentSubscriptionCreateResult;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Throwable;
 
 require __DIR__ . '/../vendor/autoload.php';

examples/demo-persistent-subscription.php

@@ -19,7 +19,6 @@
 use Prooph\EventStoreClient\Exception\InvalidOperationException;
 use Prooph\EventStoreClient\Internal\EventStorePersistentSubscription;
 use Prooph\EventStoreClient\Internal\PersistentSubscriptionCreateResult;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Throwable;
 
 require __DIR__ . '/../vendor/autoload.php';

examples/demo-subscribe-to-all-from.php

@@ -17,7 +17,6 @@
 use Amp\Promise;
 use Amp\Success;
 use Prooph\EventStoreClient\Internal\EventStoreCatchUpSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Throwable;
 
 require __DIR__ . '/../vendor/autoload.php';

examples/demo-subscribe-to-all.php

@@ -16,7 +16,6 @@
 use Amp\Loop;
 use Amp\Promise;
 use Amp\Success;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Prooph\EventStoreClient\Internal\VolatileEventStoreSubscription;
 use Throwable;
 

examples/demo-subscribe-to-stream-from-with-logger.php

@@ -17,7 +17,6 @@
 use Amp\Promise;
 use Amp\Success;
 use Prooph\EventStoreClient\Internal\EventStoreCatchUpSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Throwable;
 
 require __DIR__ . '/../vendor/autoload.php';

examples/demo-subscribe-to-stream-from.php

@@ -17,7 +17,6 @@
 use Amp\Promise;
 use Amp\Success;
 use Prooph\EventStoreClient\Internal\EventStoreCatchUpSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Throwable;
 
 require __DIR__ . '/../vendor/autoload.php';

examples/demo-subscribe-to-stream.php

@@ -16,7 +16,6 @@
 use Amp\Loop;
 use Amp\Promise;
 use Amp\Success;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Prooph\EventStoreClient\Internal\VolatileEventStoreSubscription;
 use Throwable;
 

src/ClientOperations/AbstractSubscriptionOperation.php

@@ -19,7 +19,6 @@
 use Amp\Success;
 use Generator;
 use Prooph\EventStoreClient\EndPoint;
-use Prooph\EventStoreClient\EventAppearedOnSubscription;
 use Prooph\EventStoreClient\EventStoreSubscription;
 use Prooph\EventStoreClient\Exception\AccessDeniedException;
 use Prooph\EventStoreClient\Exception\ConnectionClosedException;
@@ -34,7 +33,6 @@
 use Prooph\EventStoreClient\Messages\ClientMessages\SubscriptionDropped as SubscriptionDroppedMessage;
 use Prooph\EventStoreClient\Messages\ClientMessages\SubscriptionDropped_SubscriptionDropReason as SubscriptionDropReasonMessage;
 use Prooph\EventStoreClient\Messages\ClientMessages\UnsubscribeFromStream;
-use Prooph\EventStoreClient\SubscriptionDropped;
 use Prooph\EventStoreClient\SubscriptionDropReason;
 use Prooph\EventStoreClient\SystemData\InspectionDecision;
 use Prooph\EventStoreClient\SystemData\InspectionResult;
@@ -61,9 +59,9 @@ abstract class AbstractSubscriptionOperation implements SubscriptionOperation
     protected $resolveLinkTos;
     /** @var UserCredentials|null */
     protected $userCredentials;
-    /** @var EventAppearedOnSubscription */
+    /** @var callable */
     protected $eventAppeared;
-    /** @var SubscriptionDropped|null */
+    /** @var callable|null */
     private $subscriptionDropped;
     /** @var bool */
     private $verboseLogging;
@@ -86,8 +84,8 @@ public function __construct(
         string $streamId,
         bool $resolveLinkTos,
         ?UserCredentials $userCredentials,
-        EventAppearedOnSubscription $eventAppeared,
-        ?SubscriptionDropped $subscriptionDropped,
+        callable $eventAppeared,
+        ?callable $subscriptionDropped,
         bool $verboseLogging,
         callable $getConnection
     ) {

src/ClientOperations/ConnectToPersistentSubscriptionOperation.php

@@ -14,7 +14,6 @@
 namespace Prooph\EventStoreClient\ClientOperations;
 
 use Amp\Deferred;
-use Prooph\EventStoreClient\EventAppearedOnSubscription;
 use Prooph\EventStoreClient\EventId;
 use Prooph\EventStoreClient\EventStoreSubscription;
 use Prooph\EventStoreClient\Exception\AccessDeniedException;
@@ -60,8 +59,8 @@ public function __construct(
         int $bufferSize,
         string $streamId,
         ?UserCredentials $userCredentials,
-        EventAppearedOnSubscription $eventAppeared,
-        ?SubscriptionDropped $subscriptionDropped,
+        callable $eventAppeared,
+        ?callable $subscriptionDropped,
         bool $verboseLogging,
         callable $getConnection
     ) {

src/EventAppearedOnCatchupSubscription.php

@@ -15,7 +15,6 @@
 
 use Amp\Promise;
 use Prooph\EventStoreClient\Internal\EventStoreCatchUpSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 
 interface EventAppearedOnCatchupSubscription
 {

src/EventAppearedOnPersistentSubscription.php

@@ -15,7 +15,6 @@
 
 use Amp\Promise;
 use Prooph\EventStoreClient\Internal\EventStorePersistentSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 
 interface EventAppearedOnPersistentSubscription
 {

src/EventAppearedOnSubscription.php

@@ -14,7 +14,6 @@
 namespace Prooph\EventStoreClient;
 
 use Amp\Promise;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 
 interface EventAppearedOnSubscription
 {

src/Internal/EventStoreCatchUpSubscription.php

@@ -25,7 +25,6 @@
 use Prooph\EventStoreClient\EventAppearedOnSubscription;
 use Prooph\EventStoreClient\EventStoreConnection;
 use Prooph\EventStoreClient\EventStoreSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent as InternalResolvedEvent;
 use Prooph\EventStoreClient\LiveProcessingStarted;
 use Prooph\EventStoreClient\ResolvedEvent;
 use Prooph\EventStoreClient\SubscriptionDropped;
@@ -287,7 +286,7 @@ public function __construct(callable $callback)
 
                     public function __invoke(
                         EventStoreSubscription $subscription,
-                        InternalResolvedEvent $resolvedEvent
+                        ResolvedEvent $resolvedEvent
                     ): Promise {
                         return ($this->callback)($subscription, $resolvedEvent);
                     }

src/Internal/EventStorePersistentSubscription.php

@@ -21,17 +21,14 @@
 use Generator;
 use Prooph\EventStoreClient\ConnectionSettings;
 use Prooph\EventStoreClient\EventAppearedOnPersistentSubscription;
-use Prooph\EventStoreClient\EventAppearedOnSubscription;
 use Prooph\EventStoreClient\EventId;
-use Prooph\EventStoreClient\EventStoreSubscription;
 use Prooph\EventStoreClient\Exception\RuntimeException;
 use Prooph\EventStoreClient\Internal\Message\StartPersistentSubscriptionMessage;
 use Prooph\EventStoreClient\Internal\ResolvedEvent as InternalResolvedEvent;
 use Prooph\EventStoreClient\PersistentSubscriptionDropped;
 use Prooph\EventStoreClient\PersistentSubscriptionNakEventAction;
 use Prooph\EventStoreClient\PersistentSubscriptionResolvedEvent;
 use Prooph\EventStoreClient\ResolvedEvent;
-use Prooph\EventStoreClient\SubscriptionDropped;
 use Prooph\EventStoreClient\SubscriptionDropReason;
 use Prooph\EventStoreClient\UserCredentials;
 use Psr\Log\LoggerInterface as Logger;
@@ -53,9 +50,9 @@ class EventStorePersistentSubscription
     private $subscriptionId;
     /** @var string */
     private $streamId;
-    /** @var EventAppearedOnPersistentSubscription */
+    /** @var callable */
     private $eventAppeared;
-    /** @var PersistentSubscriptionDropped|null */
+    /** @var callable|null */
     private $subscriptionDropped;
     /** @var UserCredentials|null */
     private $userCredentials;
@@ -117,13 +114,14 @@ public function __construct(
         $this->handler = $handler;
     }
 
+    /** @internal */
     public function startSubscription(
         string $subscriptionId,
         string $streamId,
         int $bufferSize,
         ?UserCredentials $userCredentials,
-        EventAppearedOnSubscription $onEventAppeared,
-        ?SubscriptionDropped $onSubscriptionDropped,
+        callable $onEventAppeared,
+        ?callable $onSubscriptionDropped,
         ConnectionSettings $settings
     ): Promise {
         $deferred = new Deferred();
@@ -152,54 +150,21 @@ public function start(): Promise
     {
         $this->stopped->reset();
 
-        $eventAppearedCallback = function (
-            EventStoreSubscription $subscription,
+        $eventAppeared = function (
+            PersistentEventStoreSubscription $subscription,
             PersistentSubscriptionResolvedEvent $resolvedEvent
         ): Promise {
             return $this->onEventAppeared($resolvedEvent);
         };
 
-        $subscriptionDroppedCallback = function (
-            EventStoreSubscription $subscription,
+        $subscriptionDropped = function (
+            PersistentEventStoreSubscription $subscription,
             SubscriptionDropReason $reason,
             ?Throwable $exception
         ): void {
             $this->onSubscriptionDropped($reason, $exception);
         };
 
-        $eventAppeared = new class($eventAppearedCallback) implements EventAppearedOnSubscription {
-            private $callback;
-
-            public function __construct(callable  $callback)
-            {
-                $this->callback = $callback;
-            }
-
-            public function __invoke(
-                EventStoreSubscription $subscription,
-                InternalResolvedEvent $resolvedEvent
-            ): Promise {
-                return ($this->callback)($subscription, $resolvedEvent);
-            }
-        };
-
-        $subscriptionDropped = new class($subscriptionDroppedCallback) implements SubscriptionDropped {
-            private $callback;
-
-            public function __construct(callable  $callback)
-            {
-                $this->callback = $callback;
-            }
-
-            public function __invoke(
-                EventStoreSubscription $subscription,
-                SubscriptionDropReason $reason,
-                ?Throwable $exception = null
-            ): void {
-                ($this->callback)($subscription, $reason, $exception);
-            }
-        };
-
         $promise = $this->startSubscription(
             $this->subscriptionId,
             $this->streamId,

src/Internal/Message/StartPersistentSubscriptionMessage.php

@@ -14,8 +14,6 @@
 namespace Prooph\EventStoreClient\Internal\Message;
 
 use Amp\Deferred;
-use Prooph\EventStoreClient\EventAppearedOnSubscription;
-use Prooph\EventStoreClient\SubscriptionDropped;
 use Prooph\EventStoreClient\UserCredentials;
 
 /** @internal  */
@@ -31,9 +29,9 @@ class StartPersistentSubscriptionMessage implements Message
     private $bufferSize;
     /** @var UserCredentials|null */
     private $userCredentials;
-    /** @var EventAppearedOnSubscription */
+    /** @var callable */
     private $eventAppeared;
-    /** @var SubscriptionDropped|null */
+    /** @var callable|null */
     private $subscriptionDropped;
     /** @var int */
     private $maxRetries;
@@ -46,8 +44,8 @@ public function __construct(
         string $streamId,
         int $bufferSize,
         ?UserCredentials $userCredentials,
-        EventAppearedOnSubscription $eventAppeared,
-        ?SubscriptionDropped $subscriptionDropped,
+        callable $eventAppeared,
+        ?callable $subscriptionDropped,
         int $maxRetries,
         int $timeout
     ) {
@@ -87,12 +85,12 @@ public function userCredentials(): ?UserCredentials
         return $this->userCredentials;
     }
 
-    public function eventAppeared(): EventAppearedOnSubscription
+    public function eventAppeared(): callable
     {
         return $this->eventAppeared;
     }
 
-    public function subscriptionDropped(): ?SubscriptionDropped
+    public function subscriptionDropped(): ?callable
     {
         return $this->subscriptionDropped;
     }

tests/PersistentSubscriptionManagement/persistent_subscription_manager.php

@@ -23,11 +23,11 @@
 use Prooph\EventStoreClient\Exception\InvalidArgumentException;
 use Prooph\EventStoreClient\ExpectedVersion;
 use Prooph\EventStoreClient\Internal\EventStorePersistentSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Prooph\EventStoreClient\PersistentSubscriptionNakEventAction;
 use Prooph\EventStoreClient\PersistentSubscriptions\PersistentSubscriptionDetails;
 use Prooph\EventStoreClient\PersistentSubscriptions\PersistentSubscriptionsManager;
 use Prooph\EventStoreClient\PersistentSubscriptionSettings;
+use Prooph\EventStoreClient\ResolvedEvent;
 use Prooph\EventStoreClient\Util\Guid;
 use ProophTest\EventStoreClient\CountdownEvent;
 use ProophTest\EventStoreClient\DefaultData;
@@ -259,7 +259,7 @@ public function __invoke(
 
             $this->sub->stop();
 
-            $this->manager->replayParkedMessages($this->stream, 'existing');
+            yield $this->manager->replayParkedMessages($this->stream, 'existing', DefaultData::adminCredentials());
 
             $event = new CountdownEvent(2);
 

tests/a_nak_in_subscription_handler_in_autoack_mode_drops_the_subscription.php

@@ -25,9 +25,9 @@
 use Prooph\EventStoreClient\EventId;
 use Prooph\EventStoreClient\ExpectedVersion;
 use Prooph\EventStoreClient\Internal\EventStorePersistentSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Prooph\EventStoreClient\PersistentSubscriptionDropped;
 use Prooph\EventStoreClient\PersistentSubscriptionSettings;
+use Prooph\EventStoreClient\ResolvedEvent;
 use Prooph\EventStoreClient\SubscriptionDropReason;
 use Prooph\EventStoreClient\Util\Guid;
 use Throwable;

tests/catch_up_subscription_handles_errors.php

@@ -33,11 +33,11 @@
 use Prooph\EventStoreClient\EventStoreSubscription;
 use Prooph\EventStoreClient\Internal\EventStoreCatchUpSubscription;
 use Prooph\EventStoreClient\Internal\EventStoreStreamCatchUpSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Prooph\EventStoreClient\Internal\VolatileEventStoreSubscription;
 use Prooph\EventStoreClient\LiveProcessingStarted;
 use Prooph\EventStoreClient\ReadDirection;
 use Prooph\EventStoreClient\RecordedEvent;
+use Prooph\EventStoreClient\ResolvedEvent;
 use Prooph\EventStoreClient\SliceReadStatus;
 use Prooph\EventStoreClient\StreamEventsSlice;
 use Prooph\EventStoreClient\SubscriptionDropped;

tests/catchup_subscription_handles_small_batch_sizes.php

@@ -25,8 +25,8 @@
 use Prooph\EventStoreClient\EventStoreConnection;
 use Prooph\EventStoreClient\ExpectedVersion;
 use Prooph\EventStoreClient\Internal\EventStoreCatchUpSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Prooph\EventStoreClient\LiveProcessingStarted;
+use Prooph\EventStoreClient\ResolvedEvent;
 use ProophTest\EventStoreClient\Helper\TestConnection;
 use Throwable;
 use function Amp\call;

tests/connect_to_existing_persistent_subscription_with_max_one_client.php

@@ -20,8 +20,8 @@
 use Prooph\EventStoreClient\EventAppearedOnPersistentSubscription;
 use Prooph\EventStoreClient\Exception\MaximumSubscribersReachedException;
 use Prooph\EventStoreClient\Internal\EventStorePersistentSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Prooph\EventStoreClient\PersistentSubscriptionSettings;
+use Prooph\EventStoreClient\ResolvedEvent;
 use Prooph\EventStoreClient\Util\Guid;
 use Throwable;
 

tests/connect_to_existing_persistent_subscription_with_permissions.php

@@ -19,8 +19,8 @@
 use PHPUnit\Framework\TestCase;
 use Prooph\EventStoreClient\EventAppearedOnPersistentSubscription;
 use Prooph\EventStoreClient\Internal\EventStorePersistentSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Prooph\EventStoreClient\PersistentSubscriptionSettings;
+use Prooph\EventStoreClient\ResolvedEvent;
 use Prooph\EventStoreClient\Util\Guid;
 use Throwable;
 

tests/connect_to_existing_persistent_subscription_with_start_from_beginning_and_events_in_it.php

@@ -23,8 +23,8 @@
 use Prooph\EventStoreClient\EventId;
 use Prooph\EventStoreClient\ExpectedVersion;
 use Prooph\EventStoreClient\Internal\EventStorePersistentSubscription;
-use Prooph\EventStoreClient\Internal\ResolvedEvent;
 use Prooph\EventStoreClient\PersistentSubscriptionSettings;
+use Prooph\EventStoreClient\ResolvedEvent;
 use Prooph\EventStoreClient\Util\Guid;
 use Throwable;
 use function Amp\call;