Satellites are lightweight message processors that run in the same process as their owning endpoint. While they are mostly used by NServiceBus to implement infrastructure features like the Gateway, they can be used in scenarios where messages from additional queues other than the main input queue need to be processed. This is useful when the robustness of a separate queue is needed without having to create and set up a new endpoint and configure any message mappings.
Implementing a satellite
The satellite infrastructure allows the handling of messages as they become available on the satellite’s queue. To create a satellite, place the following code in the Setup method of a feature:
public class MySatelliteFeature :
Feature
{
protected override void Setup(FeatureConfigurationContext context)
{
context.AddSatelliteReceiver(
name: "MyCustomSatellite",
transportAddress: new QueueAddress("targetQueue"),
runtimeSettings: PushRuntimeSettings.Default,
recoverabilityPolicy: (config, _) => RecoverabilityAction.MoveToError(config.Failed.ErrorQueue),
onMessage: OnMessage);
}
Task OnMessage(IServiceProvider serviceProvider, MessageContext context, CancellationToken cancellationToken)
{
// Implement what this satellite needs to do once it receives a message
var messageId = context.NativeMessageId;
return Task.CompletedTask;
}
}
The call to AddSatelliteReceiver registers the action to take when a message is received. In the above example, the satellite will watch a queue named targetQueue.
The MessageContext parameter provides details about the incoming message such as the body and headers. When migrating from a satellite extension from NServiceBus version 5 and below, the implementation steps that are in the Handle method go into the func provided to the AddSatelliteReceiver method.
Managing errors
Problems processing the message need to be handled by the satellite implementation. The available options are: immediate retry, delayed retry, or moving the message to the error queue.
To request an immediate retry:
context.AddSatelliteReceiver(
name: "CustomSatellite",
transportAddress: new QueueAddress("targetQueue"),
runtimeSettings: PushRuntimeSettings.Default,
recoverabilityPolicy: (config, errorContext) =>
{
return RecoverabilityAction.ImmediateRetry();
},
onMessage: OnMessage);
To request a delayed retry:
context.AddSatelliteReceiver(
name: "CustomSatellite",
transportAddress: new QueueAddress("targetQueue"),
runtimeSettings: PushRuntimeSettings.Default,
recoverabilityPolicy: (config, errorContext) =>
{
return RecoverabilityAction.DelayedRetry(
timeSpan: TimeSpan.FromMinutes(2));
},
onMessage: OnMessage);
To request that the message be moved to the error queue:
context.AddSatelliteReceiver(
name: "CustomSatellite",
transportAddress: new QueueAddress("targetQueue"),
runtimeSettings: PushRuntimeSettings.Default,
recoverabilityPolicy: (config, errorContext) =>
{
return RecoverabilityAction.MoveToError(config.Failed.ErrorQueue);
},
onMessage: OnMessage);
Raising a critical error
In NServiceBus version 5 and below, IAdvancedSatellite offers additional customization, such as error handling or critical error processing, when receiving messages in the satellite queue by implementing the GetReceiverCustomization() method. In NServiceBus version 6 and 7, CriticalError can be resolved via the builder and used to instruct the endpoint to shut down if a critical error occurs during the processing of a message. In NServiceBus 8 and above, CriticalError can be resolved from IServiceProvider passed to the message callback.
protected override void Setup(FeatureConfigurationContext context)
{
context.AddSatelliteReceiver(
name: "CustomSatellite",
transportAddress: new QueueAddress("targetQueue"),
runtimeSettings: PushRuntimeSettings.Default,
recoverabilityPolicy: (config, _) => RecoverabilityAction.MoveToError(config.Failed.ErrorQueue),
onMessage: OnMessage);
}
Task OnMessage(IServiceProvider serviceProvider, MessageContext context, CancellationToken cancellationToken)
{
// To raise a critical error
var exception = new Exception("CriticalError occurred");
var criticalError = serviceProvider.GetRequiredService<CriticalError>();
criticalError.Raise("Something bad happened - trigger critical error", exception, cancellationToken);
return Task.CompletedTask;
}