decouple consume and publish in outbox

docs/pages/outbox.md

@@ -36,7 +36,14 @@ And then you have to define the consumer. This gets the right event bus.
 It is used to load the events to be dispatched from the database, dispatch the events and then empty the outbox table.
 
 ```php
-$consumer = new StoreOutboxConsumer($store, $realEventBus);
+use Patchlevel\EventSourcing\Outbox\EventBusPublisher;
+use Patchlevel\EventSourcing\Outbox\StoreOutboxConsumer;
+
+$consumer = new StoreOutboxConsumer(
+    $store, 
+    new EventBusPublisher($realEventBus)
+);
+
 $consumer->consume();
 ```
 

src/Outbox/EventBusPublisher.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Outbox;
+
+use Patchlevel\EventSourcing\EventBus\EventBus;
+use Patchlevel\EventSourcing\EventBus\Message;
+
+final class EventBusPublisher implements OutboxPublisher
+{
+    public function __construct(
+        private readonly EventBus $eventBus,
+    ) {
+    }
+
+    public function publish(Message $message): void
+    {
+        $this->eventBus->dispatch($message);
+    }
+}

src/Outbox/OutboxPublisher.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Outbox;
+
+use Patchlevel\EventSourcing\EventBus\Message;
+
+interface OutboxPublisher
+{
+    public function publish(Message $message): void;
+}

src/Outbox/StoreOutboxConsumer.php

@@ -4,23 +4,21 @@
 
 namespace Patchlevel\EventSourcing\Outbox;
 
-use Patchlevel\EventSourcing\EventBus\EventBus;
-
 final class StoreOutboxConsumer implements OutboxConsumer
 {
     public function __construct(
-        private readonly OutboxStore $outboxStore,
-        private readonly EventBus $eventBus,
+        private readonly OutboxStore $store,
+        private readonly OutboxPublisher $publisher,
     ) {
     }
 
     public function consume(int|null $limit = null): void
     {
-        $messages = $this->outboxStore->retrieveOutboxMessages($limit);
+        $messages = $this->store->retrieveOutboxMessages($limit);
 
         foreach ($messages as $message) {
-            $this->eventBus->dispatch($message);
-            $this->outboxStore->markOutboxMessageConsumed($message);
+            $this->publisher->publish($message);
+            $this->store->markOutboxMessageConsumed($message);
         }
     }
 }

tests/Integration/Outbox/OutboxTest.php

@@ -8,6 +8,7 @@
 use Patchlevel\EventSourcing\EventBus\DefaultEventBus;
 use Patchlevel\EventSourcing\Metadata\AggregateRoot\AttributeAggregateRootRegistryFactory;
 use Patchlevel\EventSourcing\Outbox\DoctrineOutboxStore;
+use Patchlevel\EventSourcing\Outbox\EventBusPublisher;
 use Patchlevel\EventSourcing\Outbox\OutboxEventBus;
 use Patchlevel\EventSourcing\Outbox\StoreOutboxConsumer;
 use Patchlevel\EventSourcing\Projection\Projection\Store\InMemoryStore;
@@ -63,9 +64,12 @@ public function testSuccessful(): void
             'outbox',
         );
 
+<<<<<<< Updated upstream
         $realEventBus = new DefaultEventBus();
         $realEventBus->addListener(new SendEmailProcessor());
 
+=======
+>>>>>>> Stashed changes
         $outboxEventBus = new OutboxEventBus($outboxStore);
 
         $profileProjector = new ProfileProjector($this->connection);
@@ -122,7 +126,11 @@ public function testSuccessful(): void
             $message->event(),
         );
 
-        $consumer = new StoreOutboxConsumer($outboxStore, $realEventBus);
+        $consumer = new StoreOutboxConsumer(
+            $outboxStore,
+            new EventBusPublisher($realEventBus),
+        );
+
         $consumer->consume();
 
         self::assertSame(0, $outboxStore->countOutboxMessages());

tests/Unit/Outbox/EventBusPublisherTest.php

@@ -0,0 +1,36 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Tests\Unit\Outbox;
+
+use Patchlevel\EventSourcing\EventBus\EventBus;
+use Patchlevel\EventSourcing\EventBus\Message;
+use Patchlevel\EventSourcing\Outbox\EventBusPublisher;
+use Patchlevel\EventSourcing\Tests\Unit\Fixture\Email;
+use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileCreated;
+use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId;
+use PHPUnit\Framework\TestCase;
+use Prophecy\PhpUnit\ProphecyTrait;
+
+/** @covers \Patchlevel\EventSourcing\Outbox\EventBusPublisher */
+final class EventBusPublisherTest extends TestCase
+{
+    use ProphecyTrait;
+
+    public function testPublish(): void
+    {
+        $message = new Message(
+            new ProfileCreated(
+                ProfileId::fromString('1'),
+                Email::fromString('[email protected]'),
+            ),
+        );
+
+        $eventBus = $this->prophesize(EventBus::class);
+        $eventBus->dispatch($message)->shouldBeCalled();
+
+        $publisher = new EventBusPublisher($eventBus->reveal());
+        $publisher->publish($message);
+    }
+}

tests/Unit/Outbox/StoreOutboxConsumerTest.php

@@ -4,8 +4,8 @@
 
 namespace Patchlevel\EventSourcing\Tests\Unit\Outbox;
 
-use Patchlevel\EventSourcing\EventBus\EventBus;
 use Patchlevel\EventSourcing\EventBus\Message;
+use Patchlevel\EventSourcing\Outbox\OutboxPublisher;
 use Patchlevel\EventSourcing\Outbox\OutboxStore;
 use Patchlevel\EventSourcing\Outbox\StoreOutboxConsumer;
 use Patchlevel\EventSourcing\Tests\Unit\Fixture\Email;
@@ -32,8 +32,8 @@ public function testConsume(): void
         $store->retrieveOutboxMessages(null)->willReturn([$message]);
         $store->markOutboxMessageConsumed($message)->shouldBeCalled();
 
-        $eventBus = $this->prophesize(EventBus::class);
-        $eventBus->dispatch($message)->shouldBeCalled();
+        $eventBus = $this->prophesize(OutboxPublisher::class);
+        $eventBus->publish($message)->shouldBeCalled();
 
         $consumer = new StoreOutboxConsumer($store->reveal(), $eventBus->reveal());
         $consumer->consume();
@@ -52,8 +52,8 @@ public function testConsumeWithLimit(): void
         $store->retrieveOutboxMessages(100)->willReturn([$message]);
         $store->markOutboxMessageConsumed($message)->shouldBeCalled();
 
-        $eventBus = $this->prophesize(EventBus::class);
-        $eventBus->dispatch($message)->shouldBeCalled();
+        $eventBus = $this->prophesize(OutboxPublisher::class);
+        $eventBus->publish($message)->shouldBeCalled();
 
         $consumer = new StoreOutboxConsumer($store->reveal(), $eventBus->reveal());
         $consumer->consume(100);