[MessageOutbox] Outbox relay design

[lcobucci]

We've been using eventsauce/message-outbox for the implementation of polling subscriber pattern and, although it does the job really well, there's one little detail of the OutboxRelay that makes me wonder if we really understood its design decisions.

The docs say:

A message outbox, transactional outbox, provides a solution for the double network interaction by buffering events in the database used for reconstitution. The events are persisted in a separate table in the database, used to re-dispatch them to a queue at a later time."

However, the OutboxRelay is configured using a MessageConsumer instead of a MessageDispatcher (which is a "custom message dispatching (queue)"). That makes batched dispatching a bit more complicated (especially if we want to provide some consistency over what's marked as consumed).

We can easily provide our own implementation for our use case, sure, but I was wondering if this wouldn't be useful for the package itself.

Untested implementation

<?php

namespace EventSauce\MessageOutbox;

use EventSauce\BackOff\BackOffStrategy;
use EventSauce\BackOff\ExponentialBackOffStrategy;
use EventSauce\EventSourcing\Message;
use EventSauce\EventSourcing\MessageConsumer;
use EventSauce\EventSourcing\MessageDispatcher;
use Throwable;

use Traversable;
use function count;

final readonly class OutboxRelay
{
    public function __construct(
        private OutboxRepository $repository,
        private MessageDispatcher $dispatcher,
        private BackOffStrategy $backOff = new ExponentialBackOffStrategy(100000, 25),
        private RelayCommitStrategy $commitStrategy = new MarkMessagesConsumedOnCommit(),
    ) {
    }

    public function publishBatch(int $batchSize, int $commitSize = 1): int
    {
        $numberPublished = 0;
        $messages = $this->repository->retrieveBatch($batchSize);

        foreach ($this->batchByCommitSize($messages, $commitSize) as $batch) {
            $numberPublished += $this->dispatchMessages($batch);
        }

        return $numberPublished;
    }

    /**
     * @param Traversable<Message>
     * @return iterable<list<Message>>
     */
    private function batchByCommitSize(Traversable $messages, int $size): iterable
    {
        $batch = [];

        foreach ($messages as $message) {
            $batch[] = $message;

            if (count($batch) === $size) {
                yield $batch;

                $batch = [];
            }
        }

        if (count($batch) > 0) {
            yield $batch;
        }
    }

    /** @param list<Message> $messages */
    private function dispatchMessages(array $messages): int
    {
        $tries = 0;
        start_relay:

        try {
            $tries++;
            $this->dispatcher->dispatch(...$messages);
        } catch (Throwable $throwable) {
            $this->backOff->backOff($tries, $throwable);
            goto start_relay;
        }

        $this->commitStrategy->commitMessages($this->repository, ...$messages);

        return count($messages);
    }
}

Can you please confirm if our assumptions around batching are correct and if the implementation shared is something useful? We can turn it into a PR 👍

[frankdejonge]

Hi @lcobucci, this sounds like a very much worthwhile addition. Shall we introduce this as a BatchedOutboxRelay? That way we can introduce it in a non-breaking way. The old implementation can then stay in case people use/prefer it for their own case. Alternatively, we can rename the existing to RelayOutboxThroughConsumer and call the new one RelayOutboxThroughDispatcher and create a class that extends the RelayThroughConsumer for backward compatibility. I'm OK either way, do you have any preference or ideas about this?

[lcobucci]

I like the RelayOutboxThroughConsumer and RelayOutboxThroughDispatcher idea - and the proposed migration path.
I'll send a PR with a proposed design and we can evolve the ideas there.

Thanks @frankdejonge!