Domain Services and Bounded Context using Akka - Part 2

In Part 1 of this series you saw how we can model a domain repository as an actor in Akka. It gives you declarative transaction semantics through Akka's STM and pluggable persistence engine support over a variety of data stores. As a result the domain model becomes cleaner. The repository that you design can take advantage of Akka's fault tolerance capabilities through supervisors that offer configurable lifecycle strategies.

One other important artifact of a domain model are the domain services. A domain service is not necessarily focused on any particular entity and is mostly centered around the verbs of the system. It models some actions or use cases involving multiple entities and is usually implemented as a stateless abstraction.

Using Akka you model a service as yet another actor. Domain services are coarse level abstractions and are the ones to receive requests from the clients. It can invoke other services or use any other entitites to do the job that it's supposed to do. No wonder a busy service gets requests from lots of consumers. Not only does it need to be stable, but it needs to ensure that all of it's other services with which it collaborates also stay alive while serving requests.

One of the services that it interacts with is the Domain Repository, which I discussed in the last post.

When you design a domain service using Akka actors, you can ensure that the service can make its collaborating services fault-tolerant through declarative or minimal programming effort. Akka runtime offers all the machinery to make implementation of fault tolerant services quite easy.

Consider the following domain service for management of Accounts, continuing our earlier example from the last post ..

trait AccountServer extends Actor {
  // handle crash
  faultHandler = Some(OneForOneStrategy(5, 5000))
  trapExit = List(classOf[Exception])
  
  // abstract val : the Repository Service
  val storage: AccountRepository

  // message handler
  def receive = {
    case Open(no, name) => 
      storage ! New(Account(no, name, Calendar.getInstance.getTime, None, 100))
    case msg @ Balance(_) => storage forward msg
    case msg @ Post(_, _) => storage forward msg
    case msg @ OpenM(as) => storage forward msg
  }
  
  // shutdown hook
  override def shutdown = { 
    unlink(storage)
    storage.stop
  }
}


The message handler is a standard one that forwards client requests to the repository. Note tha use of the abstract val storage: AccountRepository that helps you defer committing to the concrete implementation class till instantiation of the service object.

AccountServer plays the role of a supervisor for the repository actor. The first 2 lines of code defines the strategy of supervision. OneForOneStrategy says that only the component that has crashed will be restarted. You can make it AllForOne also when the supervising actor will restart all of the actors that it's supervising if one of them crashes. trapExit defines the list of exceptions in the linked actor for which the supervising actor will take an action.

AccountServer is the aupervising actor for the repository. When it is shutdown it has to be unlinked fro the linked actors. This we do in the shutdown hook of the AccountServer.

But how do we link the Repository actor to our domain service actor ? Note that AccountServer is not a concrete object yet. We need to instantiate a concrete implementation of AccountRepository and assign it to storage in AccountServer. And link the two during this instantiation.

We define another trait that resolves the abstract val that we defined in AccountServer and provides a concrete instance of AccountRepository. spawnLink not only starts an instance of Redis based repository implementation, it also links the repository actor with the AccountServer ..

trait RedisAccountRepositoryFactory { this: Actor =>
  val storage: AccountRepository = spawnLink(classOf[RedisAccountRepository]) 
}


Now we have all the components needed to instantiate a fault tolerant domain service object.

object AccountService extends 
  AccountServer with 
  RedisAccountRepositoryFactory {

  // start the service
  override def start: Actor = {
    super.start
    RemoteNode.start("localhost", 9999)
    RemoteNode.register("account:service", this)
    this
  }
}


Have a look at the start method that starts the service on a remote node. We start a remote node and then register the current service under the id "account:service". Any client that needs to use the service can get hold of the service actor by specifying this id .. as in the following snippet ..

class AccountClient(val client: String) { 
  import Actor.Sender.Self
  val service = RemoteClient.actorFor("account:service", "localhost", 9999)

  def open(no: String, name: String) = service ! Open(no, name)
  def balance(no: String): Option[Int] = 
    (service !! Balance(no)).getOrElse(
      throw new Exception("cannot get balance from server"))
  def post(no: String, amount: Int) = service ! Post(no, amount)
  def openMulti(as: List[(String, String)]) = service !!! OpenMulti(as)
}


