MassTransit

https://masstransit.massient.com/
Admin
MassTransit is a complete application framework for building reliable, scalable, and distributed...
Tokens:211,101
Snippets:1,990
Trust Score:6.4
Update:9 hours ago
Show doc for...
Tokens:
Raw
### Full ASP.NET Core Example - Program.cs

Source: https://masstransit.massient.com/configuration/schedulers/quartz

Complete Program.cs for a MassTransit and Quartz setup. Configures Quartz with SQL Server persistent store and clustering, adds MassTransit with Quartz consumers, and starts the Quartz hosted service.

```csharp
var builder = WebApplication.CreateBuilder(args);

var connectionString = builder.Configuration.GetConnectionString("quartz")
    ?? throw new InvalidOperationException("Connection string 'quartz' is not configured.");

// Configure Quartz with SQL Server and clustering
builder.Services.AddQuartz(q =>
{
    q.SchedulerName = "MassTransit-Scheduler";
    q.SchedulerId = "AUTO";

    q.UseDefaultThreadPool(tp =>
    {
        tp.MaxConcurrency = 10;
    });

    q.UsePersistentStore(s =>
    {
        s.UseProperties = true;
        s.RetryInterval = TimeSpan.FromSeconds(15);
        s.UseSqlServer(connectionString);

        s.UseClustering(c =>
        {
            c.CheckinMisfireThreshold = TimeSpan.FromSeconds(20);
            c.CheckinInterval = TimeSpan.FromSeconds(10);
        });
    });
});

// Add MassTransit with Quartz
builder.Services.AddMassTransit(x =>
{
    x.AddPublishMessageScheduler();
    x.AddConsumers(typeof(Program).Assembly);
    x.AddQuartzConsumers();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(builder.Configuration.GetConnectionString("RabbitMq"));

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

// Start Quartz scheduler
builder.Services.AddQuartzHostedService(options =>
{
    options.StartDelay = TimeSpan.FromSeconds(5);
    options.WaitForJobsToComplete = true;
});

var app = builder.Build();

app.UseRouting();
app.UseAuthorization();
app.MapControllers();
app.MapHealthChecks("/health/ready");

app.Run();
```

--------------------------------

### Full ASP.NET Core Example - appsettings.json

Source: https://masstransit.massient.com/configuration/schedulers/quartz

Connection strings for Quartz (SQL Server) and RabbitMQ used by the Program.cs example.

```json
{
  "ConnectionStrings": {
    "quartz": "Server=.;Database=Quartz;Trusted_Connection=True;TrustServerCertificate=true",
    "RabbitMq": "rabbitmq://localhost:5672"
  }
}
```

--------------------------------

### Example appsettings.json with PostgreSQL connection string

Source: https://masstransit.massient.com/configuration/transports/sql

Shows the appsettings.json configuration for the connection string used in the previous example. The connection string is for PostgreSQL.

```json
{
  "ConnectionStrings": {
    "Db": "Server=localhost;Port=5432;user id=postgres;password=Password12!;database=my_app;"
  },
  "AllowedHosts": "*"
}
```

--------------------------------

### Full ASP.NET Core Example with Hangfire and MassTransit

Source: https://masstransit.massient.com/configuration/schedulers/hangfire

Complete Program.cs setup for an ASP.NET Core application using Hangfire with SQL Server storage, MassTransit with Hangfire consumers, RabbitMQ, and the Hangfire dashboard. Includes health checks and dashboard authorization.

```csharp
var builder = WebApplication.CreateBuilder(args);

// Configure Hangfire with SQL Server storage
builder.Services.AddHangfire(h => h
    .UseRecommendedSerializerSettings()
    .UseSqlServerStorage(builder.Configuration.GetConnectionString("Hangfire")));

builder.Services.AddHangfireServer();

// Add MassTransit with Hangfire
builder.Services.AddMassTransit(x =>
{
    x.AddPublishMessageScheduler();
    x.AddConsumers(typeof(Program).Assembly);
    x.AddHangfireConsumers();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(builder.Configuration.GetConnectionString("RabbitMq"));

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

var app = builder.Build();

// Enable Hangfire Dashboard
app.UseHangfireDashboard("/hangfire", new DashboardOptions
{
    Authorization = new[] { new HangfireAuthFilter() }
});

app.UseHealthChecks("/health");

app.Run();
```

--------------------------------

### License key example (BEGIN/END LICENSE KEY)

