rjr

RJR uses eventmachine on the backend to process incoming messages over tcp/amqp/http/websockets/etc and dispatch json-rpc requests to registered handlers. It also employs a built in thread pool to hand off requests to the dispatcher so that the reactor isn't blocked. What results from combining the reactor pattern with a thread pool is a highly reliable concurrent event processing system.

The reactor central to eventmachine and the design pattern itself processes events one after another in a serial manner. Events are executed in the order that they arrive and no event executes until the one before it completes. Events can come from multiple sources and the reactor will take care of the serialization necessary to ensure data integrity.

A developer initializes the necessary connections and receives events by registering callbacks to be triggered on their invocation. Since the reactor blocks on any operation, the developer needs to ensure that his/her callbacks execute quickly and return control back to the reactor to continue processing events. Threads can be used to accomplish this, but spinning up a thread is an intensive operation, so a managed thread pool can be used as a nice solution that allows the developer to execute callbacks quickly and asyncronously without blocking the reactor.

Of course any resources leveraged by the callbacks will need to be protected from concurrent access but the nice thing is that if care is taken, the reactor itself can be used to handle this. In the callbacks launched by the thread pool, the developer can leverage the event machine reactor to schedule additional work that will be executed serially with the rest of the work already there. What results is an elegant / simple solution to schedule and process work concurrently with a built in mechanism to protect shared resources. Optimally the thread pool will incorporate a timeout mechanism that is able to kill jobs that exceed the timeout and restart their worker threads to keep things moving smoothly (as RJR does).

I would give this combined design pattern a name (perhaps Reactor Pool?) but I'm sure someone out there probably has already thought about this... and blogged about it... and uploaded an article to wikipedia... etc...

A few months back I gave a presentation to the Brno Ruby Users group about the JSON-RPC protocol and my implementation RJR, but didn't go into too much detail here (blog post). Recently I've been pushing many updates and improvements, including a sorely needed docs update, and figure now would be a good time to do just that.

The goal was to develop a rpc mechanism that was as extensible and pluggable as possible with the implementation being transport agnostic, eg the developer would be able to satisfy and invoke rpc requests over a variety of transport mechanisms, such as tcp, http, websockets, amqp, and more. This would provide the most outreach for developers, allowing their methods to be invoked in a wide variety of infrastructures and existing systems. It was also important that the handlers be able to determine which transport a request came in on, so as to be able to alter flow-control if desired. To accomplish this, RJR sets a variety of instances variables in the scope of the invoked method handler, things like @rjr_node_type will contain the transport which the request came in on, and other things like @rjr_callback allows the server to send json-rpc request methods back to the client, so long as the transport mechanism remains intact (eg the tcp or websocket remains open, the amqp queue is still valid, etc).

# example server from the RJR documentation:
# define a rpc method called 'hello' which takes
# one argument and returns it in upper case
RJR::Dispatcher.add_handler("hello") { |arg|
  arg.upcase
}
 
# listen for this method via amqp, websockets, http calls
amqp_node  = RJR::AMQPNode.new  :node_id => 'server', :broker => 'localhost'
ws_node    = RJR::WSNode.new    :node_id => 'server', :host   => 'localhost', :port => 8080
www_node   = RJR::WebNode.new   :node_id => 'server', :host   => 'localhost', :port => 8888
 
# start the server and block
multi_node = RJR::MultiNode.new :nodes => [amqp_node, ws_node, www_node]
multi_node.listen
multi_node.join

Since JSON-RPC is a very simple protocol I also wanted to add a mechanism to allow developers to extend the protocol easily, even if this meant that these customizations would only work against nodes running RJR. To do this RJR allows developers to set arbitrary headers to be written to the json-rpc request, so that method handlers and their invokers may process this additional metadata and do what they will with this. For example, a node being used as a server can take method arguments, authenticate them against any backend, and set a 'session-id' header on all subsequent messages. All subsequent client requests will contain this header which is available to the handlers that can authorize the user. (obviously the end user would want to use a secure transport mechanism incorporating ssl to prevent session-hijacking)

# example clients from the documentation
# invoke the method over amqp
amqp_node = RJR::AMQPNode.new :node_id => 'client', :broker => 'localhost'
puts amqp_node.invoke_request('server-queue', 'hello', 'world')
 
# invoke the method over http using rjr
client = RJR::WebNode.new :node_id => 'client'
puts client.invoke_request('http://localhost:8888', 'hello', 'mo')
 
# Invoking json-rpc requests over http using curl
# $ curl -X POST http://localhost:8888 -d '{"jsonrpc":"2.0","method":"hello","params" ["mo"],"id":"123"}'
# > {"jsonrpc":"2.0","id":"123","result":"Hello mo!"}

As far as next steps, flushing out the UDP transport mechanism and continuing to optimize performance are high on my list. At some point I would love to do a complete rewrite in a lower-level language such as C and simply write wrappers / adapters so that methods implemented in higher level languages can be invoked simultanously. But for the time being, RJR serves my purposes and will continue developing that for now.

On the Thursday before last I gave a presentation on the JSON-RPC protocol and my implementation of it to the Brno Ruby Users Group. You may find the slides here as well some code examples in Ruby and Javascript here.

RJR is a simple implementation of the JSON-RPC 2.0 standard, using eventmachine on the backend to serve JSON-RPC requests over a multitude of transport types including amqp, http, websockets, tcp/udp, and more. It permits method handlers to be registered and then invoked via any transport the end-user wishes (and the server admin permits access to).

The context of the request is made available to the handler when it is invoked, so that the developer can manage flow control accordingly and store any additional metadata in the requests and responses to extend the JSON-RPC protocol in any manner they desire.

The source code is freely available on github, any patches/issues/etc would be more than welcome. Happy hacking!