It’s been a few months already since I’ve started working for good with distributed systems using (micro)services and asynchronous processing via service bus. Many issues and question raised and one of these was how to not lose the information about commands and events being processed and even more importantly, how to notify the user once the request has completed? I’ve had to come up with some solution that seems to be sufficient (at least for now) and I’d like to share it with you.





Let’s assume that there’s a quite typical system which contains HTTP API, service bus and the set of (micro)services – in order to make it even easier, we will have only one type of such service called the User (micro)service. Whenever the request arrives at the API it will be passed further to the service bus and then to the instance of the User (micro)service which will finally consume it and produce some event. Everything is done fully asynchronously, which means that you can’t simply wait for the result and treat it as a synchronous API call.

One of the main problems that I encountered while designing such API fully based on the asynchronous operations (and I don’t mean by that only using the async keyword, but processing the commands in a way that would not return an immediate response to the client in a synchronous manner) was how to let the end-user know that the request has finished and e.g. resource was created, updated or deleted.

It was more than certain that I had to keep track of the requests and update their state. Then I had to expose some endpoint in the API that would return a status of such operation so that consumer would know if the request completed. The first thing I did was to return a 202 (Accepted) HTTP status code for any POST/PUT/DELETE request – these would be the commands being sent to the system. Of course, the GET requests are queries and they do not need to be processed via service bus, so I can return the result immediately.

Alright, that was the first step. However, 202 is merely a status code – how could the API consumer possibly know if the request has completed? Or where to fetch the new resource from? Since I’m a big fan of CQS I tend not to return any result (or body in that case) for methods that process commands (which are not idempotent). Still, I could take advantage of the HTTP Headers and put some valuable information here. And what would it be? I chose to create 2 custom headers: X-Operation that would have a unique endpoint to the operation (containing a status of the request) and X-Resource containing endpoint to the created or updated resource (if it would be deleted, then this header would remain empty).

I have a typical ICommand, IEvent, ICommandHandler and IEventHandler interfaces and their implementations like CreateUser (a command) and UserCreated (an event) etc. In case you’re not familiar with the command and event handler patterns, please read this article. Finally, there’s this special class for tracking the requests:

public class Request { public Guid Id { get; set; } = Guid.NewGuid(); public string Name { get; set; } public string Origin { get; set; } public string Resource { get; set; } public string Culture { get; set; } public DateTime CreatedAt { get; set; } public static Request Create<T>(Guid id, string origin, string culture, string resource = "") => new Request { Id = id, Name = typeof(T).Name.ToLowerInvariant(), Origin = origin.StartsWith("/") ? origin.Remove(0, 1) : origin, Culture = culture, Resource = resource, CreatedAt = DateTime.UtcNow }; } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 public class Request { public Guid Id { get ; set ; } = Guid . NewGuid ( ) ; public string Name { get ; set ; } public string Origin { get ; set ; } public string Resource { get ; set ; } public string Culture { get ; set ; } public DateTime CreatedAt { get ; set ; } public static Request Create < T > ( Guid id , string origin , string culture , string resource = "" ) = > new Request { Id = id , Name = typeof ( T ) . Name . ToLowerInvariant ( ) , Origin = origin . StartsWith ( "/" ) ? origin . Remove ( 0 , 1 ) : origin , Culture = culture , Resource = resource , CreatedAt = DateTime . UtcNow } ; }

Which I’m using within my ICommand:

public interface ICommand { Request Request { get; set; } } 1 2 3 4 public interface ICommand { Request Request { get ; set ; } }

And the corresponding RequestId inside IEvent:

public interface IEvent { Guid RequestId { get; } } 1 2 3 4 public interface IEvent { Guid RequestId { get ; } }

So, what’s going here? Actually, it’s much simpler than it looks. Imagine the following scenario – the requests comes to the API (some particular implementation of ICommand like mentioned before CreateUser). Then, I’m additionally creating the Request instance for this command (it’s being done internally within some core API logic, I don’t want to go into details here as it’s not really relevant).