Source: https://masstransit.massient.com/configuration/license

Example license key file content. Replace with your own values. The key is multi-line and includes BEGIN/END lines.

```text
-----BEGIN LICENSE KEY-----
c1SDNOXHUwMDJCTUFkSVRqTExWSzMwc0xiUkxMM0JDa3ZhWUpXWFNUbXJ3N1x1MDAyQmRFRjR0
WnZiblJ2WW1Wc0lpd2laVzFoYVd3aU9pSnpkRtWVc0dVpHRmxkSGQ1YkdWeVFIWnZiblJ2WW1W
T2lJeU1ESMVEV5TFRBNVZEQXdPakF3T2pBd1dpSXNJbVY0Y0dseVpYTWlPaUl5TURJM0xUQXhM
ZXlKV2V5SnBaQ0k2SW1OMWMxOVVXbHBIYzFCUVlrOW1jRzR4T0NJc0ltNWhiV1VpT2lKQ1lXNJRlp2Ym5SdlltVnNJbjBzSW5CeWIyUjFZM1J6SWpwN0ltMWhjOXVJam9pVFdGemMxUnlZV6YVhRZ1RHbGpaVzV6WlNJc0ltVjRjR2x5WlhNaU9pSXlNREkyTFRFeUxUQTVWREF3T2pBd09qdXaUlzSW1abFlYUjFjbVZ6SWpwdWRXeHNmWDBzSW1OeVpXRjBaV1FpVEE0VkRBd09qQXdPakF3lKOSIsInNpZ25hdHVyZSI6IkFYMVRJNWE2c1dMZXVoNi9rLzcxNk8vaXRobSI6IkVDRHNhIn0sImRhdGEiJleUpqYjI1MFlXTjBJanA3SW01aGJXVWlPaUpDWVc1cklGM04wY21GdWMybDBJanA3SW01aGJXVWlPpOWVhOelZISmhibk5wZENJc0ltUmxjMk55YVhCMGFXVWJMMTF1R0libXd2YjlWR2ZhUTlkZ1FERExseDQUdrMVBvdmhaRkN2ajdWbDlQVFNiVkFFdUE3
-----END LICENSE KEY-----
```

--------------------------------

### Short address example for GetSendEndpoint

Source: https://masstransit.massient.com/concepts/producers

Shows the short address syntax for getting a Send endpoint for a queue on RabbitMQ, which eliminates the need to specify scheme, host, port, and virtual host.

```csharp
GetSendEndpoint(new Uri("queue:input-queue"))
```

--------------------------------

### Install MassTransit.Quartz package

Source: https://masstransit.massient.com/configuration/schedulers/quartz

Run this command to add the MassTransit.Quartz NuGet package to your project.

```bash
dotnet add package MassTransit.Quartz
```

--------------------------------

### Configure AutoStart on the bus

Source: https://masstransit.massient.com/concepts/requests

Sets AutoStart = true on the RabbitMQ bus configuration to start the bus endpoint when the bus starts, reducing latency of the first request. Default is false, meaning the bus endpoint is not started until just before the first request.

```csharp
services.AddMassTransit(x =>
{
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.AutoStart = true; // default = false, starts the bus endpoint with the bus
    });
});
```

--------------------------------

### Install Serilog packages

Source: https://masstransit.massient.com/configuration/logging

Add the required Serilog packages to the project: Serilog.Extensions.Hosting, Serilog, and Serilog.Sinks.Console.

```bash
dotnet add package Serilog.Extensions.Hosting
dotnet add package Serilog
dotnet add package Serilog.Sinks.Console
```

--------------------------------

### ConvertVideoJobConsumer implementation

Source: https://masstransit.massient.com/concepts/job-consumers

Example job consumer that implements IJobConsumer<ConvertVideo>. The Run method is called when the job starts; it delays for 30 seconds using the cancellation token.

```csharp
public class ConvertVideoJobConsumer :
    IJobConsumer<ConvertVideo>
{
    public async Task Run(JobContext<ConvertVideo> context)
    {
        await Task.Delay(30000, context.CancellationToken);
    }
}
```

--------------------------------

### Install Hangfire storage packages

Source: https://masstransit.massient.com/configuration/schedulers/hangfire

Choose one of the listed storage packages based on your backend: SQL Server, Redis, or PostgreSQL.

