[STORM-138]: Pluggable serialization for multilang

conf/defaults.yaml

@@ -118,6 +118,7 @@ topology.acker.executors: null
 topology.tasks: null
 # maximum amount of time a message has to complete before it's considered failed
 topology.message.timeout.secs: 30
+topology.multilang.serializer: "backtype.storm.multilang.JsonSerializer"
 topology.skip.missing.kryo.registrations: false
 topology.max.task.parallelism: null
 topology.max.spout.pending: null

storm-core/src/jvm/backtype/storm/Config.java

@@ -499,6 +499,12 @@ public class Config extends HashMap<String, Object> {
     public static final String TOPOLOGY_DEBUG = "topology.debug";
     public static final Object TOPOLOGY_DEBUG_SCHEMA = Boolean.class;
 
+    /**
+     * The serializer for communication between shell components and non-JVM
+     * processes
+     */
+    public static final String TOPOLOGY_MULTILANG_SERIALIZER = "topology.multilang.serializer";
+    public static final Object TOPOLOGY_MULTILANG_SERIALIZER_SCHEMA = String.class;
 
     /**
      * Whether or not the master should optimize topologies by running multiple

storm-core/src/jvm/backtype/storm/multilang/BoltMsg.java

@@ -0,0 +1,63 @@
+package backtype.storm.multilang;
+
+import java.util.List;
+
+/**
+ * BoltMsg is an object that represents the data sent from a shell component to
+ * a bolt process that implements a multi-language protocol. It is the union of
+ * all data types that a bolt can receive from Storm.
+ *
+ * <p>
+ * BoltMsgs are objects sent to the ISerializer interface, for serialization
+ * according to the wire protocol implemented by the serializer. The BoltMsg
+ * class allows for a decoupling between the serialized representation of the
+ * data and the data itself.
+ * </p>
+ */
+public class BoltMsg {
+    private String id;
+    private String comp;
+    private String stream;
+    private long task;
+    private List<Object> tuple;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getComp() {
+        return comp;
+    }
+
+    public void setComp(String comp) {
+        this.comp = comp;
+    }
+
+    public String getStream() {
+        return stream;
+    }
+
+    public void setStream(String stream) {
+        this.stream = stream;
+    }
+
+    public long getTask() {
+        return task;
+    }
+
+    public void setTask(long task) {
+        this.task = task;
+    }
+
+    public List<Object> getTuple() {
+        return tuple;
+    }
+
+    public void setTuple(List<Object> tuple) {
+        this.tuple = tuple;
+    }
+}

storm-core/src/jvm/backtype/storm/multilang/ISerializer.java

@@ -0,0 +1,65 @@
+package backtype.storm.multilang;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+import backtype.storm.task.TopologyContext;
+
+/**
+ * The ISerializer interface describes the methods that an object should
+ * implement to provide serialization and de-serialization capabilities to
+ * non-JVM language components.
+ */
+public interface ISerializer extends Serializable {
+
+    /**
+     * This method sets the input and output streams of the serializer
+     *
+     * @param processIn output stream to non-JVM component
+     * @param processOut input stream from non-JVM component
+     */
+    void initialize(OutputStream processIn, InputStream processOut);
+
+    /**
+     * This method transmits the Storm config to the non-JVM process and
+     * receives its pid.
+     *
+     * @param conf storm configuration
+     * @param context topology context
+     * @return process pid
+     */
+    Number connect(Map conf, TopologyContext context) throws IOException,
+            NoOutputException;
+
+    /**
+     * This method receives a shell message from the non-JVM process
+     *
+     * @return shell message
+     */
+    ShellMsg readShellMsg() throws IOException, NoOutputException;
+
+    /**
+     * This method sends a bolt message to a non-JVM bolt process
+     *
+     * @param msg bolt message
+     */
+    void writeBoltMsg(BoltMsg msg) throws IOException;
+
+    /**
+     * This method sends a spout message to a non-JVM spout process
+     *
+     * @param msg spout message
+     */
+    void writeSpoutMsg(SpoutMsg msg) throws IOException;
+
+    /**
+     * This method sends a list of task IDs to a non-JVM bolt process
+     *
+     * @param taskIds list of task IDs
+     */
+    void writeTaskIds(List<Integer> taskIds) throws IOException;
+}

storm-core/src/jvm/backtype/storm/multilang/JsonSerializer.java

@@ -0,0 +1,164 @@
+package backtype.storm.multilang;
+
+import java.io.BufferedReader;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import org.json.simple.JSONObject;
+import org.json.simple.JSONValue;
+
+import backtype.storm.task.TopologyContext;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.utils.Utils;
+
+/**
+ * JsonSerializer implements the JSON multilang protocol.
+ */
+public class JsonSerializer implements ISerializer {
+    private DataOutputStream processIn;
+    private BufferedReader processOut;
+
+    public void initialize(OutputStream processIn, InputStream processOut) {
+        this.processIn = new DataOutputStream(processIn);
+        this.processOut = new BufferedReader(new InputStreamReader(processOut));
+    }
+
+    public Number connect(Map conf, TopologyContext context)
+            throws IOException, NoOutputException {
+        JSONObject setupInfo = new JSONObject();
+        setupInfo.put("pidDir", context.getPIDDir());
+        setupInfo.put("conf", conf);
+        setupInfo.put("context", context);
+        writeMessage(setupInfo);
+
+        Number pid = (Number) ((JSONObject) readMessage()).get("pid");
+        return pid;
+    }
+
+    public void writeBoltMsg(BoltMsg boltMsg) throws IOException {
+        JSONObject obj = new JSONObject();
+        obj.put("id", boltMsg.getId());
+        obj.put("comp", boltMsg.getComp());
+        obj.put("stream", boltMsg.getStream());
+        obj.put("task", boltMsg.getTask());
+        obj.put("tuple", boltMsg.getTuple());
+        writeMessage(obj);
+    }
+
+    public void writeSpoutMsg(SpoutMsg msg) throws IOException {
+        JSONObject obj = new JSONObject();
+        obj.put("command", msg.getCommand());
+        obj.put("id", msg.getId());
+        writeMessage(obj);
+    }
+
+    public void writeTaskIds(List<Integer> taskIds) throws IOException {
+        writeMessage(taskIds);
+    }
+
+    private void writeMessage(Object msg) throws IOException {
+        writeString(JSONValue.toJSONString(msg));
+    }
+
+    private void writeString(String str) throws IOException {
+        byte[] strBytes = str.getBytes("UTF-8");
+        processIn.write(strBytes, 0, strBytes.length);
+        processIn.writeBytes("\nend\n");
+        processIn.flush();
+    }
+
+    public ShellMsg readShellMsg() throws IOException, NoOutputException {
+        JSONObject msg = (JSONObject) readMessage();
+        ShellMsg shellMsg = new ShellMsg();
+
+        String command = (String) msg.get("command");
+        shellMsg.setCommand(command);
+
+        Object id = msg.get("id");
+        shellMsg.setId(id);
+
+        String log = (String) msg.get("msg");
+        shellMsg.setMsg(log);
+
+        String stream = (String) msg.get("stream");
+        if (stream == null)
+            stream = Utils.DEFAULT_STREAM_ID;
+        shellMsg.setStream(stream);
+
+        Object taskObj = msg.get("task");
+        if (taskObj != null) {
+            shellMsg.setTask((Long) taskObj);
+        } else {
+            shellMsg.setTask(0);
+        }
+
+        Object need_task_ids = msg.get("need_task_ids");
+        if (need_task_ids == null || ((Boolean) need_task_ids).booleanValue()) {
+            shellMsg.setNeedTaskIds(true);
+        } else {
+            shellMsg.setNeedTaskIds(false);
+        }
+
+        shellMsg.setTuple((List) msg.get("tuple"));
+
+        List<Tuple> anchors = new ArrayList<Tuple>();
+        Object anchorObj = msg.get("anchors");
+        if (anchorObj != null) {
+            if (anchorObj instanceof String) {
+                anchorObj = Arrays.asList(anchorObj);
+            }
+            for (Object o : (List) anchorObj) {
+                shellMsg.addAnchor((String) o);
+            }
+        }
+
+        return shellMsg;
+    }
+
+    private Object readMessage() throws IOException, NoOutputException {
+        String string = readString();
+        Object msg = JSONValue.parse(string);
+        if (msg != null) {
+            return msg;
+        } else {
+            throw new IOException("unable to parse: " + string);
+        }
+    }
+
+    private String readString() throws IOException, NoOutputException {
+        StringBuilder line = new StringBuilder();
+
+        // synchronized (processOut) {
+        while (true) {
+            String subline = processOut.readLine();
+            if (subline == null) {
+                StringBuilder errorMessage = new StringBuilder();
+                errorMessage.append("Pipe to subprocess seems to be broken!");
+                if (line.length() == 0) {
+                    errorMessage.append(" No output read.\n");
+                } else {
+                    errorMessage.append(" Currently read output: "
+                            + line.toString() + "\n");
+                }
+                errorMessage.append("Serializer Exception:\n");
+                throw new NoOutputException(errorMessage.toString());
+            }
+            if (subline.equals("end")) {
+                break;
+            }
+            if (line.length() != 0) {
+                line.append("\n");
+            }
+            line.append(subline);
+        }
+        // }
+        return line.toString();
+    }
+}

storm-core/src/jvm/backtype/storm/multilang/NoOutputException.java

@@ -0,0 +1,23 @@
+package backtype.storm.multilang;
+
+/**
+ * A NoOutputException states that no data has been received from the connected
+ * non-JVM process.
+ */
+public class NoOutputException extends Exception {
+    public NoOutputException() {
+        super();
+    }
+
+    public NoOutputException(String message) {
+        super(message);
+    }
+
+    public NoOutputException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public NoOutputException(Throwable cause) {
+        super(cause);
+    }
+}