Services defined using Akka can be made to run on remote nodes without much of an engineering hack. Akka runtime offers APIs for doing that. However, Akka never tries to hide from you the paradigms of distribution. You need to be aware of your distribution requirements and process and supervising hierarchies. Akka facilitates you to define them at the application level, doing the heavy lifting within its underlying implementation. This principle is inspired from Erlang's philosophy and is in sharp contrast to the RPC way of defining APIs. Along with the benefits of message based computation that decouples the sender and the receiver, Akka also enables you to handle states using its built-in STM and pluggable storage engine.

When you model a complex domain, you need to deal with multiple contexts, which Eric calls Bounded Contexts. Within a specific context you have a cohesive model with a set of domain behavior and abstractions. The interpretations of the same abstractions may change when you move to a different context within the same application. As a domain modeler you need to define your context boundaries very carefully using context maps and implement appropriate translation maps between multiple contexts.

Messaging is a great way to implement translation between contexts. You process messages within a context using domain services as above and at the end forward the same message or a translated one to the other contexts of the application. It can be in the form of push or you can also implement a publish/subscribe model between the related contexts. In the latter case, you use Akka-Camel integration that allows actors to send or receive messages through Camel end-points. In either case Akka provides you a world of options to implement loosely coupled domain contexts that form the components of your domain model.

Thinking Asynchronous - Domain Modeling using Akka Transactors - Part 1

Followers of this blog must have known by now that I am a big fan of a clean domain model. And domain driven design, espoused by Eric Evans is the way to go when you are modeling a complex domain and would like to have your model survive for quite some time in the future. Recently I have been experimenting a bit with domain driven design using some amount of asynchronous message passing techniques particularly in the services and the storage layer.

The Repository, as Eric says, is the domain centric abstraction on top of your data storage layer. It gives your model back the feeling that you are dealing with domain concepts instead of marshalling data across your storage layers. Typically you have contracts for repositories at the aggregate root level. The underlying implementation commits to a platform (like JPA) and ensures that your object graph of the aggregate root rests in peace within the relational database. It need not be a relational database - it can be file system, it can be a NoSQL database. That's the power of abstraction that Repositories add to your model.

Ever since I started playing around with Erlang, I have been toying with thoughts of making repositories asynchronous. I blogged some of my thoughts in this post and even implemented a prototype using Scala actors.

Enter Akka and its lightweight actor model that offers transaction support over an STM. Akka offers seamless integration with a variety of persistence engines like Cassandra, MongoDB and Redis. It has plans of adding to this list many of the relational stores as well. The richness of the Akka stack makes for a strong case in designing a beautiful asynchronous repository abstraction.

Consider a very simple domain model for a Bank Account ..

case class Account(no: String, 
  name: String, 
  dateOfOpening: Date, 
  dateOfClose: Option[Date],
  balance: Float)

We can model typical operations on a bank account like Opening a New Account, Querying for the Balance of an Account, Posting an amount in an Account through message dispatch. Typical messages will look like the following in Scala ..

sealed trait AccountEvent
case class Open(from: String, no: String, name: String) extends AccountEvent
case class New(account: Account) extends AccountEvent
case class Balance(from: String, no: String) extends AccountEvent
case class Post(from: String, no: String, amount: Float) extends AccountEvent


Note all messages are immutable Scala objects, which will be dispatched by the client, intercepted by a domain service, which can optionally do some processing and validation, and then finally forwarded to the Repository.

In this post we will look at the final stage in the lifecycle of a message, which is how it gets processed by the Repository. In the next post we will integrate the whole along with an abstraction for a domain service. Along the way we will see many of the goodness that Akka transactors offer including support for fault tolerant processing in the event of system crashes.

trait AccountRepository extends Actor

class RedisAccountRepository extends AccountRepository {
  lifeCycle = Some(LifeCycle(Permanent))    
  val STORAGE_ID = "account.storage"

  // actual persistent store
  private var accounts = atomic { RedisStorage.getMap(STORAGE_ID) }

