feat: add multi-stream produce to IProducer#512
Merged
alexeyzimarev merged 8 commits intodevfrom Mar 9, 2026
Merged
Conversation
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Add ProduceRequest record structs and multi-stream Produce overloads that delegate to existing single-stream methods with parallel execution. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Add multi-stream delegation methods to GatewayProducer (both generic ProduceRequest<T> and non-generic ProduceRequest overloads). Update GatewayHandler to batch messages by target stream into ProduceRequest collections instead of producing per-message, reducing overhead. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Tests validate producing to multiple streams, empty requests, multiple messages to same stream, and untyped produce requests. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Contributor
Review Summary by QodoAdd multi-stream produce support to IProducer and update Gateway
WalkthroughsDescription• Add ProduceRequest and ProduceRequest<T> record structs for batch multi-stream operations • Implement multi-stream Produce overloads on IProducer and BaseProducer<T> with parallel execution • Simplify GatewayHandler to batch messages by stream and use single multi-stream produce call • Add delegation methods to GatewayProducer for multi-stream produce support • Include comprehensive tests validating multi-stream, empty requests, and same-stream batching Diagramflowchart LR
A["ProduceRequest<br/>Record Structs"] -->|"Define input types"| B["IProducer<br/>Multi-stream Methods"]
B -->|"Implement with<br/>parallel execution"| C["BaseProducer<T><br/>Override Methods"]
C -->|"Delegate to"| D["Single-stream<br/>Produce"]
E["GatewayHandler<br/>GroupBy Messages"] -->|"Refactor to use"| F["Multi-stream<br/>Produce Call"]
G["GatewayProducer<br/>Delegation Methods"] -->|"Support"| F
H["MultiStreamProduceTests<br/>Test Suite"] -->|"Validate"| C
File Changes1. src/Core/src/Eventuous.Producers/ProduceRequest.cs
|
Contributor
Code Review by Qodo
1. Missing newline in BaseProducer.cs
|
Deploying eventuous-main with
|
| Latest commit: |
7d06e43
|
| Status: | ✅ Deploy successful! |
| Preview URL: | https://3b7f936c.eventuous-main.pages.dev |
| Branch Preview URL: | https://feat-multi-stream-produce.eventuous-main.pages.dev |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7d06e434ba
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| .Select(g => new ProduceRequest<TProduceOptions>( | ||
| g.Key, | ||
| g.Select(x => new ProducedMessage(x.Message, x.GetMeta(context), contextMeta) { OnAck = onAck, OnNack = onFail }), | ||
| g.First().ProduceOptions |
There was a problem hiding this comment.
When multiple GatewayMessages target the same stream but carry different ProduceOptions, this batching logic collapses them to g.First().ProduceOptions, so every message in that stream is produced with the first message's options. Before this change each message called producer.Produce(..., x.ProduceOptions, ...) independently, so options could vary per message; now routing keys/partitions or other producer-specific options can be silently misapplied whenever mixed options appear in one stream group.
Useful? React with 👍 / 👎.
![qodo-free-for-open-source-projects[bot]](https://avatars.githubusercontent.com/in/1068000?s=60&v=4){kind=small}
|
|
||
| return Task.WhenAll(requests.Select(r => Produce(r.Stream, r.Messages, cancellationToken))); | ||
| } | ||
| } No newline at end of file |
Contributor
There was a problem hiding this comment.
1. Missing newline in baseproducer.cs 📘 Rule violation ✓ Correctness
The modified BaseProducer.cs ends without a trailing newline, which introduces formatting that conflicts with typical .editorconfig end-of-file rules. This can cause avoidable diff noise and style enforcement failures in CI/editor tooling.
Agent Prompt
## Issue description
`src/Core/src/Eventuous.Producers/BaseProducer.cs` was modified in this PR and now ends without a trailing newline (EOF newline). This conflicts with repository formatting rules and can cause formatting checks/diff noise.
## Issue Context
Git reports `\ No newline at end of file` for the updated file; add a final newline after the last `}`.
## Fix Focus Areas
- src/Core/src/Eventuous.Producers/BaseProducer.cs[75-75]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Test Results 60 files + 40 60 suites +40 37m 45s ⏱️ + 25m 30s Results for commit cdf283c. ± Comparison against base commit 843d9ba. This pull request removes 5 and adds 19 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
- Add missing trailing newline in BaseProducer.cs - Group by (Stream, ProduceOptions) instead of just Stream in GatewayHandler to preserve per-message options when messages to the same stream have different ProduceOptions Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
WaitForInner already guards with `if (inner is not IHostedProducer) return`, so call it unconditionally. The previous boolean was inverted (true when NOT hosted), meaning the wait was never reached for actual hosted producers. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
The emulator container readiness check was hitting the default 100-second HttpClient.Timeout on CI runners, causing all Service Bus tests to fail. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
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.
Summary
ProduceRequest/ProduceRequest<TProduceOptions>record structs as input types for multi-stream produceProduce(IReadOnlyCollection<ProduceRequest>)onIProducerwith parallel execution viaTask.WhenAllProduce(IReadOnlyCollection<ProduceRequest<T>>)onBaseProducer<T>(can't go onIProducer<in T>due to variance constraints)GatewayHandlerto use single multi-streamProducecall instead ofGroupBy+ parallel per-stream callsGatewayProducerto delegate multi-stream produce to inner producerDesign notes
The generic
ProduceRequest<TProduceOptions>overload cannot live onIProducer<in TProduceOptions>because the contravariant (in) type parameter causes a CS1961 variance error when used in method parameters. The overload lives onBaseProducer<T>instead, which all concrete producers inherit from.Individual producer implementations (Kafka, RabbitMQ, PubSub, Service Bus) inherit the default parallel behavior and can override later for optimization.
Test plan
MultiStreamProduceTests: multi-stream, empty requests, same-stream batch, untyped requests🤖 Generated with Claude Code