AJ-650、AJ-651: adjust package structure for cep; extract Site from AbstractClient; rewrite EvenClient doReconnect;

Author: chengyitian

src/com/xxdb/streaming/client/AbstractClient.java

@@ -169,41 +169,6 @@ public Map<String, StreamDeserializer> getSubInfos(){
         return subInfos_;
     }
 
-    public class Site {
-        String host;
-        int port;
-        String tableName;
-        String actionName;
-        MessageHandler handler;
-        long msgId;
-        boolean reconnect;
-        Vector filter = null;
-        public boolean closed = false;
-        boolean allowExistTopic = false;
-        StreamDeserializer deserializer;
-        String userName = "";
-        String passWord = "";
-
-        boolean msgAstable = false;
-
-        Site(String host, int port, String tableName, String actionName,
-             MessageHandler handler, long msgId, boolean reconnect, Vector filter, StreamDeserializer deserializer, boolean allowExistTopic, String userName, String passWord, boolean msgAstable) {
-            this.host = host;
-            this.port = port;
-            this.tableName = tableName;
-            this.actionName = actionName;
-            this.handler = handler;
-            this.msgId = msgId;
-            this.reconnect = reconnect;
-            this.filter = filter;
-            this.allowExistTopic = allowExistTopic;
-            this.deserializer = deserializer;
-            this.userName = userName;
-            this.passWord = passWord;
-            this.msgAstable = msgAstable;
-        }
-    }
-
     abstract protected boolean doReconnect(Site site);
 
     public void setMsgId(String topic, long msgId) {
@@ -636,7 +601,7 @@ void checkServerVersion(String host, int port) throws IOException {
         }
     }
 
-    public ConcurrentHashMap<String, AbstractClient.Site[]> getTopicToSites() {
+    public ConcurrentHashMap<String, Site[]> getTopicToSites() {
         return trueTopicToSites;
     }
 }

src/com/xxdb/streaming/client/Daemon.java

