Support iterators in `MessageDispatcher::dispatch` to reduce memory footprint

[lgrossi]

Hey Frank, hope you're doing well!

I ran into something interesting while working with MessageDispatcher. I was dispatching a big batch of messages when I noticed a memory spike. After digging in, I realized it’s because dispatch(Message ...$messages) forces everything into memory, even if you're using a generator.

Would it make sense to tweak it to also accept an iterable ?
That way, it could process messages in chunks instead of loading everything upfront.

Current Issue

• Message ...$messages always expands all values into an array before calling the method.
• This means generators are fully materialized, which can cause unnecessary high memory usage for large batches.

Proposed Solution

Keep the existing variadic behavior but also support iterables efficiently:

interface MessageDispatcher
{
    /**
     * @param Message|iterable<Message> $messages
     */
    public function dispatch(Message|iterable $messages, Message ...$variadic);
}

✅ This allows:

• Comma-separated messages → dispatch($msg1, $msg2, $msg3)
• Efficient batch dispatching using generators → dispatch($generator)

Example: Batched Dispatching Implementation

final class DispatchMessagesInChunks implements MessageDispatcher
{
    public function __construct(
        private readonly MessageDispatcher $dispatcher,
        private readonly int $batchSize,
    ) {}

    /**
     * @param Message|iterable<Message> $messages
     */
    public function dispatch(Message|iterable $messages, Message ...$variadic): void
    {
        $messages = is_iterable($messages) ? $messages : [$messages];

        // The actual chunk implementation is out of scope
        foreach (chunk($messages, $this->batchSize) as $chunk) {
            $this->dispatcher->dispatch(...$chunk);
        }

        if ($variadic !== []) {
            $this->dispatcher->dispatch([], ...$variadic);
        }
    }
}

Benchmark

I ran a quick test comparing the variadic approach vs generator-based dispatching:

class BenchMarkDispatcher implements MessageDispatcher
{
    /**
     * @param Message|iterable<Message> $messages
     */
    public function dispatch(Message|iterable $messages, Message ...$variadic): void
    {
        $count = 0;
        $messages = is_iterable($messages) ? $messages : [$messages];

        foreach ([...$messages, ...$variadic] as $ignored) {
            $count++;
        }

        echo "Messages dispatched: $count\n";
    }
}

$generator = $messages = (function (int $size) {
    for ($i = 0; $i < $size; $i++) {
        yield new Message("Message #$i");
    }
});

$dispatcher = new DispatchMessagesInChunks(new BenchMarkDispatcher(), 1_000);

// $dispatcher->dispatch(...$generator(1_000_000));
// $dispatcher->dispatch($generator(1_000_000));

echo "Peak memory usage: " . memory_get_peak_usage() . " bytes\n";

Variadic Version (Forces Everything into Memory)

...
Batch finished, memory usage: 154648280 bytes
Messages dispatched: 1000000
Peak memory usage: 162721552 bytes

Generator Version (Streaming Efficiently)

...
Batch finished, memory usage: 556936 bytes
Messages dispatched: 1000000
Peak memory usage: 693984 bytes

Why This Matters?

I get that dispatching millions of messages might not be a typical use case, but having a built-in way to handle big batches efficiently means we don’t need special wrappers around the core classes. This keeps everything within the existing framework while avoiding unnecessary memory overhead.

Would love to hear your thoughts, let me know what you think! 🚀

[frankdejonge]

Hi @lgrossi! Fancy seeing you here, I think your suggestion makes sense, what's less ideal is that this is a backwards incompatible change, so would need to be released as a new major version. From a consumer point of view, I think the upgrade-path is doable, but custom implementations of this interface would need to be amended. I'd rather change to an iterable implementation entirely, so we have one way of doing things. I'm happy if you make the PR and we take it from there, or I can draft it up too, that's all good.

[frankdejonge]

I've pushed a version/4.x branch which you can target for the PR

[frankdejonge]

@lgrossi #237

[lgrossi]

Hey @frankdejonge, sorry only seeing this now, thanks for creating the PR. I see that you also applied it to the other dispatchers, which looks nice! I left some small suggestions in there.