Routing Slip Activity Configuration

Add a routing slip activity

To add a routing slip activity, use the AddActivity or AddExecuteActivity method. There are several overloads for each method.

using MassTransit;

services.AddMassTransit(x =>
{
    // Adds a compensating activity (supports execute and compensate)
    x.AddActivity<DownloadImageActivity, DownloadImageArguments, DownloadImageLog>();

    // Adds a compensating activity by type using the activity definition (optional)
    x.AddActivity(typeof(DownloadImageActivity), typeof(DownloadImageActivityDefinition));

    // Adds an execute-only activity (no compensation support)
    x.AddExecuteActivity<ProcessOrderActivity, ProcessOrderArguments>();

    // Adds an execute activity by type using the activity definition
    x.AddExecuteActivity(typeof(ProcessOrderActivity), typeof(ProcessOrderActivityDefinition));

    // Adds all activities from the specified assembly, including any activity definitions
    x.AddActivities(Assembly.GetExecutingAssembly());

    // Adds all activities from the namespace containing the type
    x.AddActivitiesFromNamespaceContaining<DownloadImageActivity>();
    x.AddActivitiesFromNamespaceContaining(typeof(DownloadImageActivity));
});

Create an activity definition

Activity definitions are used to specify the endpoint configuration of a routing slip activity so that it can be automatically configured. Definitions may be explicitly added when calling AddActivity / AddExecuteActivity or can be discovered automatically using any of the AddActivities methods.

An example activity definition is shown below.

public class DownloadImageActivityDefinition :
    ActivityDefinition<DownloadImageActivity, DownloadImageArguments, DownloadImageLog>
{
    public DownloadImageActivityDefinition()
    {
        // Configure concurrent message limit for the execute endpoint
        ConcurrentMessageLimit = 8;

        // Alternatively, configure the endpoint directly using the Endpoint method
        Endpoint(e => e.ConcurrentMessageLimit = 8);
    }

    protected override void ConfigureExecuteEndpoint(IReceiveEndpointConfigurator endpointConfigurator,
        IActivityConfigurator<DownloadImageActivity, DownloadImageArguments, DownloadImageLog> activityConfigurator,
        IBusRegistrationContext context)
    {
        endpointConfigurator.UseMessageRetry(r => r.Interval(2, 100));

        var partition = endpointConfigurator.CreatePartitioner(16);

        activityConfigurator.Arguments(x => x.UsePartitioner(partition, m => m.Message.ImageUri));
    }

    protected override void ConfigureCompensateEndpoint(IReceiveEndpointConfigurator endpointConfigurator,
        IActivityConfigurator<DownloadImageActivity, DownloadImageArguments, DownloadImageLog> activityConfigurator,
        IBusRegistrationContext context)
    {
        endpointConfigurator.UseMessageRetry(r => r.Interval(2, 100));
    }
}

The ConfigureExecuteEndpoint method is used to configure the execute endpoint for the activity when ConfigureEndpoints is called. For activities that support compensation, the ConfigureCompensateEndpoint method can be used to configure the compensate endpoint.

Execute Activity Definition

For execute only activities, the ExecuteActivityDefinition should be used.

Activity definition settings

Setting	Description
ConcurrentMessageLimit	The number of concurrent messages this activity will process at once
Endpoint	Configure the endpoint directly using the Endpoint method

Automatic endpoint configuration

When using automatic endpoint configuration via ConfigureEndpoints, MassTransit will create receive endpoints for all registered routing slip activities. For compensating activities (those implementing IActivity<TArguments, TLog>), two endpoints are created: one for execution and one for compensation.

services.AddMassTransit(x =>
{
    x.AddActivity<DownloadImageActivity, DownloadImageArguments, DownloadImageLog>();
    x.AddExecuteActivity<ProcessOrderActivity, ProcessOrderArguments>();

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

Exclude from automatic configuration

To exclude a routing slip activity from automatic configuration, use the ExcludeFromConfigureEndpoints method.

services.AddMassTransit(x =>
{
    x.AddActivity<DownloadImageActivity, DownloadImageArguments, DownloadImageLog>()
        .ExcludeFromConfigureEndpoints();
});

Or use the attribute on the activity class:

[ExcludeFromConfigureEndpoints]
public class DownloadImageActivity :
    IActivity<DownloadImageActivity, DownloadImageArguments, DownloadImageLog>
{
}

Manual endpoint configuration

When manually configuring endpoints for routing slip activities, the activity should be registered but ConfigureEndpoints should either be excluded or called after manually configured endpoints.

Configure execute activity

To manually configure an execute activity endpoint, use the ConfigureExecuteActivity method.

services.AddMassTransit(x =>
{
    x.AddExecuteActivity<ProcessOrderActivity, ProcessOrderArguments>();

    x.UsingInMemory((context, cfg) =>
    {
        cfg.ReceiveEndpoint("process-order-execute", e =>
        {
            e.UseMessageRetry(r => r.Interval(3, 1000));

            e.ConfigureExecuteActivity<ProcessOrderActivity>(context);
        });
    });
});

Configure compensating activity

For compensating activities, both execute and compensate endpoints need to be configured. This can be nested so that the compensate endpoint address can be passed to the execute activity configuration method.

services.AddMassTransit(x =>
{
    x.AddActivity<DownloadImageActivity, DownloadImageArguments, DownloadImageLog>();

    x.UsingInMemory((context, cfg) =>
    {
        // Configure the execute endpoint
        cfg.ReceiveEndpoint("download-image-execute", execute =>
        {
            execute.UseMessageRetry(r => r.Interval(2, 100));

            // Configure the compensate endpoint
            cfg.ReceiveEndpoint("download-image-compensate", compensate =>
            {
                compensate.UseMessageRetry(r => r.Interval(2, 100));

                execute.ConfigureActivity(compensate, context, typoef(DownloadImageActivity));
            });
        });

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

Configure endpoints callback

A callback can be added to customize all endpoints, including routing slip activity endpoints.

services.AddMassTransit(x =>
{
    x.AddActivity<DownloadImageActivity, DownloadImageArguments, DownloadImageLog>();
    x.AddExecuteActivity<ProcessOrderActivity, ProcessOrderArguments>();

    x.AddConfigureEndpointsCallback((name, cfg) =>
    {
        if (name.EndsWith("-execute"))
        {
            cfg.UseMessageRetry(r => r.Immediate(2));
        }
    });

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