Fixed in memory implementation of event store

Author: benatespina

src/Kreta/SharedKernel/Infrastructure/Persistence/InMemory/EventStore/InMemoryEventStore.php

@@ -15,6 +15,7 @@
 namespace Kreta\SharedKernel\Infrastructure\Persistence\InMemory\EventStore;
 
 use Kreta\SharedKernel\Domain\Model\AggregateDoesNotExistException;
+use Kreta\SharedKernel\Domain\Model\DomainEvent;
 use Kreta\SharedKernel\Domain\Model\DomainEventCollection;
 use Kreta\SharedKernel\Event\EventStore;
 use Kreta\SharedKernel\Event\Stream;
@@ -29,7 +30,7 @@ public function __construct()
         $this->store = [];
     }
 
-    public function appendTo(Stream $stream) : void
+    public function append(Stream $stream) : void
     {
         foreach ($stream->events() as $event) {
             $content = [];
@@ -52,18 +53,7 @@ public function streamOfName(StreamName $name) : Stream
         $events = new DomainEventCollection();
         foreach ($this->store as $event) {
             if ($event['stream_name'] === $name->name()) {
-                $eventData = json_decode($event['content']);
-                $eventReflection = new \ReflectionClass($event['type']);
-                $parameters = $eventReflection->getConstructor()->getParameters();
-                $arguments = [];
-                foreach ($parameters as $parameter) {
-                    foreach ($eventData as $key => $data) {
-                        if ($key === $parameter->getName()) {
-                            $arguments[] = $data;
-                        }
-                    }
-                }
-                $events->add(new $event['type'](...$arguments));
+                $events->add($this->buildEvent($event));
             }
         }
         if (0 === $events->count()) {
@@ -72,4 +62,42 @@ public function streamOfName(StreamName $name) : Stream
 
         return new Stream($name, $events);
     }
+
+    public function eventsSince(?\DateTimeInterface $since, int $offset = 0, int $limit = -1) : array
+    {
+        $since = $since instanceof \DateTimeInterface ? $since->getTimestamp() : 0;
+
+        $events = array_map(function (array $event) use ($since) {
+            $domainEvent = $this->buildEvent($event);
+            if ($domainEvent->occurredOn()->getTimestamp() >= $since) {
+                $evenContent = json_decode($event['content'], true);
+                $evenContent['occurredOn'] = $domainEvent->occurredOn()->getTimestamp();
+
+                return [
+                    'stream_name' => $event['stream_name'],
+                    'type'        => $event['type'],
+                    'content'     => $evenContent,
+                ];
+            }
+        }, $this->store);
+
+        return array_slice($events, $offset);
+    }
+
+    private function buildEvent(array $event) : DomainEvent
+    {
+        $eventData = json_decode($event['content']);
+        $eventReflection = new \ReflectionClass($event['type']);
+        $parameters = $eventReflection->getConstructor()->getParameters();
+        $arguments = [];
+        foreach ($parameters as $parameter) {
+            foreach ($eventData as $key => $data) {
+                if ($key === $parameter->getName()) {
+                    $arguments[] = $data;
+                }
+            }
+        }
+
+        return new $event['type'](...$arguments);
+    }
 }

src/Kreta/SharedKernel/Infrastructure/Persistence/Redis/EventStore/RedisEventStore.php

@@ -16,8 +16,8 @@
 
 use Kreta\SharedKernel\Domain\Model\AggregateDoesNotExistException;
 use Kreta\SharedKernel\Domain\Model\DomainEventCollection;
-use Kreta\SharedKernel\Event\StoredEvent;
 use Kreta\SharedKernel\Event\EventStore;
+use Kreta\SharedKernel\Event\StoredEvent;
 use Kreta\SharedKernel\Event\Stream;
 use Kreta\SharedKernel\Event\StreamName;
 use Kreta\SharedKernel\Serialization\Serializer;

tests/Spec/Kreta/SharedKernel/Infrastructure/Persistence/InMemory/EventStore/InMemoryEventStoreSpec.php

@@ -37,7 +37,7 @@ function it_appends_to(Stream $stream, StreamName $streamName)
         $stream->events()->shouldBeCalled()->willReturn($eventCollection);
         $stream->name()->shouldBeCalled()->willReturn($streamName);
         $streamName->name()->shouldBeCalled()->willReturn('dummy');
-        $this->appendTo($stream);
+        $this->append($stream);
     }
 
     function it_get_stream_of_name_given(Stream $stream, StreamName $streamName)
@@ -48,7 +48,7 @@ function it_get_stream_of_name_given(Stream $stream, StreamName $streamName)
         $stream->events()->shouldBeCalled()->willReturn($eventCollection);
         $stream->name()->shouldBeCalled()->willReturn($streamName);
         $streamName->name()->shouldBeCalled()->willReturn('dummy');
-        $this->appendTo($stream);
+        $this->append($stream);
 
         $this->streamOfName($streamName)->shouldReturnAnInstanceOf(Stream::class);
     }
@@ -60,4 +60,43 @@ function it_does_not_get_any_aggregate(StreamName $streamName, Id $aggregateId)
 
         $this->shouldThrow(AggregateDoesNotExistException::class)->duringStreamOfName($streamName);
     }
+
+    function it_gets_events_since_given_date(\DateTimeImmutable $since, Stream $stream, StreamName $streamName)
+    {
+        $domainEvent = new DomainEventStub('foo', 'bar');
+        $eventCollection = new DomainEventCollection([$domainEvent]);
+        $stream->events()->shouldBeCalled()->willReturn($eventCollection);
+        $stream->name()->shouldBeCalled()->willReturn($streamName);
+        $streamName->name()->shouldBeCalled()->willReturn('dummy');
+        $this->append($stream);
+
+        $this->eventsSince($since)->shouldReturn([
+            [
+                'stream_name' => 'dummy',
+                'type'        => DomainEventStub::class,
+                'content'     => [
+                    'bar'        => 'bar',
+                    'foo'        => 'foo',
+                    'occurredOn' => $domainEvent->occurredOn()->getTimestamp(),
+                ],
+            ],
+        ]);
+    }
+
+    function it_gets_empty_events_when_since_is_higher_than_persisted_events_occurred_on(
+        \DateTimeImmutable $since,
+        Stream $stream,
+        StreamName $streamName
+    ) {
+        $domainEvent = new DomainEventStub('foo', 'bar');
+        $eventCollection = new DomainEventCollection([$domainEvent]);
+        $stream->events()->shouldBeCalled()->willReturn($eventCollection);
+        $stream->name()->shouldBeCalled()->willReturn($streamName);
+        $streamName->name()->shouldBeCalled()->willReturn('dummy');
+        $this->append($stream);
+
+        $since->getTimestamp()->willReturn($domainEvent->occurredOn()->getTimestamp() + 10);
+
+        $this->eventsSince($since)->shouldReturn([null]);
+    }
 }