Store-and-forward

Component: SQL Server Transport
NuGet Package NServiceBus.SqlServer (2.x)
Target NServiceBus Version: 5.x

Prerequisites

An instance of SQL Server Express is installed and accessible as .\SqlExpress.

At startup each endpoint will create its required SQL assets including databases, tables and schemas.

The databases created by this sample are NsbSamplesStoreAndForwardReceiver and NsbSamplesStoreAndForwardSender.

Running the project

  1. Start the Sender project (right-click on the project, select the Debug > Start new instance option).
  2. Start the Receiver project.
  3. In the Sender's console is the text Press <enter> to send a message when the app is ready.
  4. Hit enter.

Verifying that the sample works correctly

  1. The Receiver displays information that the order was submitted.
  2. The Sender displays information that the order was accepted.
  3. Hit enter in the Receiver console to shut it down.
  4. Go to the SQL Server Management Studio and delete the NsbSamplesStoreAndForwardReceiver database.
  5. Hit enter again in the Sender console
  6. Notice the retry mechanism kicks in after a 10 seconds and retries sending the message to the destination database.
  7. Restart the Receiver project.
  8. Notice that the message has been delivered to the Receiver.

Code walk-through

When SQL Server transport is used in the multi-instance mode, the messages are inserted directly into the remote destination database's table. If the receiving endpoint's database is down or inaccessible, e.g. because of network failures, the sending endpoint can't send messages to it. In such situations the exception is thrown from the Send() or the Publish() methods, resulting in a potential message loss.

The message loss problem can be prevented by adding store-and-forward functionality to the SQL Server transport, as explained in this sample.

The Outbox would not help solve the issue presented in this example, because it is bypassed when sending messages from outside of a message handler, e.g. from ASP.NET MVC controller.

The sample contains three projects:

  • Shared - A class library containing common code including messages definitions.
  • Sender - A console application responsible for sending the initial OrderSubmitted message and processing the follow-up OrderAccepted message.
  • Receiver - A console application responsible for processing the OrderSubmitted message.

Sender and Receiver use different databases, just like in a production scenario where two systems are integrated using NServiceBus. Each database contains, apart from business data, queues for the NServiceBus endpoint.

Sender project

The Sender does not store any data. It mimics the front-end system where orders are submitted by the users and passed via the bus to the back-end. It is configured to use SQL Server transport.

var transport = busConfiguration.UseTransport<SqlServerTransport>();
var sender = @"Data Source=.\SqlExpress;Database=NsbSamplesStoreAndForwardSender;Integrated Security=True";
transport.ConnectionString(sender);
var receiver = @"Data Source=.\SqlExpress;Database=NsbSamplesStoreAndForwardReceiver;Integrated Security=True";
transport.UseSpecificConnectionInformation(
    EndpointConnectionInfo.For("Samples.SqlServer.StoreAndForwardReceiver")
        .UseConnectionString(receiver)
);
busConfiguration.UsePersistence<InMemoryPersistence>();
var pipeline = busConfiguration.Pipeline;
pipeline.Register<ForwardBehavior.Registration>();
pipeline.Register<SendThroughLocalQueueBehavior.Registration>();

The Sender registers two custom behaviors, one for the send pipeline and one for the receive pipeline.

Send pipeline

The new behavior is added at the beginning of the send pipeline (in Version 2) or in the routing stage (in Version 3).

