finished simplified drpc...

Nathan Marz · Nathan Marz · commit 576577b3daf5 · 2011-10-21T21:29:37.000-07:00
diff --git a/src/clj/backtype/storm/LocalDRPC.clj b/src/clj/backtype/storm/LocalDRPC.clj
@@ -31,7 +31,10 @@
   (.failRequest (:handler (. this state)) id)
   )
   
+(defn -getServiceId [this]
+  (:service-id (. this state)))  
+
 (defn -shutdown [this]
-  (ServiceRegistry/unregisterService {:service-id (. this state)})
-  (.shutdown {:handler (. this state)})
+  (ServiceRegistry/unregisterService (:service-id (. this state)))
+  (.shutdown (:handler (. this state)))
   )
diff --git a/src/clj/backtype/storm/daemon/drpc.clj b/src/clj/backtype/storm/daemon/drpc.clj
@@ -7,7 +7,7 @@
   (:import [java.util.concurrent Semaphore ConcurrentLinkedQueue])
   (:import [backtype.storm.daemon Shutdownable])
   (:import [java.net InetAddress])
-  (:use [backtype.storm bootstrap config])
+  (:use [backtype.storm bootstrap config log])
   (:gen-class))
 
 (bootstrap)
@@ -19,8 +19,9 @@
 (defn acquire-queue [queues-atom function]
   (swap! queues-atom
     (fn [amap]
-      (when-not (amap function))
+      (if-not (amap function)
         (assoc amap function (ConcurrentLinkedQueue.))
+        amap)
         ))
   (@queues-atom function))
 
@@ -54,6 +55,7 @@
               req (DRPCRequest. args id)
               ^ConcurrentLinkedQueue queue (acquire-queue request-queues function)
               ]
+          (log-message "Received DRPC request for " function " " args)
           (swap! id->start assoc id (current-time-secs))
           (swap! id->sem assoc id sem)
           (.add queue req)
diff --git a/src/jvm/backtype/storm/drpc/DRPCSpout.java b/src/jvm/backtype/storm/drpc/DRPCSpout.java
@@ -27,7 +27,7 @@ public class DRPCSpout implements IRichSpout {
     public static Logger LOG = Logger.getLogger(DRPCSpout.class);
     
     SpoutOutputCollector _collector;
-    List<DRPCClient> _clients;
+    List<DRPCClient> _clients = new ArrayList<DRPCClient>();
     String _function;
     String _local_drpc_id = null;
     
@@ -69,7 +69,6 @@ public void open(Map conf, TopologyContext context, SpoutOutputCollector collect
             if(servers.isEmpty()) {
                 throw new RuntimeException("No DRPC servers configured for topology");   
             }
-            _clients = new ArrayList<DRPCClient>();
             if(numTasks < servers.size()) {
                 for(String s: servers) {
                     _clients.add(new DRPCClient(s, port));
@@ -111,7 +110,7 @@ public void nextTuple() {
                 }
             }
         } else {
-            ILocalDRPC drpc = (ILocalDRPC) ServiceRegistry.getService(_local_drpc_id);
+            DistributedRPC.Iface drpc = (DistributedRPC.Iface) ServiceRegistry.getService(_local_drpc_id);
             try {
                 DRPCRequest req = drpc.fetchRequest(_function);
                 if(req.get_request_id().length() > 0) {
diff --git a/src/jvm/backtype/storm/drpc/ReturnResults.java b/src/jvm/backtype/storm/drpc/ReturnResults.java
@@ -35,7 +35,7 @@ public void execute(Tuple input) {
             String id = (String) retMap.get("id");
             DistributedRPC.Iface client;
             if(local) {
-                client = (ILocalDRPC) ServiceRegistry.getService(host);
+                client = (DistributedRPC.Iface) ServiceRegistry.getService(host);
             } else {
                 client = new DRPCClient(host, port);
             }

Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,10 @@`
`31`	`31`	`(.failRequest (:handler (. this state)) id)`
`32`	`32`	`)`
`33`	`33`
	`34`	`+(defn -getServiceId [this]`
	`35`	`+ (:service-id (. this state)))`
	`36`	`+`
`34`	`37`	`(defn -shutdown [this]`
`35`		`- (ServiceRegistry/unregisterService {:service-id (. this state)})`
`36`		`- (.shutdown {:handler (. this state)})`
	`38`	`+ (ServiceRegistry/unregisterService (:service-id (. this state)))`
	`39`	`+ (.shutdown (:handler (. this state)))`
`37`	`40`	`)`
Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,7 @@ public void execute(Tuple input) {`
`35`	`35`	`String id = (String) retMap.get("id");`
`36`	`36`	`DistributedRPC.Iface client;`
`37`	`37`	`if(local) {`
`38`		`- client = (ILocalDRPC) ServiceRegistry.getService(host);`
	`38`	`+ client = (DistributedRPC.Iface) ServiceRegistry.getService(host);`
`39`	`39`	`} else {`
`40`	`40`	`client = new DRPCClient(host, port);`
`41`	`41`	`}`