Using Akka Dispatchers In Scala

Akka dispatchers are extremely important in Akka framework. They are directly responsible for optimal performance, throughput and scalability.

Akka supports dispatchers for both event-driven lightweight threads and thread-based Actors. For thread-based Actors each dispatcher is bound to a dedicated OS thread.

Default dispatcher is a single event-based dispatcher for all Actors created. The dispatcher used is this one:

Dispatchers.globalExecutorBasedEventDrivenDispatcher

For many cases it becomes mandatory to group Actors together for a dedicated dispatcher, then we can override the defaults and define our own dispatcher.

Setting the Dispatcher
Normally we set the dispatcher in Actor itself

class EchoActor extends Actor {
     self.dispatcher = ..... // set the dispatcher
}

Or we can set it in the ActorRef

actorRef.dispatcher = dispatcher

There are different kind of dispatchers

• Thread-based
• Event-based
• Priority event-based
• Work-stealing

Thread-based
It binds dedicated OS thread to each Actor. The messages are posted to LinkedBlockingQueue which feeds messages to dispatcher one by one. It has worst performance and scalability. We also cannot share it among actors. Although Actors do not block for threads in this case.
Code example

class EchoActor extends Actor {
     self.dispatcher = Dispatchers.newThreadBasedDispatcher(self)
     ....
}

Event-based
The ExecutorBasedEventDrivenDispatcher binds a set of Actors to a thread pool backed up by a BlockingQueue. The dispatcher must be shared among Actors. This dispatcher is highly configurable and here we can specify things like ‘type of queue’, ‘max items’ , ‘rejection-policy’.
Code example

class EchoActor extends Actor {
self.dispatcher = Dispatchers.newExecutorBasedEventDrivenDispatcher(name)
 .withNewThreadPoolWithLinkedBlockingQueueWithCapacity(100)
 .setCorePoolSize(16)
 .setMaxPoolSize(128)
 .setKeepAliveTimeInMillis(60000)
 .setRejectionPolicy(new CallerRunsPolicy)
 .build
 ....
}

Priority event-based
It is meant for handling messages when priorities are assigned to messages. It is done by using PriorityExecutorBasedEventDrivenDispatcher. It requires a PriorityGenerator as an attribute in its constructor.

Let’s look at an example where we have a PriorityExecutorBasedEventDrivenDispatcher used for a group of messages fired on an actor.

package com.meetu.akka.dispatcher

import akka.actor.Actor.actorOf
import akka.actor.Actor
import akka.dispatch.PriorityExecutorBasedEventDrivenDispatcher
import akka.dispatch.PriorityGenerator

object PriorityDispatcherExample extends App {
  val actor = Actor.actorOf(
    new Actor {
      def receive = {
        case x => println(x)
      }
    })

  val priority = PriorityGenerator {
    case "high priority" => 0
    case "low priority" => 100
    case _ => 50
  }

  actor.dispatcher = new PriorityExecutorBasedEventDrivenDispatcher("foo", priority)

  actor.start
  actor.dispatcher.suspend(actor)

  actor ! "low priority"
  actor ! "others"
  actor ! "low priority"
  actor ! "high priority"

  actor.dispatcher.resume(actor)
  Actor.registry.shutdownAll
}

Manage Akka Actors With Supervisors

We are building a concurrent and massively scalable system using Akka framework. Akka uses light-weight Actor paradigm of managing concurrency.

Most of the times a bunch of Akka actors are required for a common functionality; therefore we do tend to manage them as a group.

This is where Akka Supervisor comes into the picture. Akka Supervisors are responsible for starting, stopping and monitoring child Akka actors.

Lets look now how to use supervisor over Akka Actors. Just to keep things simple we have an Actor which simply echoes the message it receives. Here is the code listing for EchoActor:

/**
 * An Actor that echoes everything you send to it
 */
class EchoActor extends Actor {
  def receive = {
    case msg => {
      println(msg)
    }
  }
}

