Store-and-forward

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

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

Verifying that the sample works correctly

The Receiver displays information that the order was submitted.
The Sender displays information that the order was accepted.
Hit enter in the Receiver console to shut it down.
Go to the SQL Server Management Studio and delete the NsbSamplesStoreAndForwardReceiver database.
Hit enter again in the Sender console
Notice the retry mechanism kicks in after a 10 seconds and retries sending the message to the destination database.
Restart the Receiver project.
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 = endpointConfiguration.UseTransport<SqlServerTransport>();
transport.EnableLegacyMultiInstanceMode(GetConnecton);

var pipeline = endpointConfiguration.Pipeline;
pipeline.Register(
    stepId: "Forward",
    behavior: new ForwardBehavior(),
    description: "Forwards messages to destinations.");
pipeline.Register("Store",
    factoryMethod: builder =>
    {
        var localAddress = builder.Build<ReadOnlySettings>().LocalAddress();
        return new SendThroughLocalQueueRoutingToDispatchConnector(localAddress);
    },
    description: "Send messages through local endpoint.");

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).

class SendThroughLocalQueueRoutingToDispatchConnector :
    ForkConnector<IRoutingContext, IDispatchContext>
{
    public SendThroughLocalQueueRoutingToDispatchConnector(string localAddress)
    {
        this.localAddress = localAddress;
    }

    public override Task Invoke(IRoutingContext context, Func<Task> next, Func<IDispatchContext, Task> fork)
    {
        IncomingMessage incomingMessage;
        var fromHandler = context.Extensions.TryGet(out incomingMessage);

        DelayedDeliveryConstraint constraint;
        if (context.Extensions.TryGetDeliveryConstraint(out constraint) || fromHandler)
        {
            return next();
        }
        var operations = context.RoutingStrategies
            .Select(s => RouteThroughLocalEndpointInstance(s, context))
            .ToArray();
        return fork(new DispatchContext(operations, context));
    }

    TransportOperation RouteThroughLocalEndpointInstance(RoutingStrategy routingStrategy, IRoutingContext context)
    {
        var outgoingMessage = context.Message;
        var headers = new Dictionary<string, string>(outgoingMessage.Headers);
        var originalTag = routingStrategy.Apply(headers);
        var unicastTag = originalTag as UnicastAddressTag;
        if (unicastTag == null)
        {
            var multicastTag = originalTag as MulticastAddressTag;
            if (multicastTag == null)
            {
                throw new Exception($"Unsupported type of address tag: {originalTag.GetType().FullName}");
            }
            headers["$.store-and-forward.eventtype"] = multicastTag.MessageType.AssemblyQualifiedName;
        }
        else
        {
            headers["$.store-and-forward.destination"] = unicastTag.Destination;
        }
        var message = new OutgoingMessage(outgoingMessage.MessageId, headers, outgoingMessage.Body);
        return new TransportOperation(
            message: message,
            addressTag: new UnicastAddressTag(localAddress),
            requiredDispatchConsistency: DispatchConsistency.Default,
            deliveryConstraints: context.Extensions.GetDeliveryConstraints());
    }

    string localAddress;

    class DispatchContext :
        ContextBag,
        IDispatchContext
    {
        TransportOperation[] operations;
        public ContextBag Extensions => this;
        public IBuilder Builder => Get<IBuilder>();

        public DispatchContext(TransportOperation[] operations, IBehaviorContext parentContext)
            : base(parentContext?.Extensions)
        {
            this.operations = operations;
        }

        public IEnumerable<TransportOperation> Operations => operations;
    }
}

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).

string destination;
var message = context.Message;
var headers = message.Headers;
var body = message.Body;
if (headers.TryGetValue("$.store-and-forward.destination", out destination))
{
    var operation = new TransportOperation(
        message: new OutgoingMessage(context.MessageId, headers, body),
        addressTag: new UnicastAddressTag(destination));
    return fork(new DispatchContext(operation, context));

}
if (headers.TryGetValue("$.store-and-forward.eventtype", out destination))
{
    var messageType = Type.GetType(destination, true);
    var operation = new TransportOperation(
        message: new OutgoingMessage(context.MessageId, headers, body),
        addressTag: new MulticastAddressTag(messageType));
    return fork(new DispatchContext(operation, context));
}
return 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.

var recoverability = endpointConfiguration.Recoverability();
recoverability.Delayed(
    customizations: delayed =>
    {
        delayed.NumberOfRetries(100);
        delayed.TimeIncrease(TimeSpan.FromSeconds(10));
    });

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 = endpointConfiguration.UseTransport<SqlServerTransport>();
var routing = transport.Routing();
routing.RegisterPublisher(
    eventType: typeof(OrderSubmitted),
    publisherEndpoint: "Samples.SqlServer.StoreAndForwardSender");

transport.EnableLegacyMultiInstanceMode(async address =>
{
    string connectionString;
    if (address.StartsWith("Samples.SqlServer.StoreAndForwardReceiver") ||
        address == "error")
    {
        connectionString = Connections.ReceiverConnectionString;
    }
    else
    {
        connectionString = Connections.SenderConnectionString;
    }

    var connection = new SqlConnection(connectionString);
    await connection.OpenAsync()
        .ConfigureAwait(false);
    return connection;
});

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.