Posts for the month of October 2008

Install Scala Plugin for NetBeans 6.5 RC2

>>> Updated Nov 8
Content of this blog is out of date, please look at New Scala Plugin for NetBeans 6.5 RC2 Is Packed
===

NetBeans 6.5 RC2 released. The Scala plugin on NetBeans' Plugins Portal is not compilable with RC2. To get Scala plugin working with RC2, do:

# Open NetBeans 6.5 RC2, go to "Tools" -> "Plugins", check "Setting" -> "Add", add new update center as "Last Development Build" with url: http://deadlock.netbeans.org/hudson/job/nbms-and-javadoc/lastStableBuild/artifact/nbbuild/nbms/updates.xml.gz

# Then in the "Available Plugins" tab, you can find the "Scala" category (or, you can click on "Name" in "Available Plugins" tab to find them. You may need to click "Reload Catalog" to get the latest available modules), check "Scala Kit" and click "Install", following the instructions. Restart IDE.

I'll re-pack a new version of Scala plugin for Plugins Portal when NetBeans 6.5 is officially released.

Scala for NetBeans Screenshot#14: Refined Color Theme

With more Scala coding experience, I refined color theme of Scala plugin for NetBeans. By desalinating Type name a bit, I found I can concentrate on logic a bit better. And all function calls are highlighted now, so, for multiple-line expression, when error-prone op is put on wrong new line, you can get some hints at once. It also gives you hint if a val/var is with implicit calling, which will be highlighted as a function call.

There are still some bugs when infer var/val's correct type in some cases.

Now the editor is much informative with highlighting and bubble pop display of type information (move mouse on it with CTRL/COMMAND pressed).

You'll need to update to newest 1.9.0 Editing module, please wait for it appealing on Development Update Center.

Click on the picture to enlarge it

nn

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
   }
}

FOR, WHILE Is Too Easy, Let's Go Looping

With several 10k code in Erlang, I'm familiar with functional style coding, and I found I can almost rewrite any functions in Erlang to Scala, in syntax meaning.

Now, I have some piece of code written in Java, which I need to translate them to Scala. Since "for", "while", or "do" statement is so easy in Java, I can find a lot of them in Java code. The problem is, should I keep them in the corresponding "for", "while", "do" in Scala, or, as what I do in Erlang, use recursive function call, or, "loop"?