```bash
# Choose one of the following:
dotnet add package Hangfire.SqlServer       # SQL Server
dotnet add package Hangfire.Redis.StackExchange  # Redis
dotnet add package Hangfire.PostgreSQL      # PostgreSQL
```

--------------------------------

### Message contracts for broker topology example

Source: https://masstransit.massient.com/configuration/transports/amazon-sqs

Defines the command contract for processing a file and the event contract for a consumer receiving files from a customer. These are used in the default broker topology examples.

```csharp
namespace Acme;

public record ProcessFile
{
}
```

```csharp
namespace Acme;

public record FileReceivedEvent
{
}
```

--------------------------------

### Schedule Job using IRequestClient.ScheduleJob

Source: https://masstransit.massient.com/concepts/job-consumers

Schedules a job to run at a specified start time. New in MassTransit v8.3.0. The example schedules the job 15 minutes from now.

```csharp
[HttpPost("{path}")]
public async Task<IActionResult> SubmitJob(string path, [FromServices] IRequestClient<SubmitJob<ConvertVideo>> client)
{
    await client.ScheduleJob(DateTimeOffset.Now.AddMinutes(15), new ConvertVideo
    {
        Path = path
    });

    return Ok(new
    {
        jobId,
        path
    });
}
```

--------------------------------

### Run SampleSignalR.Mvc with two launch profiles

Source: https://masstransit.massient.com/configuration/integrations/signalr

Commands to start the sample MVC application with two different launch profiles, simulating horizontal scaling. Run each command in a separate terminal, then open the specified URLs in two browser tabs to see messages appear in both.

```bash
> cd (your cloned Sample-SignalR)\src\SampleSignalR.Mvc
> dotnet run --launch-profile sample1
> dotnet run --launch-profile sample2
```

--------------------------------

### Get Send Endpoint and Send a Message

Source: https://masstransit.massient.com/concepts/producers

Shows how to obtain an ISendEndpoint from an ISendEndpointProvider and send a message. The GetSendEndpoint method is async and must be awaited. The example defines a SubmitOrder record and sends it with an OrderId.

```csharp
public record SubmitOrder
{
    public string OrderId { get; init; }
}

public async Task SendOrder(ISendEndpointProvider sendEndpointProvider)
{
    var endpoint = await sendEndpointProvider.GetSendEndpoint(_serviceAddress);

    await endpoint.Send(new SubmitOrder { OrderId = "123" });
}
```

--------------------------------

### Minimal unit test with AddMassTransitTestHarness

Source: https://masstransit.massient.com/guides/unit-testing

The minimal setup for a MassTransit unit test. It uses the ServiceCollection extension AddMassTransitTestHarness to add the test harness, then starts it with StartTestHarness. The harness is automatically stopped when the ServiceProvider is disposed.

```csharp
[Test]
public async Task A_complete_unit_test()
{
    await using var provider = new ServiceCollection()
        .AddMassTransitTestHarness(x =>
        {
        })
        .BuildServiceProvider(true);

    var harness = await provider.StartTestHarness();
}
```

--------------------------------

### Create a class library project for message contracts

Source: https://masstransit.massient.com/guides/message-contract

Commands to create a new class library project and add it to the solution. Run these in a terminal.

```bash
dotnet new classlib -o ./src/MyProject.Contracts
dotnet sln add ./src/MyProject.Contracts
```

--------------------------------

### Valid message type URN examples

Source: https://masstransit.massient.com/configuration/serialization

Examples of valid message type URNs. The last example shows a nested class, indicated by the '+' symbol.

```text
urn:message:MyProject.Messages:UpdateAccount
urn:message:MyProject.Messages.Events:AccountUpdated
urn:message:MyProject:ChangeAccount
urn:message:MyProject.AccountService:MyService+AccountUpdatedEvent
```

--------------------------------

### Bus Observer Implementation

Source: https://masstransit.massient.com/configuration/observability

Example class implementing IBusObserver to monitor bus life cycle events: PreStart, PostStart, PreStop, PostStop, and BusFault. Each method returns Task.CompletedTask.

