MassTransit

# 7.0.4

Version 7.0.4 includes several new features which are detailed below. There were also numerous bug fixes, as well as tweaks to job consumers and batch consumers.

# General

MassTransit now references the .NET Core 2.1 version of Microsoft.Extensions.Logging.Abstractions, which should now make v7 usable with that SDK version.

The endpoint name formatters now support adding a prefix to all endpoint names, including the namespace in addition to the type name, and creating instance-specific queues for consumers.

services.AddMassTransit(x =>
{
    x.AddConsumer<CommonConsumer>()
        .Endpoint(e => e.InstanceId = "SomeUniqueValue");
});

This would format the endpoint name as either CommonSomeUniqueValue, common_some_unique_value, or common-some-unique-value.

The prefix can be specified on the endpoint name formatter, as well as whether or not to include the namespace in the endpoint name.

services.AddSingleton<IEndpointNameFormatter>(provider => new KebabCaseEndpointNameFormatter("Dev", true));

This would format the endpoint name as dev-my-service-contacts-common, assuming that CommonConsumer was in the MyService.Contracts namespace.

# Transports

# Amazon SNS

Topics are no longer created for published message types that include/implement other message types. Since SNS does not support polymorphic message routing, these topics were extraneous.

Queue attributes are now copied to error/skipped queues.

# RabbitMQ, Azure Service Bus

Message types can now be excluded from being created as exchanges or topics. Some other frameworks have stereotype interfaces such as IAmAnEvent or IAmACommand, or even base type interfaces like IShouldBeTreatedAsAMessage, and those interfaces would be used to create exchanges/topics for polymorphic message support. These interfaces can now be excluded using the publish topology configuration.

Using an attribute, a type can be excluded:

[ExcludeFromTopology]
public interface IEvent
{
    Guid TransactionId { get; }
}

The alternative is to configure the publish topology message type:

Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    cfg.Publish<IEvent>(p => p.Exclude = true);
});

# Containers

# In-Memory Test Harness

Registration for the test harness has been added, making it easier to test consumers using a container. To configure the in-memory test harness, see the example below.

 var provider = new ServiceCollection()
    .AddMassTransitInMemoryTestHarness(cfg =>
    {
        cfg.AddConsumer<PingRequestConsumer>();
    })
    .BuildServiceProvider(true);

var harness = provider.GetRequiredService<InMemoryTestHarness>();

await harness.Start();
try
{
    var bus = provider.GetRequiredService<IBus>();

    IRequestClient<PingMessage> client = bus.CreateRequestClient<PingMessage>();

    await client.GetResponse<PongMessage>(new PingMessage());

    Assert.That(await harness.Consumed.Any<PingMessage>());
}
finally
{
    await harness.Stop();

    await provider.DisposeAsync();
}

To include a consumer test harness for a consumer, additional configuration is required.

  var provider = new ServiceCollection()
    .AddMassTransitInMemoryTestHarness(cfg =>
    {
        cfg.AddConsumer<PingRequestConsumer>();

        cfg.AddConsumerTestHarness<PingRequestConsumer>();
    })
    .BuildServiceProvider(true);

var harness = provider.GetRequiredService<InMemoryTestHarness>();

await harness.Start();
try
{
    var bus = provider.GetRequiredService<IBus>();

    IRequestClient<PingMessage> client = bus.CreateRequestClient<PingMessage>();

    await client.GetResponse<PongMessage>(new PingMessage());

    Assert.That(await harness.Consumed.Any<PingMessage>());

    var consumerHarness = provider.GetRequiredService<IConsumerTestHarness<PingRequestConsumer>>();

    Assert.That(await consumerHarness.Consumed.Any<PingMessage>());
}
finally
{
    await harness.Stop();

    await provider.DisposeAsync();
}

Saga test harnesses can also be used, as shown below.

 var provider = new ServiceCollection()
    .AddMassTransitInMemoryTestHarness(cfg =>
    {
        cfg.AddSaga<TestSaga>()
            .InMemoryRepository();

        cfg.AddSagaTestHarness<TestSaga>();
    })
    .BuildServiceProvider(true);

var harness = provider.GetRequiredService<InMemoryTestHarness>();

await harness.Start();
try
{
    _sagaId = Guid.NewGuid();
    _testValueA = "TestValueA";

    await harness.Bus.Publish(new A
    {
        CorrelationId = _sagaId,
        Value = _testValueA
    });

    Assert.That(await harness.Published.Any<A>());

    Assert.That(await harness.Consumed.Any<A>());

    var sagaHarness = provider.GetRequiredService<ISagaTestHarness<TestSaga>>();

    Assert.That(await sagaHarness.Consumed.Any<A>());

    Assert.That(await sagaHarness.Created.Any(x => x.CorrelationId == _sagaId));

    var saga = sagaHarness.Created.Contains(_sagaId);
    Assert.That(saga, Is.Not.Null);
    Assert.That(saga.ValueA, Is.EqualTo(_testValueA));

    Assert.That(await harness.Published.Any<Aa>());

    Assert.That(await harness.Published.Any<B>(), Is.False);
}
finally
{
    await harness.Stop();

    await provider.DisposeAsync();
}

# Scoped Filters

Container registration for Scoped filters is now optional.

# State Machine Activities

Container registration for state machine activities is now optional for Microsoft Extensions Dependency Injection and Autofac.

← Configuration →