Scheduling Configuration
Time is important, particularly in distributed applications. Sophisticated systems need to schedule things, and MassTransit has extensive scheduling support.
MassTransit supports two different methods of message scheduling:
- Scheduler-based, using either Quartz.NET or Hangfire, where the scheduler runs in a service and schedules messages using a queue.
- Transport-based, using the transports built-in message scheduling/delay capabilities. In some cases, such as RabbitMQ, this requires an additional plug-in to be installed and configured.
Recurring schedules are only supported by Quartz.NET or Hangfire.
Configuration
Depending upon the scheduling method used, the bus must be configured to use the appropriate scheduler.
services.AddMassTransit(x =>
{
Uri schedulerEndpoint = new Uri("queue:scheduler");
x.AddMessageScheduler(schedulerEndpoint);
x.UsingRabbitMq((context, cfg) =>
{
cfg.UseMessageScheduler(schedulerEndpoint);
cfg.ConfigureEndpoints(context);
});
});
RabbitMQ
Azure Service Bus
Amazon SQS
SQL Transport
Usage
To use the message scheduler (outside of a consumer), resolve IMessageScheduler from the container.
Consumer
To schedule messages from a consumer, use any of the ConsumeContext extension methods, such as ScheduleSend, to schedule messages.
services.AddMassTransit(x =>
{
Uri schedulerEndpoint = new Uri("queue:scheduler");
x.AddMessageScheduler(schedulerEndpoint);
x.AddConsumer<ScheduleNotificationConsumer>();
x.UsingRabbitMq((context, cfg) =>
{
cfg.UseMessageScheduler(schedulerEndpoint);
cfg.ConfigureEndpoints(context);
});
});
public class ScheduleNotificationConsumer :
IConsumer<ScheduleNotification>
{
public async Task Consume(ConsumeContext<ScheduleNotification> context)
{
Uri notificationService = new Uri("queue:notification-service");
await context.ScheduleSend<SendNotification>(notificationService,
context.Message.DeliveryTime, new()
{
EmailAddress = context.Message.EmailAddress,
Body = context.Message.Body
});
}
}
public record ScheduleNotification
{
public DateTime DeliveryTime { get; init; }
public string EmailAddress { get; init; }
public string Body { get; init; }
}
public record SendNotification
{
public string EmailAddress { get; init; }
public string Body { get; init; }
}
The message scheduler, specified during bus configuration, will be used to schedule the message.
Scope
To schedule messages from a bus, use IMessageScheduler from the container (or create a new one using the bus and appropriate scheduler).
services.AddMassTransit(x =>
{
Uri schedulerEndpoint = new Uri("queue:scheduler");
x.AddMessageScheduler(schedulerEndpoint);
x.UsingRabbitMq((context, cfg) =>
{
cfg.UseMessageScheduler(schedulerEndpoint);
cfg.ConfigureEndpoints(context);
});
});
await using var scope = provider.CreateAsyncScope();
var scheduler = scope.ServiceProvider.GetRequiredService<IMessageScheduler>();
await scheduler.SchedulePublish<SendNotification>(
DateTime.UtcNow + TimeSpan.FromSeconds(30), new()
{
EmailAddress = "frank@nul.org",
Body = "Thank you for signing up for our awesome newsletter!"
});
public record SendNotification
{
public string EmailAddress { get; init; }
public string Body { get; init; }
}
Recurring Messages
You can also schedule a message to be send to you periodically. This functionality uses the Quartz.Net periodic schedule feature and requires some knowledge of cron expressions.
To request a recurring message, you need to use ScheduleRecurringSend
extension method, which is available
for both Context
and SendEndpoint
. This message requires a schedule object as a parameter, which must
implement RecurringSchedule
interface. Since this interface is rather broad, you can use the default
abstract implementation DefaultRecurringSchedule
as the base class for your own schedule.
public class PollExternalSystemSchedule : DefaultRecurringSchedule
{
public PollExternalSystemSchedule()
{
CronExpression = "0 0/1 * 1/1 * ? *"; // this means every minute
}
}
public class PollExternalSystem {}
var schedulerEndpoint = await bus.GetSendEndpoint(_schedulerAddress);
var scheduledRecurringMessage = await schedulerEndpoint.ScheduleRecurringSend(
InputQueueAddress, new PollExternalSystemSchedule(), new PollExternalSystem());
When you stop your service or just have any other need to tell Quartz service to stop sending you
these recurring messages, you can use the return value of ScheduleRecurringSend
to cancel the recurring schedule.
await bus.CancelScheduledRecurringMessage(scheduledRecurringMessage);
You can also cancel using schedule id and schedule group values, which are part of the recurring schedule object.
Quartz.NET
To host Quartz.NET with MassTransit, configure Quartz and MassTransit as shown below.
services.AddQuartz(q =>
{
q.UseMicrosoftDependencyInjectionJobFactory();
});
services.AddMassTransit(x =>
{
x.AddPublishMessageScheduler();
x.AddQuartzConsumers();
x.UsingRabbitMq((context, cfg) =>
{
cfg.UsePublishMessageScheduler();
cfg.ConfigureEndpoints(context);
});
});
Hangfire
services.AddHangfire(h =>
{
h.UseRecommendedSerializerSettings();
h.UseMemoryStorage();
});
services.AddMassTransitTestHarness(x =>
{
x.AddPublishMessageScheduler();
x.AddHangfireConsumers();
x.UsingInMemory((context, cfg) =>
{
cfg.UsePublishMessageScheduler();
cfg.ConfigureEndpoints(context);
});
})