We will build supervision on a bunch of EchoActor, lets say our supervisor is EchoActorSupervisor. But before we actually look at its code listing we understand its constituent elements first.

The Supervisor first require SupervisorConfig. This SupervisorConfig in turn requires fault handling strategy for the group of actors. They are:

• AllForOneStrategy
• OneForOneStrategy

The AllForOne fault handler will restart all the components and OneForOneStrategy fault handler will restart just the component which died. Fault Handler takes as parameter a List of Exceptions which will be handled, maximum number of restart tries and within-time in millis

Lets have a look how does our SupervisorConfig looks like:

...
val supervisorConfig = SupervisorConfig(OneForOneStrategy(List(classOf[Exception]), 3, 500), Nil)
...

Supervisor can now be created by passing in SupervisorConfig created in previous listing:

...
val supervisorConfig = SupervisorConfig(OneForOneStrategy(List(classOf[Exception]), 3, 500), Nil)
val supervisor = Supervisor(supervisorConfig)
...

now we can link and unlink actors to the supervisor. In our case since we are using EchoActor we can link two Actor Reference to the supervisor. In the code below we are linking and starting two EchoActor.

...
val echoActor1 = actorOf[EchoActor]
val echoActor2 = actorOf[EchoActor]

val supervisorConfig = SupervisorConfig(OneForOneStrategy(List(classOf[Exception]), 3, 500), Nil)
val supervisor = Supervisor(supervisorConfig)

supervisor.link(echoActor1)
supervisor.link(echoActor2)

echoActor1.start
echoActor2.start
...

Starting Akka Project With SBT 0.10

I was starting with Akka project with SBT but found that the latest SBT is quite different from before.

I tried to create AKKA project with latest SBT but got stuck. Old SBT used to ask to create a new project in case it did not find any in the directory. With new SBT it is not the case. If you want to know how to go about creating new Akka project with SBT read on.

After installing SBT if we type in sbt in the command prompt in an empty directory, this is what we are likely to see.

In order to create project execute the following commands in the sbt session.

> set name := "AkkaQuickStart"
> set version := "1.0"
> set scalaVersion := "2.9.0-1"
> session save
> exit

We should get the following output if we type the above mentioned commands in sbt session.

SBT creates files and directories when we executed the commands. It creates build.sbt and it contains the same values we typed in sbt session. Other directories like target and project are of little consequence to us.

Project directory will become important later when we will try to add sbteclipse plugin. My project directory contains the following subdirectories and files.

Getting Started With Scala

I and Vikas presented on “Getting Started With Scala” few years back. It was presented at OSS camp 2009 in New Delhi. I still remember the interest it created at the conference.

Today I happened to have a look at it on Slideshare. Looking at the number of views it seems to be doing good. Out of my previous employer presentations its having highest hits. Scala is really doing good in terms of popularity.

Presentation is still old. To get started concepts are still the same. Enjoy the presentation…

Experiences at Cloud Computing Conference Pune 2011

I was one of the speaker of the second IndicThreads conference held at Pune on 3-4th June 2011.

Sessions at the conference dealt with key topics like Cloud Security, Amazon Elastic Beanstalk, Legal Issues in Cloud Computing, OpenStack, Xen Cloud Platform, Rails and CouchDB on the cloud, CloudFoundry, Gigapaces PAAS, Monitoring Cloud Applications, ORM with Objectify-Appengine, Scalable Architecture on Amazon AWS Cloud, Cloud Lock-in, Cloud Interoperability, Apache Hadoop, Map Reduce and Predictive Analysis.

My talk focussed on managing persistence on GAE. It dealt with choices available to a developer and then focussed on doing it with Objectify-Appengine.

Demo application is present here for download. For the instructions on how to run it please read the wiki.

Working With Scala Collections

We have a monthly iBAT (Inphina Beer and Technology Sessions). We look forward to this day and it was Scala day this time. I presented on Scala collections.