  def receive = {
    case New(a) => 
      atomic {
        accounts.+=(a.no.getBytes, toByteArray[Account](a))
      }

    case Balance(from, no) =>
      val b = atomic { accounts.get(no.getBytes) }
      b match {
        case None => reply(None)
        case Some(a) => 
          val acc = fromByteArray[Account](a).asInstanceOf[Account]
          reply(Some(acc.balance))
      }

      //.. other message handlers
  }

  override def postRestart(reason: Throwable) = {
    accounts = RedisStorage.getMap(STORAGE_ID)  
  }
}


The above snippet implements a message based Repository abstraction with an underlying implementation in Redis. Redis is an advanced key/value store that offers persistence for a suite of data structures like Lists, Sets, Hashes and more. Akka offers transparent persistence to a Redis storage through a common set of abstractions. In the above code you can change RedisStorage.getMap(STORAGE_ID) to CassandraStorage.getMap(..) and switch your underlying storage to Cassandra.

The above Repository works through asynchronous message passing modeled with Akka actors. Here are some of the salient points in the implementation ..

1. Akka is based on the let-it-crash philosophy. You can design supervisor hierarchies that will be responsible for controlling the lifecycles of your actors. In the Actor abstraction you can configure how you would like to handle a crash. LifeCycle(Permanent) means that the actor will always be restarted by the supervisor in the event of a crash. It can also be Lifecycle(Temporary), which means that it will not be restarted and will be shut down using the shutdown hook that you provide. In our case we make the Repository resilient to crashes.


2. accounts is the handle to a Map that gets persisted in Redis. Here we store all accounts that the clients open hashed by the account number. Have a look at the New message handler in the implementation.


3. With Akka you can also provide a restart hook when you repository crashes and gets restarted automatically by the supervisor. postRestart is the hook where we re-initialize the Map structure.


4. Akka uses multiverse, a Java based STM implementation for transaction handling. In the code mark your transactions using atomic {} and the underlying STM will take care of the rest. Instead of atomic, you can also use monadic for-comprehensions for annotating your transaction blocks. Have a look at Akka documentation for details.

Asynchronous message based implementations decouple the end points, do not block and offer more manageability in distribution of your system. Typical implementations of actor based models are very lightweight. Akka actors take around 600 bytes which means that you can have millions of actors even in a commodity machine. Akka supports various types of message passing semantics which you can use to organize interactions between your collaborating objects.

In the next post we will see how the Repository interacts with Domain Services to implement client request handling. And yeah, you got it right - we will use more of Akka actors.

Dependency Injection as Function Currying

Dependency Injection is one of the techniques that I use regularly when I am programming in Java. It's a nice way of making an application decoupled from concrete implementations and localize object creation logic within specific bootstrapping modules. Be it in the form of Spring XML or Guice Modules, the idea is to keep it configurable so that specific components of your application can choose to work with specific implementations of an abstraction.

It so happens that these days possibly I have started looking at things a bit differently. I have been programming more in Scala and Clojure and being exposed to many of the functional paradigms that they encourage and espouse, it has stated manifesting in the way I think of programming. In this post I will look into dependency injection on a different note. At the end of it may be we will see that this is yet another instance of a pattern melding into the chores of a powerful language's idiomatic use.

In one of my projects I have a class whose constructor has some of its parameters injected and the others manually provided by the application. Guice has a nice extension that does this for you - AssistedInject. It writes the boilerplate stuff by generating an implementation of the factory. You just need to annotate the implementation class' constructor and the fields that aren't known to the injector. Here's an example from the Guice page ..

public class RealPayment implements Payment {
  @Inject
  public RealPayment(
        CreditService creditService,  // injected
        AuthService authService,  // injected
        @Assisted Date startDate, // caller to provide
        @Assisted Money amount);  // aller to provide
  }
  ...
}


Then in the Guice module we bind a Provider<Factory> ..

bind(PaymentFactory.class).toProvider(
  FactoryProvider.newFactory(
    PaymentFactory.class, RealPayment.class));


The FactoryProvider maps the create() method's parameters to the corresponding @Assisted parameters in the implementation class' constructor. For the other constructor arguments, it asks the regular Injector to provide values.

