remove 0MQ and replace with netty

revans2 · Dec 10, 2013 · b63ed13 · b63ed13
1 parent 1bcc169
commit b63ed13
Show file tree

Hide file tree

Showing 23 changed files with 36 additions and 382 deletions.
diff --git a/MODULES b/MODULES
@@ -1,4 +1,3 @@
 storm-console-logging
 storm-core
-storm-netty
 
diff --git a/bin/install_zmq.sh b/bin/install_zmq.sh
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
@@ -18,7 +18,7 @@ storm.zookeeper.retry.intervalceiling.millis: 30000
 storm.cluster.mode: "distributed" # can be distributed or local
 storm.local.mode.zmq: false
 storm.thrift.transport: "backtype.storm.security.auth.SimpleTransportPlugin"
-storm.messaging.transport: "backtype.storm.messaging.zmq"
+storm.messaging.transport: "backtype.storm.messaging.netty.Context"
 
 ### nimbus.* configs are for the master
 nimbus.host: "localhost"

diff --git a/storm-core/project.clj b/storm-core/project.clj
@@ -10,7 +10,6 @@
                  [clj-time "0.4.1"]
                  [com.netflix.curator/curator-framework "1.0.1"
                   :exclusions [log4j/log4j]]
-                 [backtype/jzmq "2.1.0"]
                  [com.googlecode.json-simple/json-simple "1.1"]
                  [compojure "1.1.3"]
                  [hiccup "0.3.6"]
@@ -27,6 +26,7 @@
                  [com.google.guava/guava "13.0"]
                  [ch.qos.logback/logback-classic "1.0.6"]
                  [org.slf4j/log4j-over-slf4j "1.6.6"]
+                 [io.netty/netty "3.6.3.Final"]
                  ]
 
   :source-paths ["src/clj"]

diff --git a/storm-core/src/clj/backtype/storm/messaging/zmq.clj b/storm-core/src/clj/backtype/storm/messaging/zmq.clj
diff --git a/storm-core/src/clj/backtype/storm/testing.clj b/storm-core/src/clj/backtype/storm/testing.clj
@@ -96,7 +96,7 @@
 ;; local dir is always overridden in maps
 ;; can customize the supervisors (except for ports) by passing in map for :supervisors parameter
 ;; if need to customize amt of ports more, can use add-supervisor calls afterwards
-(defnk mk-local-storm-cluster [:supervisors 2 :ports-per-supervisor 3 :daemon-conf {} :inimbus nil :supervisor-slot-port-min 1]
+(defnk mk-local-storm-cluster [:supervisors 2 :ports-per-supervisor 3 :daemon-conf {} :inimbus nil :supervisor-slot-port-min 1024]
   (let [zk-tmp (local-temp-path)
         [zk-port zk-handle] (zk/mk-inprocess-zookeeper zk-tmp)
         daemon-conf (merge (read-storm-config)

diff --git a/storm-core/src/clj/zilch/mq.clj b/storm-core/src/clj/zilch/mq.clj
diff --git a/...acktype/storm/messaging/netty/Client.java → ...acktype/storm/messaging/netty/Client.java b/...acktype/storm/messaging/netty/Client.java → ...acktype/storm/messaging/netty/Client.java
@@ -1,5 +1,16 @@
 package backtype.storm.messaging.netty;
 
+import backtype.storm.Config;
+import backtype.storm.messaging.IConnection;
+import backtype.storm.messaging.TaskMessage;
+import backtype.storm.utils.Utils;
+import org.jboss.netty.bootstrap.ClientBootstrap;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFactory;
+import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.Random;
@@ -9,18 +20,6 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.jboss.netty.bootstrap.ClientBootstrap;
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelFactory;
-import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import backtype.storm.Config;
-import backtype.storm.messaging.IConnection;
-import backtype.storm.messaging.TaskMessage;
-import backtype.storm.utils.Utils;
-
 class Client implements IConnection {
     private static final Logger LOG = LoggerFactory.getLogger(Client.class);
     private final int max_retries;

diff --git a/...cktype/storm/messaging/netty/Context.java → ...cktype/storm/messaging/netty/Context.java b/...cktype/storm/messaging/netty/Context.java → ...cktype/storm/messaging/netty/Context.java
@@ -1,11 +1,11 @@
 package backtype.storm.messaging.netty;
 
-import java.util.Map;
-import java.util.Vector;
-
 import backtype.storm.messaging.IConnection;
 import backtype.storm.messaging.IContext;
 
+import java.util.Map;
+import java.util.Vector;
+
 public class Context implements IContext {
     @SuppressWarnings("rawtypes")
     private Map storm_conf;

diff --git a/...storm/messaging/netty/ControlMessage.java → ...storm/messaging/netty/ControlMessage.java b/...storm/messaging/netty/ControlMessage.java → ...storm/messaging/netty/ControlMessage.java
@@ -3,7 +3,6 @@
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.buffer.ChannelBufferOutputStream;
 import org.jboss.netty.buffer.ChannelBuffers;
-import org.jboss.netty.channel.Channel;
 
 enum ControlMessage {
     CLOSE_MESSAGE((short)-100),

diff --git a/...e/storm/messaging/netty/MessageBatch.java → ...e/storm/messaging/netty/MessageBatch.java b/...e/storm/messaging/netty/MessageBatch.java → ...e/storm/messaging/netty/MessageBatch.java
@@ -1,13 +1,11 @@
 package backtype.storm.messaging.netty;
 
-import java.util.ArrayList;
-
+import backtype.storm.messaging.TaskMessage;
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.buffer.ChannelBufferOutputStream;
 import org.jboss.netty.buffer.ChannelBuffers;
-import org.jboss.netty.channel.Channel;
 
-import backtype.storm.messaging.TaskMessage;
+import java.util.ArrayList;
 
 class MessageBatch {
     private int buffer_size;

diff --git a/...storm/messaging/netty/MessageDecoder.java → ...storm/messaging/netty/MessageDecoder.java b/...storm/messaging/netty/MessageDecoder.java → ...storm/messaging/netty/MessageDecoder.java
@@ -1,10 +1,10 @@
 package backtype.storm.messaging.netty;
 
+import backtype.storm.messaging.TaskMessage;
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelHandlerContext;
 import org.jboss.netty.handler.codec.frame.FrameDecoder;
-import backtype.storm.messaging.TaskMessage;
 
 public class MessageDecoder extends FrameDecoder {    
     /*

diff --git a/...storm/messaging/netty/MessageEncoder.java → ...storm/messaging/netty/MessageEncoder.java b/...storm/messaging/netty/MessageEncoder.java → ...storm/messaging/netty/MessageEncoder.java
diff --git a/...acktype/storm/messaging/netty/Server.java → ...acktype/storm/messaging/netty/Server.java b/...acktype/storm/messaging/netty/Server.java → ...acktype/storm/messaging/netty/Server.java
@@ -1,10 +1,9 @@
 package backtype.storm.messaging.netty;
 
-import java.net.InetSocketAddress;
-import java.util.Map;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Executors;
-
+import backtype.storm.Config;
+import backtype.storm.messaging.IConnection;
+import backtype.storm.messaging.TaskMessage;
+import backtype.storm.utils.Utils;
 import org.jboss.netty.bootstrap.ServerBootstrap;
 import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelFactory;
@@ -14,10 +13,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import backtype.storm.Config;
-import backtype.storm.messaging.IConnection;
-import backtype.storm.messaging.TaskMessage;
-import backtype.storm.utils.Utils;
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 
 class Server implements IConnection {
     private static final Logger LOG = LoggerFactory.getLogger(Server.class);