```csharp
public class BusObserver :
    IBusObserver
{
    /// <summary>
    /// Called before the bus is started, allowing observers to prepare
    /// </summary>
    public Task PreStart(IBus bus)
    {
        return Task.CompletedTask;
    }

    /// <summary>
    /// Called after the bus has been started successfully
    /// </summary>
    public Task PostStart(IBus bus, BusStartedContext context)
    {
        return Task.CompletedTask;
    }

    /// <summary>
    /// Called before the bus is stopped, allowing observers to complete any pending operations
    /// </summary>
    public Task PreStop(IBus bus)
    {
        return Task.CompletedTask;
    }

    /// <summary>
    /// Called after the bus has been stopped
    /// </summary>
    public Task PostStop(IBus bus, BusStoppedContext context)
    {
        return Task.CompletedTask;
    }

    /// <summary>
    /// Called when an unhandled exception occurs in the bus
    /// </summary>
    public Task BusFault(IBus bus, BusFaultContext context)
    {
        return Task.CompletedTask;
    }
}
```

--------------------------------

### NewId.NextGuid()

Source: https://masstransit.massient.com/reference/newid

Generates a new Guid from a NewId. This method returns a Guid representation of the generated identifier.

```APIDOC
## NewId.NextGuid()

### Description
Generates a new Guid from a NewId. This method returns a Guid representation of the generated identifier.

### Method
NextGuid()

### Return Value
- **Guid** - A new Guid instance.

### Example
```csharp
Guid guid = NewId.NextGuid();
```
```

--------------------------------

### Create a new class library project

Source: https://masstransit.massient.com/guides/message-consumer

Commands to create a new class library project and add it to the solution. Run in a terminal window.

```bash
dotnet new classlib -o ./src/MyProject.Components
dotnet sln add ./src/MyProject.Components
```

--------------------------------

### Generate NewId and Guid

Source: https://masstransit.massient.com/reference/newid

Shows the two primary generation methods: NewId.Next() returns a NewId, and NewId.NextGuid() returns a Guid.

```csharp
NewId newId = NewId.Next();

Guid guid = NewId.NextGuid();
```

--------------------------------

### Create an activity definition for DownloadImageActivity

Source: https://masstransit.massient.com/configuration/routing-slip-activity

Example activity definition that configures the execute and compensate endpoints. Sets ConcurrentMessageLimit to 8, configures retry, and uses a partitioner on the execute endpoint.

```csharp
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));
    }
}
```

--------------------------------

### Basic Quartz.NET configuration

Source: https://masstransit.massient.com/configuration/schedulers/quartz

Adds Quartz.NET services for development or simple deployments.

```csharp
services.AddQuartz();
```

--------------------------------

### Raw JSON message example

Source: https://masstransit.massient.com/configuration/serialization

Example of a raw JSON message that would be received when using a serializer that doesn't wrap the message in an envelope (application/json).

```json
{
  "orderId": "181c0000-6393-3630-36a4-08daf4e7c6da",
  "timestamp": "2023-01-12T21:55:53.714Z"
}
```

--------------------------------

### Run the SignalR Console Sample

Source: https://masstransit.massient.com/configuration/integrations/signalr

Commands to run the console sample from the cloned Sample-SignalR repository. After running, type a message to broadcast to all connections and observe it in the browser chat.

```bash
> cd (your cloned Sample-SignalR)\src\SampleSignalR.Service
> dotnet run
```

--------------------------------

### Configure Serilog in Program.cs

Source: https://masstransit.massient.com/configuration/logging

Configure Serilog with minimum levels for MassTransit and Microsoft, enrich from log context, and write to console. Then use it with the host and configure MassTransit with Azure Service Bus.

