Ray Roestenburg | Code

Testing oneway actors with the stackable trait pattern

noreply@blogger.com (Raymond Roestenburg) — Thu, 01 Sep 2011 18:25:00 +0000

Just a quick post on something quite simple our team came up with recently for testing a particular type of Actor.
It comes in handy when you want to test an Actor that receives one-way messages and does not send out any messages, or reply to messages. Maybe after processing you want to assert some state somewhere.

Actors that respond to request/response are very easy to test. You don't have to use any Barriers or Latches or even the (very handy) TestKit, you just ask the Actor to send a reply using !!, and assert on the reply, fail on timeout.

Would it not be nice to test one-way Actors in the same way? What we would like is for the Actor to just send the exact same message back that was sent to it after it is done processing the message. Of course only in a test scenario. That way you would be able to test one-way messages in the same way as request/response messages.

The idea was to dynamically add some code to wrap around the Actor that you would like to test, in some way delegate to the Actor, and always do a reply to the sender with the original message, after the Actor is done with the processing of that message. That way the test becomes very easy, you just send a 'asking' !! to the Actor, the Actor does it's normal thing as if it received a 'telling' ! after which the extra bit of code kicks in and sends the original message back to the unit test, so you know its done its job.

Akka has a nice way of handling the difference between one-way and request/response. You can use the reply_? message from within an Actor, which replies back to the sender if the sender asked for a response, or does nothing in the event the sender 'told' the Actor something in one-way style.

The easiest way we found to dynamically add a sort of template method was to use a stackable trait (which is described on the artima site here). The below ReplyAfterProcessing trait overrides the receive of the Actor, first applies the Actor's receive (which is a PartialFunction) and chains it with an anonymous partial function that always replies the message.

That way you can test a one-way Actor like this. The "with ReplyAfterProcessing" mixes in the trait and replaces the standard receive processing "dynamically".

Happy hAkking!

Unit testing Akka Actors with the TestKit

noreply@blogger.com (Raymond Roestenburg) — Fri, 11 Feb 2011 23:13:00 +0000

Just a quick post to show some examples of how you can use the akka.util.TestKit to unit test Actors. I'm not sure when it was added to Akka but it is really handy. Instead of using CyclicBarriers and CountDownLatches, you just mixin this very handy trait into your spec, akka.util.TestKit.

I added some examples of test scenarios that pop up frequently when working with Actors:

You send something to an Actor and you would like to know if it received the message
You send something to an Actor and that in turn does some processing and on success sends something to another Actor and you would like to know if that Actor received a specific message
You send some messages and you are not interested in every message the Actor receives and would like to ignore them until some message is received
You would like to know for sure that an Actor does not send through certain messages to others
You would like to know what an Actor sends back as a response based on a certain message

Testing this by hand with the correct concurrent latches and barriers can be brittle and quite error prone, not to mention quite complex.

So how does it work? The TestKit sort of swaps out the senderOption of an Actor through an implicit value scoping trick, which means that when your Actor sends back a message (to "the sender", for instance with a reply), it automatically sends it back to a testActor in TestKit, which queues up messages for you so you can inspect stuff. You can also use that same testActor when you have an Actor that sends messages to some ActorRef. You just switch out the ActorRef you would normally pass to your Actor with the testActor Ref.

I added some very simple Actors for the above scenarios, which touch a lot of typical interactions:

EchoActor - an Actor that just echoes any message you send to it
ForwardingActor - an Actor that forwards a message to another Actor(Ref)
FilteringActor - an Actor that forwards only certain messages (of type String)
SequencingActor - an Actor that forwards some random amount of uninteresting messages, an interesting message, and then some random amount of uninteresting messages

As you can see in the code below, you use a within block to indicate that some piece of code must complete with a certain duration. Every within block gets handled sequentially to keep things simple. Some nice methods and implicits in akka.util.duration make it possible to just write 'within(100 millis)' to indicate a duration of 100 milliseconds, or 'within(50 millis, 100 millis)' to indicate a min and max for the duration.
Then you just bang out some messages on an ActorRef. After that you can use expectMsg to assert that a specific message has been received, or expectNoMsg to assert that no message should have been received. 'ignoreMsg' takes a partial function that ignores all messages for which the partial function returns true (Funny, the scala code explains it better than that sentence just now). in the test with the SequencingActor, I ignore all messages that are not "something", than assert the "something" message and than ignore everything again that is not "something", making sure there are no messages after that. Make sure you stop all actors after all.
Anyway, enough rambling, check out this gist :)

Starting with Akka 1.0

noreply@blogger.com (Raymond Roestenburg) — Wed, 09 Feb 2011 22:47:00 +0000

Since version 1.0 of Akka is coming out very soon, I thought it might be handy to write a quick new post on getting started with Akka again since a couple of things have gotten quite a bit easier.

Development setup
I am using the following tools:

sbt (0.7.4)
Intellij IDEA Community Edition (10.0) with Scala plugin
sbt-idea (latest)