@@ -120,7 +120,7 @@ public void run() {
             while (!pThread.isInterrupted()) {
                     for (String site : this.dispatcher.getAllReconnectSites()) {
                         if (dispatcher.getNeedReconnect(site) == 1) {
-                            AbstractClient.Site s = dispatcher.getSiteByName(site);
+                            Site s = dispatcher.getSiteByName(site);
                             dispatcher.activeCloseConnection(s);
                             String lastTopic = "";
                             for (String topic : dispatcher.getAllTopicsBySite(site)) {
@@ -133,7 +133,7 @@ public void run() {
                             // try reconnect after 3 second when reconnecting stat
                             long ts = dispatcher.getReconnectTimestamp(site);
                             if (System.currentTimeMillis() >= ts + 3000) {
-                                AbstractClient.Site s = dispatcher.getSiteByName(site);
+                                Site s = dispatcher.getSiteByName(site);
                                 dispatcher.activeCloseConnection(s);
                                 for (String topic : dispatcher.getAllTopicsBySite(site)) {
                                     log.info("try to reconnect topic " + topic);

src/com/xxdb/streaming/client/MessageDispatcher.java

@@ -30,17 +30,17 @@ interface MessageDispatcher {
 
     List<String> getAllReconnectSites();
 
-    AbstractClient.Site getSiteByName(String site);
+    Site getSiteByName(String site);
 
-    void activeCloseConnection(AbstractClient.Site site);
+    void activeCloseConnection(Site site);
 
     List<String> getAllTopicsBySite(String site);
 
     Set<String> getAllReconnectTopic();
 
     Map<String, StreamDeserializer> getSubInfos();
 
-    ConcurrentHashMap<String, AbstractClient.Site[]> getTopicToSites();
+    ConcurrentHashMap<String, Site[]> getTopicToSites();
 }
 
 

src/com/xxdb/streaming/client/MessageParser.java

@@ -45,7 +45,7 @@ private Boolean isListenMode(){
 
     public void run() {
         Map<String, StreamDeserializer> subinfos = dispatcher.getSubInfos();
-        ConcurrentHashMap<String, AbstractClient.Site[]> topicToSites = dispatcher.getTopicToSites();
+        ConcurrentHashMap<String, Site[]> topicToSites = dispatcher.getTopicToSites();
         Socket socket = null;
         try {
             DBConnection conn;
@@ -121,7 +121,7 @@ public void run() {
                 } else if (body.isVector()) {
                     BasicAnyVector dTable = (BasicAnyVector) body;
 
-                    AbstractClient.Site[] sites = topicToSites.get(topic);
+                    Site[] sites = topicToSites.get(topic);
                     int colSize = dTable.rows();
                     int rowSize = dTable.getEntity(0).rows();
                     if (sites != null && sites[0].msgAstable) {

src/com/xxdb/streaming/client/Site.java

@@ -0,0 +1,98 @@
+package com.xxdb.streaming.client;
+
+import com.xxdb.data.Vector;
+
+public class Site {
+    String host;
+    int port;
+    String tableName;
+    String actionName;
+    MessageHandler handler;
+    long msgId;
+    boolean reconnect;
+    Vector filter = null;
+    boolean closed = false;
+    boolean allowExistTopic = false;
+    StreamDeserializer deserializer;
+    String userName = "";
+    String passWord = "";
+
+    boolean msgAstable = false;
+
+    Site(String host, int port, String tableName, String actionName,
+         MessageHandler handler, long msgId, boolean reconnect, Vector filter, StreamDeserializer deserializer, boolean allowExistTopic, String userName, String passWord, boolean msgAstable) {
+        this.host = host;
+        this.port = port;
+        this.tableName = tableName;
+        this.actionName = actionName;
+        this.handler = handler;
+        this.msgId = msgId;
+        this.reconnect = reconnect;
+        this.filter = filter;
+        this.allowExistTopic = allowExistTopic;
+        this.deserializer = deserializer;
+        this.userName = userName;
+        this.passWord = passWord;
+        this.msgAstable = msgAstable;
+    }
+
+    public String getHost() {
+        return host;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    public String getTableName() {
+        return tableName;
+    }
+
+    public String getActionName() {
+        return actionName;
+    }
+
+    public MessageHandler getHandler() {
+        return handler;
+    }
+
+    public long getMsgId() {
+        return msgId;
+    }
+
+    public boolean isReconnect() {
+        return reconnect;
+    }
+
+    public Vector getFilter() {
+        return filter;
+    }
+
+    public boolean isClosed() {
+        return closed;
+    }
+
+    public void setClosed(boolean closed) {
+        this.closed = closed;
+    }
+
+    public boolean isAllowExistTopic() {
+        return allowExistTopic;
+    }
+
+    public StreamDeserializer getDeserializer() {
+        return deserializer;
+    }
+
+    public String getUserName() {
+        return userName;
+    }
+
+    public String getPassWord() {
+        return passWord;
+    }
+
+    public boolean isMsgAstable() {
+        return msgAstable;
+    }
+}

src/com/xxdb/streaming/client/ThreadedClient.java

@@ -12,9 +12,7 @@
 import java.net.SocketException;
 import java.text.DateFormat;
 import java.text.SimpleDateFormat;
-import java.time.Duration;
 import java.util.*;
-import java.time.LocalTime;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
 

src/com/xxdb/streaming/client/TopicManager.java

@@ -80,9 +80,9 @@ public List<String> getAllTopic() {
         return re;
     }
 
-    public AbstractClient.Site[] getSites(String topic) {
+    public Site[] getSites(String topic) {
         TopicInfo ti = getTopicInfo(topic);
-        AbstractClient.Site[] q = ti.sites;
+        Site[] q = ti.sites;
         return q;
     }
 
@@ -91,7 +91,7 @@ public String getTopic(String HATopic) {
         return ti.originTopic;
     }
 
-    public void setSites(String topic, AbstractClient.Site[] sites) {
+    public void setSites(String topic, Site[] sites) {
         TopicInfo ti = getTopicInfo(topic);
         ti.sites = sites;
     }
@@ -101,7 +101,7 @@ private class TopicInfo {
         private String originTopic; // originTopic is same to
         private BlockingQueue<List<IMessage>> messageQueue;
         private ConcurrentHashMap<String, Integer> nameToIndex;
-        private AbstractClient.Site[] sites;
+        private Site[] sites;
         private int reconnectStat;
 
         public TopicInfo(String topic) {
@@ -133,11 +133,11 @@ public int getReconnectStat() {
             return this.reconnectStat;
         }
 
-        public void setSites(AbstractClient.Site[] sites) {
+        public void setSites(Site[] sites) {
             this.sites = sites;
         }
 
-        public AbstractClient.Site[] getSites() {
+        public Site[] getSites() {
             return this.sites;
         }
     }

…b/streaming/cep/AttributeSerializer.java …ming/client/cep/AttributeSerializer.javasrc/com/xxdb/streaming/cep/AttributeSerializer.java renamed to src/com/xxdb/streaming/client/cep/AttributeSerializer.java src/com/xxdb/streaming/cep/AttributeSerializer.java renamed to src/com/xxdb/streaming/client/cep/AttributeSerializer.java

@@ -1,4 +1,4 @@
-package com.xxdb.streaming.cep;
+package com.xxdb.streaming.client.cep;
 
 import com.xxdb.data.Entity;
 import com.xxdb.io.ExtendedDataOutput;

…/com/xxdb/streaming/cep/EventClient.java …db/streaming/client/cep/EventClient.javasrc/com/xxdb/streaming/cep/EventClient.java renamed to src/com/xxdb/streaming/client/cep/EventClient.java src/com/xxdb/streaming/cep/EventClient.java renamed to src/com/xxdb/streaming/client/cep/EventClient.java

@@ -1,4 +1,4 @@
-package com.xxdb.streaming.cep;
+package com.xxdb.streaming.client.cep;
 
 import com.xxdb.DBConnection;
 import com.xxdb.comm.ErrorCodeInfo;
@@ -9,12 +9,16 @@
 import com.xxdb.streaming.client.AbstractClient;
 import com.xxdb.streaming.client.IMessage;
 import com.xxdb.streaming.client.MessageHandler;
+import com.xxdb.streaming.client.Site;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import java.io.IOException;
 import java.net.SocketException;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Date;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 
@@ -118,7 +122,7 @@ protected void unsubscribeInternal(String host, int port, String tableName, Stri
                 if (sites == null || sites.length == 0)
                     ;
                 for (int i = 0; i < sites.length; i++)
-                    sites[i].closed = true;
+                    sites[i].setClosed(true);
             }
             synchronized (queueManager) {
                 queueManager.removeQueue(topic);
@@ -135,6 +139,19 @@ protected void unsubscribeInternal(String host, int port, String tableName, Stri
 
     @Override
     protected boolean doReconnect(Site site) {
-        return false;
+        try {
+            site.getHost();
+            subscribe(site.getHost(), site.getPort(), site.getTableName(), site.getActionName(), site.getHandler(), site.getMsgId() + 1, true, site.getUserName(), site.getPassWord());
+            Date d = new Date();
+            DateFormat df = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
+            log.info(df.format(d) + " Successfully reconnected and subscribed " + site.getHost() + ":" + site.getPort() + "/" + site.getTableName() + "/" + site.getActionName());
+            return true;
+        } catch (Exception ex) {
+            Date d = new Date();
+            DateFormat df = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
+            log.error(df.format(d) + " Unable to subscribe table. Will try again after 1 seconds." + site.getHost() + ":" + site.getPort() + "/" + site.getTableName() + "/" + site.getActionName());
+            ex.printStackTrace();
+            return false;
+        }
     }
 }

…com/xxdb/streaming/cep/EventHandler.java …b/streaming/client/cep/EventHandler.javasrc/com/xxdb/streaming/cep/EventHandler.java renamed to src/com/xxdb/streaming/client/cep/EventHandler.java src/com/xxdb/streaming/cep/EventHandler.java renamed to src/com/xxdb/streaming/client/cep/EventHandler.java

@@ -1,4 +1,4 @@
-package com.xxdb.streaming.cep;
+package com.xxdb.streaming.client.cep;
 
 import com.xxdb.comm.ErrorCodeInfo;
 import com.xxdb.data.*;