```csharp
using Serilog;
using Serilog.Events;

Log.Logger = new LoggerConfiguration()
    .MinimumLevel.Information()
    .MinimumLevel.Override("MassTransit", LogEventLevel.Information)
    .MinimumLevel.Override("Microsoft", LogEventLevel.Warning)
    .MinimumLevel.Override("Microsoft.Hosting", LogEventLevel.Information)
    .MinimumLevel.Override("Microsoft.EntityFrameworkCore", LogEventLevel.Fatal)
    .MinimumLevel.Override("Microsoft.EntityFrameworkCore.Database.Command", LogEventLevel.Fatal)
    .Enrich.FromLogContext()
    .WriteTo.Console()
    .CreateLogger();

var builder = WebApplication.CreateBuilder(args);

builder.Host.UseSerilog();

builder.Services.AddMassTransit(x =>
{
    x.AddServiceBusMessageScheduler();

    x.UsingAzureServiceBus((context, cfg) =>
    {
        cfg.Host(serviceBusConnectionString);

        cfg.UseServiceBusMessageScheduler();

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

--------------------------------

### SubmitOrderConsumer with retryable ADOException

Source: https://masstransit.massient.com/configuration/middleware/retry

Example consumer that may throw an ADOException due to transient conditions like database deadlocks. This is the consumer used in the retry configuration examples.

```csharp
public class SubmitOrderConsumer :
    IConsumer<SubmitOrder>
{
    ISessionFactory _sessionFactory;

    public async Task Consume(ConsumeContext<SubmitOrder> context)
    {
        using(var session = _sessionFactory.OpenSession())
        using(var transaction = session.BeginTransaction())
        {
            var customer = session.Get<Customer>(context.Message.CustomerId);

            // continue with order processing

            transaction.Commit();
        }
    }
}
```

--------------------------------

### ConfigureSaga with IRegistrationContext and UseInMemoryOutbox

Source: https://masstransit.massient.com/reference/obsolete

Shows the new overload of ConfigureSaga that accepts IRegistrationContext, which should be used instead of the obsolete overload. The example configures message retry and uses the new UseInMemoryOutbox overload that requires IRegistrationContext.

```csharp
protected override void ConfigureSaga(IReceiveEndpointConfigurator configurator,
    ISagaConfigurator<JobAttemptSaga> sagaConfigurator,
    IRegistrationContext context)
{
    configurator.UseMessageRetry(r => r.Intervals(100, 1000, 2000, 5000));

    // use the new overload, not the obsolete one
    configurator.UseInMemoryOutbox(context);
}
```

--------------------------------

### Create a consumer class implementing IConsumer<T>

Source: https://masstransit.massient.com/guides/message-consumer

Basic consumer class that implements IConsumer<CreateCustomer> with a Consume method. Place in the Consumers folder of the MyProject.Components project.

```csharp
namespace MyProject.Components.Consumers;

using MassTransit;
using MyProject.Contracts;

public class CreateCustomerConsumer :
    IConsumer<CreateCustomer>
{
    public async Task Consume(ConsumeContext<CreateCustomer> context)
    {
        // Handle the message
    }
}
```

--------------------------------

### Configure WebApplicationFactory with AddMassTransitTestHarness

Source: https://masstransit.massient.com/guides/unit-testing/web-application-factory

Sets up the Web Application Factory to use MassTransit's test harness by adding the test harness services. This is the initial setup required before creating the test harness and sending requests.

```csharp
await using var application = new WebApplicationFactory<Program>()
    .WithWebHostBuilder(builder =>
        builder.ConfigureServices(services =>
            services.AddMassTransitTestHarness()));

var testHarness = application.Services.GetTestHarness();

using var client = application.CreateClient();

var orderId = NewId.NextGuid();

var submitOrderResponse = await client.PostAsync("/Order", JsonContent.Create(new Order
{
    OrderId = orderId
}));

var consumerTestHarness = testHarness.GetConsumerHarness<SubmitOrderConsumer>();

Assert.That(await consumerTestHarness.Consumed.Any<SubmitOrder>(x => x.Context.Message.OrderId == orderId), Is.True);
```

--------------------------------

### SubmitOrderConsumerDefinition example

Source: https://masstransit.massient.com/concepts/consumers

Example consumer definition that overrides the default endpoint name to 'ha-submit-order', sets a concurrent message limit of 4, and configures retry and in-memory outbox in the ConfigureConsumer method.

```csharp
public class SubmitOrderConsumerDefinition :
    ConsumerDefinition<SubmitOrderConsumer>
{
    public SubmitOrderConsumerDefinition()
    {
        // override the default endpoint name, for whatever reason
        EndpointName = "ha-submit-order";

        // limit the number of messages consumed concurrently
        // this applies to the consumer only, not the endpoint
        ConcurrentMessageLimit = 4;
    }

    protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator,
        IConsumerConfigurator<DiscoveryPingConsumer> consumerConfigurator, IRegistrationContext context)
    {
        endpointConfigurator.UseMessageRetry(r => r.Interval(5, 1000));
        endpointConfigurator.UseInMemoryOutbox(context);
    }
}
```

--------------------------------

### Declare event using CorrelatedBy<Guid> interface

Source: https://masstransit.massient.com/guides/saga-state-machines/event

Declares an event using the CorrelatedBy<Guid> interface. The event declaration is not required because the default convention automatically configures events with this interface.

```csharp
public interface OrderCanceled :
    CorrelatedBy<Guid>
{
}

