I am currently working on a project for a warehousing system, where CQRS and Event Sourcing are being factored into the system. Part of this work is building the concept of sagas using a slip-based routing table. Enterprise Integration Patterns has a great write-up from a simplistic point of view to explain what this is.
From my understanding, the concept of slip-based routing (or sagas, if you will) is that you take a request, build a document, then forward this document to several recipients that will process this document until the process succeeds, or fails, posting the success or failure somewhere for use by the system user at a later date. During this process, when the document is received, the recipient can do one of two things:
- Process the request and forward it immediately
- Pre-process and/or simply store the request and wait for a system operator to view the information, make a decision, and move forward.
Consider this as your typical coding structure to handle such a workflow (basis of this was after a discussion with Greg Young on the topic, who is advising on this project.):
public class Route {
public Uri Uri { get; set; }
public string Description { get; set; }
public bool HasCompleted { get; set; } = false
public DateTime CompletionTimestamp { get; set; }
}
public class RoutingTable {
List<Route> _routes;
public IReadOnlyList<Route> Routes => _routes;
public Route OnError { get; }
public Route OnComplete { get; }
public RoutingTable(IEnumerable<Route> routes, Route onError, Route onComplete) {
_routes = new List<Route>(routes);
OnError = onError;
OnComplete = onComplete;
}
public Route GetRoute() => _routes.Where(r => !r.HasCompleted).FirstOrDefault() ?? OnComplete;
public void MarkCompleted() {
var route = GetRoute();
if (route != null) {
route.HasCompleted = true;
route.CompletionTimestamp = DateTime.UtcNow;
}
}
}
public interface IDocument {
Guid DocumentId { get; }
Guid CorelationId { get; }
RoutingTable Routes { get; }
}
public interface IProtocol {
Task ForwardAsync<TDocument>(Uri uri, TDocument document, CancellationToken token = default) where TDocument : IDocument
}
Note: You can create various ways of building the implementations of
IProtocolfor your use. Within this article, I am going to assume that you have one forwarding protocol.
So, a possible use of this framework may look like the following:
Note 1: I am not providing details of the document, just basis for discussion purposes. consider this as pseudocode.
Note 2: Observe that all elements of the document are primitive types or basic classes. Nothing within the document implements any infrastructure of any application.
var routes = new Route[] {
new Route { Uri = new Uri("app://fulfillment/process-payments") },
new Route { Uri = new Uri("app://fulfillment/pick-items") },
new Route { Uri = new Uri("app://fulfillment/pack-order") },
new Route { Uri = new Uri("app://fulfillment/ship") },
}
var errorRoute = new Route { Uri = new Uri("app://fulfillment/failure") };
var fulfillmentCompletedRoute = new Route { Uri = new Uri("app://fulfillment/completed") };
var table = new RoutingTable(routes, errorRoute, fulfillmentCompletedRoute);
public class Fulfillment : IDocument {
public Guid DocumentId { get; }
public Guid CorelationId { get; }
public RoutingTable Routes { get; }
public string SaleId { get; }
public List<object> Items { get; }
public List<object> Payments { get; }
public string ShippingAddress { get; }
public Fulfillment(..., ..., RoutingTable table, ..., ...) {
Routes = table;
}
}
Let’s make an assumption that within your application, you have implemented a single IProtocol for internal application routing; with the idea that a protocol can be anything from http/https, tcp streams, writing to a file share, etc.
public class InternalRoutingProtocol : IProtocol {
Dictionary<Uri, Func<object, Task>> _handlers = new Dictionary<Uri, Func<object, Task>>;
public Register<TDocument>(Uri uri, Func<TDocument, Task> handler) where TDocument : IDocument {
_handlers.Add(uri, handler);
}
public async Task ForwardAsync<TDocument>(Uri uri, TDocument document, CancellationToken token = default) where TDocument : IDocument {
if (_handlers.TryGetValue(uri, out Func<object, Task> handler)) {
await handler.Invoke(document, token);
return;
}
throw new RouteNotFoundException("...");
}
}
And within your main method [or your Startup.cs class, for web applications… can be within an IoC container as well, etc., etc., etc.] you would implement the above class hierarchy as follows:
public class Program {
public async Task Main(string [] args) {
var processPayments = new Uri("app://fulfillment/process-payments");
var pickItems = new Uri("app://fulfillment/pick-items");
var packOrder = new Uri("app://fulfillment/pack-order");
var shipOrder = new Uri("app://fulfillment/ship");
var errors = new Uri("app://fulfillment/failure");
var completed = new Uri("app://fulfillment/completed");
var routingProtocol = new InternalRoutingProtocol();
routingProtocol.Register<Fulfillment>(processPayments, async (document) => {
try {
// process the payment.
document.Routes.MarkCompleted();
var nextRoute = document.Routes.GetRoute();
await routingProtocol.ForwardAsync(nextRoute.Uri, document);
} catch (Exception exc) {
await routingProtocol.ForwardAsync(document.Routes.OnError.Uri, document);
}
});
routingProtocol.Register<Fulfillment>(pickItems, async (document) => {
// store document.
// create view for user.
});
routingProtocol.Register<Fulfillment>(packOrder, async (document) => {
// store document.
// create view for user.
});
routingProtocol.Register<Fulfillment>(shipOrder, async (document) => {
try {
// prepare shipment document
// print label
document.Routes.MarkCompleted();
var nextRoute = document.Routes.GetRoute();
await routingProtocol.ForwardAsync(nextRoute.Uri, document);
} catch (Exception exc) {
await routingProtocol.ForwardAsync(document.Routes.OnError.Uri, document);
}
});
routingProtocol.Register<Fulfillment>(completed, async (document) => {
// perform cleanup as necessary.
// send email, ping SignalR clients, etc.
});
routingProtocol.Register<Fulfillment>(errors, async (document) => {
// determine error.
// if can be resolved, resolve and restart processing.
// if cannot be resolved, store document and prepare view for manual intervention.
});
var orderToFulfill = new Fulfillment {
// build the details here for the fulfillment.
};
var start = orderToFulfill.Routes.GetRoute();
await routingProtocol.ForwardAsync(start.Uri, orderToFulfill);
}
}
Looking above, you can see that each handler is focused on a specific thing that they have to do, such as processing one or more payments, picking items from the warehouse, packaging those items for shipment, etc.
The overall moral of the story here is that the concept of a slip-based Saga is to break-down a large process to several distinct areas of concern, where the end result of handling that document should have your system into a consistent state, of some sort. With event sourcing, this may mean that a handler may work with several aggregates during the handling of each task, as you naturally would for picking items (each item is its own aggregate, most likely) to processing payments, etc.