added receiveRecover for persistence

src/Phluxor/Persistence/Message/ReplayComplete.php → src/Phluxor/Persistence/Message/ReplayCompleted.php

@@ -4,6 +4,6 @@
 
 namespace Phluxor\Persistence\Message;
 
-final class ReplayComplete
+final class ReplayCompleted
 {
 }

src/Phluxor/Persistence/Mixin.php

@@ -9,7 +9,8 @@
 use Phluxor\ActorSystem\Context\ReceiverInterface;
 use Phluxor\ActorSystem\Context\ReceiverPartInterface;
 use Phluxor\ActorSystem\Message\MessageEnvelope;
-use Phluxor\Persistence\Message\ReplayComplete;
+use Phluxor\Persistence\Message\OfferSnapshot;
+use Phluxor\Persistence\Message\ReplayCompleted;
 use Phluxor\Persistence\Message\RequestSnapshot;
 use RuntimeException;
 
@@ -58,19 +59,26 @@ public function init(
         $result = $this->providerState->getSnapshot($this->name());
         if ($result->isOk()) {
             $this->eventIndex = $result->getEventIndex();
-            $this->receiver->receive(new MessageEnvelope(header: null, message: $result->getSnapshot()));
+            $messageEnvelope = new MessageEnvelope(header: null, message: $result->getSnapshot());
+            $this->receiver->receive($messageEnvelope);
+            $offerSnapshot = new OfferSnapshot($result->getSnapshot());
+            $this->receiveRecover($offerSnapshot);
         }
         $this->providerState->getEvents(
             $this->name(),
             $this->eventIndex,
             0,
             function (mixed $event) use ($receiver) {
-                $this->receiver->receive(new MessageEnvelope(header: null, message: $event));
+                $messageEnvelope = new MessageEnvelope(header: null, message: $event);
+                $this->receiver->receive($messageEnvelope);
+                $this->receiveRecover($messageEnvelope->getMessage());
                 $this->eventIndex++;
             }
         );
         $this->recovering = false;
-        $this->receiver->receive(new MessageEnvelope(header: null, message: new ReplayComplete()));
+        $messageEnvelope = new MessageEnvelope(header: null, message: new ReplayCompleted());
+        $this->receiver->receive($messageEnvelope);
+        $this->receiveRecover($messageEnvelope->getMessage());
     }
 
     public function persistenceReceive(Message $message): void
@@ -93,4 +101,14 @@ public function name(): string
     {
         return $this->name;
     }
+
+    /**
+     * @param mixed $message
+     * @return void
+     */
+    public function receiveRecover(
+        mixed $message
+    ): void {
+        // Implement this method
+    }
 }

src/Phluxor/Persistence/PersistentInterface.php

@@ -32,4 +32,12 @@ public function persistenceSnapshot(Message $snapshot): void;
     public function recovering(): bool;
 
     public function name(): string;
+
+    /**
+     * @param mixed $message
+     * @return void
+     */
+    public function receiveRecover(
+        mixed $message
+    ): void;
 }

tests/Test/Persistence/Messages.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Persistence;
+
+final class Messages
+{
+}

tests/Test/Persistence/PersistenceTest.php

@@ -7,9 +7,14 @@
 use Phluxor\ActorSystem;
 use Phluxor\Persistence\EventSourcedBehavior;
 use Phluxor\Persistence\InMemoryProvider;
+use Phluxor\Persistence\Message\RequestSnapshot;
+use Phluxor\Persistence\Mixin;
+use Phluxor\Persistence\PersistentInterface;
 use PHPUnit\Framework\TestCase;
 use Test\Persistence\ProtoBuf\TestMessage;
 
+use Test\Persistence\ProtoBuf\TestSnapshot;
+
 use function Swoole\Coroutine\run;
 
 class PersistenceTest extends TestCase
@@ -69,4 +74,90 @@ public function testRecovers(): void
             });
         });
     }
+
+    public function testReceiveRecovery(): void
+    {
+        run(function () {
+            go(function () {
+                $system = ActorSystem::create();
+                $deadLetter = false;
+                $system->getEventStream()?->subscribe(function (mixed $event) use (&$deadLetter): void {
+                    if ($event instanceof ActorSystem\DeadLetterEvent) {
+                        $deadLetter = true;
+                    }
+                });
+                $props = ActorSystem\Props::fromProducer(fn() => $this->receiveRecoverActor(),
+                    ActorSystem\Props::withReceiverMiddleware(
+                        new EventSourcedBehavior(new InMemoryStateProvider(new InMemoryProvider(2)))
+                    ));
+                $ref = $system->root()->spawnNamed($props, 'test.actor');
+                $this->assertNull($ref->isError());
+                $system->root()->send($ref->getRef(), new TestMessage(['message' => 'hello']));
+                $f = $system->root()->requestFuture($ref->getRef(), new Query(), 1);
+                $this->assertSame('hello', $f->result()->value());
+
+                $system->root()->poisonFuture($ref->getRef())?->wait();
+                $system->root()->send($ref->getRef(), new TestMessage(['message' => 'world']));
+                $ref = $system->root()->spawnNamed($props, 'test.actor');
+                $this->assertNull($ref->isError());
+                $f = $system->root()->requestFuture($ref->getRef(), new Query(), 1);
+                $this->assertSame('hello', $f->result()->value());
+                $system->root()->send($ref->getRef(), new TestMessage(['message' => 'hello2']));
+                $system->root()->send($ref->getRef(), new TestMessage(['message' => 'hello3']));
+                $this->assertTrue($deadLetter);
+                $f = $system->root()->requestFuture($ref->getRef(), new Query(), 1);
+                $this->assertSame('hello3', $f->result()->value());
+                $f = $system->root()->requestFuture($ref->getRef(), new Messages(), 1);
+                $this->assertSame(
+                    [
+                        'Phluxor\Persistence\Message\OfferSnapshot',
+                        'Test\Persistence\ProtoBuf\TestMessage',
+                        'Phluxor\Persistence\Message\ReplayCompleted'
+                    ],
+                    $f->result()->value()
+                );
+            });
+        });
+    }
+
+    private function receiveRecoverActor(): ActorSystem\Message\ActorInterface
+    {
+        return new class() implements ActorSystem\Message\ActorInterface, PersistentInterface {
+            use Mixin;
+
+            private string $state = '';
+            /** @var string[] */
+            private array $receiveRecoverMessages = [];
+
+            public function receive(ActorSystem\Context\ContextInterface $context): void
+            {
+                $msg = $context->message();
+                switch (true) {
+                    case $msg instanceof RequestSnapshot:
+                        $this->persistenceSnapshot(new TestSnapshot(['message' => $this->state]));
+                        break;
+                    case $msg instanceof TestSnapshot:
+                        $this->state = $msg->getMessage();
+                        break;
+                    case $msg instanceof TestMessage:
+                        if (!$this->recovering()) {
+                            $this->persistenceReceive($msg);
+                        }
+                        $this->state = $msg->getMessage();
+                        break;
+                    case $msg instanceof Query:
+                        $context->respond($this->state);
+                        break;
+                    case $msg instanceof Messages:
+                        $context->respond($this->receiveRecoverMessages);
+                        break;
+                }
+            }
+
+            public function receiveRecover(mixed $message): void
+            {
+                $this->receiveRecoverMessages[] = get_debug_type($message);
+            }
+        };
+    }
 }