public class OrderStateMachine :
    MassTransitStateMachine<OrderState>
{
    public OrderStateMachine()
    {
        Event(() => OrderCanceled); // not required, as it is the default convention
    }
}
```

--------------------------------

### Schedule Recurring Job with Start and End Dates

Source: https://masstransit.massient.com/concepts/job-consumers

Uses AddOrUpdateRecurringJob with a cron expression builder to schedule a recurring job confined to a time period. The job runs every 30 minutes between the specified start and end dates.

```csharp
public async Task ConfigureRecurringJobs(IPublishEndpoint endpoint)
{
    await endpoint.AddOrUpdateRecurringJob("RoutineMaintenance",
        new RoutineMaintenanceCommand(), x =>
        {
            x.Start = new DateTimeOffset(2024, 1, 1, 0, 0, 0, TimeSpan.Zero);
            x.End = new DateTimeOffset(2025, 1, 1, 0, 0, 0, TimeSpan.Zero);
            x.Every(minutes: 30);
        });
}
```

--------------------------------

### Using IClientFactory to create a request client

Source: https://masstransit.massient.com/concepts/requests

Shows how to resolve IClientFactory from the service provider, create a request client with a fixed destination address, and send a request.

```csharp
public async Task WorkerMethod(IServiceProvider provider)
{
    var clientFactory = provider.GetRequiredService<IClientFactory>();

    var serviceAddress = new Uri("exchange:check-order-status");

    var client = clientFactory.CreateRequestClient<CheckOrderStatus>(serviceAddress);

    var response = await client.GetResponse<OrderStatusResult>(new { OrderId = id});
}
```

--------------------------------

### Install MassTransit.Hangfire package

Source: https://masstransit.massient.com/configuration/schedulers/hangfire

Adds the MassTransit.Hangfire package, which automatically includes Hangfire.Core.

```bash
dotnet add package MassTransit.Hangfire
```

--------------------------------

### Create a Partitioner instance with Murmur3UnsafeHashGenerator

Source: https://masstransit.massient.com/configuration/middleware/partitioner

Instantiates a Partitioner with 16 partitions and the recommended Murmur3UnsafeHashGenerator for distributing partition keys. This instance can be shared across multiple UsePartitioner calls.

```csharp
var partition = new Partitioner(16, new Murmur3UnsafeHashGenerator());
```

--------------------------------

### Build a routing slip with activities

Source: https://masstransit.massient.com/guides/routing-slips/monitor-using-a-saga

Creates a RoutingSlipBuilder, adds activities, and sets a variable. This is the initial step to create a routing slip for tracking.

```csharp
var builder = new RoutingSlipBuilder(NewId.NextGuid());

// add your activities as normal
builder.AddActivity("DownloadImage", new Uri("queue:download-image_execute?bind=false"),
    new
    {
        ImageUri = new Uri("https://images.google.com/someImage.jpg")
    });
builder.AddActivity("FilterImage", new Uri("queue:filter-image_execute?bind=false"));
builder.AddVariable("WorkPath", @"\dfs\work");
```

--------------------------------

### Define OrderConsumer class

Source: https://masstransit.massient.com/configuration/transports/rabbitmq

Defines a consumer for SubmitOrder messages. The Consume method is empty in this example.

```csharp
public class OrderConsumer :
    IConsumer<SubmitOrder>
{
    public async Task Consume(ConsumeContext<SubmitOrder> context)
    {

    }
}
```

--------------------------------

### Building and executing a routing slip

Source: https://masstransit.massient.com/skills/routing-slip-activity-foundation.md

Creates a RoutingSlipBuilder, adds an activity with arguments, adds a TraceId variable, builds the slip, and executes it on the bus.

```csharp
var builder = new RoutingSlipBuilder(NewId.NextGuid());

builder.AddActivity("DownloadImage", new Uri("exchange:download-image_execute"), new
{
    ImageUri = "https://example.com/image.png",
    WorkPath = "/tmp/images"
});

builder.AddVariable("TraceId", Guid.NewGuid().ToString("N"));

