MassTransit

# Event Hub

Azure Event Hub is included as a Rider, and supports consuming and producing messages from/to Azure event hubs.

Uses MassTransit.Azure.ServiceBus.Core (opens new window), MassTransit.EventHub (opens new window), MassTransit.Extensions.DependencyInjection (opens new window)

To consume messages from an event hub, configure a Rider within the bus configuration as shown.

namespace EventHubConsumer
{
    using System;
    using System.Threading.Tasks;
    using MassTransit;
    using MassTransit.Azure.ServiceBus.Core;
    using MassTransit.EventHubIntegration;
    using Microsoft.Extensions.DependencyInjection;

    public class Program
    {
        public static async Task Main()
        {
            var services = new ServiceCollection();

            services.AddMassTransit(x =>
            {
                x.UsingAzureServiceBus((context, cfg) =>
                {
                    cfg.Host("connection-string");

                    cfg.ConfigureEndpoints(context);
                });
              
                x.AddRider(rider =>
                {
                    rider.AddConsumer<EventHubMessageConsumer>();

                    rider.UsingEventHub((context, k) =>
                    {
                        k.Host("connection-string");

                        k.Storage("connection-string");

                        k.ReceiveEndpoint("input-event-hub", c =>
                        {
                            c.ConfigureConsumer<EventHubMessageConsumer>(context);
                        });
                    });
                });
            });
        }

        class EventHubMessageConsumer :
            IConsumer<EventHubMessage>
        {
            public Task Consume(ConsumeContext<EventHubMessage> context)
            {
                return Task.CompletedTask;
            }
        }

        public interface EventHubMessage
        {
            string Text { get; }
        }
    }
}

The familiar ReceiveEndpoint syntax is used to configure an event hub. The consumer group specified should be unique to the application, and shared by a cluster of service instances for load balancing. Consumers and sagas can be configured on the receive endpoint, which should be registered in the rider configuration. While the configuration for event hubs is the same as a receive endpoint, there is no implicit binding of consumer message types to event hubs (there is no pub-sub using event hub).

# Producers

Producing messages to event hubs uses a producer. In the example below, a messages is produced to the event hub.

namespace EventHubProducer
{
    using System;
    using System.Threading;
    using System.Threading.Tasks;
    using MassTransit;
    using MassTransit.Azure.ServiceBus.Core;
    using MassTransit.EventHubIntegration;
    using Microsoft.Extensions.DependencyInjection;

    public class Program
    {
        public static async Task Main()
        {
            var services = new ServiceCollection();

            services.AddMassTransit(x =>
            {
                x.UsingAzureServiceBus((context, cfg) =>
                {
                    cfg.Host("connection-string");

                    cfg.ConfigureEndpoints(context);
                });
              
                x.AddRider(rider =>
                {
                    rider.UsingEventHub((context, k) =>
                    {
                        k.Host("connection-string");

                        k.Storage("connection-string");
                    });
                });
            });

            var provider = services.BuildServiceProvider(true);

            var busControl = provider.GetRequiredService<IBusControl>();

            await busControl.StartAsync(new CancellationTokenSource(10000).Token);

            var serviceScope = provider.CreateScope();

            var producerProvider = serviceScope.ServiceProvider.GetRequiredService<IEventHubProducerProvider>();
            var producer = await producerProvider.GetProducer("some-event-hub");

            await producer.Produce<EventHubMessage>(new { Text = "Hello, Computer." });
        }

        public interface EventHubMessage
        {
            string Text { get; }
        }
    }
}

← Mediator →