I sure choose to loop, and since Scala supports recursive function call on functions defined in function body (Erlang doesn't), I choose define these functions' name as "loop", and I tried to write code let "loop" looks like a replacement of "for", "while" etc.

Here's a piece of code that is used to read number string and convert to double, only piece of them.

The Java code:

public class ReadNum {

    private double readNumber(int fstChar, boolean isNeg) {
        StringBuilder out = new StringBuilder(22);
        out.append(fstChar);
        
        double v = '0' - fstChar;
        // the maxima length of number stirng won't exceed 22
        for (int i = 0; i < 22; i++) {
            int c = getChar();
            switch (c) {
                case '0':
                case '1':
                case '2':
                case '3':
                case '4':
                case '5':
                case '6':
                case '7':
                case '8':
                case '9':
                    v = v * 10 - (c - '0');
                    out.append(c);
                    continue;
                case '.':
                    out.append('.');
                    return readFrac(out, 22 - i);
                case 'e':
                case 'E':
                    out.append(c);
                    return readExp(out, 22 - i);
                default:
                    if (c != -1) backup(1);
                    if (!isNeg) return v; else return -v
            }
        }
        return 0;
    }
}

The Scala code:

class ReadNum {
   private
   def readNumber(fstChar:Char, isNeg:Boolean) :Double = {
      val out = new StringBuilder(22)
      out.append(fstChar)

      val v:Double = '0' - fstChar
      def loop(c:Char, v:Double, i:Int) :Double = c match {
         // the maxima length of number stirng won't exceed 22
         case _ if i > 21 =>
            0
         case '0' | '1' | '2' | '3' | '4' | '5' | '6' | '7' | '8' | '9' =>
            out.append(c)
            val v1 = v * 10 - (c - '0')
            loop(getChar, v1, i + 1)
         case '.' =>
            out.append('.')
            readFrac(out, 22 - i)
         case 'e' | 'E' =>
            out.append(c)
            readExp(out, 22 - i)
         case _ =>
            if (c != -1) backup(1)
            if (isNeg) v else -v
      }; loop(getChar, v, 1)
   }
}

As you can see in line 25, the loop call is put at the position immediately after the "loop" definition, following "}; ", I don't put it to another new line, it makes me aware of the "loop" function is just used for this call.

And yes, I named all these embedded looping function as "loop", every where.

An Example Syntax in Haskell, Erlang and Scala

>>> Updated Oct 16:

I found some conventions of coding style make code more readable for Scala. For example, use <pre>{ x => something } instead of (x => dosomething)</pre> <p>for anonymous function; Use x, y, z as the names of arguments of anonymous functions; Put all modifiers to the ahead line etc. That makes me can pick up these functions by eyes quickly.</br>

======

It's actually my first time to write true Scala code, sounds strange? Before I write Scala code, I wrote a Scala IDE first, and am a bit familiar with Scala syntax now. And I've got about 1.5 year experience on Erlang, it began after I wrote ErlyBird.

Now it's time to write some real world Scala code, I choose to port  Paul R. Brown's perpubplat blog engine, which is written in Haskell. And I have also some curiosities on how the syntax looks in Erlang, so I tried some Erlang code too.

Here's some code piece of entry module in Haskell, Erlang and Scala:

Original Haskell code piece

empty :: Model
empty = Model M.empty M.empty M.empty [] 0

build_model :: [Item] -> Model
build_model [] = empty
build_model items = Model (map_by permatitle sorted_items)
                    bid                    
                    (build_child_map sorted_items)
                    (sorted_items)
                    (n+1)
    where
      sorted_items = sort_by_created_reverse items
      bid = (map_by internal_id sorted_items)
      n = fst . M.findMax $ bid

build_child_map :: [Item] -> M.Map Int [Int]
build_child_map i = build_child_map_ (M.fromList $ (map (\x -> (internal_id x,[])) i)) i

-- Constructed to take advantage of the input being in sorted order.
build_child_map_ :: M.Map Int [Int] -> [Item] -> M.Map Int [Int]
build_child_map_ m [] = m
build_child_map_ m (i:is) = if (parent i == Nothing) then
                                build_child_map_ m is
                            else
                                build_child_map_ (M.insertWith (++) (unwrap $ parent i) [internal_id i] m) is

sort_by_created_reverse :: [Item] -> [Item]
sort_by_created_reverse = sortBy created_sort_reverse

created_sort_reverse :: Item -> Item -> Ordering
created_sort_reverse a b = compare (created b) (created a)

In Erlang:

% @spec empty :: Model
empty() -> #model{}.

% @spec build_model :: [Item] -> Model
build_model([]) -> empty();
build_model(Is) -> 
    SortedIs = sort_by_created_reverse(Is),
    Bid = dict:from_list([{I#item.internal_id, I} || I <- SortedIs]),
    N = lists:max(dict:fetch_keys(Bid)),
    
    #model{by_permatitle = dict:from_list([{X#item.permatitle, X} || X <- SortedIs]),
           by_int_id = Bid,               
           child_map = build_child_map(SortedIs),
           all_items = SortedIs,
           next_id = N + 1}.


% @spec build_child_map :: [Item] -> M.Map Int [Int]
build_child_map(Is) -> build_child_map_(dict:from_list(lists:map(fun (X) -> {X#item.internal_id, []} end), Is), Is).

%% Constructed to take advantage of the input being in sorted order.
% @spec build_child_map_ :: M.Map Int [Int] -> [Item] -> M.Map Int [Int]
build_child_map_(D, []) -> D;
build_child_map_(D, [I|Is]) -> 
    case I#item.parent of 
        undefined ->                
            build_child_map_(D, Is);
        P_Id ->
            build_child_map_(dict:append(unwrap(P_Id), I#item.internal_id, D), Is)
    end.

% @spec sort_by_created_reverse :: [Item] -> [Item]
sort_by_created_reverse(Is) -> lists:sort(fun created_sort_reverse/2, Is).

% @spec created_sort_reverse :: Item -> Item -> Ordering
created_sort_reverse(A, B) -> compare(B#item.created, A#item.created).

In Scala

object Entry {
    def empty = new Model()

    def build_model(is:List[Item]) = is match {
        case Nil => empty
        case _ =>
            val sortedIs = sortByCreatedReverse(is)
            val bid = Map() ++ sortedIs.map{ x => (x.internalId -> x) }
            val n = bid.keys.toList.sort{ (x, y) => x > y }.head // max

            new Model(Map() ++ sortedIs.map{ x => (x.permatitle -> x) },
                      bid,
                      buildChildMap(sortedIs),
                      sortedIs,
                      n + 1)
    }

    def buildChildMap(is:List[Item]) = buildChildMap_(Map() ++ is.map{ x => (x.internalId -> Nil) }, is)

    private
    def buildChildMap_(map:Map[Int, List[Int]], is:List[Item]) = {
        map ++ (for (i <- is if i.parent.isDefined; pid = i.parent.get; cids = map.getOrElse(pid, Nil)) 
                yield (pid -> (cids + i.internalId)))
    }

    def sortByCreatedReverse(is:List[Item]) = is.sort{ (x, y) => x.created before y.created }
}

>>> Updated Oct 16:

Per Martin's suggestion, the above code can be written more Scala style (the reasons are in the comments). Thanks, Martin.

object Entry {
   def empty = new Model()

   def build_model(is:List[Item]) = is match {
       case Nil => empty
       case _ =>
           val sortedIs = sortByCreatedReverse(is)
           val bid = Map() ++ sortedIs.map{ x => (x.internalId -> x) }
           // use predefined max in Iterable
           val n = Iterable.max(bid.keys.toList)   

           new Model(Map() ++ sortedIs.map{ x => (x.permatitle -> x) },
                     bid,
                     buildChildMap(sortedIs),
                     sortedIs,
                     n + 1)
   }

   // you can use a wildcard anonymousfunction here
   def buildChildMap(is:List[Item]) = buildChildMap_(Map() ++ is.map(_.internalId -> Nil), is)

   private
   def buildChildMap_(map:Map[Int, List[Int]], is:List[Item]) =
       map ++ {  // rewrite for so that definitions go into body -- it's more efficient.
           for (i <- is if i.parent.isDefined) yield {
               val pid = i.parent.get
               val cids = map.getOrElse(pid, Nil)
               pid -> (cids + i.internalId)
           }
       }
       
   // you can use a wildcard anonymous function here
   def sortByCreatedReverse(is:List[Item]) = is.sort{ _.created before _.created } 
}

======

I use ErlyBird for Erlang coding, and Scala for NetBeans for Scala coding. The experience seems that IDE is much aware of Scala, and I can get the typing a bit faster than writing Erlang code.

If you are not familiar with all these 3 languages, which one looks more understandable?

RPC Server for Erlang, In Java

We are using Erlang to do some serious things, one of them is indeed part of a banking system. Erlang is a perfect language in concurrent and syntax (yes, I like its syntax), but lacks static typing (I hope new added -spec and -type attributes may be a bit helping), and, is not suitable for processing massive data (performance, memory etc). I tried parsing a 10M size XML file with xmerl, the lib for XML in OTP/Erlang, which causes terrible memory disk-swap and I can never get the parsed tree out.

It's really a need to get some massive data processed in other languages, for example, C, Java etc. That's why I tried to write RPC server for Erlang, in Java.

There is a jinterface lib with OTP/Erlang, which is for communication between Erlang and Java. And there are docs for how to get it to work. But, for a RPC server that is called from Erlang, there are still some tips for real world:

1. When you send back the result to caller, you need set the result as a tuple, with caller's tag Ref as the first element, and the destination should be the caller's Pid. It's something like:

OtpErlangTuple msg = new OtpErlangTuple(new OtpErlangObject[] {call.tag, tResult});
sConnection.send(call.to, msg); 

where, call.tag is a OtpErlangRef?, and tResult can be any OtpErlangObject?, call.to is a OtpErlangPid?.

2. If you need to send back a massive data back to caller, the default buffer size of OtpErlangOutputStream? is not good, I set it to 1024 * 1024 * 10

3. Since there may be a lot of concurrent callers call your RPC server, you have to consider the concurrent performance of your server, I choose using thread pool here.

The RPC server in Java has two class, RpcNode?.java, and RpcMsg?.java:

package net.lightpole.rpcnode;

import com.ericsson.otp.erlang.OtpErlangAtom;
import com.ericsson.otp.erlang.OtpErlangList;
import com.ericsson.otp.erlang.OtpErlangObject;
import com.ericsson.otp.erlang.OtpErlangPid;
import com.ericsson.otp.erlang.OtpErlangRef;
import com.ericsson.otp.erlang.OtpErlangTuple;

/**
 *
 * @author Caoyuan Deng
 */
public class RpcMsg {

    public OtpErlangAtom call;
    public OtpErlangAtom mod;
    public OtpErlangAtom fun;
    public OtpErlangList args;
    public OtpErlangPid user;
    public OtpErlangPid to;
    public OtpErlangRef tag;

    public RpcMsg(OtpErlangTuple from, OtpErlangTuple request) throws IllegalArgumentException {
        if (request.arity() != 5) {
            throw new IllegalArgumentException("Not a rpc call");
        }

        /* {call, Mod, Fun, Args, userPid} */
        if (request.elementAt(0) instanceof OtpErlangAtom && ((OtpErlangAtom) request.elementAt(0)).atomValue().equals("call") &&
                request.elementAt(1) instanceof OtpErlangAtom &&
                request.elementAt(2) instanceof OtpErlangAtom &&
                request.elementAt(3) instanceof OtpErlangList &&
                request.elementAt(4) instanceof OtpErlangPid &&
                from.elementAt(0) instanceof OtpErlangPid &&
                from.elementAt(1) instanceof OtpErlangRef) {

            call = (OtpErlangAtom) request.elementAt(0);
            mod = (OtpErlangAtom) request.elementAt(1);
            fun = (OtpErlangAtom) request.elementAt(2);
            args = (OtpErlangList) request.elementAt(3);
            user = (OtpErlangPid) request.elementAt(4);
            to = (OtpErlangPid) from.elementAt(0);
            tag = (OtpErlangRef) from.elementAt(1);

        } else {
            throw new IllegalArgumentException("Not a rpc call.");
        }
    }

    /* {'$gen_call', {To, Tag}, {call, Mod, Fun, Args, User}} */
    public static RpcMsg tryToResolveRcpCall(OtpErlangObject msg) {
        if (msg instanceof OtpErlangTuple) {
            OtpErlangTuple tMsg = (OtpErlangTuple) msg;
            if (tMsg.arity() == 3) {
                OtpErlangObject[] o = tMsg.elements();
                if (o[0] instanceof OtpErlangAtom && ((OtpErlangAtom) o[0]).atomValue().equals("$gen_call") &&
                        o[1] instanceof OtpErlangTuple && ((OtpErlangTuple) o[1]).arity() == 2 &&
                        o[2] instanceof OtpErlangTuple && ((OtpErlangTuple) o[2]).arity() == 5) {
                    OtpErlangTuple from = (OtpErlangTuple) o[1];
                    OtpErlangTuple request = (OtpErlangTuple) o[2];

                    try {
                        return new RpcMsg(from, request);
                    } catch (IllegalArgumentException ex) {
                        ex.printStackTrace();
                    }
                }
            }
        }
        
        return null;
    }
}
package net.lightpole.rpcnode;

import com.ericsson.otp.erlang.OtpAuthException;
import com.ericsson.otp.erlang.OtpConnection;
import com.ericsson.otp.erlang.OtpErlangAtom;
import com.ericsson.otp.erlang.OtpErlangExit;
import com.ericsson.otp.erlang.OtpErlangObject;
import com.ericsson.otp.erlang.OtpErlangString;
import com.ericsson.otp.erlang.OtpErlangTuple;
import com.ericsson.otp.erlang.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;

/**
 *
 * Usage:
 *   $ erl -sname clientnode -setcookie mycookie
 *   (clientnode@cmac)> rpc:call(xnodename@cmac, 'System', currentTimeMillis, []).
 * 
 * @author Caoyuan Deng
 */
public abstract class RpcNode {

    public static final OtpErlangAtom OK = new OtpErlangAtom("ok");
    public static final OtpErlangAtom ERROR = new OtpErlangAtom("error");
    public static final OtpErlangAtom STOPED = new OtpErlangAtom("stoped");
    private static final int THREAD_POOL_SIZE = 100;
    private OtpSelf xSelf;
    private OtpConnection sConnection;
    private ExecutorService execService;

    public RpcNode(String xnodeName, String cookie) {
        this(xnodeName, cookie, THREAD_POOL_SIZE);
    }

    public RpcNode(String xnodeName, String cookie, int threadPoolSize) {
        execService = Executors.newFixedThreadPool(threadPoolSize);

        startServerConnection(xnodeName, cookie);
        loop();
    }

    private void startServerConnection(String xnodeName, String cookie) {
        try {
            xSelf = new OtpSelf(xnodeName, cookie);
            boolean 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 (IOException ex) {
            Logger.getLogger(RpcNode.class.getName()).log(Level.SEVERE, null, ex);
        } catch (OtpAuthException ex) {
            Logger.getLogger(RpcNode.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    private void loop() {
        while (true) {
            try {
                final int[] flag = {0};

                final OtpErlangTuple msg = (OtpErlangTuple) sConnection.receive();

                Runnable task = new Runnable() {

                    public void run() {
                        RpcMsg call = RpcMsg.tryToResolveRcpCall(msg);

                        if (call != null) {
                            long t0 = System.currentTimeMillis();

                            flag[0] = processRpcCall(call);

                            System.out.println("Rpc time: " + (System.currentTimeMillis() - t0) / 1000.0);
                        } else {
                            try {
                                sConnection.send(sConnection.peer().node(), new OtpErlangString("unknown request"));
                            } catch (IOException ex) {
                                Logger.getLogger(RpcNode.class.getName()).log(Level.SEVERE, null, ex);
                            }
                        }
                    }
                };

                execService.execute(task);

                if (flag[0] == -1) {
                    System.out.println("Exited");
                    break;
                }

            } catch (OtpErlangExit ex) {
                Logger.getLogger(RpcNode.class.getName()).log(Level.SEVERE, null, ex);
            } catch (IOException ex) {
                Logger.getLogger(RpcNode.class.getName()).log(Level.SEVERE, null, ex);
            } catch (OtpAuthException ex) {
                Logger.getLogger(RpcNode.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

    protected void sendRpcResult(RpcMsg call, OtpErlangAtom head, OtpErlangObject result) throws IOException {
        OtpErlangTuple tResult = new OtpErlangTuple(new OtpErlangObject[] {head, result});

        // Should specify call.tag here
        OtpErlangTuple msg = new OtpErlangTuple(new OtpErlangObject[]{call.tag, tResult});
        // Should specify call.to here
        sConnection.send(call.to, msg, 1024 * 1024 * 10); 
    }

    public abstract int processRpcCall(RpcMsg call);
    

    // ------ helper
    public static String getShortLocalHost() {
        return getLocalHost(false);
    }

    public static String getLongLocalHost() {
        return getLocalHost(true);
    }

    private static String getLocalHost(boolean longName) {
        String localHost;
        try {
            localHost = InetAddress.getLocalHost().getHostName();
            if (!longName) {
                /* Make sure it's a short name, i.e. strip of everything after first '.' */
                int dot = localHost.indexOf(".");
                if (dot != -1) {
                    localHost = localHost.substring(0, dot);
                }
            }
        } catch (UnknownHostException e) {
            localHost = "localhost";
        }

        return localHost;
    }
}

As you can see, the RpcNode? is an abstract class, by implement int processRpcCall(RpcMsg? call), you can get your what ever wanted features. For example:

package net.lightpole.xmlnode;

import basexnode.Main;
import com.ericsson.otp.erlang.OtpErlangAtom;
import com.ericsson.otp.erlang.OtpErlangList;
import com.ericsson.otp.erlang.OtpErlangObject;
import com.ericsson.otp.erlang.OtpErlangString;
import java.io.IOException;
import net.lightpole.rpcnode.RpcMsg;
import net.lightpole.rpcnode.RpcNode;

/**
 *
 * @author dcaoyuan
 */
public class MyNode extends RpcNode {

    public MyNode(String xnodeName, String cookie, int threadPoolSize) {
        super(xnodeName, cookie, threadPoolSize);
    }

    @Override
    public int processRpcCall(RpcMsg call) {
        final String modStr = call.mod.atomValue();
        final String funStr = call.fun.atomValue();
        final OtpErlangList args = call.args;

        try {
            OtpErlangAtom head = ERROR;
            OtpErlangObject result = null;

            if (modStr.equals("xnode") && funStr.equals("stop")) {
                head = OK;
                sendRpcResult(call, head, STOPED);
                return -1;
            }

            if (modStr.equals("System") && funStr.equals("currentTimeMillis")) {
                head = OK;
                long t = System.currentTimeMillis();
                result = new OtpErlangLong(t);
            } else {
                result = new OtpErlangString("{undef,{" + modStr + "," + funStr + "}}");
            }

            if (result == null) {
                result = new OtpErlangAtom("undefined");
            }

            sendRpcResult(call, head, result);
        } catch (IOException ex) {
            ex.printStackTrace();
        } catch (Exception ex) {
        }

        return 0;
    }
}

I tested MyNode? by:

$ erl -sname clientnode -setcookie mycookie
...
(clientnode@cmac)> rpc:call(xnodename@cmac, 'System', currentTimeMillis, []).

And you can try to test its concurrent performance by:

%% $ erl -sname clientnode -setcookie mycookie
%% > xnode_test:test(10000)
-module(xnode_test).

-export([test/1]).

test(ProcN) ->
    Workers = [spawn_worker(self(), fun rpc_parse/1, {})        
               || I <- lists:seq(0, ProcN - 1)],
    Results = [wait_result(Worker) || Worker <- Workers].

rpc_parse({}) ->
    rpc:call(xnodename@cmac, 'System', currentTimeMillis, []).

spawn_worker(Parent, F, A) ->
    erlang:spawn_monitor(fun() -> Parent ! {self(), F(A)} end).

wait_result({Pid, Ref}) ->
    receive
        {'DOWN', Ref, _, _, normal} -> receive {Pid, Result} -> Result end;
        {'DOWN', Ref, _, _, Reason} -> exit(Reason)
    end.

I spawned 10000 calls to it, and it run smoothly.

I'm also considering to write a more general-purpose RPC server in Java, which can dynamically call any existed methods of Java class.