cleanup code, fix receive thread killing code, fix zmq url in remote mode

Nathan Marz · Nathan Marz · commit d30ad355181e · 2012-04-28T15:43:39.000-07:00
diff --git a/src/clj/backtype/storm/daemon/worker.clj b/src/clj/backtype/storm/daemon/worker.clj
@@ -213,8 +213,7 @@
                                    port
                                    receive-queue-map
                                    :kill-fn (fn [t]
-                                              (halt-process! 11))
-                                   :valid-tasks task-ids))
+                                              (halt-process! 11))))
                                                               
         shutdown* (fn []
                     (log-message "Shutting down worker " storm-id " " supervisor-id " " port)
diff --git a/src/clj/backtype/storm/messaging/loader.clj b/src/clj/backtype/storm/messaging/loader.clj
@@ -1,10 +1,10 @@
 (ns backtype.storm.messaging.loader
-  (:use [clojure.contrib.def :only [defnk]])
   (:use [backtype.storm.bootstrap])
   (:import [java.util.concurrent LinkedBlockingQueue])
   (:require [backtype.storm.messaging.local :as local]))
 
 (bootstrap)
+
 (defn mk-local-context []
   (local/mk-local-context))
 
@@ -19,26 +19,20 @@
   [context storm-id port receive-queue-map
    :daemon true
    :kill-fn (fn [t] (System/exit 1))
-   :priority Thread/NORM_PRIORITY
-   :valid-tasks nil]
-  (let [valid-tasks (set (map short valid-tasks))
-        vthread (async-loop
+   :priority Thread/NORM_PRIORITY]
+  (let [vthread (async-loop
                  (fn [socket receive-queue-map]
                    (let [[task msg] (msg/recv socket)]
                      (if (= task -1)
                        (do
                          (log-message "Receiving-thread:[" storm-id ", " port "] received shutdown notice")
                          (.close socket)
                          nil )
-                       (if (or (nil? valid-tasks) (contains? valid-tasks task))
-                         (let [task (int task)
-                               ^LinkedBlockingQueue receive-queue (receive-queue-map task)]
-                           ;; TODO: probably need to handle multi-part messages here or something
-                           (.put receive-queue msg)
-                           0
-                           )
-                         (log-message "Receiving-thread:[" storm-id ", " port "] received invalid message for unknown task " task ". Dropping...")
-                         ))))
+                       (do
+                         (if (contains? receive-queue-map task)
+                           (.put ^LinkedBlockingQueue (receive-queue-map task) msg)
+                           (log-message "Receiving-thread:[" storm-id ", " port "] received invalid message for unknown task " task ". Dropping..."))
+                          0 ))))
                  :args-fn (fn [] [(msg/bind context storm-id port) receive-queue-map])
                  :daemon daemon
                  :kill-fn kill-fn
@@ -49,9 +43,9 @@
         (msg/send kill-socket
                   -1
                   (byte-array []))
-        (.close kill-socket)
         (log-message "Waiting for receiving-thread:[" storm-id ", " port "] to die")
         (.join vthread)
+        (.close kill-socket)
         (log-message "Shutdown receiving-thread: [" storm-id ", " port "]")
         ))))
 
diff --git a/src/clj/backtype/storm/messaging/zmq.clj b/src/clj/backtype/storm/messaging/zmq.clj
@@ -20,11 +20,17 @@
     [port msg]
     ))
 
-(defn get-zmq-url [local? port]
+(defn get-bind-zmq-url [local? port]
   (if local?
     (str "ipc://" port ".ipc")
     (str "tcp://*:" port)))
 
+(defn get-connect-zmq-url [local? host port]
+  (if local?
+    (str "ipc://" port ".ipc")
+    (str "tcp://" host ":" port)))
+
+
 (defprotocol ZMQContextQuery
   (zmq-context [this]))
 
@@ -43,14 +49,14 @@
   (bind [this storm-id port]
     (-> context
         (mq/socket mq/pull)
-        (mq/bind (get-zmq-url local? port))
+        (mq/bind (get-bind-zmq-url local? port))
         (ZMQConnection.)
         ))
   (connect [this storm-id host port]
     (-> context
         (mq/socket mq/push)
         (mq/set-linger linger-ms)
-        (mq/connect (get-zmq-url local? port))
+        (mq/connect (get-connect-zmq-url local? host port))
         (ZMQConnection.)))
   (term [this]
     (.term context))
diff --git a/src/clj/backtype/storm/testing.clj b/src/clj/backtype/storm/testing.clj
@@ -87,8 +87,7 @@
     ))
 
 (defn mk-shared-context [conf]
-  (if (and (= (conf STORM-CLUSTER-MODE) "local")
-           (not (conf STORM-LOCAL-MODE-ZMQ)))
+  (if-not (conf STORM-LOCAL-MODE-ZMQ)
     (msg-loader/mk-local-context)
     ))
 
