[Messenger][WIP] Sending messages in batches #62600
Open
+1,101
−10
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
carsonbot
added
Status: Needs Review
Feature
Messenger
RFC
carsonbot
changed the title
JoppeDC
force-pushed
the
batch-messenger-poc
branch
2 times, most recently
from
59d8706 to
90f61db
Compare
JoppeDC
force-pushed
the
batch-messenger-poc
branch
from
90f61db to
f77ca1b
Compare
Some transports will have a max batch limit (eg. SQS) Either the transport has to make sub batches, or the messageBus should fail when trying to send bigger batches The issue with the transport making sub batches, is that you have the chance that half the batch fails and half succeeds. eg. dispatchBatch(20 messages) - transport sends 10 - transport tries to send 10 again, but fails now half the batch got sent, but we have no clue which ones. IMO, best to block this.
ro0NL
reviewed
Dec 1, 2025
| private string $batchId, | ||
| private int $batchIndex, | ||
| private int $batchSize, | ||
| private ?BatchCollector $collector = null, |
| public function addPendingSend(string $alias, SenderInterface $sender, Envelope $envelope, int $index): void | ||
| { | ||
| $this->pending[$alias] ??= ['sender' => $sender, 'items' => []]; | ||
| $this->pending[$alias]['items'][] = ['envelope' => $envelope, 'index' => $index]; |
Contributor
There was a problem hiding this comment.
Suggested change
| $this->pending[$alias]['items'][] = ['envelope' => $envelope, 'index' => $index]; | |
| $this->pending[$alias]['items'][$index] = $envelope; |
?
| * | ||
| * @author Joppe De Cuyper <[email protected]> | ||
| * | ||
| * @internal |
Contributor
There was a problem hiding this comment.
im fine with making it publicly available, for other maybe-implementions
otherwise we can argue whole BatchStamp is internal
Contributor
|
it might be interesting to update DoctrineTransport as such |
Contributor
Author
Yeah that was the next thing i was looking at. This interface sounds nice and easy on paper, but already ran into trouble when implementing. Not all transports that support batching, also support returning the ids of each item in the badge. So TransportMessageIdStamp can't be promised depending on the transport used |
Contributor
it sounds generally reasonable, there's no current contract enforcing it either as well it SHOULD be implemented if possible |
wouterj
changed the title
carsonbot
changed the title
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Note: This PR is a proof of concept. It is in no way, shape or form supposed to be a final ready for production fix. This PR is supposed to get the conversation going. Please, provide feedback where needed, as this feature turned out to be more complex than i initially imagined 😀 This is also the reason the code is littered with comments
Batch Dispatch for Symfony Messenger
Allows dispatching multiple messages in a single batch, enabling transports to optimize delivery (e.g., bulk inserts to a queue). This allows you to optimize message sending (Eg. Single SQS request for 10 messages, where you pay per-request)
How it works
$bus->dispatchBatch([new OrderCreated(1), new OrderCreated(2), new OrderCreated(3)]);Step by step:
Notes