Changeset 532:7057fc0c1459 for kraken-logdb/src/main/java/org/krakenapps/logdb/query/command/Rpc.java

Timestamp:

11/14/11 02:31:08 (6 months ago)

Author:

xeraph

Branch:

default

Message:

added mapreduce query (not yet completed, and rpc has race condition bug)

File:

• kraken-logdb/src/main/java/org/krakenapps/logdb/query/command/Rpc.java (modified) (4 diffs)

kraken-logdb/src/main/java/org/krakenapps/logdb/query/command/Rpc.java

@@ -2 +2 @@
 
 import java.util.Map;
-
+import org.krakenapps.api.Primitive;
 import org.krakenapps.logdb.LogQueryCommand;
+import org.krakenapps.rpc.RpcClient;
+import org.krakenapps.rpc.RpcConnection;
+import org.krakenapps.rpc.RpcConnectionProperties;
+import org.krakenapps.rpc.RpcException;
+import org.krakenapps.rpc.RpcSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class Rpc extends LogQueryCommand {
+        private final Logger logger = LoggerFactory.getLogger(Rpc.class.getName());
+
+        private String agentGuid;
+        private RpcConnection upstream;
+
         /**
          * dist query guid
@@ -13 +25 @@
         private boolean sender;
 
-        public Rpc(String guid, boolean sender) {
+        private RpcConnection datastream;
+        private RpcSession datasession;
+
+        public Rpc(String agentGuid, RpcConnection upstream, String guid, boolean sender) {
+                this.agentGuid = agentGuid;
+                this.upstream = upstream;
                 this.guid = guid;
                 this.sender = sender;
@@ -20 +37 @@
         @Override
         public void push(Map<String, Object> m) {
-                write(m);
+                if (sender) {
+                        if (datastream == null) {
+                                RpcClient client = new RpcClient(agentGuid);
+                                RpcConnectionProperties props = new RpcConnectionProperties(upstream.getRemoteAddress());
+                                props.setPassword((String) upstream.getProperty("password"));
+                                datastream = client.connect(props);
+                                try {
+                                        datasession = datastream.createSession("logdb-mapreduce");
+                                        datasession.call("setLogStream", guid);
+                                } catch (RpcException e) {
+                                        logger.error("kraken logdb: cannot set log stream", e);
+                                } catch (InterruptedException e) {
+                                        logger.error("kraken logdb: cannot set log stream", e);
+                                }
+
+                                logger.info("kraken logdb: opened rpc data stream for query guid [{}]", guid);
+                        }
+
+                        datasession.post("push", m);
+                } else {
+                        write(m);
+                }
+
+                if (logger.isDebugEnabled())
+                        logger.debug("kraken logdb: rpc [{}]", Primitive.stringify(m));
         }
 
@@ -34 +75 @@
 
         @Override
+        public void eof() {
+                if (datasession != null) {
+                        datastream.close();
+                        datastream = null;
+                        datasession = null;
+                }
+
+                logger.info("kraken logdb: closed rpc data stream for query guid [{}]", guid);
+                super.eof();
+        }
+
+        @Override
         public String toString() {
                 return "RPC " + (sender ? "Output" : "Input") + guid;