Model correctly and write less code, using Akka Streams

One of my most productive days was throwing away 1000 lines of code.

-- Ken Thompson

More than two decades of writing software programs to earn some humble money and to keep the home hearth warm, has taught me that writing simple, elegant and readable programs result in much more maintainable and extensible software. Importantly, stakeholders prefer to have maintainable code. One way to ensure that  these properties exist in the code, is to write fewer lines of code, wherever possible. Generations of greats in the world of computer science and programming have been reminding us about this little yet undeniable truth. Yet, it is surprising how often we tend to ignore this, knowingly and to projects’ detriment.

If we have to write programs that are simple to understand, we have to understand the problem well, before anything else. If the problem is redefined in simpler and more fundamental terms, then it becomes easier to model the problem in the programming terms. The solution that emerges is likely to be simple and small (but no smaller than required). John Bentley’s Programming Pearls offers excellent treatise of this matter. I still refer to it from time to time. I hope, others do too.

It is my considered view that a well-modeled (i.e., its crux is understood) problem, leads to compact and resilient code. It is much easier to reason about. The verbosity of the language and accidental complexities of the technologies involved may bring in necessary planks and poles, but to an alert pair of eyes, the theme remains easily discernible. A beautiful codebase is a smaller codebase.

The case at hand

Allow me to take you through a greatly abridged version of a recent assignment I have been

associated with. I have been helping a team rewrite their server stack, which is meant to allow users

play quiz like games. Shorn of all business functionality, what happens is this:

• A player begins a session, chooses a game (quiz) - from several available - to play
• Player answers questions one at a time 
• The Server, at the end of the game (all questions attempted), calculates and declares total she has scored

The server updates/informs other important functional units, of player’s accomplishments. Any number of players can play games simultaneously. The server has to keep up with this load.

Simple enough, when one strips it of layers of HTTP, Configuration, Database, Timers, Queues and what have you! I have dropped from the list, other business requirements, as well.

Focus on the interaction

For the purpose of this blog, let us consider a particular portion of the abovementioned functionality: what does the server do, when a player contacts the server? Assuming that the player has already been authenticated - which is a precondition of and not a part of the play-related interaction mentioned above - her interaction has two distinct parts:

Interaction 1 (StartARound)

Hey Server, I want to play a game. Here’s my Session Identifier (obtained earlier from AUTH component) and give me a question.

Interaction 2 (PlayARound)

Hey Server, here’s my answer to the previous question and give me the next one (let’s assume that server offers 3 questions only, in three consecutive rounds; then the game ends.)

Let’s ignore all the network, protocol, routing and JSONfication details for the time being. Then, let’s take a look at what is it that the Server does and then, how we can describe that in a manner that is terse, yet conveys the right meaning.

Modeling the interaction is the key

The Server takes in a piece of data and keeps on transforming it, till another piece of data is ready to be handed over to the client. This transformation consists of one or more steps and in each step, one piece gives rise to another piece. Also, in each step, help from other components or services may be summoned and used, before emitting the resultant piece. Therefore, if we can identify each step, what goes into it and what comes out of it, we can easily model how the whole server works! Moreover, with such a model in hand, we can also verify the behaviour of the server.

An obvious first question is how do we identify these pieces? In the world of OO and Functional Programming, where I manage to reside (and so far, have not been evicted), it is quite natural to identify these pieces by their Types! Every step takes in a Type and gives rise to the same or different Type. Given this tenet, how can we represent the way the Server responds to the player?

The Type-borne behaviour

The diagram below elucidates the scheme of things. The rightmost box shows the transformations that are happening inside the Server.

In terms of Types, one way to describe the flow ‘Start A Round’ (above) is:

Let’s elaborate

The logic of confirming the correctness of sessionID passed by the player (is it existing and valid, or not), is encased in the transformer named sessionExistenceChecker. Because the server stipulates that every message reaching its shores, must have a valid sessionID, every message has to pass through sessionExistenceChecker. However, the important observation is this:

sessionExistenceChecker  understands SessionCarrier only. Therefore, in order to be recognized by the checker, every message must also be a SessionCarrier. In OO terms, every message entering sessionExistenceChecker must be subtype (IS-A) of SessionCarrier.

There are three benefits of this approach:

- Type is self-documenting: the model is self-explanatory. The constraints are obvious. If I want to know what do I need to gather before I can ask sessionExistenceChecker to flag me off OK, I have to look no further than the type it expects.

- Compiler helps to reduce defect: if I am unmindful and pass a message which is not session-id-checkable, the compiler will block my progress with a grim message. A defect will be forestalled much before the code is readied for testing. That’s a substantial gain.

- Code is intuitive and readable: it is quite easy - in many cases straightforward - to translate this model into code (we will see a little later).

