Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/Eventuous/eventuous/llms.txt

Use this file to discover all available pages before exploring further.

The Eventuous Gateway is a lightweight infrastructure component that connects a subscription on one transport to a producer on another. It reads events from a source (e.g. the KurrentDB $all stream), optionally transforms or filters them, and publishes the results to a target (e.g. a Kafka topic or an Azure Service Bus queue). It is the standard pattern for building event-driven integration pipelines in Eventuous without writing bespoke polling loops.

​
Installation

dotnet add package Eventuous.Gateway

​
Core concepts

TypeRole
IGatewayTransform<TProduceOptions>Your code: maps one incoming event to zero or more outgoing messages
RouteAndTransform<TProduceOptions>Delegate form of the same contract
GatewayMessage<TProduceOptions>The output value: target stream + message + options
GatewayHandler<TProduceOptions>Internal IEventHandler that wires transform → producer
GatewayProducer<TProduceOptions>Wraps a producer, waits for it to be ready if it’s an IHostedProducer

​
The transform interface

Implement IGatewayTransform<TProduceOptions> to control routing and translation: 
public interface IGatewayTransform<TProduceOptions> {
    ValueTask<GatewayMessage<TProduceOptions>[]> RouteAndTransform(
        IMessageConsumeContext context
    );
}
Return an empty array to drop the event silently. Return multiple GatewayMessage<T> records to fan out a single incoming event to several targets. GatewayMessage<TProduceOptions> is a simple record: 
public record GatewayMessage<TProduceOptions>(
    StreamName      TargetStream,
    object          Message,
    Metadata?       Metadata,
    TProduceOptions ProduceOptions
);

​
Implementing a gateway transform

The example below bridges KurrentDB to Kafka. Events of type BookingRegistered are translated into an integration event and routed to the booking-events topic. All other event types are dropped. 
public class BookingToKafkaTransform
    : IGatewayTransform<KafkaProduceOptions> {

    public ValueTask<GatewayMessage<KafkaProduceOptions>[]> RouteAndTransform(
        IMessageConsumeContext context
    ) {
        // Drop everything except BookingRegistered
        if (context.Message is not BookingRegistered evt) {
            return new(Array.Empty<GatewayMessage<KafkaProduceOptions>>());
        }

        // Build the outgoing integration event
        var integrationEvent = new BookingIntegrationEvent(
            evt.BookingId,
            evt.RoomId,
            evt.CheckIn,
            evt.CheckOut
        );

        var message = new GatewayMessage<KafkaProduceOptions>(
            TargetStream   : new StreamName("booking-events"),
            Message        : integrationEvent,
            Metadata       : null,
            ProduceOptions : new KafkaProduceOptions(evt.BookingId) // use bookingId as partition key
        );

        return new(new[] { message });
    }
}

​
Registering the gateway

Use AddGateway to wire the subscription, transform, and producer together in one call. The registration automatically handles:
  • Registering the producer as a singleton and starting it as a hosted service (if it implements IHostedProducer)
  • Building the consume pipeline
  • Wrapping the producer in GatewayProducer<T> so it waits until the broker connection is established before producing

​
With a class-based transform

services.AddGateway<
    AllStreamSubscription,
    AllStreamSubscriptionOptions,
    KafkaBasicProducer,
    KafkaProduceOptions,
    BookingToKafkaTransform  // registered as a singleton automatically
>("BookingToKafkaGateway");

​
With an inline delegate

services.AddGateway<
    AllStreamSubscription,
    AllStreamSubscriptionOptions,
    KafkaBasicProducer,
    KafkaProduceOptions
>(
    subscriptionId    : "BookingToKafkaGateway",
    routeAndTransform : context => {
        if (context.Message is not BookingRegistered evt)
            return new(Array.Empty<GatewayMessage<KafkaProduceOptions>>());

        return new(new[] {
            new GatewayMessage<KafkaProduceOptions>(
                new StreamName("booking-events"),
                new BookingIntegrationEvent(evt.BookingId),
                null,
                new KafkaProduceOptions(evt.BookingId)
            )
        });
    },
    configureSubscription: opts => opts.StartFrom = InitialPosition.Latest
);

​
Configuring the subscription

Pass configureSubscription and configureBuilder to adjust subscription options and add additional handlers: 
services.AddGateway<
    AllStreamSubscription,
    AllStreamSubscriptionOptions,
    KafkaBasicProducer,
    KafkaProduceOptions,
    BookingToKafkaTransform
>(
    subscriptionId        : "BookingToKafkaGateway",
    configureSubscription : opts => {
        opts.StartFrom    = InitialPosition.Latest;
        opts.EventFilter  = EventTypeFilter.ExcludeSystemEvents();
    },
    awaitProduce          : true  // wait for Kafka ACK before acknowledging the subscription
);

​
awaitProduce

When awaitProduce is true (the default), the gateway waits for the producer to confirm delivery before the subscription considers the event handled. Set it to false for fire-and-forget semantics, where the subscription acknowledges immediately and produce happens asynchronously.

​
How metadata flows

GatewayProducer<T> preserves correlation and causation IDs from the inbound context by injecting them into the ProducedMessage metadata before forwarding to the underlying producer. This means tracing identifiers propagate transparently across transport boundaries without any extra code in your transform.

Build docs developers (and LLMs) love

Get started for freeTalk to us