So the basic issue that AssistedInject solves is to finalize (close) some of the parameters at the module level to be provided by the injector, while keeping the abstraction open for the rest to be provided by the caller.

On a functional note this sounds a lot like currying .. The best rationale for currying is to allow for partial application of functions, which does the same thing as above in offering a flexible means of keeping parts of your abstraction open for later pluggability.

Consider the above abstraction modeled as a case class in Scala ..

trait CreditService
trait AuthService

case class RealPayment(creditService: CreditService,
                       authService: AuthService,
                       startDate: Date,
                       amount: Int)


One of the features of a Scala case class is that it generates a companion object automatically along with an apply method that enables you to invoke the class constructor as a function object ..

val rp = RealPayment( //..


is in fact a syntactic sugar for RealPayment.apply( //.. that gets called implicitly. But you know all that .. right ?

Now for a particular module , say I would like to finalize on PayPal as the CreditService implementation, so that the users don't have to pass this parameter repeatedly - just like the injector of your favorite dependency injection provider. I can do this as follows in a functional way and pass on a partially applied function to all users of the module ..

scala> case class PayPal(provider: String) extends CreditService
defined class PayPal

scala> val paypalPayment = RealPayment(PayPal("bar"), _: AuthService, _: Date, _: Int)
paypalPayment: (AuthService, java.util.Date, Int) => RealPayment = <function>



Note how the Scala interpreter now treats paypalPayment as a function from (AuthService, java.util.Date, Int) => RealPayment. The underscore acts as the placeholder that helps Scala create a new function object with only those parameters. In our case the new functional takes only three parameters for whom we used the placeholder syntax. From your application point of view what it means is that we have closed the abstraction partially by finalizing the provider for the CreditService implementation and left the rest of it open. Isn't this precisely what the Guice injector was doing above injecting some of the objects at module startup ?

Within the module I can now invoke paypalPayment with only the 3 parameters that are still open ..

scala> case class DefaultAuth(provider: String) extends AuthService
defined class DefaultAuth

scala> paypalPayment(DefaultAuth("foo"), java.util.Calendar.getInstance.getTime, 10000)
res0: RealPayment = RealPayment(PayPal(foo),DefaultAuth(foo),Sun Feb 28 15:22:01 IST 2010,10000)



Now suppose for some modules I would like to close the abstraction for the AuthService as well in addition to freezing PayPal as the CreditService. One alternative will be to define another abstraction as paypalPayment through partial application of RealPayment where we close both the parameters. A better option will be to reuse the paypalPayment abstraction and use explicit function currying. Like ..

scala> val paypalPaymentCurried = Function.curried(paypalPayment)
paypalPaymentCurried: (AuthService) => (java.util.Date) => (Int) => RealPayment = <function>



and closing it partially using the DefaultAuth implementation ..

scala> val paypalPaymentWithDefaultAuth = paypalPaymentCurried(DefaultAuth("foo"))
paypalPaymentWithDefaultAuth: (java.util.Date) => (Int) => RealPayment = <function>



The rest of the module can now treat this as an abstraction that uses PayPal for CreditService and DefaultAuth for AuthService. Like Guice we can have hierarchies of modules that injects these settings and publishes a more specialized abstraction to downstream clients.

DSL : Grow your syntax on top of a clean semantic model

A DSL primarily has two components - a semantic model that abstracts the underlying domain and a linguistic abstraction on top that speaks the dialect of the user. The semantic model is the model of the domain where you can apply all the principles of DDD that Eric Evans espouses. And the linguistic abstraction is a thin veneer on top of the underlying model. The more well abstracted your model is, easier will be the construction of the layer on top of it. Here's a general architecture of a DSL engineering stack :-

It's interesting to observe that the two components of the stack evolve somewhat orthogonally.

The Semantic Model evolves Bottom Up

The semantic model usually evolves in a bottom up fashion - larger abstractions are formed from smaller abstractions using principles of composition. It can be through composition of traits or objects or it can be through composition of functions as well. How beautiful your compositions can be depends a lot on the language you use. But it's important that the semantic model also speaks the language of the domain.

Here's an example code snippet from my upcoming book DSLs In Action that models the business rules for a trading DSL. When you do a trade on a stock exchange you get charged a list of tax and fee components depending on the market where you execute the trade. The following snippet models a business rule using Scala that finds out the list of applicable tax/fee heads for a trade ..

class TaxFeeRulesImpl extends TaxFeeRules {
  override def forTrade(trade: Trade): List[TaxFee] = {
    (forHKG orElse 
       forSGP orElse 
         forAll)(trade.market)
  }
   
  val forHKG: PartialFunction[Market, List[TaxFee]] = { 
    case HKG => 
      // in real life these can come from a database
      List(TradeTax, Commission, Surcharge)
  }
    
  val forSGP: PartialFunction[Market, List[TaxFee]] = {
    case SGP => 
      List(TradeTax, Commission, Surcharge, VAT)
  }
    
  val forAll: PartialFunction[Market, List[TaxFee]] = {
    case _ => List(TradeTax, Commission)
  }

  //..
}


The method forTrade clearly expresses the business rule, which reads almost as expressive as the English version ..

"Get the Hong Kong specific list for trades executed on the Hong Kong market OR Get the Singapore specific list for trades executed on the Singapore market OR Get the most generic list valid for all other markets"

Note how Scala PartialFunction s can be chained together to give the above model an expressive yet succinct syntax.

The Language Interface evolves Top Down

Here you start with the domain user. What dialect does he use on the trading desk ? And then you try to build an interpreter around that which uses the services that the semantic model publishes. I call this thin layer of abstraction a DSL Facade that sits between your DSL script and the underlying domain model and acts as the glue.

It also depends a lot on the host language as to how you would like to implement the facade. With a language like Lisp, macros can come in very handy in designing an interpreter layer for the facade. And with macros you do bottom up programming, bending the host language to speak your dialect.

When you are developing an external DSL, the EBNF rules that you specify act as the DSL Facade for growing your syntax. Within the rules you can use foreign code embedding to interact with your semantic model.

In summary, when you design a DSL, the semantic model is as important as the dialect that it speaks. Having a well designed semantic model is an exercise in designing well-engineered abstractions. And as I mention in my book, the four qualities of good abstractions are minimalism, distillation, extensibility and composability.

Why I don't like ActiveRecord for Domain Model Persistence

When it comes to a rich domain modeling, I am not a big fan of the ActiveRecord model. The biggest problem that it entails is invasiveness - the persistence model invades into my domain model. And this was also my first reaction to the Lift-CouchDB integration module which was released recently.

The moment I say ..

class Person extends CouchRecord[Person] {
  //..
}


my domain model becomes tied to the persistence concerns.

When I started scouchdb, my very first thought was to make it non-invasive. The Scala objects must be pure and must remain pure and completely oblivious of the underlying persistence model. In the age of polyglot persistence there is every possibility that you may need to persist a domain model across multiple storage engines. I may be using a JPA backed relational store as the enterprise online database, which gets synchronized with some offline processing that comes from a CouchDB backend. I need the domain model persistence in both my Oracle and my CouchDB engines. The ActiveRecord pattern is difficult to scale to such requirements.

Here's what I implemented in scouchdb ..

// Scala abstraction : pure
case class ItemPrice(store: String, item: String, price: Number)

// specification of the db server running
val couch = Couch("127.0.0.1")
val item_db = Db("item_db")

// create the database
couch(item_db create)

// create the Scala object : a pure domain object
val s = ItemPrice("Best Buy", "mac book pro", 3000)

// create a document for the database with an id
val doc = Doc(item_db, "best_buy")

// add
couch(doc add s)

// query by id to get the id and revision of the document
val id_rev = couch(item_db by_id "best_buy")

// query by id to get back the object
// returns a tuple3 of (id, rev, object)
val sh = couch(item_db by_id("best_buy", classOf[ItemPrice]))

// got back the original object
sh._3.item should equal(s.item)
sh._3.price should equal(s.price)


It's a full cycle session of interaction with the CouchDB persistence engine without any intrusion into the domain abstraction. I am free to use ItemPrice domain abstraction for a relational storage as well.

CouchDB offers a model of persistence where the objects that we store should be close to the granularity of domain abstractions. I should be able to store the entire Aggregate Root of my model components directly as JSON. ActiveRecord model offers a lower level of abstraction and makes you think more in terms of persistence of the individual entities. The thought process is so relational that you ultimately end up with a relational model both in terms of persistence and domain. With CouchDB you need to think in terms of documents and views and NOT in terms of relations and tables. I blogged on this same subject some time back.

The philosophy that I adopted in scouchdb was to decouple the domain entities from the persistence layer. You hand over a pure Scala object to the driver, it will extract a JSON model from it and write it to CouchDB. I use sjson for this serialization. sjson works totally based on reflection and can transparently serialize and deserialize Scala objects that you hand over to it. From this point of view, the three aspects of managing domain abstractions, JSON serialization and persistence into CouchDB are totally orthogonal. I think this is difficult to get with an ActiveRecord based model.

Scala Self-Type Annotations for Constrained Orthogonality

I talked about orthogonality in design in one of my earlier posts. We had a class Address in Scala and we saw how we can combine it with other orthogonal concerns without polluting the core abstraction. We could do this because Scala offers a host of capabilities to compose smaller abstractions and build larger wholes out of them. A language is orthogonal when it allows such capabilities of composition without overlaps in functionalities between the composing featuresets.

The design of Scala offers many orthogonal features - I showed some of them in my earlier post. The power of mixins for adding orthogonal features ..

val a = new Address(..) with LabelMaker {
  override def toLabel = {
    //..
  }
}

and the power of Scala views with implicits ..

object Address {
  implicit def AddressToLabelMaker(addr: Address) = new LabelMaker {
    def toLabel =
    //..
  }
}

Here Address and LabelMaker are completely unrelated and offers truly orthogonal capabilities when mixed in. However there can be some cases where the mixins themselves are not completely orthogonal to the core abstractions, but really optional extensions to them. In fact the mixins implement some functionalities that may depend on the core abstraction as well. Let's see yet another feature of the Scala type system that makes this modeling wholesome.

Consider the following abstraction for a security trade that takes place in a stock exchange ..

// details ellided for clarity
case class Trade(refNo: String, account: String, 
                 instrument: String, quantity: Int,
                 unitPrice: Int) {
  // principal value of the trade
  def principal = quantity * unitPrice
}


For every trade executed on the exchange we need to have a set of tax and fees associated with it. The exact set of tax and fees depend on a number of factors like type of trade, instruments traded, the exchange where it takes place etc. Let's have a couple of tax/fee traits that model this behavior ..

trait Tax { 
  def calculateTax = //..
}
  
trait Commission { 
  def calculateCommission = //..
}


In the above definitions both methods calculateTax and calculateCommission depends on the trade being executed. One option is to keep them abstract in the above trait and provide their implementations after mixing in with Trade ..

val t = new Trade(..) with Tax with Commission {
  // implementations
  def calculateTax = principal * 0.2
  def calculateCommission = principal * 0.15
}

I did it at the instance level. You can very well use this idiom at the class level and define ..

class RichTrade extends Trade with Tax with Commission {
  //..
}


However the above composition does not clearly bring out the fact that the domain rules mandate that the abstractions Tax and Commission should be constrained to be used with the Trade abstraction only.

Scala offers one way of making this knowledge explicit at the type level .. using self-type annotations ..

trait Tax { this: Trade =>
  // refers to principal of trade
  def calculateTax = principal * 0.2
}
  
trait Commission { this: Trade =>
  // refers to principal of trade
  def calculateCommission = principal * 0.15
}


The traits are still decoupled. But using Scala's self type annotations you make it explicit that Tax and Commission are meant to be used *only* by mixing them with Trade.

val t = new Trade(..) with Tax with Commission
t.calculateTax
t.calculateCommission


Can I call this constraining the orthogonality of abstractions ? Tax and Commission provide orthogonal attributes to Trade optionally and publish their constraints explicitly in their definitions. It's not much of a difference from the earlier implementations. But I prefer to use this style to make abstractions closer to what the domain speaks.