public class SendThroughLocalQueueBehavior :
    IBehavior<OutgoingContext>
{
    Configure configure;

    public SendThroughLocalQueueBehavior(Configure configure)
    {
        this.configure = configure;
    }

    public void Invoke(OutgoingContext context, Action next)
    {
        // If processing an incoming message (in a handler), skip this behavior
        if (context.IncomingMessage != null)
        {
            next();
            return;
        }

        var deliveryOptions = context.DeliveryOptions;
        var sendOptions = deliveryOptions as SendOptions;
        var outgoingHeaders = context.OutgoingLogicalMessage.Headers;
        if (sendOptions == null)
        {
            var publishOptions = deliveryOptions as PublishOptions;
            if (publishOptions == null)
            {
                // Should never get here as is makes no sense to reply from outside of a handler
                throw new Exception($"Not supported delivery option: {deliveryOptions.GetType().Name}");
            }
            // Technically it is not necessary to store the actual type,
            // just a marker that this is a Publish operation
            outgoingHeaders["$.store-and-forward.eventtype"] =
                publishOptions.EventType.AssemblyQualifiedName;
        }
        else
        {
            outgoingHeaders["$.store-and-forward.destination"] =
                sendOptions.Destination.ToString();
            sendOptions.Destination = configure.LocalAddress;
            // Could as well store other properties of the SendOptions
            // to handle things like delayed delivery
        }
        context.Set<DeliveryOptions>(new SendOptions(configure.LocalAddress));
        next();
    }

    public class Registration :
        RegisterStep
    {
        public Registration()
            : base(
                stepId: "SendThroughLocalQueue",
                behavior: typeof(SendThroughLocalQueueBehavior),
                description: "Put the outgoing message into this endpoint's input queue")
        {
            InsertBefore(WellKnownStep.MutateOutgoingMessages);
            InsertAfter(WellKnownStep.EnforceBestPractices);
        }
    }
}

The behavior ignores:

  • Messages sent from a handler, because the incoming message will be retried ensuring the outgoing messages are eventually delivered.
  • Deferred messages, because their destination is the local timeout manager satellite.

The behavior captures the destination of the message in a header and overrides the original value so that the message is actually sent to the local endpoint (put at the end of the endpoint's incoming queue).

NOTICE: In Version 3 of this sample some properties of a message (such as defer time) are not handled. In order to use similar feature in production, make sure to add code to handle all possible situations or refrain from using deferred messages in endpoints where store-and-forward is used.

Receive pipeline

In the receive pipeline the new behavior is placed just before loading the message handlers (in Version 2) or in the physical processing stage (in Version 3).

var logicalMessage = context.IncomingLogicalMessage;
var headers = logicalMessage.Headers;
if (headers.TryGetValue("$.store-and-forward.destination", out var destination))
{
    // Ultimate destination is in the header - send there (via the outbox) and skip the processing.
    bus.Send(destination, logicalMessage.Instance);
}
else if (headers.TryGetValue("$.store-and-forward.eventtype", out var eventType))
{
    // The event that to be re-published - publish it and skip the processing.
    bus.Publish(logicalMessage.Instance);
}
else
{
    // Normal message, process it.
    next();
}

If the message contains the headers used by the send-side behavior, it is forwarded to the ultimate destination instead of being processed locally. This is the first time the remote database of the Receiver endpoint is contacted. Should it be down, the retry mechanism kicks in and ensures the message is eventually delivered to the destination. In this example the retry mechanism is configured to retry every 10 seconds for up to 100 times.

<SecondLevelRetriesConfig Enabled="true"
                          TimeIncrease="00:00:10"
                          NumberOfRetries="100" />
When both sender's and receiver's databases cannot be accessed in a distributed transaction, the ForwardBehavior has to include a TransactionScope that suppresses the ambient transaction before forwarding the message.

Receiver project

The Receiver mimics a back-end system. The following code configures it to use the multi-instance mode of the SQL Server transport.

var transport = busConfiguration.UseTransport<SqlServerTransport>();
var receiver = @"Data Source=.\SqlExpress;Database=NsbSamplesStoreAndForwardReceiver;Integrated Security=True";
transport.ConnectionString(receiver);
var sender = @"Data Source=.\SqlExpress;Database=NsbSamplesStoreAndForwardSender;Integrated Security=True";
var connectionInfo = EndpointConnectionInfo.For("Samples.SqlServer.StoreAndForwardSender")
    .UseConnectionString(sender);
transport.UseSpecificConnectionInformation(connectionInfo);

busConfiguration.UsePersistence<InMemoryPersistence>();
Multi-instance mode is deprecated in Version 3 of SQL Server transport and will be removed in the next major version. By that time an alternative store-and-forward solution will be provided. For more information refer to the SQL Server transport Version 2 to Version 3 upgrade guide.

Related Articles

  • Outbox
    Reliable messaging without distributed transactions.
  • SQL Server Transport
    High-level description of NServiceBus SQL Server Transport.

Last modified