Implementing Content-Based Router Pattern in Akka

Question

I am trying to implement a content-based router in my Akka actor system and according to this document the ConsistentHashingRouter is the way to go. After reading through its official docs, I still find myself confused as to how to use this built-in hashing router. I think that’s because the router itself is hash/key-based, and the example the Akka doc author chose to use was a scenario involving key-value based caches…so I can’t tell which keys are used by the cache and which ones are used by the router!

Let’s take a simple example. Say we have the following messages:

interface Notification {
    // Doesn’t matter what’s here.
}

// Will eventually be emailed to someone.
class EmailNotification implements Notification {
    // Doesn’t matter what’s here.
}

// Will eventually be sent to some XMPP client and on to a chatroom somewhere.
class ChatOpsNotifications implements Notification {
    // Doesn’t matter what’s here.
}

etc. In theory we might have 20 Notification impls. I’d like to be able to send a Notification to an actor/router at runtime and have that router route it to the correct NotificationPubisher:

interface NotificationPublisher<NOTIFICATION implements Notification> {
    void send(NOTIFICATION notification)
}

class EmailNotificationPublisher extends UntypedActor implements NotificationPubisher<EmailNotification> {
    @Override
    void onReceive(Object message) {
        if(message instanceof EmailNotification) {
            send(message as EmailNotification)
        }
    }

    @Override
    void send(EmailNotification notification) {
        // Use Java Mail, etc.
    }
}

class ChatOpsNotificationPublisher extends UntypedActor implements NotificationPubisher<ChatOpsNotification> {
    @Override
    void onReceive(Object message) {
        if(message instanceof ChatOpsNotification) {
            send(message as ChatOpsNotification)
        }
    }

    @Override
    void send(ChatOpsNotification notification) {
        // Use XMPP/Jabber client, etc.
    }
}

Now I could do this routing myself, manually:

class ReinventingTheWheelRouter extends UntypedActor {
    // Inject these via constructor
    ActorRef emailNotificationPublisher
    ActorRef chatOpsNotificationPublisher
    // ...20 more publishers, etc.

    @Override
    void onReceive(Object message) {
        ActorRef publisher
        if(message instanceof EmailNotification) {
            publisher = emailNotificationPublisher
        } else if(message instanceof ChatOpsNotification) {
            publisher = chatOpsNotificationPublisher
        } else if(...) { ... } // 20 more publishers, etc.

        publisher.tell(message, self)
    }
}

Or I could use the Akka-Camel module to defined a Camel-based router and send Notifications off to the Camel router, but it seems that Akka aready has this built-in solution, so why not use it? I just cant figure out how to translate the Cache example from those Akka docs to my Notification example here. What’s the purpose of the “key” in the ConsistentHashingRouter? What would the code look like to make this work?

Of course I would appreciate any answer that helps me solve this, but would greatly prefer Java-based code snippets if at all possible. Scala looks like hieroglyphics to me.

---

I agree that a Custom Router is more appropriate than ConsistentHashingRouter. After reading the docs on custom routers, it seems I would:

1. Create a GroupBase impl and send messages to it directly (notificationGroup.tell(notification, self)); then
2. The GroupBase impl, say, NotificationGroup would provide a Router instance that was injected with my custom RoutingLogic impl
3. When NotificationGroup receives a message, it executes my custom RoutingLogic#select method which determines which Routee (I presume some kind of an actor?) to send the message to

If this is correct (and please correct me if I’m wrong), then the routing selection magic happens here:

class MessageBasedRoutingLogic implements RoutingLogic {
    @Override
    Routee select(Object message, IndexedSeq<Routee> candidates) {
        // How can I query the Routee interface and deterine whether the message at-hand is in fact
        // appropriate to be routed to the candidate?
        //
        // For instance I'd like to say "If message is an instance of
        // an EmailNotification, send it to EmailNotificationPublisher."
        //
        // How do I do this here?!?
        if(message instanceof EmailNotification) {
            // Need to find the candidate/Routee that is
            // the EmailNotificationPublisher, but how?!?
        }
    }
}

But as you can see I have a few mental implementation hurdles to cross. The Routee interface doesn’t really give me anything I can intelligently use to decide whether a particular Routee (candidate) is correct for the message at hand.

So I ask: (1) How can I map messages to Routees (effectively performing the route selection/logic)? (2) How do I add my publishers as routees in the first place? And (3) Do my NotificationPublisher impls still need to extend UntypedActor or should they now implement Routee?

Answer 1

Here is a simple little A/B router in Scala. I hope this helps even though you wanted a Java based answer. First the routing logic:

class ABRoutingLogic(a:ActorRef, b:ActorRef) extends RoutingLogic{
  val aRoutee = ActorRefRoutee(a)
  val bRoutee = ActorRefRoutee(b)

  def select(msg:Any, routees:immutable.IndexedSeq[Routee]):Routee = {
    msg match{
      case "A" => aRoutee
      case _ => bRoutee
    }
  }
}

The key here is that I am passing in my a and b actor refs in the constructor and then those are the ones I am routing to in the select method. Then, a Group for this logic:

case class ABRoutingGroup(a:ActorRef, b:ActorRef) extends Group { 
  val paths = List(a.path.toString, b.path.toString)

  override def createRouter(system: ActorSystem): Router =
    new Router(new ABRoutingLogic(a, b))

  val routerDispatcher: String = Dispatchers.DefaultDispatcherId
}

Same thing here, I am making the actors I want to route to available via the constructor. Now a simple actor class to act as a and b:

class PrintingActor(letter:String) extends Actor{
  def receive = {
    case msg => println(s"I am $letter and I received letter $msg") 
  }
}

I will create two instances of this, each with a different letter assignment so we can verify that the right ones are getting the right messages per the routing logic. Lastly, some test code:

object RoutingTest extends App{
  val system = ActorSystem()
  val a = system.actorOf(Props(classOf[PrintingActor], "A"))
  val b = system.actorOf(Props(classOf[PrintingActor], "B"))
  val router = system.actorOf(Props.empty.withRouter(ABRoutingGroup(a,b)))

  router ! "A"
  router ! "B"

}

If you ran this, you would see:

I am A and I received letter A
I am B and I received letter B

It's a very simple example, but one that shows one way to do what you want to do. I hope you can bridge this code into Java and use it to solve your problem.