var routingSlip = builder.Build();
await bus.Execute(routingSlip);
```

--------------------------------

### Configure middleware for a consumer with retry, redelivery, and outbox

Source: https://masstransit.massient.com/concepts/middleware

Shows how to configure MassTransit middleware for a consumer using AddConfigureEndpointsCallback. The order is intentional: immediate retry handles brief failures, redelivery schedules later attempts, and the outbox holds outgoing messages until an attempt succeeds.

```csharp
services.AddMassTransit(x =>
{
    x.AddConsumer<SubmitOrderConsumer>();

    x.AddConfigureEndpointsCallback((context, name, cfg) =>
    {
        cfg.UseDelayedRedelivery(r => r.Intervals(TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(15)));
        cfg.UseMessageRetry(r => r.Immediate(3));
        cfg.UseInMemoryOutbox(context);
    });
});
```

--------------------------------

### Configure request client for a message type

Source: https://masstransit.massient.com/concepts/requests

Configures MassTransit to add a consumer on a specific endpoint and sends requests to that endpoint instead of publishing. The request client is added with AddRequestClient<CheckOrderStatus> pointing to the exchange:order-status address. The bus endpoint is a temporary queue created only after the bus starts; if requests time out with RequestTimeoutException, ensure the bus is started.

```csharp
services.AddMassTransit(x =>
{
    // configure the consumer on a specific endpoint address
    x.AddConsumer<CheckOrderStatusConsumer>()
        .Endpoint(e => e.Name = "order-status");

    // Sends the request to the specified address, instead of publishing it
    x.AddRequestClient<CheckOrderStatus>(new Uri("exchange:order-status"));

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

--------------------------------

### UpdateUserStatus message record

Source: https://masstransit.massient.com/configuration/transports/azure-service-bus

Defines a message type with a UserId and Status. Used in the SessionId convention example.

```csharp
public record UpdateUserStatus
{
    public Guid UserId { get; init; }
    public string Status { get; init; }
}
```

--------------------------------

### Configure SignalR backplane with MassTransit and RabbitMQ in Startup.cs

Source: https://masstransit.massient.com/configuration/integrations/signalr

Add this code to your ASP.NET Core Startup.cs file to configure SignalR and MassTransit with RabbitMQ. It registers the SignalR hub and sets up the bus with RabbitMQ, then configures endpoints for consumers and hubs. The ReceiveEndpoints without a queue name are non-durable and auto-deleting.

```csharp
public void ConfigureServices(IServiceCollection services)
{
    // other config...

    services.AddSignalR();

    // Other config perhaps...

    // creating the bus config
    services.AddMassTransit(x =>
    {
        // Add this for each Hub you have
        x.AddSignalRHub<ChatHub>(cfg => {/*Configure hub lifetime manager*/});

        x.UsingRabbitMq((context, cfg) =>
        {
            cfg.Host("localhost", "/", h =>
            {
                h.Username("guest");
                h.Password("guest");
            });

            // register consumer' and hub' endpoints
            cfg.ConfigureEndpoints(context);
        }));
    });
}
```

--------------------------------

### SubmitOrder message record

Source: https://masstransit.massient.com/configuration/transports/azure-service-bus

Defines a message type with a CustomerId and TransactionId. Used in the PartitionKey convention example.

```csharp
public record SubmitOrder
{
    public string CustomerId { get; init; }
    public Guid TransactionId { get; init; }
}
```

--------------------------------

### Create a compensating activity with IActivity<TArguments, TLog>

Source: https://masstransit.massient.com/guides/routing-slips

Implements the IActivity<TArguments, TLog> interface to create an activity that supports compensation. The first generic argument specifies the argument type, the second specifies the log type. Type parameters may be an interface, class, or record; for classes or records, proper accessors like { get; set; } or { get; init; } should be specified.

```csharp
public class DownloadImageActivity :
    IActivity<DownloadImageArguments, DownloadImageLog>
{
    Task<ExecutionResult> Execute(ExecuteContext<DownloadImageArguments> context);
    Task<CompensationResult> Compensate(CompensateContext<DownloadImageLog> context);
}
```

--------------------------------

### Basic MassTransit configuration with Quartz consumers

Source: https://masstransit.massient.com/configuration/schedulers/quartz

Configures MassTransit with publish message scheduler and Quartz consumers, using RabbitMQ.

```csharp
services.AddMassTransit(x =>
{
    x.AddPublishMessageScheduler();

    x.AddQuartzConsumers();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.UsePublishMessageScheduler();

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

--------------------------------

### Configure Prometheus with OpenTelemetry in ASP.NET Core

Source: https://masstransit.massient.com/configuration/observability

Uses OpenTelemetry.Extensions.Hosting, OpenTelemetry.Exporter.Prometheus, and OpenTelemetry.Exporter.Prometheus.AspNetCore. Configures metrics with AddPrometheusExporter and maps the scraping endpoint with UseOpenTelemetryPrometheusScrapingEndpoint. Requires the builder variable from the hosting environment.

```csharp
void ConfigureResource(ResourceBuilder r)
{
    r.AddService("Service Name",
        serviceVersion: "Version",
        serviceInstanceId: Environment.MachineName);
}

builder.Services.AddOpenTelemetry()
    .ConfigureResource(ConfigureResource)
    .WithMetrics(b => b
        .AddMeter(InstrumentationOptions.MeterName) // MassTransit Meter
        .AddPrometheusExporter()
    );

var app = builder.Build();

app.UseOpenTelemetryPrometheusScrapingEndpoint(); // Map prometheus metrics endpoint
```

--------------------------------

### ExcludeFromImplementedTypes attribute example

Source: https://masstransit.massient.com/configuration/topology

Shows how to use the ExcludeFromImplementedTypes attribute to prevent scope filters from being created for the message type.

```csharp
[ExcludeFromImplementedTypes]
public interface ICommand
{
}

public record ReformatHardDrive :
    ICommand
{
}
```

--------------------------------

### Configure global partition key conventions

Source: https://masstransit.massient.com/configuration/transports/sql

Configures transport-specific topology conventions at a global level using GlobalTopology. This example adds a PartitionKeySendTopologyConvention and sets a formatter for a base interface.

```csharp
GlobalTopology.Send.TryAddConvention(new PartitionKeySendTopologyConvention());

GlobalTopology.Send.UsePartitionKeyFormatter<ICanHasPartitionKey>(x => x.Message.PartitionKey.ToString());
```

--------------------------------

### EntityName attribute example

Source: https://masstransit.massient.com/configuration/topology

Shows how to use the EntityName attribute to override the default entity name for a message type.

```csharp
[EntityName("order-submitted")]
public record LegacyOrderSubmittedEvent
{
}
```

--------------------------------

### PublishObserver implementation

Source: https://masstransit.massient.com/configuration/observability

Implements IPublishObserver to observe publish operations. PrePublish is called before publishing, PostPublish after acknowledgment, and PublishFault on exceptions. Note: not called for sends; use both observers to observe all outbound messages.

```csharp
public class PublishObserver :
    IPublishObserver
{
    /// <summary>
    /// Called right before the message is published to the transport (exchange/topic).
    /// All headers are configured and the message is serialized.
    /// Note: This is for publish operations (IPublishEndpoint), not for sends.
    /// Use both PublishObserver and SendObserver to observe all outbound messages.
    /// </summary>
    public Task PrePublish<T>(PublishContext<T> context)
        where T : class
    {
        return Task.CompletedTask;
    }

    /// <summary>
    /// Called after the message is published and acknowledged by the broker.
    /// The message is now available to all subscribers.
    /// </summary>
    public Task PostPublish<T>(PublishContext<T> context)
        where T : class
    {
        return Task.CompletedTask;
    }

    /// <summary>
    /// Called if an exception occurs while publishing the message.
    /// This includes serialization errors, routing failures, or transport errors.
    /// </summary>
    public Task PublishFault<T>(PublishContext<T> context, Exception exception)
        where T : class
    {
        return Task.CompletedTask;
    }
}
```

--------------------------------

### ScheduleJob

Source: https://masstransit.massient.com/concepts/job-consumers

Schedules a job to run at a specified start time using the ScheduleJob method on IRequestClient<SubmitJob<TJob>>.

```APIDOC
## ScheduleJob

### Description
Schedules a job to run at a specified start time using the `ScheduleJob` method on `IRequestClient<SubmitJob<TJob>>`.

### Method
POST

### Endpoint
`{path}`

### Parameters
#### Path Parameters
- **path** (string) - Required - The path to the video to convert.

### Request Example
```csharp
[HttpPost("{path}")]
public async Task<IActionResult> SubmitJob(string path, [FromServices] IRequestClient<SubmitJob<ConvertVideo>> client)
{
    await client.ScheduleJob(DateTimeOffset.Now.AddMinutes(15), new ConvertVideo
    {
        Path = path
    });

    return Ok(new
    {
        jobId,
        path
    });
}
```

### Response
#### Success Response (200)
- **jobId** (Guid) - The generated job ID.
- **path** (string) - The path to the video.

#### Response Example
```json
{
  "jobId": "...",
  "path": "..."
}
```
```
Tokens:
Raw