Now, let’s look at the result of transformation. The transformer applies its logic and emits either of these Types:

IncorrectSessionIDProvided	This indicates that not everything is correct with the SessionCarrier that has been passed.
StartARound	This indicates that StartARound type - which IS-A SessionCarrier - has come out of the checker, unscathed!

The key understanding, again, is that these are not values but Types! The actual (runtime) objects moving in and out of the checker may carry anything, but they must conform to these Types.

The next step (refer to the shaded region of the Flow diagram, above) is to choose a number for the player, associate that with a Round Identifier and then get back to her with an appropriate message. This logic is encased in the transformer named guessNumberPreparator. Because it  is next to sessionExistenceChecker, it has but got to be capable of consuming either of IncorrectSessionIDProvided and StartARound. Then, it emits either of these Types:

IncorrectSessionIDProvided	This indicates that not everything is correct with the SessionCarrier that has been passed.
RoundStarted	This carries A confirmation that the Session Identifier passed with StartARound has been found to be correct (by the checker earlier) A confirmation that the Server has chosen a number for the player A identifier of the round given to the player to guess

That’s it. We have the blueprint of the Server’s implementation of Interaction[1], available.
Translating this into code - when implemented using Akka Streams - we get this:

val serverSays =        Source            .single(StartARound("A123")) // A123 is a session id, obtained from AUTH service            .via(sessionExistenceChecker)            .via(guessNumberPreparator)

The diagram below illustrates Server’s implementation of Interaction[2]: when Player makes a guess, and the server gives her points for correct guessing and shows the latest score.

Recall that the Server ends the game after 3 rounds. The way error is handled is the same as that in the previous flow (StartARound). Also, the transformer that checks correctness of Session Identifier is reused here.

I am not depicting the flow of types and transformers for this flow, for space’s sake. The code segment that implements this flow is small, crisp and perhaps, quite intuitive as well:

val serverSays =     Source         .single(GuessSubmittedByPlayer(sessionID,roundID, guessedNumber)))         .via(sessionExistenceChecker)         .via(roundCorrectnessChecker)         .via(guessedNumberVerifier)         .via(pointsAssigner)         .via(scoreBoardUpdater)         .via(currentScorePreparator)         .via(gameTerminationDecider)         .via(nextGuessGenerator)

That’s what our Server does, to implement Interaction[2]. That’s all there is to it, really!

An Akka-streams based implementation brings in many other benefits, However, the aim of this blog is not to explore and discuss, many and very beneficial aspects of Akka Streams. A number of blogs already exist which do the job very, very well (Colin Breck’s are here, my personal favourite), not to mention Akka Stream’s own site and numerous discussions on StackOverFlow. Therefore, I will rather bring your attention to other aspects of this approach of modeling:

- If we can model the pathway of processing of any message as a series of transformations, then translation of the same in code becomes decidedly easier.

- If the model is clear, the code is small, crisp and readable. This is with what we have begun this blog, haven’t we? The code does what the model depicts; nothing more, nothing less. No code exists, that has no reason to exist. Brevity matters.

- Every transformation demands that types are provided to indicate what it takes and gives. If and when, our code fails to satisfy transformers  - by passing wrong types - compiler stops us in the track. Because it is a compile-time prevention of potential defects, this approach saves time and effort, in a very significant way. That’s an undeniable gain. Type-Driven Development, did  you say?

- When it comes to testing the behaviour, it is possible to write separate testcases for every transformer separately and for a series of them as specified in the flow. For example, it is quite easy to test if the points have been assigned correctly:

val serverSays =         Source           .single(GuessSubmittedByPlayer(sessionID,roundID, guessedNumber)))           .via(sessionExistenceChecker)           .via(roundCorrectnessChecker)           .via(guessedNumberVerifier)           .via(pointsAssigner)           // subsequent transformations dropped because we want to test this outcome

Using akka stream’s testkit and scalatest, it is straightforward to test such partial outcomes (refer to the github codebase). So, we can test functional behaviour of the Server without having to set up the planks and poles around it.

- Take a good look at what your Server is supposed to do, and spend time in modeling the expected behaviour.

- Depend on types to strengthen the correctness and let the compiler help you save time to spend on testing.

- Write quick and automated test cases.

- Above all, do yourself a favour: write less code!

All accompanying code reside here, on github.

Remember the programming maxim: the best code I ever wrote, was the code that I never wrote! (This blog is also posted here: I work as a Principal at Swanspeed Consulting)

Akka's CircuitBreaker: how to incorporate in your application, by passing messages and handling futures

Background

These days, it is quite common to build an application, which makes use of various other services, to augment its own service. In fact, the value of such a service lies in this aggregation: an aggregation, if and when thoughtfully done - and its result is brought to the users - lifts the usefulness of the application by a few notches.