Scala collection is elegant and concise. Scala Collections like java is object-oriented and we can work with generic types. It is optionally persistent i.e can be mutable and immutable. It provides higher order methods like foreach, map and filter.

Scala collections follow uniform return type principle. Which basically means that when you perform an operation on a collection they return a collection of the same type.

The root Trait in Scala collections is Traversable. It may take some time to get used to it as Scala collections root Traits like Traversable have a big bunch of methods.

The for notation is more readable. It is basically like a for loop with a return type. Look for yourself how readable it is compared to the original code.

This code listing is without a for notation

scala> val listOfNumbers = List(1, 2, 3)
listOfNumbers: List[Int] = List(1, 2, 3)

scala> val flattenedListOfNumbers = listOfNumbers flatMap (0 to _)
flattenedListOfNumbers: List[Int] = List(0, 1, 0, 1, 2, 0, 1, 2, 3)

After using for notation

scala> val listOfNumbers = List(1, 2, 3)
listOfNumbers: List[Int] = List(1, 2, 3)

scala> val flattenedList = for(number <- listOfNumbers; range <- 0 to number) yield range
flattenedList: List[Int] = List(0, 1, 0, 1, 2, 0, 1, 2, 3)

One more interesting thing we can do with a Map is that we can reverse key and value with the following code

scala> val aMap = Map("India" -> "Delhi", "France" -> "Paris", "Italy" -> "Rome")
aMap: scala.collection.immutable.Map[java.lang.String,java.lang.String] = Map(India -> Delhi, France -> Paris, Italy -> Rome)

scala> val reverseMap = aMap map{case(k, v) => (v, k)}
reverseMap: scala.collection.immutable.Map[java.lang.String,java.lang.String] = Map(Delhi -> India, Paris -> France, Rome -> Italy)

scala> val alsoReverseMap = for((k, v) <- aMap) yield (v -> k) 
alsoReverseMap: scala.collection.immutable.Map[java.lang.String,java.lang.String] = Map(Delhi -> India, Paris -> France, Rome -> Italy)

Reference were from Martin’s Odersky talk on Future Proofing Collections and Scala Collections API documentation. Enjoy the presentation …

Working With Scala Test

Scala Test is an open source Test framework for the java platform. With Scala test we can either test java or Scala code. It integrates with popular existing tools like JUnit, TestNG and it is designed to do different styles of testing like Behavior Driven Design for example.

Let’s have a look at a simple JUnit4 Test in Scala Test

package com.inphina.ibat.junit

import org.scalatest.junit.AssertionsForJUnit
import org.junit.Assert._
import org.junit.Test
import  org.junit.Before

class SimpleJunit4Demo extends AssertionsForJUnit {

  var sb: StringBuilder = _

  @Before def initialize() {
    sb = new StringBuilder("Welcome To ")
  }

  @Test
  def verifyEasy() {
    sb.append("ScalaTest!")
    assertEquals("Welcome To ScalaTest!", sb.toString)
  }

}

This Test implementation is not very different from a normal JUnit Test we are accustomed to. In order to understand ScalaTest there are three concepts we need to be aware of.

• Suite: A collection of tests. A Test is anything that has a name and can succeed or fail
• Runner: ScalaTest provides a Runner application that can run Suites of Tests
• Reporter: As Tests are run events are fired to a reporter, it takes care of presenting results back to the user

Let’s have a bit more detailed look at the Suite which is a Trait.

Working With Spring Data JPA

SpringSource have recently released the first milestone of Spring Data JPA project. What are its features? and how can we use them. Spring framework already provides support to build a JPA based data access layer. So, what does Spring Data adds to it. I used it for one small application and found significant reduction of code. In this post we will use an Employee entity and EmployeeRepository as an example to work with Spring Data JPA.

We have a domain Employee :

@Entity
public class Employee {

  @Id
  @GeneratedValue(strategy = GenerationType.AUTO)
  private Long id;

  private String name;

