Akka AMQP RPC example

gistfile1.scala

import akka.amqp.AMQP.{ToBinary, FromBinary}
import rpc.RPC.{RpcClientSerializer, RpcServerSerializer}
import rpc.RPC
import akka.actor.ActorRegistry

object RPCExample {

  def main(args: Array[String]) {
    // shared between client and server
    val connection = AMQP.newConnection()
    val exchangeName = "my_rpc_exchange"
    val routingKey = Some("rpc.in.key");

    // simple server request handler sending back the length of the string request
    def requestHandler(request: String): Int = {
      request.length
    }

    // start the rpc server
    val rpcServer = RPC.newRpcServer(connection, exchangeName, rpcServerSerializer, requestHandler _, routingKey,
      queueName = Some("rpc.in.key.queue"))

    // start the rpc client
    val rpcClient = RPC.newRpcClient(connection, exchangeName, rpcClientSerializer, routingKey)

    // start calling the server via the client and get some response
    val response = rpcClient.call("rpc_request")
    println("Response: " + response)

    AMQP.shutdownAll
    ActorRegistry.shutdownAll
    System.exit(0)
  }


  /** client serializer to send string and receive int */
  val clientToBinary = new ToBinary[String] {
    def toBinary(t: String) = t.getBytes
  }
  val clientFromBinary = new FromBinary[Int] {
    def fromBinary(bytes: Array[Byte]) = bytes.head.toInt
  }
  val rpcClientSerializer = new RpcClientSerializer[String, Int](clientToBinary, clientFromBinary)

  /** server serializer to receive string and send int */
  val serverFromBinary = new FromBinary[String] {
    def fromBinary(bytes: Array[Byte]) = new String(bytes)
  }
  val serverToBinary = new ToBinary[Int] {
    def toBinary(t: Int) = Array(t.toByte)
  }
  val rpcServerSerializer = new RpcServerSerializer[String, Int](serverFromBinary, serverToBinary)
}