Add builder pattern API for multiple Event Hubs in Aspire.Azure.Messaging.EventHubs
- Dominant language
- C#
- Stars
- 6.3k
- Forks
- 991
- Avg merge
- 2d 15h
- Merged PRs (30d)
- 196
Description
## Is there an existing issue for this?
- [x] I have searched the existing issues
## Is your feature request related to a problem? Please describe the problem.
Currently, if you want to work with multiple Event Hubs from the same namespace, you must register each client separately, potentially creating duplicate connections:
```csharp
// Current approach - separate registrations
builder.AddAzureEventHubProducerClient("telemetry");
builder.AddAzureEventHubProducerClient("logs");
builder.AddAzureEventHubProducerClient("metrics");
builder.AddAzureEventHubConsumerClient("orders");
builder.AddAzureEventProcessorClient("payments");
// Each registration may create separate connections
// No clear way to group clients by namespace
// Can't easily inject specific producers/consumers by Event Hub name
```
The Cosmos DB integration already solves this with `AddAzureCosmosDatabase()` and `CosmosDatabaseBuilder`, and similar proposals exist for Blob Storage (#12179) and Service Bus (#12180), but Event Hubs lacks an equivalent pattern.
## Describe the solution you'd like
Add a **builder pattern API** that allows registering multiple producers/consumers for different Event Hubs within the same namespace:
```csharp
// Proposed API
builder.AddAzureEventHubNamespace("events")
.AddKeyedProducer("telemetry")
.AddKeyedProducer("logs")
.AddKeyedProducer("metrics")
.AddKeyedConsumer("orders", consumerGroup: "$Default")
.AddKeyedProcessor("payments", consumerGroup: "$Default", checkpointStore: "checkpoints")
.AddKeyedBufferedProducer("analytics", options =>
{
options.MaximumWaitTime = TimeSpan.FromSeconds(5);
});
// Usage in services
public class TelemetryService(
[FromKeyedServices("telemetry")] EventHubProducerClient telemetryProducer,
[FromKeyedServices("logs")] EventHubProducerClient logsProducer,
[FromKeyedServices("metrics")] EventHubProducerClient metricsProducer)
{
// All producers share connection to the same Event Hubs namespace
public async Task LogEvent(string eventData)
{
await telemetryProducer.SendAsync([new EventData(eventData)]);
}
}
public class OrderProcessor(
[FromKeyedServices("orders")] EventHubConsumerClient ordersConsumer) : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await foreach (var partition in ordersConsumer.ReadEventsAsync(stoppingToken))
{
// Process order events
}
}
}
```
### Benefits
1. **Resource Efficiency**: Share connection/namespace configuration across multiple Event Hubs
2. **Better DI Integration**: Producers/consumers registered as keyed services by Event Hub name
3. **Cleaner Code**: Fluent builder API for declaring multiple Event Hubs in same namespace
4. **Logical Grouping**: Clear namespace-level organization of Event Hubs
5. **Consistency**: Matches patterns in Cosmos DB, Blob Storage (#12179), and Service Bus (#12180)
6. **Type Safety**: Keyed services allow injecting specific clients by Event Hub name
### Implementation Requirements
1. **New `EventHubNamespaceBuilder` class**:
- Constructor accepting `IHostApplicationBuilder`, connection name, settings
- Store namespace-level configuration (fully qualified namespace, credentials)
- `AddKeyedProducer(string eventHubName, Action?)` - Register producer for specific Event Hub
- `AddKeyedBufferedProducer(string eventHubName, Action?)` - Register buffered producer
- `AddKeyedConsumer(string eventHubName, string consumerGroup, Action?)` - Register consumer
- `AddKeyedProcessor(string eventHubName, string consumerGroup, string checkpointStoreConnectionName, Action?)` - Register processor with checkpoint store
- `AddKeyedPartitionReceiver(string eventHubName, string consumerGroup, string partitionId, Action?)` - Register partition receiver
2. **New extension methods** in `AspireEventHubsExtensions`:
- `AddAzureEventHubNamespace()` - Returns `EventHubNamespaceBuilder`
- `AddKeyedAzureEventHubNamespace()` - Returns `EventHubNamespaceBuilder` with keyed namespace
3. **Playground sample enhancement**:
- Update `AspireEventHub` playground to use new builder pattern
- Demonstrate multiple Event Hubs in same namespace
- Show producer, consumer, and processor registration
- Example with checkpoint store integration
### API Consistency
**Cosmos DB (existing):**
```csharp
builder.AddAzureCosmosDatabase("db")
.AddKeyedContainer("entries")
.AddKeyedContainer("users");
```
**Blob Storage (proposed - #12179):**
```csharp
builder.AddAzureBlobService("storage")
.AddKeyedContainer("images")
.AddKeyedContainer("documents");
```
**Service Bus (proposed - #12180):**
```csharp
builder.AddAzureServiceBus("messaging")
.AddKeyedSender("orders")
.AddKeyedProcessor("commands");
```
**Event Hubs (this proposal):**
```csharp
builder.AddAzureEventHubNamespace("events")
.AddKeyedProducer("telemetry")
.AddKeyedConsumer("orders", consumerGroup: "$Default")
.AddKeyedProcessor("payments", "$Default", "checkpoints");
```
### Example Usage Scenarios
#### Scenario 1: Multiple Producers in Same Namespace
```csharp
builder.AddAzureEventHubNamespace("monitoring")
.AddKeyedProducer("telemetry")
.AddKeyedProducer("logs")
.AddKeyedProducer("metrics")
.AddKeyedProducer("traces");
```
#### Scenario 2: Producer + Consumer for Same Event Hub
```csharp
builder.AddAzureEventHubNamespace("orders")
.AddKeyedProducer("order-events")
.AddKeyedConsumer("order-events", "$Default");
```
#### Scenario 3: Event Processor with Checkpoint Store
```csharp
builder.AddAzureBlobServiceClient("checkpoints");
builder.AddAzureEventHubNamespace("transactions")
.AddKeyedProcessor("payments", "$Default", "checkpoints", options =>
{
options.LoadBalancingUpdateInterval = TimeSpan.FromSeconds(10);
})
.AddKeyedProcessor("refunds", "$Default", "checkpoints");
```
#### Scenario 4: Multiple Consumer Groups
```csharp
builder.AddAzureEventHubNamespace("analytics")
.AddKeyedProducer("raw-events")
.AddKeyedConsumer("raw-events", "aggregator")
.AddKeyedConsumer("raw-events", "archiver")
.AddKeyedConsumer("raw-events", "real-time");
```
#### Scenario 5: Mixed Producers, Consumers, and Processors
```csharp
builder.AddAzureEventHubNamespace("platform")
.AddKeyedProducer("commands")
.AddKeyedConsumer("events", "$Default")
.AddKeyedBufferedProducer("metrics", options =>
{
options.MaximumWaitTime = TimeSpan.FromSeconds(1);
})
.AddKeyedProcessor("notifications", "email-group", "checkpoints");
```
## Additional context
This feature would bring the Event Hubs integration to feature parity with Cosmos DB and aligns with proposed enhancements for Blob Storage (#12179) and Service Bus (#12180). It provides a consistent API pattern across all Azure integrations in Aspire.
**Additional Considerations:**
- **Checkpoint Store Requirement**: `EventProcessorClient` requires a checkpoint store (typically Blob Storage). The builder should validate that a checkpoint store connection is provided
- **Consumer Groups**: Each consumer/processor requires a consumer group. Should default to `$Default` but allow override
- **Connection String Parsing**: Event Hub names are typically in the connection string (`EntityPath=myhub`), but the builder should allow explicit specification for clarity
- **Buffered Producer Lifecycle**: `EventHubBufferedProducerClient` has automatic flushing - users should be aware of disposal/lifetime management
**Migration Path:**
Existing code continues to work unchanged. The builder pattern is purely additive:
```csharp
// Still supported
builder.AddAzureEventHubProducerClient("telemetry");
// New pattern (optional)
builder.AddAzureEventHubNamespace("events")
.AddKeyedProducer("telemetry");
```
Reference implementation:
- `src/Components/Aspire.Microsoft.Azure.Cosmos/CosmosDatabaseBuilder.cs`
- `src/Components/Aspire.Microsoft.Azure.Cosmos/AspireMicrosoftAzureCosmosExtensions.cs`
- `playground/AspireEventHub/EventHubsConsumer/Program.cs`
- `playground/AspireEventHub/EventHubsApi/Program.cs`
Related issues: #12179 (Blob Storage), #12180 (Service Bus)
Contributor guide
Assessment
This issue has not been assessed yet.