MassTransit

# 7.1.8

# Resolved Issues

  • Responding from a state machine through the InMemoryOutbox wasn't working as expected (opens new window)
  • Kafka producer options can now be configured
  • IsResponseAccepted will now return false if a ResponseAddress is not present
  • Azure Service Bus receive endpoints should now allow PrefetchCount = 0
  • RabbitMQ Quorum queues can now be configured using .SetQuorumQueue(), which sets the appropriate queue attributes
  • RabbitMQ queue attributes are now copied to the _error and _skipped queues

# MassTransit Templates, Getting Started

A set of dotnet new templates were released, which can be installed using dotnet new -i MassTransit.Templates. The Getting Started documentation was updated to use the new templates and a new sample was added (opens new window) based on the documentation. Most of the new documentation will be written from a container-first perspective, starting with AddMassTransit and subsequently documenting the container-free versions where available.

# Delayed Message Delivery / Redelivery

It is now possible to configure delayed redelivery using the message transport. This approach works directly with the message transport, and does not use any configured message scheduler. A message scheduler can still be configured and used for regular message scheduling, but the transport message delay feature will be used for redelivery. This can significantly reduce the load on Quartz/Hangfire scheduling and separates the concerns of message redelivery and message scheduling.

The configuration syntax is the same as scheduled redelivery, but with a new transport-independent extension method:

cfg.ReceiveEndpoint("submit-order", e =>
{
    e.UseDelayedRedelivery(r => r.Intervals(TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(15), TimeSpan.FromMinutes(30)));
    e.UseMessageRetry(r => r.Immediate(5));
    e.UseInMemoryOutbox();

    e.ConfigureConsumer<SubmitOrder>();
});

The in-memory transport was updated to support delayed message delivery, making it easier to test delayed message scenarios.

The new delayed support can also be used a message scheduler, regardless of transport. To configure the delayed exchange scheduler, see the example below.

namespace SchedulingDelayed
{
    using System;
    using MassTransit;
    using Microsoft.Extensions.DependencyInjection;

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

            services.AddMassTransit(x =>
            {
                x.AddDelayedMessageScheduler();

                x.UsingInMemory((context, cfg) => 
                {
                    cfg.UseDelayedMessageScheduler();

                    cfg.ConfigureEndpoints(context);
                });
            });
        }
    }
}

# ExceptionInfo Data

The ExceptionInfo type has a new property, Data, which is a IDictionary<string, object>. This extra property supports the inclusion of application/exception specific data with the exception details. Since MassTransit does not support the directly serialization of Exception types (nor should it), this makes it possible to include additional values when an exception is thrown.

This is an opt-in approach, and by default no exception data will be added. This was decided to avoid accidental leakage of sensitive data.

A new exception type, MassTransitApplicationException, can be used to opt-in to the automatic propagation of Data properties into ExceptionInfo.Data. For example, data values can be explicitly added when the exception is caught and thrown.

class SubmitOrderConsumer :
    IConsumer<SubmitOrder>
{
    public Task Consume(ConsumeContext<SubmitOrder> context)
    {
        try
        {
            throw new IntentionalTestException("This was intentional");
        }
        catch (Exception exception)
        {
            throw new MassTransitApplicationException(exception, new
            {
                context.Message.OrderNumber,
                context.Message.CustomerNumber,
            });
        }
    }
}

The built-in Data property of Exception will also be used as a source when using MassTransitApplicationException. In the example below, the Dataproperties are added to the exception (framework and other libraries use this to store information, this example just does it to show the usage).

class SubmitOrderConsumer :
    IConsumer<SubmitOrder>
{
    public Task Consume(ConsumeContext<SubmitOrder> context)
    {
        try
        {
            var exception = new IntentionalTestException("This was intentional");
                exception.Data.Add("Username", "Frank");
                exception.Data.Add("CustomerId", 27);
                throw exception;
        }
        catch (Exception exception)
        {
            throw new MassTransitApplicationException(exception);
        }
    }
}

# Raw JSON Message Headers

The Raw JSON message serializer/deserializer was updated to support transport message headers, offering better support for messages without a regular MassTransit message envelope. The supported message headers include:

Name Type Property Notes
MessageId Guid MessageId
CorrelationId Guid CorrelationId
ConverationId Guid ConversationId
RequestId Guid RequestId
MT-InitiatorId Guid InitiatorId
MT-Source-Address Uri SourceAddress
MT-Response-Address Uri ResponseAddress
MT-Fault-Address Uri FaultAddress
MT-MessageType string SupportedMessageTypes ;delimited
MT-Host-Info JSON Host serialized Host

The default configuration doesn't change the existing behavior. When configuring the use of the Raw JSON serializer, new options are available:

Option Description
AnyMessageType Default behavior, any T is allowed. Do not specify to check the transport message type header
AddTransportHeaders Default behavior, headers are written to the transport and used
CopyHeaders Optional, will copy non-MassTransit headers to outbound messages

These options are specified as a parameter to cfg.UseRawJsonSerializer(options).

# State Machine Delay Provider

When configuring a message schedule in an Automatonymous state machine, instead of specifying a fixed delay, a DelayProvider can be configured. This enables dynamic delay periods based upon the contents of the state machine instance. The delay can still be dynamically specified when using .Schedule() based upon the instance and/or message contents.

# Azure Blob Storage Package Change

The Azure Blob Storage Message Data repository was changed to use the new standard Azure packages. This may require slight reconfiguration, but should work as it did previously.

← 7.1.7 →