public class CreateUser : ICommand { public Request Request { get; set; } public string Email { get; set; } public string Password { get; set; } } 1 2 3 4 5 6 public class CreateUser : ICommand { public Request Request { get ; set ; } public string Email { get ; set ; } public string Password { get ; set ; } }

You may include whatever you want inside the Request class. For me, the most important is Id, but I’m also keeping track of the command name, origin (what was the originally invoked URL) or resource (the unique endpoint of the resource included in the X-Resource HTTP header).

On the other hand, the IEvent contains only the id of the original request, so everything can be composed into a flow (it can be just a single command producing a single event, or much more sophisticated workflow/saga).

Surely, we want to store the request details somewhere in a database:

public static class States { public static string Accepted => "accepted"; public static string Processing => "processing"; public static string Completed => "completed"; public static string Rejected => "rejected"; } public class Operation { public Guid Id { get; protected set; } public Guid RequestId { get; protected set; } public string Name { get; protected set; } public string UserId { get; protected set; } public string Origin { get; protected set; } public string Resource { get; protected set; } public string State { get; protected set; } public string Message { get; protected set; } public string Code { get; protected set; } public DateTime CreatedAt { get; protected set; } public DateTime UpdatedAt { get; protected set; } protected Operation() { } public Operation(Guid requestId, string name, string userId, string origin, string resource, DateTime createdAt) { Id = Guid.NewGuid(); Name = name; RequestId = requestId; UserId = userId; Origin = origin; Resource = resource; CreatedAt = createdAt; State = States.Accepted; } public void Complete(string message = null) { if (State.EqualsCaseInvariant(States.Rejected)) { throw new InvalidOperationException($"Operation: {Id} has been rejected and can not be completed."); } SetCode(OperationCodes.Success); SetMessage(message); SetState(States.Completed); } public void Reject(string code, string message) { if (State.EqualsCaseInvariant(States.Completed)) { throw new InvalidOperationException($"Operation: {Id} has been completed and can not be rejected."); } SetCode(code); SetMessage(message); SetState(States.Rejected); } public void Process() { if (State.EqualsCaseInvariant(States.Completed)) { throw new InvalidOperationException($"Operation: {Id} has been completed and can not be processed."); } if (State.EqualsCaseInvariant(States.Rejected)) { throw new InvalidOperationException($"Operation: {Id} has been rejected and can not be processed."); } SetState(States.Processing); } public void SetMessage(string message) { if (message?.Length > 500) { throw new ArgumentException("Operation message can not have more than 500 characters.", nameof(message)); } Message = message; UpdatedAt = DateTime.UtcNow; } private void SetState(string state) { if(State.EqualsCaseInvariant(state)) return; State = state; UpdatedAt = DateTime.UtcNow; } public void SetCode(string code) { if(Code.EqualsCaseInvariant(code)) return; Code = code; UpdatedAt = DateTime.UtcNow; } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 public static class States { public static string Accepted = > "accepted" ; public static string Processing = > "processing" ; public static string Completed = > "completed" ; public static string Rejected = > "rejected" ; } public class Operation { public Guid Id { get ; protected set ; } public Guid RequestId { get ; protected set ; } public string Name { get ; protected set ; } public string UserId { get ; protected set ; } public string Origin { get ; protected set ; } public string Resource { get ; protected set ; } public string State { get ; protected set ; } public string Message { get ; protected set ; } public string Code { get ; protected set ; } public DateTime CreatedAt { get ; protected set ; } public DateTime UpdatedAt { get ; protected set ; } protected Operation ( ) { } public Operation ( Guid requestId , string name , string userId , string origin , string resource , DateTime createdAt ) { Id = Guid . NewGuid ( ) ; Name = name ; RequestId = requestId ; UserId = userId ; Origin = origin ; Resource = resource ; CreatedAt = createdAt ; State = States . Accepted ; } public void Complete ( string message = null ) { if ( State . EqualsCaseInvariant ( States . Rejected ) ) { throw new InvalidOperationException ( $ "Operation: {Id} has been rejected and can not be completed." ) ; } SetCode ( OperationCodes . Success ) ; SetMessage ( message ) ; SetState ( States . Completed ) ; } public void Reject ( string code , string message ) { if ( State . EqualsCaseInvariant ( States . Completed ) ) { throw new InvalidOperationException ( $ "Operation: {Id} has been completed and can not be rejected." ) ; } SetCode ( code ) ; SetMessage ( message ) ; SetState ( States . Rejected ) ; } public void Process ( ) { if ( State . EqualsCaseInvariant ( States . Completed ) ) { throw new InvalidOperationException ( $ "Operation: {Id} has been completed and can not be processed." ) ; } if ( State . EqualsCaseInvariant ( States . Rejected ) ) { throw new InvalidOperationException ( $ "Operation: {Id} has been rejected and can not be processed." ) ; } SetState ( States . Processing ) ; } public void SetMessage ( string message ) { if ( message ? . Length > 500 ) { throw new ArgumentException ( "Operation message can not have more than 500 characters." , nameof ( message ) ) ; } Message = message ; UpdatedAt = DateTime . UtcNow ; } private void SetState ( string state ) { if ( State . EqualsCaseInvariant ( state ) ) return ; State = state ; UpdatedAt = DateTime . UtcNow ; } public void SetCode ( string code ) { if ( Code . EqualsCaseInvariant ( code ) ) return ; Code = code ; UpdatedAt = DateTime . UtcNow ; } }

As you can see there’s an Operation class created for each new request which may have a different state. And it’s being updated whenever an event containing given RequestId occurs. The final state of the request (or operation in that example) depends whether the published event was successful e.g. UserCreated (then the state would be equal to completed) or not (rejected status in that case + some internal operation code like “email_in_use” in order to make it clearer what exactly happened if there was an error).

public class UserCreated : IEvent { public Guid RequestId { get; } public string UserId { get; } public string Email { get; } protected UserCreated() { } public UserCreated(Guid requestId, string userId, string email) { RequestId = requestId; UserId = userId; Email = email; } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 public class UserCreated : IEvent { public Guid RequestId { get ; } public string UserId { get ; } public string Email { get ; } protected UserCreated ( ) { } public UserCreated ( Guid requestId , string userId , string email ) { RequestId = requestId ; UserId = userId ; Email = email ; } }

public class CreateUserRejected : IRejectedEvent { public Guid RequestId { get; } public string UserId { get; } public string Code { get; } public string Reason { get; } protected CreateUserRejected() { } public CreateUserRejected(Guid requestId, string userId, string code, string reason) { RequestId = requestId; UserId = userId; Code = code; Reason = reason; } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 public class CreateUserRejected : IRejectedEvent { public Guid RequestId { get ; } public string UserId { get ; } public string Code { get ; } public string Reason { get ; } protected CreateUserRejected ( ) { } public CreateUserRejected ( Guid requestId , string userId , string code , string reason ) { RequestId = requestId ; UserId = userId ; Code = code ; Reason = reason ; } }

The important thing here is that you’d have the User (micro)service handling all of the commands related to the users (and publishing the events), but on top of that there would be a separate (micro)service (let’s call it an Operation service) that would subscribe to all of the available commands and requests in your system (including the ones from the other microservices) and just keep track of the Request details included in your commands and update them accordingly based on the RequestId properties within the events.

Eventually, there’s an API endpoint with the following path: /operations/{requestId} which does return the the operation details – status, code, resource URL and so on. The end-user can fetch this object in order to find out if the request already finished. However, there’s even a smarter way to do it – instead of making API consumer to pull the operation status, it’s better to push it for example via web sockets using SignalR.

{ "state": "completed", "success": false, "code": "invalid_email", "message": "Provided email was invalid.", "resource": "" } 1 2 3 4 5 6 7 { "state" : "completed" , "success" : false , "code" : "invalid_email" , "message" : "Provided email was invalid." , "resource" : "" }

It was difficult at first to switch thinking from a typical request – (synchronous) response pattern, yet once I realized that there’s a way to track all of the asynchronous requests and notify the end-user what’s happening, I enjoyed using this pattern, as it provides more flexibility and scalability while designing your API. Thus such API acts just as a gateway that has no business logic whatsoever and is not a bottleneck anymore, as you’re not obliged to return a response in a synchronous way.