  // … methods omitted
}

We also have a EmployeeRepository handling the Employee entities. The interface and the implementation are given below.

package com.inphina.springdata.repository;

public interface EmployeeRepository {
	
	Employee save(Employee employee);
	
	List<Employee> findAll();
	
	Employee findById(Long id);
	
	List<Employee> findByName(String name, int page, int pageSize);

}

Configuring Spring Application Using Configuration Annotation

A Spring Module called Java Config has been there for some time now. With the release of Spring 3.0 SpringSource has pulled this module into the core Spring Framework as @Configuration.

How can we use this @Configuration in our application? and should we use it when we have XML and annotations for configuration already?

In Spring 2.5 we find that there is a support for Property Externalization. Which was good in several ways than before. What it meant was, that the properties file could be modified for different build environments by the administrator.

One good thing about this approach is that the administrator need not edit the application context file. Editing verbose XML has plenty of room for errors. Editing the properties file on the other hand is a whole lot simpler and is less error prone.

For example for configuring the datasource this is what we do typically. We used Property externalization in Spring 2.5 using PropertyPlaceHolderConfigurer.

<context:property-placeholder location="classpath:datasource.properties" />
<bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
		<property name="driverClassName"><value>${dataSource.driverClassName}</value></property>
		<property name="url"><value>${dataSource.url}</value></property>
		<property name="username"><value>${dataSource.username}</value></property>
		<property name="password"><value>${dataSource.password}</value></property>
</bean>

Mailing Google Docs As Attachment On Google App Engine

Our application works with Google Spreadsheet API to update spreadsheets. The application now needed to send mail with an attachment of a google spreadsheet. We posted our query on google docs community here and on Google app engine here and there were no clear directions of implementing it.

Let’s see how we went about implementing this feature.

Google provides its spreadsheet API and it allows you to create and retrieve data in Google Spreadsheets, but it does not allow us to either create or manage their permissions. Google also has Google Documents List API which can be used to create and retrieve a list of Google Spreadsheets. For our case Google Documents List API provides exporting functionality such that we can export Google documents in common formats like pdf, rtf, xls and others. Google also provides java mail api and with its Multi-Part Messages feature we can send attachment as well.

For implementing this, we used spreadsheet API to retrieve spreadsheet entry. Using the spreadsheet entry we used Google Documents List API to export it in the specified format and we used java mail api to send exported data as an attachment.

We used play framework for our implementation but it can also be applied on other java frameworks as well. The class EmailSender first creates a google spreadsheet service and document service.

public class EmailSender extends Controller {

private static final String META_FEED_URL = "https://spreadsheets.google.com/feeds/spreadsheets/private/full";
private static String spreadSheetName = "spreadsheetname";
private static String yourApplicationName = "emailSender-version1";
private static String userEmail = "yourmail@gmail.com";
private static String recepientEmail = "recepientmail@gmail.com";
private static String password = "password";


public static void emailSpreadsheetAsAttachment() throws AuthenticationException, MessagingException, IOException {
	SpreadsheetService spreadSheetService = createSpreadSheetService();
	DocsService documentService = createDocumentService();
	SpreadsheetEntry spreadsheetEntry = loadSpreadSheet(spreadSheetService, spreadSheetName);
	byte[] spreadSheetData = getSpreadSheetData(spreadSheetService, documentService, spreadsheetEntry);
	sendMailWithAttachment(spreadSheetData);
	}
. . .
. . .
private static SpreadsheetService createSpreadSheetService() throws AuthenticationException {
	SpreadsheetService spreadSheetService = new SpreadsheetService(yourApplicationName);
	spreadSheetService.setUserCredentials(userEmail, password);
	return spreadSheetService;
	}

private static DocsService createDocumentService() throws AuthenticationException {
	DocsService docsService = new DocsService(yourApplicationName);
	docsService.setUserCredentials(userEmail, password);
	return docsService;
       }
. . .
. . .
}