nathanmarz/storm — pull request #518 by anfeng (2013-03-24 07:27:17+00:00)
Title: Plugin mechanism for messaging layer (Issue #372)

This pull request proposes a plugin mechanism for messaging layer.

We are introducing a new Storm config parameter, storm.messaging.transport, to specify a desired plugin. For now, defaults.yaml states that zmq is such a transport.

A messaging plugin needs to implements a simple interface, ITransport, and should have a default constructor and implements, newContext(), to return an IContext.
public interface ITransport {
public IContext newContext();
}

IContext is defined to have the following methods, where prepare(storm_conf) is invoked immediately after IContext is constructed by Storm core (TransportFactory).
public interface IContext {
public void prepare(Map storm_conf);

public IConnection bind(String storm_id, int port);
public IConnection connect(String storm_id, String host, int port);
public void term();
};

IConnection is defined as below, where TaskMessage is a simple Java class contains task ID and message.
public interface IConnection {
public TaskMessage recv();
public TaskMessage recv_with_flags(int flags);
public void send(int task, byte[] message);
public void close();
}

I have modified zmq.clj to implements all these interfaces. local.clj has been revised to implement IContext and IConnection.

worker.clj has been revised slightly to leverage the plugin mechanism. It does not knows ZMQ anymore :-)