Install these tools by following the links. I am using sbt-idea as a sbt processor (see sbt-idea link). idea-sbt-plugin is a plugin that you can use to run sbt from within IDEA. Add the Scala plugin and idea-sbt-plugin (look for SBT) in IDEA by using the plugin manager (Settings-> Plugins).
After you have installed these tools you first need to create an sbt project. So lets do that first. ($ means I'm typing in the console, > means I'm in sbt, scala> means I'm in scala interpreter)

Minimal setup

$ mkdir akkastart

$ cd akkastart
$ sbt
press Y to start a new project, fill in some details like name, organization, version, scala version (2.8.1), sbt version (0.7.4) and wait for sbt to download scala and sbt dependencies.
This should create a lib, project, src and target directory. Create a project/build/ directory, add Project.scala to it with the following content:

import sbt._
import sbt_akka_bivy._

class MyProject(info: ProjectInfo) extends DefaultProject(info) with AkkaProject{
}

Create a project/plugins directory and add a Plugins.scala to it with the following content:

import sbt._

class Plugins(info: ProjectInfo) extends PluginDefinition(info) {
  val bumRepo = "Bum Networks Release Repository" at "http://repo.bumnetworks.com/releases"
  val AkkaRepo = "Akka Repository" at "http://akka.io/repository"
  val akkaPlugin = "se.scalablesolutions.akka" % "akka-sbt-plugin" % "1.0-RC6"
  val sbtAkkaBivy = "net.evilmonkeylabs" % "sbt-akka-bivy" % "0.2.0"
}

The akka plugin makes getting all the dependencies and modules for akka extremely easy. The bivy plugin is just really handy for deployment.
Then run sbt update. this will add a lib_managed directory with the minimal dependencies for Akka. add akka modules to the Project.scala file if you expect to use them, for instance akka-camel:

import sbt._
import sbt_akka_bivy._

class MyProject(info: ProjectInfo) extends DefaultProject(info) with AkkaProject{
  val akkaCamel = akkaModule("camel")
}

run sbt update after adding akka modules.
Now we can generate an idea project from the sbt project by running sbt idea: (if you followed installation instructions for sbt-idea as processor from the sbt-idea page)

$ sbt idea

That creates a .iml file your directory and a .idea directory with all the stuff Idea needs. (it also creates a .iml for building the sbt sources like Project.scala)

Now open the .iml file with Idea. Everything should just work (Don't you just love it when that happens? :)
Making your module (Build->Make Module akkastart) builds in the same target directories as sbt would.
Add an akka.conf to src/main/resources (so it gets on your classpath) if you don't like the defaults, check here on akka.io for more information. At first you probably don't need an akka.conf yet, when Akka finds no config it sets up defaults.

That's all there is to it, now you can start adding your Actors in src/main/scala as you see fit.
Of course you can also just do something like this to get that quick scala console fix (output in italics):

$ sbt
> console

scala> import akka.actor.Actor
import akka.actor.Actor
scala> class PrintActor extends Actor {
   | def receive = {
   | case s:String => {
   | println("received:"+s)
   | }
   | }
   | }
defined class PrintActor

scala> val actorRef = Actor.actorOf(new PrintActor)

23:28:45.830 [run-main] WARN akka.config.Config$ -

Can't load 'akka.conf'.

One of the three ways of locating the 'akka.conf' file needs to be defined:

1. Define the '-Dakka.config=...' system property option.

2. Put the 'akka.conf' file on the classpath.

3. Define 'AKKA_HOME' environment variable pointing to the root of the Akka distribution.

I have no way of finding the 'akka.conf' configuration file.

Using default values everywhere.

actorRef: akka.actor.ActorRef = Actor[PrintActor:f55f3eb0-349b-11e0-ae19-0019d2b39ec9]

scala> actorRef.start

23:28:59.710 [run-main] DEBUG a.d.Dispatchers$globalExecutorBasedEventDrivenDispatcher$ - Starting up Dispatchers$globalExecutorBasedEventDrivenDispatcher$[akka:event-driven:dispatcher:global]

with throughput [5]

res1: akka.actor.ActorRef = Actor[PrintActor:f55f3eb0-349b-11e0-ae19-0019d2b39ec9]

scala> actorRef ! "test"

23:29:03.412 [run-main] INFO a.d.LazyExecutorServiceWrapper - Lazily initializing ExecutorService for

scala> 23:29:03.428 [akka:event-driven:dispatcher:global-1] DEBUG akka.dispatch.MonitorableThread - Created thread akka:event-driven:dispatcher:global-1

received:test

For some more examples you can check these here on github, I've updated them to 1.0-RC6.

Happy hAkking!

Repeater and Idempotent Receiver implementation in Akka

noreply@blogger.com (Raymond Roestenburg) — Mon, 27 Sep 2010 18:59:00 +0000

Update to this post: I have upgraded the examples to Akka 1.0-RC1. Check the updated code at github here.

After a couple of examples to get started with Akka, I thought it would be nice to show a more involved example.
In this post I am going to show how you can build your own simple reliable messaging (Guaranteed Delivery) on top of Akka Remote Actors. It is an implementation of an Idempotent Receiver and a Repeater. I'm currently using the master branch of Akka (1.0-SNAPSHOT). I have put this example up on github here along with my other Akka examples from previous posts, so you can skip the lengthy explanation below completely and go straight to the source and check out the eip.idempotent package :)
Instead of pygmentizing all my code into this post, I'll just link to github instead.

Update: You might ask yourself why I am implementing this so high in the stack (which some very smart people did :) At the moment I need this to work on an unmodified version of Akka and I needed it quick. Building it in for instance TCP would have probably made it necessary to modify the Akka source itself and how it builds on top of JBoss Netty. I'm leaving that for another day, stay tuned :)

And I thought it would be nice to show the use of the LifeCycleEvent features of Akka, how it inherits all the Netty goodness and how it is a bit easier to work with Actors instead of hand-rolled threading.

Anyway, back to the example.

What I wanted to achieve with the code:

Recovery from error conditions on the client or the server side.
Guaranteed delivery of messages, even if connection loss occurs (reconnects, errors, shutdown/startup).
A simple lightweight protocol.
It should work with Google Protobuf serialization, since I need a very compact wire protocol. (It shouldn't be to hard to add others)
It should be a drop-in solution and stay out of my way. I should just be able to bang (!) on an ActorRef on the client side, and receive on an Actor on the server side.
I don't need the order of messages maintained.
It only has to work for one way messaging (for now), not for synchronous request-response. (It shouldn't be too hard to add).
It should work on top of Akka's Remote Actors.
It should work in a network balanced infrastructure.
The goal is to improve the reliability of a very lean protocol, not to implement a full featured message queuing protocol or to compete with existing protocols (Because there are plenty of options already available). Keepin it simple.
Show that Akka takes away a lot of the complexities in building this type of stuff.

Idempotent Receiver

An Idempotent Receiver is an Enterprise Integration Pattern that is used to safely receive duplicate messages. The main reason for the use of this pattern is for solving common problems in communicating over a network:

Network components between servers might duplicate messages
Connection errors might only be known on one side of the communication link

For instance, a client sends a message, receives no error on the connection and assumes the message is received by the server. Some network component between the client and the server routes the message to the server, where a connection error occurs. The client is unaware of this and the server never receives the message.
Even in the case where the client uses request-response to communicate with the server, the client cannot be sure that the request has been handled or not, when the client does not receive a reply. The server might have handled the request and failed to send back a response. Resending a message to the server on no reply could result in the operation being applied twice.

The Idempotent Receiver pattern allows the client to repeat messages, because it can identify duplicate messages. I've implemented the Idempotent Receiver as a Remote Actor. It passes messages on to the actual Actor that is supposed to receive the messages, unless the message is a duplicate of course. The Idempotent Receiver checks every incoming message on an Envelopes implementation. I've just included some in-memory implementations, it is quite easy to add a persistent one, for instance using the Akka Persistence module.

Repeater
On the client side I've implemented an Actor that communicates with the Idempotent Receiver on the server side. The Repeater keeps track of all messages that are sent to the server in a RepeatBuffer and keeps a copy of the messages, which it can repeat if needed to the server. The user of the repeater can just get an ActorRef to it and send the messages as you normally would. The Repeater wraps every message in an Envelope, and sends this Envelope to the Idempotent Receiver using the Remote Actor Protocol.

Protocol
I've cheated a bit (in Dutch we say "Beter goed gestolen dan slecht bedacht" :) and looked at how the Remote Actor Protocol works, and borrowed a bit from TCP and other message standards and came up with the following.
Every message will be sent inside an Envelope. An Envelope consists of a Frame ID and an Envelope ID, and a Payload part that can contain any message. I've created protobuf message definition for this here.
Because the Repeater and Idempotent Receiver are keeping messages in memory or on disk, you need a way to tell which messages can be removed. That is where the Frame comes in. Envelopes are always part of a Frame, which is like a 'sticky session' for a number of Envelopes. The Frame consists of a source address, a destination address, an ID and a size (the amount of envelopes).

The Repeater first requests a Frame from the Idempotent Receiver and then starts sending messages for that Frame. The Idempotent Receiver checks if it has received every (unique) message for a Frame based on the Frame size. When a Frame is complete, the Idempotent Receiver sends a CompleteFrameRequest out of band to the Repeater and removes the envelopes from storage. The Repeater receives the request and also removes the envelopes from storage.

As from Akka 0.10 you can register Lifecycle event listeners on both the RemoteClient and RemoteServer (By the way, check out the awesome Akka User List, the guys there are extremely helpful).

I use the Lifecycle event Listeners to be notified of connection errors and successful reconnects.When the connection has been restored after a disconnect or error, a repeat of the Frame is triggered on the client side, or a repeat request of the Frame is triggered on the server side to the Repeater with some selective acknowledgement of the already received Envelope IDs.

Since the Envelope message is just another protobuf encoded message, Akka handles it as usual in the RemoteClient and RemoteServer using the Akka Remote Protocol. The Idempotent Receiver receives the Envelope message and deserializes the payload to the correct protobuf message that the Actor is expecting, and the Envelope itself. Right now it only works for protobuf messages because that's all I use, but the EnvelopeSerializer can be easily changed to handle all types of messages (If that code looks familiar it's because it's basically a copy of the MessageSerializer class in Akka :)

Scale out
A problem with the Idempotent Receiver is that it is not easy to scale out. If you would have RemoteActors distributed over many Servers and use a network load balancer, you can't just put an Idempotent Receiver on all of the servers, since they would not know about each other. Sharing state over distributed nodes is not really what you want, it gets complicated very quickly. Instead of distributing the Envelope and Frame state over several servers, I have implemented a JGroups based Envelopes implementation where every node in the cluster 'owns' the Frames that is has created. If it receives a message for a Frame that it doesn't own, it sends it into the cluster, so that the Idempotent Receiver that owns the Frame can handle it. You can specify a range for the Frame IDs in the Envelopes so that different servers will never create the same Frame.

Testing
You can check out the unit tests here. To make sure that everything works, I have written a very simple and crude Network Proxy (Blocking I/O) for testing, that will be used in the unit tests between the Repeater and Idempotent Receiver. The Network Proxy can be started, stopped, and connection errors can be caused by registering a function in the loop that receives from the client and passes this to the RemoteServer, and in the loop that receives responses from the RemoteServer and passes these back to the client.

In the repository you can also find this unit test that tests if all LifeCycle events (both RemoteClient and RemoteServer) are triggered by the Akka framework, using this simple proxy.

I've written this unit test here to test the Repeater and Idempotent Receiver. The JGroups version is tested in this unit test here. Let me know what you think!

Upgrading examples to Akka master (0.10) and Scala 2.8.0 Final

noreply@blogger.com (Raymond Roestenburg) — Tue, 27 Jul 2010 06:23:00 +0000

Update on this post: I have upgraded the AkkaExampels git repo to version 1.0-RC1. Camel references are upgraded to 2.5.0. I have also included a reference to the scalablesolutions Akka maven repository, so that the akka-sbt-plugin is automatically downloaded at sbt update. AkkaExamples on github

In this post I am going to write about upgrading from Akka 0.8.1 to Akka 0.10. (at the moment 0.10 is not ready yet, but for simplicity I will refer to 0.10, To be exact it's git commit c737e2edf4b9ad980d6922f169ca4fb19f8c8284 :)
Also to make things a bit simpler, I have pushed my previous Akka examples here to github, so you can just clone from there and check it out. (You will need to build Akka as shown below first until 0.10 comes out, and be sure to run 'sbt update' before you try to build)

Building Akka
At the time of writing there is no 0.10 release yet that you can just download, so I built Akka from master and published it to my local ivy repository. the sbt build has really improved a lot, the update is quicker and so is the build:

$ git clone git://github.com/jboner/akka.git
$ cd akka
$ sbt update
$ sbt dist
$ sbt publish-local
$ cd akka-sbt-plugin
$ sbt publish-local

I am also building and publishing the akka-sbt-plugin to local, so I can use it later.
Akka now uses Google Protobuf 2.3.0. If you are going to use Google Protobuf for your own messages, you will need to install the 2.3.0 protoc compiler, get it here and follow instructions in the readme file.
In case you are going to use Apache Camel, you need to upgrade the Camel dependencies to 2.4.0 (Akka 0.8.1 used Camel 2.2.0).

SBT Project file plugins
Akka 0.10 comes with a very nice addition for SBT, the AkkaProject sbt plugin. This makes it a lot more succinct and easy to start off with an Akka project, as you can see below:

import sbt._
import Process._
import sbt_akka_bivy._
class Project(info: ProjectInfo) extends DefaultProject(info) with AkkaProject with AkkaKernelDeployment{
  val akkaCamel = akkaModule("camel")
  val akkaKernel = akkaModule("kernel")
  val junit = "junit" % "junit" % "4.8.1" % "test->default"
  val camelFtp= "org.apache.camel" % "camel-ftp" % "2.4.0" % "compile"
  val camelMina= "org.apache.camel" % "camel-mina" % "2.4.0" % "compile"
  val camelJetty= "org.apache.camel" % "camel-jetty" % "2.4.0" % "compile"

  val scalatest = "org.scalatest" % "scalatest" % "1.2-for-scala-2.8.0.final-SNAPSHOT" % "test->default"
  override def repositories = Set( "scala-tools-snapshots" at 
      "http://scala-tools.org/repo-snapshots/"
  )
}

As you can see I have added some Camel components to play with, and I've added the scalatest and junit dependencies for unit testing. The Project file now uses the AkkaProject plugin and the Bivy Sack plugin (sbt-akka-bivy) for microkernel deployments (check here, really cool) which I am going to use in a later post for quick and easy deployment when I actually have something to deploy, the project only contains unit tests for now. Before you can use this Project file, you need to add a Plugins.scala file to the project/plugins directory, which describes the akka-sbt-plugin that we built, and it also pulls in the sbt-akka-bivy plugin for deployment:

import sbt._

class Plugins(info: ProjectInfo) extends PluginDefinition(info) {
  val bumRepo = "Bum Networks Release Repository" at "http://repo.bumnetworks.com/releases"
  val akkaPlugin = "se.scalablesolutions.akka" % "akka-sbt-plugin" % "0.10"
  val sbtAkkaBivy = "net.evilmonkeylabs" % "sbt-akka-bivy" % "0.2.0"
}

Run 'sbt update' to pull the dependencies in. After that you can setup a new project in IDEA. (I'm hoping an sbt IDEA plugin will get really good soon cause this always means some fiddling with the classpath).

Intellij IDEA

I am using the Intellij IDEA Community EAP (95.390 to be precise) which you can get from here.

Once you have installed it, update the Scala plugin in the plugins section if you had a previous version, otherwise install from the available list. (File-Settings-Plugins)

Choose a new project, from existing source, make sure you select the compile and test jars (in lib_managed) and the scala 2.8.0 jars, leave out the 2.7.7 ones from sbt. You can add a Scala Facet yourself, or IDEA will autodetect it and you can choose that it will create it for you, just make sure to use the 2.8.0 final ones. The same goes for the src 'IDEA module', you can add it in the new project wizard or in the Project Structure settings (set the src/main/scala as source and src/test/scala as test). Make sure you set IDEA to use the 2.8.0 final compiler and library in the Scala Facet settings, to the jars found in project/boot/scala-2.8.0/lib. Put the scala 2.8.0 jars right after your JDK in your module dependencies.

Project Structure - src module dependencies

Project Structure - src module sources

Project Structure - scala 2.8.0 project library

I have updated the examples from previous posts about Akka in the project on github. The biggest difference from Akka 0.8 is that the identity and the instance of an Actor are separated respectively into an ActorRef trait and an Actor trait, as described here. There is an Actor object as well which is used as a factory module. The Actor.actorOf method makes sure that you cannot access the Actor instance methods directly to prevent threading problems if you would do such a thing. In the examples I import all the Actor object methods. You can only send messages to the Actor implementation through the API exposed through the ActorRef class. Within the Actor instance, this is done through the self reference.

Below the updated code for 0.10 of the previous simple example in 0.8, to show the usage of actorOf in the specs and the self reference in the Worker Actor.

import org.scalatest.Spec
import org.scalatest.matchers.MustMatchers

import se.scalablesolutions.akka.actor.{Actor}
import se.scalablesolutions.akka.actor.Actor._

/**
 * A Spec for an example Worker Actor.
 */
class WorkerSpecs extends Spec with MustMatchers {
  describe("A Worker") {

    describe("(when it queues commands)") {
      it("should have the correct number of commands queued") {

        val command = new Command("test", "data")
        val actorRef = actorOf(new Worker())
        actorRef.start
        var reply: CountResponse = (actorRef !! new CountCommandsQueued()).getOrElse(fail()).asInstanceOf[CountResponse]
        reply.count must be === 0

        actorRef ! command
        actorRef ! command
        actorRef ! command
        reply = (actorRef !! new CountCommandsQueued()).getOrElse(fail()).asInstanceOf[CountResponse]
        reply.count must be === 3
        actorRef.stop
      }
    }
    describe("(when it executes all queued commands)") {
      it("should have no commands queued after executing") {
        val command = new Command("test", "data")
        val actorRef = actorOf(new Worker())
        actorRef.start
        actorRef ! command
        actorRef ! command
        var reply: CountResponse = (actorRef !! new CountCommandsQueued()).getOrElse(fail()).asInstanceOf[CountResponse]
        reply.count must be === 2
        actorRef ! new Execute()
        reply = (actorRef !! new CountCommandsQueued()).getOrElse(fail()).asInstanceOf[CountResponse]
        reply.count must be === 0
        actorRef.stop
      }
    }
  }
}

/**A command message */
case class Command(name: String, data: String)
/** A message to get the amount of commands queued*/
case class CountCommandsQueued()
/** A message to execute all commands queued */
case class Execute()
/** A message reply on the CountCommandsQueued message */
case class CountResponse(count: Int)

/**
 * A Worker actor that receives Commands, queues the commands, and executes commands on the Execute message,
 * as example for some simple Actor testing
 */
class Worker extends Actor {
  var commands: List[Command] = Nil

  def receive = {
    case msg: Command => {
      commands = msg :: commands
    }
    case msg: Execute => {
      for (command <- commands) {
        commands = commands.dropRight(1)
      }
    }
    case msg: CountCommandsQueued => {
      self.reply(new CountResponse(commands.size))
    }
  }
}

Self reference
In the simple example above I updated the code that uses the Actor API from within the Actor, by prefixing the calls with self. What you can also do is import the 'self methods' in the Actor class, in case you want to save time typing 'self' all the time. I've done this in the 0.10 update to my previous example to join messages in 0.8 (FirstMessageHandler Actor):

import org.scalatest.matchers.MustMatchers
import org.scalatest.Spec
import se.scalablesolutions.akka.actor.Actor._
import se.scalablesolutions.akka.actor.{ActorRef, ActorRegistry, Actor}
/**
 * A Spec for the Aggregator
 */
class AggregatorSpecs extends Spec with MustMatchers {
  var aggregateMessageReceived: Boolean = false
  var aggregateMessage: AggregateMessage = null

  describe("An Aggregator") {

    describe("(when it receives FirstMessage and then SecondMessage)") {
      val firstRef = actorOf(new FirstMessageHandler())
      val secondRef = actorOf(new SecondMessageHandler())
      val receiveTestRef = actor {
        case msg: AggregateMessage => {
          aggregateMessageReceived = true
          aggregateMessage = msg

        }
      }
      val aggregator = actorOf(new Aggregator(receiveTestRef))
      firstRef.start
      secondRef.start
      aggregator.start

      it("should send an AggregateMessage containing data of FirstMessage and SecondMessage to the passed in actor") {
        firstRef ! new FirstMessage("id-1", "name-1")
        Thread.sleep(200)
        secondRef ! new SecondMessage("data-1")
        Thread.sleep(1000)
        aggregateMessageReceived must be === true
        aggregateMessage.id must be === "id-1"
        aggregateMessage.name must be === "name-1"
        aggregateMessage.data must be === "data-1"
        firstRef.stop
        secondRef.stop
        aggregator.stop
      }
    }
  }
}

/** A message that is expected to arrive first*/
case class FirstMessage(id: String, name: String)
/** A message that is expected to arrive second*/
case class SecondMessage(data: String)
/** An aggregated message, from first and second */
case class AggregateMessage(id: String, name: String, data: String)
/** A command to get the last message*/
case class GiveMeLastMessage()

/**
 * An Aggregator actor that aggregates a first and second message type
 */
class Aggregator(pass: ActorRef) extends Actor {
  def receive = {
    case msg: SecondMessage => {
      println("Aggregator, my data is " + msg.data)
      val firstMessageHandler: ActorRef = ActorRegistry.actorsFor(classOf[FirstMessageHandler]).head
      var reply: Option[Any] = firstMessageHandler !! new GiveMeLastMessage
      if (reply.isDefined) {
        val first: FirstMessage = reply.get.asInstanceOf[FirstMessage]
        println("Aggregator, my first message is " + first.id)
        val ag = new AggregateMessage(first.id, first.name, msg.data)
        pass ! ag
      }
    }
  }
}

/**
 * A Message Handler for the SecondMessage type
 */
class SecondMessageHandler extends Actor {
  def receive = {
    case msg: SecondMessage => {
      // do some processing
      println("Secondmessage, my data is " + msg.data)
      // then call the aggregator
      val aggregator: ActorRef = ActorRegistry.actorsFor(classOf[Aggregator]).head
      aggregator ! msg
    }
  }
}

/**
 * A Message Handler for the FirstMessage type
 */
class FirstMessageHandler extends Actor {
  import self._
  var lastMessage: Option[FirstMessage] = None
  var lastRequestor: Option[Any] = None

  def receive = {
    case msg: FirstMessage => {
      // do some processing
      println("Firstmessage, my name is " + msg.name)

      lastMessage = Some(msg)
      if (lastRequestor != None) {
        val a = lastRequestor.asInstanceOf[ActorRef]
        a ! msg
        lastMessage = None
      }
    }
    case msg: GiveMeLastMessage => {
      if (!lastMessage.isDefined) {
        lastRequestor = senderFuture
      } else {
        reply(lastMessage.get)
      }
    }
  }
}

Protobuf Serialization
The Serializable.Protobuf trait has been removed in 0.10, and Akka 0.10 now uses protobuf 2.3.0. Below the updated code for 0.10 for the previous simple protobuf serialization example in 0.8:

package unit.akka

import se.scalablesolutions.akka.actor.Actor
import se.scalablesolutions.akka.actor.Actor._
import org.scalatest.{BeforeAndAfterAll, Spec}
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
import se.scalablesolutions.akka.remote.{RemoteClient, RemoteServer}
import org.scalatest.matchers.{MustMatchers, ShouldMatchers}
import unit.test.proto.Commands
import com.google.protobuf.Message
import unit.test.proto.Commands.WorkerCommand
import unit.akka.CommandBuilder._

/**
 * Test to check if communicating with protobuf serialized messages works.
 */
@RunWith(classOf[JUnitRunner])
class ProtobufSpecs extends Spec with BeforeAndAfterAll with MustMatchers {
  var server: RemoteServer = new RemoteServer()

  override def beforeAll(configMap: Map[String, Any]) {
    server.start("127.0.0.1", 8091)
  }

  override def afterAll(configMap: Map[String, Any]) {
    RemoteClient.shutdownAll
    server.shutdown
  }

  describe("Send using Protobuf protocol") {

    it("should receive local protobuf pojo and reply") {
      // send a local msg, check if everything is ok
      val actor = actorOf(new TestProtobufWorker())
      actor.start
      val msg = Worker(1, "my-name-1", "my-data-1")
      val result: Option[Any] = actor !! msg
      result match {
        case Some(reply: Commands.WorkerCommand) => {
          // test actor changes name to uppercase
          reply.getName must be === "MY-NAME-1"
        }
        case None => fail("no response")
      }
      actor.stop
    }

    it("should receive remote protobuf pojo and reply") {
      //start a remote server
      val actor = actorOf(new TestProtobufWorker())
      //register the actor that can be remotely accessed
      server.register("protobuftest", actor)
      val msg = Worker(2, "my-name-2", "my-data-2")
      val result: Option[Any] = RemoteClient.actorFor("protobuftest", "127.0.0.1", 8091) !! msg

      result match {
        case Some(reply: Commands.WorkerCommand) => {
          // test actor changes name to uppercase
          reply.getName must be === "MY-NAME-2"
        }
        case None => fail("no response")
      }
      actor.stop
    }
  }
}


/**
 * Actor that sends back the message uppercased.
 */
class TestProtobufWorker extends Actor {
  def receive = {
    case msg: Commands.WorkerCommand => {
      log.info("received protobuf command pojo:" + msg.getId)
      val r = Worker(2, msg.getName.toUpperCase, msg.getData.toUpperCase)
      log.info("sending back a renamed pojo:" + r.getName)
      self.reply(r)
    }
  }
}

/**
 * Shortcuts for creating WorkerCommands
 */
object CommandBuilder {
  def Worker(id: Int, name: String, data: String): WorkerCommand = {
    Commands.WorkerCommand.newBuilder.setId(id).setName(name).setData(data).build
  }

  def Worker(command: WorkerCommand): WorkerCommand = {
    Commands.WorkerCommand.newBuilder.setId(command.getId).setName(command.getName).setData(command.getData).build
  }
}

You now just work directly with the protobuf messages, instead of creating a case class that extends the Serializable.Protobuf trait and that serializes to and from the protobuf message. In the test I added the CommandBuilder object just for convenience.

If you want to check out the source code go to my github repo here.

Scala 2.8.0, Akka 0.8.1 and cassandra 0.6.2

noreply@blogger.com (Raymond Roestenburg) — Wed, 23 Jun 2010 15:09:00 +0000

In this post I'm going to install cassandra 0.6.2 on ubuntu karmic koala. I'm going to use the akka persistence module to access it and write a unit test to check if everything is working.

I've used the installation instructions for cassandra as shown here, with the following changes:

3. Add the following lines in /etc/apt/sources.list instead:

deb http://www.apache.org/dist/cassandra/debian unstable main
deb-src http://www.apache.org/dist/cassandra/debian unstable main

That's pointing to version 0.6.2 at the moment.

9. Instead change the JMX port in /etc/init.d/cassandra, it doesn't look like cassandra.in.sh is used. When I changed it in cassandra.in.sh, Cassandra kept bumping into my local Tomcat server.
The storage-conf.xml is in /etc/cassandra. The storage-conf.xml file that comes with akka does not seem to work and needs a small change, I have copied the below KeySpace definition in the /ect/cassandra/storage-conf.xml file:

<Keyspace Name="akka">
      <KeysCachedFraction>0.01</KeysCachedFraction>
        <ColumnFamily CompareWith="UTF8Type" Name="map" />
        <ColumnFamily CompareWith="UTF8Type" Name="vector"/>
        <ColumnFamily CompareWith="UTF8Type" Name="ref"/>

      <ReplicaPlacementStrategy>org.apache.cassandra.locator.RackUnawareStrategy</ReplicaPlacementStrategy>

      <!-- Number of replicas of the data -->
      <ReplicationFactor>1</ReplicationFactor>

      <!--
       ~ EndPointSnitch: Setting this to the class that implements
       ~ AbstractEndpointSnitch, which lets Cassandra know enough
       ~ about your network topology to route requests efficiently.
       ~ Out of the box, Cassandra provides org.apache.cassandra.locator.EndPointSnitch,
       ~ and PropertyFileEndPointSnitch is available in contrib/.
      -->
      <EndPointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndPointSnitch>
</Keyspace>

Cassandra complains if you don't add the ReplicaPlacementStrategy. I'm not sure yet what these settings mean exactly but I didn't let that stop me :)

You should now be able to start cassandra and use it from akka (sudo /init.d/etc/cassandra start). You can check if the akka keyspace is present, with the 'show keyspaces' command in cassandra-cli.

Be sure to check /var/log/cassandra/*.log if everything works, and you can start jconsole and connect to casssandra through JMX, check out some MBeans. (localhost, port 10036 in my case)

And then a simple unit test. What is important is that you use the lazy declaration for the storage, and use the atomic blocks. You can also use the persistent map straight from an Actor, but I like to do it this way because you can switch out the implementation (for instance from Cassandra to Redis), by passing in the below PersistentMap trait to for instance an Actor constructor.

You need to convert keys and values to byte arrays, so it is quite a low level interface for storing data. This unit test is very simple on purpose, I would probably use the Akka Serializable interfaces on case classes that I would want to store.

package unittest

import se.scalablesolutions.akka.stm.Transaction.Local._
import se.scalablesolutions.akka.util.Logging
import org.scalatest.{Spec, BeforeAndAfterAll}
import org.scalatest.matchers.ShouldMatchers
import se.scalablesolutions.akka.persistence.cassandra.CassandraStorage
import org.scalatest.junit.JUnitRunner
import org.junit.runner.RunWith
import org.junit.Assert._
import org.apache.cassandra.service.NotFoundException

@RunWith(classOf[JUnitRunner])
class AkkaCassandraTest extends Spec with ShouldMatchers with BeforeAndAfterAll with Logging {
  describe("Store stuff in Cassandra using Akka persistence abstraction") {
    it("should be able to store some stuff and find it again") {
      val map = new CassandraPersistentMap("test-map")
      map.put("test-key", "test-value")

      val someValue = map.get("test-key")
      if (someValue.isDefined) {
        val value = someValue.get
        log.info("found " + value)
      } else {
        fail("test-key not found")
      }
      val anotherValue = map.get("no-key")
      assertFalse(anotherValue.isDefined)
    }
  }
}

/**
 * Simple map trait  
 */
trait PersistentMap {
  def put(key: String, value: String)

  def get(key: String): Option[String]
}

/**
 * Simple Map that uses Cassandra
 */
class CassandraPersistentMap(storageKey: String) extends PersistentMap {
  //Using atomic we define the transaction span.
  private lazy val storage = atomic {CassandraStorage.getMap(storageKey)}

  def put(key: String, value: String): Unit = {
    atomic {
      storage.put(key.getBytes, value.getBytes)
      storage.commit
    }
  }

  def get(key: String): Option[String] = {
    atomic {
      val result = storage.get(key.getBytes)
      result match {
        case Some(bytes) => {
          Some(new String(bytes))
        }
        case None => {
          None
        }
      }
    }
  }
}

Testing Akka Remote Actor using Serializable.Protobuf

noreply@blogger.com (Raymond Roestenburg) — Thu, 27 May 2010 20:42:00 +0000

Just a short post to show how you can test a remote actor, where the messages are serialized using Google protobuf. Protobuf is automatically used for the internal akka protocols (to make remote actors possible for instance), but the messages that you define yourself are not automatically encoded using protobuf of course, (you would need to do quite a bit of reflection and include the protoc compiler in the framework, and some translation or mapping for that to automagically work, that's a path down the rabbit hole I'm glad the akka team didn't take).

By default java serialization is used. There is a setting in akka.conf for serialization, but that's only for the cluster protocol. I'm using akka 0.8 by the way.

Let's get back to the task at hand.

First you need to install protobuf 2.2.0. (check here). 2.3.0 gave me some issues, the generated java classes did not compile. (and akka 0.8 ships with the 2.2.0 jar, so you might be asking for trouble with 2.3.0)

Create a .proto file that describes the message (in my case I named the file ProtobufCommandPojo.proto):

package unit.test.proto;

message workercommand {
required uint64 id = 1;
required string name = 2;
required string data = 3;
}

Then generate the java code for the protobuf message (in my case to a java directory):

$ protoc ProtobufCommandPojo.proto --java_out ../../java

This generates quite a few java classes, mainly ProtobufCommandPojo.

Create a message case class that mixes in the Serializable.Protobuf trait:

package unit

import se.scalablesolutions.akka.serialization.Serializable
import test.proto.ProtobufCommandPojo

/**
 * message class that uses Serializable.Protobuf trait.
 * uses generated protobuf classes.
 */
case class ProtobufMixedIn(id: Long, name :String, data :String) extends Serializable.Protobuf[ProtobufMixedIn] {

     def getMessage : com.google.protobuf.Message = {
       ProtobufCommandPojo.workercommand.getDefaultInstance().toBuilder().setId(id).setData(data).setName(name).build()
     }
}

As you can see you use the generated class to do the actual protobuf message building. this is basically the mapping between protobuf and the message class.
Note: the Serializable.Protobuf class has a default implementation for fromBytes, which assumes that you do a standard mapping between your class and the protobuf message definition:

def fromBytes(bytes: Array[Byte]): T = getMessage.toBuilder.mergeFrom(bytes).asInstanceOf[T]

I'm expecting that if you build a protobuf message in a different way than is 'expected' in the getMessage method, you might not get the result you want. I'm guessing that there is some reflection or naming convention going on under the hood to make this default implementation work, I'll leave that investigation to you if you feel like you want to know how this works exactly.

And then the unit test. You simply work with the case class ProtobufMixedIn, as you would with a normal message. Akka does the rest.

package unit

import se.scalablesolutions.akka.actor.Actor
import org.scalatest.{BeforeAndAfterAll, Spec}
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
import se.scalablesolutions.akka.remote.{RemoteClient, RemoteServer}
import org.scalatest.matchers.{MustMatchers, ShouldMatchers}

/**
 * Test to check if communicating with protobuf serialized messages works.
 */
@RunWith(classOf[JUnitRunner])
class ProtobufTest extends Spec with MustMatchers {
  describe("Send using Protobuf protocol") {

    it("should receive local protobuf pojo and reply") {
      // send a local msg, check if everything is ok
      val actor = new TestProtobufWorker()
      actor.start
      val msg = new ProtobufMixedIn(1, "my-name-1", "my-data-1")
      val result: Option[ProtobufMixedIn] = actor !! msg
      result match {
        case Some(reply: ProtobufMixedIn) => {
          // test actor changes name to uppercase
          reply.name must be === "MY-NAME-1"
        }
        case None => fail("no response")
      }
      actor.stop
    }

    it("should receive remote protobuf pojo and reply") {
      //start a remote server
      val server = new RemoteServer()
      val actor = new TestProtobufWorker()
      server.start("127.0.0.1", 8091)
      //register the actor that can be remotely accessed
      server.register("protobuftest", actor)
      val msg = new ProtobufMixedIn(2, "my-name-2", "my-data-2")

      val result: Option[ProtobufMixedIn] = RemoteClient.actorFor("protobuftest", "127.0.0.1", 8091) !! msg

      result match {
        case Some(reply: ProtobufMixedIn) => {
          // test actor changes name to uppercase
          reply.name must be === "MY-NAME-2"
        }
        case None => fail("no response")
      }
      //shutdown server and client processes
      server.shutdown
      RemoteClient.shutdownAll
    }
  }
}

/**
 * Actor that sends back the message uppercased.
 */
class TestProtobufWorker extends Actor {
  def receive = {
    case msg: ProtobufMixedIn => {
      log.info("received protobuf command pojo:" + msg.id)
      val r = new ProtobufMixedIn(2, msg.name.toUpperCase, msg.data.toUpperCase)
      log.info("sending back a renamed pojo:" + r.name)
      reply(r)
    }
  }
}

Join messages with Akka

noreply@blogger.com (Raymond Roestenburg) — Thu, 06 May 2010 21:20:00 +0000

Update to this post: I have upgraded the examples to Akka 1.0-RC1. Check the updated code at github here.

Akka raises the abstraction level for writing concurrent systems. In this post I want to give a simple example of this.

Let's say that you have two types of messages coming into your system (of let's say type FirstMessage and SecondMessage), and you would like to:
1. First do some processing on them separately/independently and after that:
2. 'join' or correlate these two messages when they come in one after the other in a pair, based on the time they are received by the system and send out a new message.
3. All processing has to be non-blocking.
4 only make aggregates of the messages that come in as [FirstMessage, SecondMessage], ignoring consecutive FirstMessages.

In this example, the only thing you know is that one type of message always comes in a little bit later than the other, and when this happens, you want to combine the information in these two messages into one outgoing message. And, this is not the only thing happening in the system, you also want to handle these messages asynchronously, do some type of processing before aggregation.

Something like this (apologies for the crappy graphics, got no time to waste on it :)

I tried to sort of use the enterprise integration patterns diagrams. On the left side the two types of messages that come in (let's call them of type FirstMessage and SecondMessage), on the right side the result (AggregatedMessage). I was so lazy to make this diagram, that I left out the two processes that do something with FirstMessage and SecondMessage on just before t=1 and t = 2, but I guess you get the drift.

Without something like Akka, I would probably dive into the java concurrent package.. (what follows is a braindump, you can add your own pinch of salt :)

Anyway, grab out my favourites: ExecutorService, LinkedBlockingQueue and others, create a Runnable class, run in a while loop, poll (or take) from the queue, and based on the type of message in the queue, store that in some state in the Runnable implementing class, let's call that the Aggregator.

(would have to be thread safe, so I need to think about locking, or use a concurrent map, key=type of message, value = list of some wrapper that holds the message plus the time you received it). Then if the message that comes in is of type FirstMessage, put it in the list in the map, if the message is of type SecondMessage, get the last put FirstMessage out of the map, combine the two, creating an AggregatedMessage, remove the handled messages from the data structure. Then I would be able to send that AggregatedMessage to something else, for that I would need to think of a way to call for instance some remoting API, web services or whatever you like, get a proxy to that from inside the Aggregator, need to wire that up in some kind of way. And of course I would need to find a way to get the Aggregator up and running in an ExecutorService, and be able to call it from the outside in a reasonable way. Of course I would need to think of how to start and stop the Aggregator, monitor it's health, be able to restart it if it fails, stuff like that.

And then I need to unit test this on thread safety, calling the code from two threads to simulate the two streams of message types, making sure it really works, mock out the web service in and out, oh yeah write some WSDL, do some config of JAX-WS (oh crap that's synchronous by nature), probably finding out along the way that sometimes SecondMessage comes before FirstMessage, rounding the whole thing out...

And maybe I will use the ConcurrentLinkedQueue instead of the BlockingQueue, just to find out that my thread is tightly spinning in my while(!stop) loop when it's in production.. (yes that does happen.. time to add a Thread.sleep?)

And that's only the aggregator, I haven't covered the two processes yet that need to take place before it (the ones I didn't draw in the diagram..)

I'm actually already getting tired writing about it, let alone writing the actual code!

Anyway, this post was about akka, so how would I do something like this using Actors?

Well, a first try at it would be something like this:

import java.util.Date
import org.scalatest.matchers.MustMatchers
import org.scalatest.Spec
import se.scalablesolutions.akka.actor.{SupervisorFactory, ActorRegistry, Actor}
import se.scalablesolutions.akka.actor.Actor._
import se.scalablesolutions.akka.config.ScalaConfig._

case class FirstMessage(id: String, name: String)
case class SecondMessage(data: String)
case class AggregateMessage(id: String, name: String, data: String)
case class GiveMeLastMessage()


class Aggregator(pass: Actor) extends Actor {
  def receive = {
    case msg: SecondMessage => {
      println("Aggregator, my data is " + msg.data)
      val firstMessageHandler: Actor = ActorRegistry.actorsFor(classOf[FirstMessageHandler]).head
      var reply: Option[FirstMessage] = firstMessageHandler !! new GiveMeLastMessage
      if (reply.isDefined) {
        val first: FirstMessage = reply.get
        println("Aggregator, my first message is " + first.id)
        val ag = new AggregateMessage(first.id, first.name, msg.data)
        pass ! ag
      }
    }
  }
}

class SecondMessageHandler extends Actor {
  def receive = {
    case msg: SecondMessage => {
      // do some processing
      println("Secondmessage, my data is " + msg.data)
      // then call the aggregator
      val aggregator: Actor = ActorRegistry.actorsFor(classOf[Aggregator]).head
      aggregator ! msg
    }
  }
}

class FirstMessageHandler extends Actor {
  var lastMessage: Option[FirstMessage] = None
  var lastRequestor: Option[Any] = None

  def receive = {
    case msg: FirstMessage => {
      // do some processing
      println("Firstmessage, my name is " + msg.name)

      lastMessage = Some(msg)
      if (lastRequestor != None) {
        val a = lastRequestor.asInstanceOf[Actor]
        a ! msg
        lastMessage = None
      }
    }
    case msg: GiveMeLastMessage => {
      if (!lastMessage.isDefined) {
        lastRequestor = senderFuture
      } else {
        reply(lastMessage.get)
      }
    }
  }
}


class AggregatorSpec extends Spec with MustMatchers {
  var aggregateMessageReceived: Boolean = false
  var aggregateMessage: AggregateMessage = null

  describe("An Aggregator") {

    describe("(when it receives FirstMessage and then SecondMessage)") {
      val first = new FirstMessageHandler()
      val second = new SecondMessageHandler()
      val receiveTest = actor {
        case msg: AggregateMessage => {
          aggregateMessageReceived = true
          aggregateMessage = msg

        }
      }
      val aggregator = new Aggregator(receiveTest)
      first.start
      second.start
      aggregator.start

      it("should send an AggregateMessage containing data of FirstMessage and SecondMessage to the passed in actor") {
        first ! new FirstMessage("id-1", "name-1")
        Thread.sleep(200)
        second ! new SecondMessage("data-1")
        Thread.sleep(1000)
        aggregateMessageReceived must be === true
        aggregateMessage.id must be === "id-1"
        aggregateMessage.name must be === "name-1"
        aggregateMessage.data must be === "data-1"
        first.stop
        second.stop
        aggregator.stop
      }
    }
  }
}

What I really like about it is that I can try something like this out in very few lines of code. And, the solution is quite simple. When receiving the SecondMessage type in Aggregator you can ask the FirstMessageHandler to give you the last message it received, and inside that Actor, you use the senderFuture to respond back to the Aggregator! this is of course a very rough first try at it and should expand the test to run from many threads etc, but it hardly took me any time to write.

Starting with Akka part 2, Intellij IDEA, Test Driven Development

noreply@blogger.com (Raymond Roestenburg) — Tue, 04 May 2010 21:06:00 +0000

Update to this post: I have upgraded the examples to Akka 1.0-RC1. this newer post might be more helpful, or check directly at github here.

In this post I'm going to expand on the previous post's simple project, add the use of an IDE, and do some Test Driven Development with akka (with scalatest+junit)

First, check if your Project.scala file has the dependencies for scalatest and junit:

import sbt._

class Project(info: ProjectInfo) extends DefaultWebProject(info){
  override def repositories = Set(
  "Java.Net" at "http://download.java.net/maven/2",
  "jBoss" at "http://repository.jboss.org/maven2",
  "service mix" at "http://svn.apache.org/repos/asf/servicemix/m2-repo/",
  "Apache Camel" at "https://repository.apache.org/content/repositories/releases/",
  "Akka Maven Repository" at "http://scalablesolutions.se/akka/repository",
  "Multiverse Releases" at "http://multiverse.googlecode.com/svn/maven-repository/releases/",
  "GuiceyFruit" at "http://guiceyfruit.googlecode.com/svn/repo/releases/",
  "DataBinder" at "http://databinder.net/repo",
  "Configgy" at "http://www.lag.net/repo",
  ScalaToolsSnapshots)

  override def libraryDependencies = Set(

  /* servlet implementation */
  "org.eclipse.jetty"  % "jetty-server"   % "7.0.1.v20091125" % "test",
  "org.eclipse.jetty"  % "jetty-webapp"   % "7.0.1.v20091125" % "test",
                "org.scalatest" % "scalatest" % "1.0.1-for-scala-2.8.0.Beta1-with-test-interfaces-0.3-SNAPSHOT" % "test->default",
  /* camel */
  "org.apache.camel" % "camel-ftp" % "2.2.0" % "compile",
  /* akka dependencies */
  "se.scalablesolutions.akka" % "akka-kernel_2.8.0.Beta1"  % "0.8.1" % "compile",
  "se.scalablesolutions.akka" % "akka-core_2.8.0.Beta1"    % "0.8.1" % "compile",
  "se.scalablesolutions.akka" % "akka-servlet_2.8.0.Beta1" % "0.8.1" % "compile",
  "se.scalablesolutions.akka" % "akka-rest_2.8.0.Beta1"    % "0.8.1" % "compile")
  override def jettyPort = 9012
  val junit = "junit" % "junit" % "4.5" % "test"
}

As you can see I have also added a dependency to Camel FTP2, a Camel component for FTP, which I will be using in my next post.

At the moment I am using Intellij IDEA Community Edition EAP with the Scala Plugin for 2.8.0.Beta1. The EAP edition does come with a warning from jetbrains: "Please note that the quality of EAP versions may at times be way below even usual beta standards".

The Eclipse plugin is not usable at the moment, at least it didn't work for me. I did notice that IDEA was storing a +800MB index file somewhere in my home directory, which is kind of freaky, and compilation is a bit slow. I'm going to investigate later what I will lose by using a simpler editor and doing everything through sbt scripting. But for now I'm happy to use IDEA.

(Here is a good tip to add scala highlighting to gedit.)

First download the IDEA Community Edition EAP and install:

$ wget http://download.jetbrains.com/idea/ideaIC-95.54.tar.gz
$ sudo mkdir /usr/share/idea
$ sudo cp ideaIC-95.54.tar.gz /usr/share/idea
$ cd /usr/share/idea
$ tar xvf ideaIC-95.54.tar.gz
$ cd /usr/local/bin
$ sudo ln -s /usr/share/idea/idea-IC-95.54/bin/idea.sh

Then run idea so that it creates a .IdeaIC90 directory in your home directory.
add a plugins directory to this directory:

$ cd ~/.IdeaIC90/config
$ mkdir plugins

Then download and install the latest scala plugin (at the moment scala-intellij-bin-0.3.1156.zip):

$ wget 'http://plugins.intellij.net/plugin/?action=download&id=8152' -O scala-plugin.zip
$ cp scala-plugin.zip .IdeaIC90/config/plugins
$ cd .IdeaIC90/config/plugins/
$ unzip scala-plugin.zip

Now run Idea, you should see the Scala plugin in the Plugins section. Create a new project from scratch, create it in the directory where your project is (in my case ~/akka_start):

Before I had some issues that IDEA would take the wrong scala facets and version, if that happens to you, check here:
http://www.jetbrains.net/devnet/thread/284354

Also, if the facet is not automatically included in your project, or you didn't check the Scala tickbox in the project wizard, open the .iml file in an editor (not in IDEA, it will be replaced) and add the facets so you get something like this (add the FacetManager part):

<?xml version="1.0" encoding="UTF-8"?>
<module type="JAVA_MODULE" version="4">
  <component name="FacetManager">
    <facet type="Scala" name="Scala">
      <configuration>
        <option name="takeFromSettings" value="true" />
        <option name="myScalaCompilerJarPaths">
          <array>
            <option value="$MODULE_DIR$/lib/scala-compiler.jar" />
          </array>
        </option>
        <option name="myScalaSdkJarPaths">
          <array>
            <option value="$MODULE_DIR$/lib/scala-library.jar" />
          </array>
        </option>
      </configuration>
    </facet>
  </component>
  <component name="NewModuleRootManager" inherit-compiler-output="true">
    <exclude-output />
    <content url="file://$MODULE_DIR$">
      <sourceFolder url="file://$MODULE_DIR$/src/main/scala" isTestSource="false" />
      <sourceFolder url="file://$MODULE_DIR$/src/test/scala" isTestSource="true" />
    </content>
    <orderEntry type="inheritedJdk" />
    <orderEntry type="sourceFolder" forTests="false" />
    <orderEntry type="library" name="scala-2.8.0.Beta1" level="application" />
    <orderEntry type="library" name="akka" level="project" />
  </component>
</module>

Now add the source and test folders in the project structure dialog:

And add the dependency to the jars in lib_managed/scala_2.8.0.Beta1/compile
If you followed the previous post, rebuild project should now complete with no errors, code completion should work and you are ready to go :)

As I said before, the IDEA scala compiler is a bit slow, so at the moment I don't compile after every change in IDEA. I keep this cool sbt feature running in a terminal:

$ sbt
$ ~test-quick

The ~ makes test-quick running as a triggered action, and compiles and runs all failed or dependent tests when any source modification is done.

I wrote a small test just to check if everything is working. (You would normally of course never put the test code and the scala code together in one file, but its simpler to read here)

The test starts a "Worker" Actor, queues a few Commands, and then checks the amount of commands that are queued in the worker, and checks if all commands are executed. I'm using the MustMatchers for the nice syntax ('must be' matcher). The red-green-refactor flow is very nice while using the ~test-quick command.

import org.scalatest.Spec
import org.scalatest.matchers.MustMatchers

import se.scalablesolutions.akka.actor.{Actor}


case class Command(name: String, data: String)

case class CountCommandsQueued()

case class Execute()

case class CountResponse(count: Int)


class Worker extends Actor {
  var commands: List[Command] = Nil

  def receive = {
    case msg: Command => {
      commands = msg :: commands
    }
    case msg: Execute => {
      for (command <- commands) {
        commands = commands.dropRight(1)
      }
    }
    case msg: CountCommandsQueued => {
      reply(new CountResponse(commands.size))
    }
  }
}

class WorkerSpec extends Spec with MustMatchers {
  describe("A Worker") {

    describe("(when it queues commands)") {
      it("should have the correct number of commands queued") {

        val command = new Command("test", "data")
        val actor = new Worker()
        actor.start
        var reply: CountResponse = (actor !! new CountCommandsQueued()).getOrElse(fail())
        reply.count must be === 0

        actor ! command
        actor ! command
        actor ! command
        reply = (actor !! new CountCommandsQueued()).getOrElse(fail())
        reply.count must be === 3
        actor.stop
      }
    }
    describe("(when it executes all queued commands)") {
      it("should have no commands queued after executing") {
        val command = new Command("test", "data")
        val actor = new Worker()
        actor.start
        actor ! command
        actor ! command
        var reply: CountResponse = (actor !! new CountCommandsQueued()).getOrElse(fail())
        reply.count must be === 2
        actor ! new Execute()
        reply = (actor !! new CountCommandsQueued()).getOrElse(fail())
        reply.count must be === 0
        actor.stop
      }
    }
  }
}

Starting with Akka and Scala

noreply@blogger.com (Raymond Roestenburg) — Sat, 17 Apr 2010 12:49:00 +0000

Update to this post: I have upgraded the examples to Akka 1.0-RC1. this newer post might be more helpful.

I recently started playing around with Scala. I had read a lot about Scala before but never got around really doing anything with it. I was looking for some (probably Java) libraries that would make a couple of things really easy:

Easy concurrency
Interop with Java or in Java
Distributed computing nodes
Event driven
Highly scalable
Low bandwidth usage for communicating between nodes (so no XML)
Pluggable data format
Easy development of message exchanges, asynchronous, one way, two way, synchronous, etc
Possible (but hopefully not necessary) distributed state/database
Reliable when failures happen
Easy integration with a range of communication protocols / endpoints

At first I found BSON as a data format, as part of the mongodb project. Quick after that I found Google Protobuf which was a better candidate for me. That lead me to JBoss Netty, which supports protobuf out of the box. Netty was a bit too low level for my needs this time and by chance I saw a tweet come by from Dean Wampler, 'Akka 0.7 released'. Didn't know what it was (Akka is apparently a mountain in Sweden) checked it out, and to my surprise Akka uses JBoss Netty and protobuf for Remote Actors! If you look at the many (pluggable) features of Akka it's easy to get enthusiastic about it very quickly. I read through the source code, which (as far as I can tell at this moment of course) looks good. Plenty of tests, well structured, easy to read. There are some smart people behind this framework with the right know-how, which is very important as well.

Akka does have a Java API, but once you see the Scala version, I don't think you will want to use it. That's got nothing to do with the quality of the Java API by the way, more the succinct way of expressing code and the amount of typing that you will save yourself from :). Well, for some years now I follow the rule to learn at least one new programming language every year, so this year I'm adding Scala to the list.

So I started with an Akka project, did some playing around with Eclipse, Intellij, maven, sbt, Scala of course, Apache Camel integration. I used the akka-template-rest template first to get started quickly. I quickly found out that the Scala Eclipse plugin is not really ready yet, so now I'm using Intellij IDEA. in the next post I'll talk about the use of an IDE.

Disclaimer: Since I'm not a Scala expert (yet :) and I don't always work on linux I just got things to work to get the job done, if you have any tips for improvement let me know. I'm using Ubuntu 9.10 Karmic Koala.

Install sbt (0.7.3):
Follow the setup here.

sbt will install scala inside the project that we will create. Create a new project by running sbt in a newly created project:

$ mkdir akka_start
$ cd akka_start
$ sbt
Project does not exist, create new project? (y/N/s) y
Name: akka_start
Organization: agilesquad
Version [1.0]: 
Scala version [2.7.7]: 2.8.0.Beta1 
sbt version [0.7.3]:

It's very important to use 2.8.0.Beta1, the RC1 still has some issues. sbt itself is running on Scala 2.7.7.

I find sbt a very nice tool so far, it is very similar to maven, can use maven dependencies, but you write your build file in Scala in an internal DSL, which is a lot nicer than XML.

Lets write a simple project file:

$ cd ~/akka_start
$ mkdir project/build
$ gedit project/build/Project.scala

Project.scala:

import sbt._

class Project(info: ProjectInfo) extends DefaultWebProject(info){
 override def repositories = Set(
  "Java.Net" at "http://download.java.net/maven/2",
  "jBoss" at "http://repository.jboss.org/maven2",
  "service mix" at "http://svn.apache.org/repos/asf/servicemix/m2-repo/",
  "Apache Camel" at "https://repository.apache.org/content/repositories/releases/",
  "Akka Maven Repository" at "http://scalablesolutions.se/akka/repository",
  "Multiverse Releases" at "http://multiverse.googlecode.com/svn/maven-repository/releases/",
  "GuiceyFruit" at "http://guiceyfruit.googlecode.com/svn/repo/releases/",
  "DataBinder" at "http://databinder.net/repo",
  "Configgy" at "http://www.lag.net/repo",
  ScalaToolsSnapshots)

 override def libraryDependencies = Set(

  /* servlet implementation */
  "org.eclipse.jetty"  % "jetty-server"   % "7.0.1.v20091125" % "test",
  "org.eclipse.jetty"  % "jetty-webapp"   % "7.0.1.v20091125" % "test",
    "org.scalatest" % "scalatest" % "1.0.1-for-scala-2.8.0.Beta1-with-test-interfaces-0.3-SNAPSHOT" % "test->default",
  /* camel */
  "org.apache.camel" % "camel-ftp" % "2.2.0" % "compile",
  /* akka dependencies */
  "se.scalablesolutions.akka" % "akka-kernel_2.8.0.Beta1"  % "0.8.1" % "compile",
  "se.scalablesolutions.akka" % "akka-core_2.8.0.Beta1"    % "0.8.1" % "compile",
  "se.scalablesolutions.akka" % "akka-servlet_2.8.0.Beta1" % "0.8.1" % "compile",
  "se.scalablesolutions.akka" % "akka-rest_2.8.0.Beta1"    % "0.8.1" % "compile")
 override def jettyPort = 9012
  val junit = "junit" % "junit" % "4.5" % "test"
}

I am not using any distributed database yet so I removed those dependencies (redis etc). I have already added some dependencies that I will use in the next post (camel-ftp2, junit, scalatest). As you can see I am using akka 0.8.1 which is dependent on Scala 2.8.0.Beta1 at the moment.

update the project dependencies:

$ cd ~/akka_start
$ sbt update

You will now find a lib_managed directory in your project directory.
Add a webapp:

$ cd ~/akka_start/src/main
$ mkdir webapp
$ mkdir webapp/WEB-INF
$ cd webapp/WEB-INF
$ gedit web.xml

web.xml:

<?xml version="1.0"?>

<!DOCTYPE web-app PUBLIC "-//Sun Microsystems, Inc.//DTD Web Application 2.3//EN"
 "http://java.sun.com/dtd/web-app_2_4.dtd">

<web-app>

 <listener>
  <listener-class>se.scalablesolutions.akka.servlet.Initializer</listener-class>
 </listener>

 <servlet>
  <servlet-name>AkkaServlet</servlet-name>
  <servlet-class>se.scalablesolutions.akka.rest.AkkaServlet</servlet-class>
 </servlet>

 <servlet-mapping>
  <servlet-name>AkkaServlet</servlet-name>
  <url-pattern>/*</url-pattern>
 </servlet-mapping>

</web-app>

Check if a WAR is being created:

$ cd ~/akka_start
$ sbt package

Add a akka.conf to the resources:

$ cd ~/akka_start/src/main/resources
$ gedit akka.conf

akka.conf:

####################
# Akka Config File #
####################

# This file has all the default settings, so all these could be removed with no visible effect.
# Modify as needed.

<log>
  filename = "./logs/akka.log"
  roll = "daily"  # Options: never, hourly, daily, sunday/monday/...
  level = "info" # Options: fatal, critical, error, warning, info, debug, trace
  console = on
  # syslog_host = ""
  # syslog_server_name = ""
</log>

<akka>
  version = "0.8.1"

  # FQN to the class doing initial active object/actor
  # supervisor bootstrap, should be defined in default constructor
  boot = ["com.agilesquad.Boot"]

  <actor>
    timeout = 5000              # default timeout for future based invocations
    serialize-messages = off    # does a deep clone of (non-primitive) messages to ensure immutability
  </actor>

  <stm>
    service = on
    fair = on                     # should transactions be fair or non-fair (non fair yield better performance)
    max-nr-of-retries = 1000      # max nr of retries of a failing transaction before giving up
    timeout = 10000               # transaction timeout; if transaction has not committed within the timeout then it is aborted
    distributed = off             # not implemented yet
  </stm>

  <remote>
    compression-scheme = "zlib" # Options: "zlib" (lzf to come), leave out for no compression
    zlib-compression-level = 6  # Options: 0-9 (1 being fastest and 9 being the most compressed), default is 6

    <cluster>
      service = on
      name = "default"                                                        # The name of the cluster
      actor = "se.scalablesolutions.akka.cluster.jgroups.JGroupsClusterActor" # FQN of an implementation of ClusterActor
      serializer = "se.scalablesolutions.akka.serialization.Serializer$Java$" # FQN of the serializer class
    </cluster>

    <server>
      service = on
      hostname = "localhost"
      port = 9999
      connection-timeout = 1000 # in millis (1 sec default)
    <server>

    <client>
      reconnect-delay = 5000    # in millis (5 sec default)
      read-timeout = 10000      # in millis (10 sec default)
    <client>
  </remote>
</akka>

In the akka.conf you need to specify your boot class, in this case com.agilesquad.Boot
(boot = ["com.agilesquad.Boot"])

Now write a very simple Actor to test if it runs (inside jetty). I have chosen to write an Actor that integrates with an Apache Camel file endpoint.

$ cd src/main/scala/
$ gedit Boot.scala

Boot.scala:

package com.agilesquad

import se.scalablesolutions.akka.actor.{Actor, SupervisorFactory}
import se.scalablesolutions.akka.camel.{Message, Consumer}
import se.scalablesolutions.akka.config.ScalaConfig._

class Boot {
   val factory = SupervisorFactory(
  SupervisorConfig(
   RestartStrategy(OneForOne, 3, 100, List(classOf[Exception])),
   Supervise(new CamelConsumer(), LifeCycle(Permanent)) :: Nil))

  factory.newInstance.start
}

class CamelConsumer extends Actor with Consumer {
  
  def endpointUri = "file:///home/rroestenburg/input"
  def receive = {
    case msg:Message => {
      log.info("received camel message...")
    }
  }
}

For simplicity I have just added all the classes in the Boot.scala file.

Now run sbt without arguments, execute compile, execute jetty-run

$ sbt 
$ compile
$ jetty-run

You should now have a running akka server! as shown above the endpointUri is pointing to an input directory of your choice (in my case /home/rroestenburg/input), copy a file in there and check if the message ends up in the logging ("received camel message).

That's it for now, in the next post I'll get into the use of Intellij IDEA Community Edition with the latest Scala Plugin for 2.8.0.Beta1, as well as some Test Driven Development with Scala and Akka, and how to use some extra Camel components.

Executing XPath from the command line

noreply@blogger.com (Raymond Roestenburg) — Fri, 16 Oct 2009 18:48:00 +0000

I had to check some large (SOAP) XML files the other day and was wondering if there was an easy command line tool to execute XPath expressions. I found xmllint in my Cygwin install, and it happened to be installed on the linux servers I was working on.

It's actually quite easy to use:
xmllint --noent --shell file.xml

That opens up a shell in which you can do many things, but at the time I was only interested in running some XPath expressions. The first thing you run into is that namespaces can get in your way with XPath. You need to register the namespaces in the XML to prefixes that can be used in the XPath context.
In the shell you can use this command to for instance register soap:
/ > setns soap=http://schemas.xmlsoap.org/soap/envelope/

after that you can run xpath expressions, some simple examples:

/ > xpath soap:Envelope
/ > xpath count(//mynamespace:myElement)

Quite a handy little tool.

Quick notes to self on java.io

noreply@blogger.com (Raymond Roestenburg) — Fri, 08 May 2009 09:20:00 +0000

Just a quick post so that I don't forget next time:

java.io.FilterOutputStream

If you extend java.io.FilterOutputStream, don't forget to override write(byte b[], int off, int len), since the default implementation falls back on looping write(byte b) which is terribly slow.
Also, when you also override write(byte[] b), make sure to call write(b,0,b.length); and not the super method:


@Override
public void write(byte[] b) throws IOException {
  write(b,0,b.length);
}

Readers, Writers, Streams, that delegate to underlying source.
Always check if the implementation of the close() method, closes the underlying source automatically. Don't assume that it always will.
XMLStreamReader for instance, doesn't close the underlying source. FilterOutputStream does. BufferedWriter does. BufferedReader does. GZIPInputStream does.
(In my opinion this should be consistent, but it is not.)

Firefox and stylesheet link mime-type

noreply@blogger.com (Raymond Roestenburg) — Wed, 08 Oct 2008 15:55:00 +0000

I found out that Firefox is a bit more strict than Internet Explorer when it comes to stylesheet links. I had one of the stylesheet files up on googlecode, in the downloads/files section. I'm hosting the blog completely on free servers, so I had to find a place to put the Syntax Highlighter stuff. For a while now I saw that the code on the blog was not shown correctly in Firefox, although it did show correctly in IE and thought 'pfff'. Until someone nagged me about it (thanks Marc).

Seams I had an error on my page, Firefox didn't like the mime-type of the file that was hosted on googlecode downloads. Reported that the mime-type was ‘text/x-c’, not ‘text/css’. Must be the default mime-type that is used to serve files from googlecode.

As far as I know, you can't control the mime-type for files in the googlecode downloads section. The solution is quite simple though, I moved the css file to the source section (svn) and added an 'svn:mime-type' property to the file. I used tortoisesvn for it, right click on the file, properties, add property with name 'svn:mime-type' and value 'text/css', commit. That way the file is served with the correct mime-type.

Note on WCF behavior extension element

noreply@blogger.com (Raymond Roestenburg) — Tue, 16 Sep 2008 07:43:00 +0000

As described here: about configuration errors when defining behavior extensions, there is a bug that makes it necessary to fully specifiy the class and assembly name in your behavior extension:


<?xml version="1.0" encoding="utf-8"?>
<!-- configuration, servicemodel etc. left out for clarity-->
<extensions>
  <behaviorExtensions>
  <!-- Because of a bug in .NET framework you need to fully specify the behavior class and assembly
-->
    <add name="springBehavior"
            type="AgileSquad.Wcf.Spring.SpringServiceBehavior, AgileSquad.Wcf.Spring, Version=1.0.0.0, Culture=neutral, PublicKeyToken=a44c692614ed00fa"/>
  </behaviorExtensions>
</extensions>

Update on WCF service dependency injection using Spring.NET

noreply@blogger.com (Raymond Roestenburg) — Sat, 13 Sep 2008 07:21:00 +0000

I have just uploaded the code that I use for DI in combination with WCF, using Spring.Net.
(better late than never!)

It's up on http://code.google.com/p/agilesquadblog/.
When you svn checkout it is in the AgileSquad.Wcf subfolder.

The instance provider itself is basically a copy/paste out of Oran Dennison's blog.
I have only changed the ReleaseInstance method to dispose of the instance (the service implementation) if it implements IDisposable:


public void ReleaseInstance(InstanceContext instanceContext, object instance)
{
  IDisposable disposable = instance as IDisposable;
  if(disposable!=null)
  {
    disposable.Dispose();
  }
}

Additional to that the source code includes a working example (visual studio 2008 solution) of the instance provider, test service and configuration.
It includes unit tests to test the spring instance provider.

It also includes some interfaces that I use to abstract the use of WCF channel factory proxies and generated client classes.

The idea behind that is that I don't want my code to reference implementation directly, cause it makes testing more difficult.

So instead of something like:


using (var generatedWcfClient = new GeneratedWcfClient("endPoint"))
{
  generatedWcfClient.SendMessage(message);
}

I use something like:


using (IServiceProxy<IMyService> proxy = factory.CreateProxy())
{
  proxy.Service.SendMessage(message);
}

In my Service Consumer code. (The factory in the above example is injected through DI of course, see the code)

That means that the code that uses the proxies can be easily configured to use mock or in-memory implementations of the interfaces. It also decouples the use of the specific solution for communicating with a service (WCF in this case). The chance that I will change to some other implementation than WCF is obviously very small. But for testing and decoupling it is great in my opinion. It also decouples the choice of using a generated client or a channel factory.

Check out the code for more details.

The code also includes a ServiceTestEnvironment class that I use to easily test WCF Services. It hosts the WCF Service in a ServiceHost, and allows the unit test to call the service using a proxy and SetUp and TearDown the "environment". In the Test project I use the above mentioned IServiceProxy and IServiceProxyFactory interfaces. The ChannelFactory implementation is used, as a generated proxy would start conflicting (ambiguous references)with the service code because you would have to reference both the client and the server in this way of testing.

Firefox

noreply@blogger.com (Raymond Roestenburg) — Sat, 13 Sep 2008 07:17:00 +0000

I just found out my blog layout looks really weird on Firefox 3. Anybody any ideas?

Ruby for building and testing C# code

noreply@blogger.com (Raymond Roestenburg) — Fri, 08 Aug 2008 12:24:00 +0000

Check out http://www.infoq.com/interviews/IronRuby-John-Lam for an interview with John Lam on IronRuby.

What is interesting (amongst other things in the interview) is the view that you could start using Ruby in testing and building applications that are not necessarily written in Ruby, meaning using RSpec and rake to build and test for instance .NET applications . Because of the integration that IronRuby has with the .NET framework, if you needed to, you could do more .NET specific things in your build or test if you needed to.

Sounds like a nice idea to build a continuous integration solution with this...

VisualSVN Server

noreply@blogger.com (Raymond Roestenburg) — Sat, 12 Jul 2008 17:48:00 +0000

After trying to get Subversion + Apache to work with Active Directory on Windows Server 2003 for quite a while, a colleague of mine pointed me at http://www.visualsvn.com/server/. It's for free and comes with a nice user friendly management console. But more importantly (for me at least) is that it supports Active Directory out of the box, which saved me a lot of time figuring out how to setup and configure those Apache auth mods!

The only thing I found strange is that the management console doesnt allow you to create repositories with a '.' in the name. Creating a repository with for instance underscores in the management console, and then renaming the repository folder on disk to use dots instead works fine though.

A simple XNA game

noreply@blogger.com (Raymond Roestenburg) — Fri, 11 Jul 2008 09:45:00 +0000

The company I work for (CSC) has updated their company logo and corporate image. Within this context I developed a little videogame with XNA with my colleague Mo'in Creemers.

Since CSC sponsors a team in the Tour de France, the marketing division thought it was a nice idea to make a videogame that uses a virtual trainer (http://www.tacx.com/), which looks like the picture here on the right. The videogame will be used for a giveaway of a racing bicycle.

First challenge we had to tackle is to communicate with the Tacx Virtual Trainer. While we were discussing reverse-engineering the device, sniffing the USB port and more 'high tech' ideas, someone had a brilliant idea: Why not just call Tacx and ask them if they can't send us the API for it?

Luckily Tacx was so kind to send us the DLL's to communicate with the virtual trainer, along with some testing tools!

One challenge down, one to go: To build a playable game in less than 2 weeks.
We still needed some cool graphics & audio and a game development environment that doesnt get in the way. And of course we still had to actually build something.

The idea was to make a topdown view 2D racer game, where you ride around a little island, pickup gems as you go along. Every picked up gem fills in a piece of the new logo. The goal of the game is to complete the logo in time. You get a score for how fast you cycle and the gems randomly multiply your existing score from 1/8 till 8 times. So your final score is actually quite random (you need to pickup at least 9 gems), that way everyone has a chance to win the giveaway bicycle.

We decided to keep it simple and use Microsoft XNA and go for 2D. XNA really is a great library for game development. It simplifies life in the right places (more on that later) and still gives you a lot of performance. We chose to use XNA Game Studio 2.0 (which installs on top of Visual Studio 2005). Game development environment, check!

Since our graphic design skills are questionable at the least, and we really didn't have time to develop the code and design graphics in 2 weeks, we used some of Lostgarden/danc's graphics (Big up Danc!)

http://lostgarden.com/2007/05/dancs-miraculously-flexible-game.html

As you can see the graphics are excellent.

The cool thing is that Danc lets you download the graphics in Microsoft Expression Design format, which is vector based. Makes it a lot easier to modify than bitmaps, and the resulting quality is a lot better too. We tweaked the graphics a bit, probably for the worse :) I took the character in the middle of the picture above, took his face off, drew a cap in its place, gave him a bicycle and some legs (or at least thats what it's supposed to look like).
Graphics, check!

Now we need to get some nice audio. The graphics are really cute and bright and we needed some audio that would suit the graphics. So the first thing that came to mind was Super Mario. downloaded some samples and created a xap file. Audio, check!

Now for development. We used a lot of the XNA creator club examples, like the SpaceWar example, Tiled Sprites, Collision Detection Series and SpriteSheet examples to get into XNA. The learning curve actually is not so steep at all. I built our own pipeline component to load the tiled layers. We use the SpriteSheet example pipeline and based our own pipelone project on it. I used Mappy (http://www.tilemap.co.uk/mappy.php) to create the layers, and exported it to text for easy loading of the 'baddies' in the game and the playing field. I linked the id's in the 2-dimensional array to the id's in the spritesheet for passive layers. I linked the id's in a interactive layer to an enum, based on which a GameObjectFactory creates interactive objects in the games (basically the bugs, the road, water, gems, rocks, dirtroad and so on).

All the interactive objects implement the following interface:


    public interface IInteractable : ITransformable, ISpriteDrawable
    {
        bool InteractWith(Player player);

        void Update(GameTime gameTime);
    }

Most of the interactors extend an AbstractSpriteGameObject class, which draws itself and delegates to a PixelCollidable class to do collision detection.

The Player class then has an InteractWith(..) for every type of object,
double dispatch is sweet. The interactive objects can update themselves on every frame. The InteractWith method implementation on the interactive objects is mostly like:


if(player.Collidable.CollidesWith(this.Collidable)
{
   // change state
   ..
   player.InteractWith(this);
   return true;
}
else
{
   // collision has stopped
   return false;
}

The interactive objects have an internal state, the Update method does different things based on that state. When there is interaction the player gets called with the interactive object so that the player can react to the situation.

The nice thing about the Tacx virtual trainer is that it can actually generate resistance, and make it harder for you to cycle. We used that in the game as well, if you stay on the road it is quite easy to cycle, and it gets harder if you go off track. when you hit a bug or a rock we increase the resistance exponentially for a short period for extra effect.

Here are some screenshots from the final result:

In the bottom of the screen is the new CSC logo that you need to complete, by picking up the yellow gems. The bugs will try and interfere with you.

We put a layer of clouds over all the layers to create a bit of an extra effect. If you fall off the island (by basically riding into the blue tiles) you drown and start over again. (By the way, the score stays zero on these screens cause at the time of creating the screenshots I didn't have the tacx vr bicycle attached. Riding on the street is easier than riding on the dirtroad, or the grass next to it.

You obviously have to try to stay on the road and try not to bump into the rocks and the bugs on the road..
In this case I'm picking up a gem and increasing my score times 8.

As you can see you can zoom out while you are playing (with the controls on the handle bar of the bicycle) to help find the gems on the island. Once you have picked up enough gems to complete the logo, the Logo zooms in to the middle of the screen and you can fill in your name on the high-scores.

No WOW for WOA

noreply@blogger.com (Raymond Roestenburg) — Sun, 01 Jun 2008 18:33:00 +0000

I was at an architect community meeting when I first heard of the term WOA, and later looked up some references on the internet. I am not sure who coined the term and why. WOA stands for Web Oriented Architecture. The first couple of questions that came to mind were "Sounds like SOA, what's the difference?" and "Do we really need another acronym?". After googling a bit (for instance http://www.ddj.com/web-development/207600577) I found the following:
"Web-Oriented Architecture is an alternative to Service-Oriented Architecture. WOA is fundamentally about "open" systems that communicate over "open" and established protocols and formats."
Hm. SOA uses open formats, and services are open, use of established protocols and formats are fine as well. Why is WOA an "alternative"?

"In WOA, HTTP takes center-stage as the communication protocol between applications and web-services. Beyond that, the exchange format is also web-centric and is often XHTML, RSS, RDF, or other such formats."

Hm. Ok fine. you can have a SOA with HTTP on the center stage. You could have CORBA for that matter. SOA does not say that HTTP is excluded from the protocols it can work with. SOA does not say that RSS is not allowed or something like that.

Maybe a good time to try and explain what I understand as SOA. At this point I was almost going to try and write it down, but instead of repeating what is already on the web, I'll just refer you to the OASIS reference model of SOA: http://www.oasis-open.org/committees/download.php/18486/pr-2changes.pdf.

By the way, RSS is actually a very good example of a service implementation if you look at the OASIS paper of what a service is.

"In WOA, web-services embrace the same technological design principles that web-applications have been following for years: communication between services is stateless, formats are open and extensible, and data exchanges are cacheable. All desirable properties for creating large scale, ad-hoc networks of connected, heterogenous web-services. "

That just sounds like really good things to do when you design services. No difference again here with SOA. SOA says nothing about not-caching, statefull, closed formats, not-extensible. A lot of the articles floating around try to make a connection between SOA and SOAP. Although there is alot implemented today with SOAP, this is not a prerequisite for SOA.

Then I read the following:
http://weblog.infoworld.com/realworldsoa/archives/2008/04/why_woasoa_is_m.html

"SOA should extend out of the firewall to the Internet, if you're to provide real value to your enterprise. However, this was not universally accepted by the rank-and-file SOA guys. Indeed, generally speaking most viewed SOA as something that occurred within the firewall exclusively, and extending the reach of their SOA to Internet-based resources was taboo."

That is opinion, not definition. How you do your security is implementation detail. (Obviously you will have far different requirements in a SOA environment)
From the OASIS reference model:

"Service Oriented Architecture (SOA) is a paradigm for organizing and utilizing distributed capabilities that may be under the control of different ownership domains."

Sounds like cross firewall to me.

"Not sure if anyone is selling this as a replacement to SOA, or traditional enterprise architecture, but it's really an approach to architecture where there is a core acceptance that Web-based resources may provide the most speed to delivery, the largest number of resources, and with minimum amount cost. That's it."

And in my opinion, all of that is a narrowing down of a specific implementation of a service oriented architecture style that doesnt really add more meaning, at least not enough for me to start using the term.

Configure WCF service dependency injection using Spring.NET

noreply@blogger.com (Raymond Roestenburg) — Fri, 23 Nov 2007 10:16:00 +0000

I am a big fan of Spring, Spring.NET and Dependency Injection, specifically DI that can be done transparantly to the objects it configures.

I am using Spring.NET dependency injection in my WCF services as described by Oran Dennison:
http://orand.blogspot.com/2006/10/wcf-service-dependency-injection.html

I only didn't like to 'take the ServiceHost route' that Oran took, and found out it is quite easy to 'take the configuration route':


<?xml version="1.0" encoding="utf-8"?>
<configuration>
  <!-- Put the necessary Spring.NET config here
  --> 
  <system.serviceModel>
    <services>
      <service behaviorConfiguration="InjectionBehavior"
               name="AgileSquad.Service.MyService">
        <endpoint address="http://localhost/MyService"
                  binding="basicHttpBinding"
                  name="basicProfileBinding"
                  contract="AgileSquad.Service.IMyServiceInterface" />
      </service>
    </services>
    <behaviors>
      <serviceBehaviors>
        <behavior name="InjectionBehavior">
          <springBehavior/>
        </behavior>
      </serviceBehaviors>
    </behaviors>
    <!-- Behavior extensions -->
    <extensions>
      <behaviorextensions>
        <!-- springBehavior extension
             for injecting using Spring using
             custom WCF InstanceProvider
        -->
        <add name="springBehavior"
             type="AgileSquad.Wcf.SpringServiceBehavior,AgileSquad.Wcf,
             Version=1.0.0.0, Culture=neutral, PublicKeyToken=mykeytoken"/>
        </behaviorextensions>
    </extensions>
  </system.serviceModel>
</configuration>

The behaviorConfiguration attribute on the service element points to a behavior configuration,
found in the behaviors element.
You add your own behavior extension through WCF's extensions mechanism. My service behavior implementation is strongly named, so replace the PublicKeyToken with a real keytoken instead of 'mykeytoken', and obviously you would still have to add your Spring.NET configuration as described in Oran's post.