RPC Server for Erlang, In Scala

There has been Java code in my previous blog: RPC Server for Erlang, In Java, I’m now try to rewrite it in Scala. With the pattern match that I’ve been familiar with in Erlang, write the Scala version is really a pleasure. You can compare it with the Java version.

I do not try Scala’s actor lib yet, maybe late.

And also, I should port Erlang’s jinterface to Scala, since OtpErlangTuple, OtpErlangList should be written in Scala’s Tuple and List.

The code is auto-formatted by NetBeans’ Scala plugin, and the syntax highlighting is the same as in NetBeans, oh, not exactly.

/* * RpcMsg.scala * */
package net.lightpole.rpcnode import com.ericsson.otp.erlang.{OtpErlangAtom, OtpErlangList, OtpErlangObject, OtpErlangPid, OtpErlangRef, OtpErlangTuple} class RpcMsg(val call:OtpErlangAtom, val mod :OtpErlangAtom, val fun :OtpErlangAtom, val args:OtpErlangList, val user:OtpErlangPid, val to :OtpErlangPid, val tag :OtpErlangRef) {
} object RpcMsg { def apply(msg:OtpErlangObject) : Option[RpcMsg] = msg match { case tMsg:OtpErlangTuple => tMsg.elements() match { /* {'$gen_call', {To, Tag}, {call, Mod, Fun, Args, User}} */ case Array(head:OtpErlangAtom, from:OtpErlangTuple, request:OtpErlangTuple) => if (head.atomValue.equals("$gen_call")) { (from.elements, request.elements) match { case (Array(to :OtpErlangPid, tag:OtpErlangRef), Array(call:OtpErlangAtom, mod :OtpErlangAtom, fun :OtpErlangAtom, args:OtpErlangList, user:OtpErlangPid)) => if (call.atomValue.equals("call")) { Some(new RpcMsg(call, mod, fun, args, user, to, tag)) } else None case _ => None } } else None case _ => None } case _ => None }
}

/* * RpcNode.scala * * To change this template, choose Tools | Template Manager * and open the template in the editor. */
package net.lightpole.rpcnode import com.ericsson.otp.erlang.{OtpAuthException, OtpConnection, OtpErlangAtom, OtpErlangExit, OtpErlangObject, OtpErlangString, OtpErlangTuple, OtpSelf}
import java.io.IOException
import java.net.InetAddress
import java.net.UnknownHostException
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import java.util.logging.Level
import java.util.logging.Logger trait Cons { val OK = new OtpErlangAtom("ok") val ERROR = new OtpErlangAtom("error") val STOPED = new OtpErlangAtom("stoped") val THREAD_POOL_SIZE = 100
} /** * * Usage: * $ erl -sname clientnode -setcookie mycookie * (clientnode@cmac)> rpc:call(xnodename@cmac, xnode, parse, []). * * @author Caoyuan Deng */
abstract class RpcNode(xnodeName:String, cookie:String, threadPoolSize:Int) extends Cons { def this(xnodeName:String, cookie:String) = this(xnodeName, cookie, 100) private var xSelf:OtpSelf = _ private var sConnection:OtpConnection = _ private var execService:ExecutorService = Executors.newFixedThreadPool(threadPoolSize) private val flags = Array(0) startServerConnection(xnodeName, cookie) loop def startServerConnection(xnodeName:String, cookie:String ) = { try { xSelf = new OtpSelf(xnodeName, cookie); // The node then publishes its port to the Erlang Port Mapper Daemon. // This registers the node name and port, making it available to a remote client process. // When the port is published it is important to immediately invoke the accept method. // Forgetting to accept a connection after publishing the port would be the programmatic // equivalent of false advertising val registered = xSelf.publishPort(); if (registered) { System.out.println(xSelf.node() + " is ready."); /** * Accept an incoming connection from a remote node. A call to this * method will block until an incoming connection is at least * attempted. */ sConnection = xSelf.accept(); } else { System.out.println("There should be an epmd running, start an epmd by running 'erl'."); } } catch { case ex:IOException => case ex:OtpAuthException => } } def loop : Unit = { try { val msg = sConnection.receive val task = new Runnable() { override def run = RpcMsg(msg) match { case None => try { sConnection.send(sConnection.peer.node, new OtpErlangString("unknown request")); } catch { case ex:IOException => } case Some(call) => val t0 = System.currentTimeMillis flag(0) = processRpcCall(call) System.out.println("Rpc time: " + (System.currentTimeMillis() - t0) / 1000.0) } } execService.execute(task) if (flag(0) == -1) { System.out.println("Exited") } else loop } catch { case ex:IOException => loop case ex:OtpErlangExit => case ex:OtpAuthException => } } /** @throws IOException */ def sendRpcResult(call:RpcMsg, head:OtpErlangAtom, result:OtpErlangObject) = { val tResult = new OtpErlangTuple(Array(head, result)) // Should specify call.tag here val msg = new OtpErlangTuple(Array(call.tag, tResult)) // Should specify call.to here sConnection.send(call.to, msg, 1024 * 1024 * 10) } /** @abstact */ def processRpcCall(call:RpcMsg) : Int
} object RpcCall { def getShortLocalHost : String = getLocalHost(false) def getLongLocalHost : String = getLocalHost(true) def getLocalHost(longName:Boolean) : String = { var localHost = "localhost" try { localHost = InetAddress.getLocalHost.getHostName; if (!longName) { /* Make sure it's a short name, i.e. strip of everything after first '.' */ val dot = localHost.indexOf(".") if (dot != -1) localHost = localHost.substring(0, dot) } } catch { case ex:UnknownHostException => } localHost }
}

Categories: Blogs  Caoyuan Blog  

Comments

Add comment