So, in a typical case of such aggregation, a call to our application give rise to a number of calls to services external to itself. Let's assume that all such calls follow the REST style, over HTTP. Our  application makes a series of calls (may be concurrently) , waits for them to return, processes/transforms the values that are returned and possibly, performs an aggregation. Regular stuff, nothing surprising here.

What if one or more of these external calls fail (unreachable host, as an example)?

The call from our application times out eventually, and our fallback logic takes over; perhaps, a default or an exception-indicating value is returned. Again, this is regular stuff too. We all encounter such cases and take requisite actions. This is certainly not a blocker, by itself.

However, if our application is widely popular (why shouldn't it be?), then it is quite likely to be the target of numerous simultaneous calls from its users; perhaps the result of a sudden surge. All these incoming calls are going to result in outbound calls to the external services. If an external service is unavailable, each of these incoming calls (call chains, one may say) is going to waste some time waiting for the responses.

The problem at hand

The point is this: if 2 or 3 successive incoming calls find an outbound call failing, then it is reasonable to assume that a persistent problem exists at the external service's endpoint. Therefore, it is to our application's (and to the users of our application, by extension) benefit that we instruct the subsequent incoming calls to bypass this.  This decision may result in disappointing (or irritating) a few users with missing pieces of information but overall throughput of our application is not compromised.

A Circuit Breaker is a handy pattern to apply to put this approach in practice.

Experts’ take on CircuitBreaker

I came to learn about the CircuitBreaker pattern, from Martin Fowler's blog on the subject: https://martinfowler.com/bliki/CircuitBreaker.html This still is a very good point to start, in my opinion.

Wikipedia has a detailed treatise on it. I also found this blog, to be quite informative:  

CircuitBreaker in Akka

This blog is about the support that Akka has, for this particular pattern. From whatever I have seen so far, a good example of how a CircuitBreaker works in Akka is hard to come by. Akka's own documentation on the subject, is woefully short of what one looks for. Here's my attempt to fill in the void. Let's jump into that.

Akka provides a CircuitBreaker. If an actor calls a service using a CircuitBreaker, then successful and failed calls are tracked. If a certain number of successive calls fails, CircuitBreaker prevents any further request - opens the circuit as it were - to the service, till a certain amount of time elapses. The first request after this duration is allowed to call the service. If this is successful, the CircuitBreaker opens the circuit temporarily; we say the circuit is half-open now. If the next request is successful, then the CircuitBreaker assumes that the external service is now available and the circuit is closed; if the next request fails again, the circuit becomes  open again.

To demonstrate this behaviour, let us implement a small application. It makes use of the information about leading soccer clubs, available at http://clubinfo.com. If we place a call to this REST api, to retrieve information about a soccer club we form a HTTP GET request:

http://api.football-data.org/v1/teams/7

where 7 represents a particular club (in this case, Hamburger SV)

The response is a JSON string, which looks like this:

We employ the following actors to implement the flow:

• Requestor
• SoccerClubInfoGetter
• SysAdminConsole and
• CallWastePreventor

Various actors and messages which our sample application employs

The code is here: https://github.com/nsengupta/Akka-CircuitBreaker-Demo

Requestor is our point of entry; anybody looking for the details of a club, must send a message to this actor. Requestor plays the role of a mediator; it calls either SoccerClubInfoGetter or CallWastePreventor. Let me explain.

SoccerClubInfoGetter encapsulates the actual call made to the external REST endpoint, asynchronously:

CompletableFuture.supplyAsync( new Supplier<ClubDetailsFromXternalSource>() {         @Override          public ClubDetailsFromXternalSource get() { String s = Http.get(clubInfoAskedFor).text(); return ( new                             InteractionProtocol.ClubDetailsFromXternalSource(s,originalSender)                         );                     } }), getContext().system().dispatcher()

I am using a simple REST endpoint accessing tool, named javalite (Http.get()) Any such tool can be used here.

SoccerClubInfoGetter simply pipes the response to javalite's (Http.get()) call above (available as a JSON):

pipe ( CompletableFuture.supplyAsync(                // As shown above            )         ).to(getSender()     );

In our case, Requestor is the recipient of the piped response above, because it is the sender.

In order to emulate a failing call to the external REST endpoint, we feign a longish sleep, so that the caller decides to give up and return, convinced that the external service is unavailable. Because all clubs are identified by a non-zero integer, we take such a step when the identifier of the club is passed as zero!

else { // Emulating a failed call to the external service Thread.sleep(2000); getSender().tell(   new InteractionProtocol.UnavailableClubDetails("timed out"),   getSelf()       }

Take a look at SoccerClubInfoGetter to understand how is it enacting the role of a ‘failure generator’, if you will.

Requestor relies on the SoccerClubInfoGetterActor for getting the job done. If SoccerClubInfoGetterActor sleeps for 2 seconds, Requestor's ask() times out (ASK_TIMEOUT value is set to 1 second).

We convert the Object it returns, to a ClubDetailsFromXternalSource message using a Function object. If ask() fails, we absorb the Exception and return a TimedOutClubDetails message. In both of these cases, the Requestor feeds the message to itself, by calling a getSelf().

Requestor's receive() function is equipped with the logic of handling either of the two messages mentioned in the paragraph above:

Requestor remembers the ActorRef of the actor who has sought the club's information in the first place, which it uses to provide the appropriate response. In case, the ask() has timed out, the response is a piece of helpful information ("Service unresponsive, try again later"). In both the cases, we are returning a String type; so, the compiler is happy too.

Take a look at RESTDriver, to understand the behaviour of Requestor and SoccerClubInfoGetter.

Failure is emulated

Let's revisit the following code snippet from SoccerClubInfoGetter:

It is an emulation of an unpredictable situation: the call to the external API may or may not fail; we don't know. To bring in an element of determinism, we are targeting a clubID of zero to force the logic follow this path. By making it sleep, we are delaying the thread long enough to cause the ask() from Requestor to time out. In a real-life situation, this determinism is absent; so, it is possible that a series of calls to the external API fails after timing out. This is a waste for sure. Our application's throughput is adversely affected and users are unhappy.

A CircuitBreaker comes as an aid

A CircuitBreaker helps in a situation like this. When we instantiate a CircuitBreaker, we give it a number that represents the tolerance level of failures and a duration for which it remains . If we set it to 3, then we are telling the CircuitBreaker that

1. If 3 successive calls through you fail, consider that target of these calls is in trouble.
2. Cause the circuit to open and leave it at that for the finite duration you are initialized with
3. Any call that reaches you during this duration, send it back immediately, citing a failure
4. Once the duration elapses, be ready the for next call and let it go through.
5. If this call fails, go back to step 2 above.

To execute this cycle predictably and flawlessly, a CircuitBreaker follows a series of well-defined State Transitions. This page from Akka's documentation provides a clear illustration.

If you follow the diagram, you can see that the step '4' above, leaves the story, tad incomplete. What actually happens is that at the step 4, the CircuitBreaker causes the circuit to open in half. If the very next call (step '5' above) is successful, then the CircuitBreaker causes the circuit to close; otherwise, it causes the circuit to be open again (step '2' again).

In our application, we implement a CircuitBreaker in the CallWastePreventor actor (code here).

The key aspect of CallWastePreventor's receive() function is a block, which is - unsurprisingly - very similar to SoccerClubInfoGetter:

workingCallable parameter to the circuit breaker is an instance of JDK 8's Callable. As the  pipe() call expects a CompletionStage, this Callable has to produce a CompletionStage. Moreover, as the pipe() call has to finally pipe a message of the type ClubDetailsFromXternalSource, the Callable has to produce a CompletionStage of ClubDetailsFromXternalSource:

nonWorkingCallable is initialized in an equivalent manner.

In its role as the mediator, the Requestor makes no difference between a SoccerClubInfoGetter and a CallWastePreventor. The mechanism it resorts to is the same: it asks whichever ActorRef it is injected with, (the construction parameter circuitBreakerJeeves) for the information about a particular club and deals with the response it eventually receives.

To get an idea of the way the application works, take a look at RESTDriver (uses SoccerClubInfoGetter) and CBDriver (uses CallWastePreventor).

CircuitBreaker’s Transition handlers

A CircuitBreaker provides a facility to register callback functions, namely onOpen, onHalfOpen and onClose, to mark each of the state-transitions mentioned earlier. We employ a SysAdminConsole actor; this actor is notified every time the CircuitBreaker's state changes. Using a specific message called AdminQueryMessage, we get to know what is the current state, the CircuitBreaker is in, at the moment. We use this to our advantage while testing the application. Take a look at CircuitBreakerTest (in the usual location of src/test/java).

Importance of durations

It is a common (and important) knowledge to realize the importance of duration in the behaviour of an Actor-based application. In our demonstrative application, it assumes even greater significance, because the basis of the emulation of failure in it, is occurrence of a Time-Out. For the sake of simplicity (and, some laziness too, I admit), some of these durations are initialized using hard-coded values. As a result, other components, whose behaviour depends on the aforementioned durations, also use hard-coded values. This understanding will help, when you go through CircuitBreakerTest, CBDriver and RESTDriver.

Let me know if you find this article helpful, or if you think something specific is missing, whose presence would have helped in grasping the behaviour of Akka’s CircuitBreaker.