Merge remote-tracking branch 'origin/dev' into dev

Author: chengyitian

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

@@ -118,6 +118,8 @@ public void setRunningThread(Thread runningThread){
         @Override
         public void run() {
             while (!pThread.isInterrupted()) {
+                if (!AbstractClient.ifUseBackupSite) {
+                    // original logic:
                     for (String site : this.dispatcher.getAllReconnectSites()) {
                         if (dispatcher.getNeedReconnect(site) == 1) {
                             Site s = dispatcher.getSiteByName(site);
@@ -150,10 +152,47 @@ public void run() {
                         }
                     }
 
-                try {
-                    Thread.sleep(1000);
-                } catch (InterruptedException e) {
-                    break;
+                    try {
+                        Thread.sleep(1000);
+                    } catch (InterruptedException e) {
+                        break;
+                    }
+                } else {
+                    // if set backupSite
+                    for (String site : this.dispatcher.getAllReconnectSites()) {
+                        if (dispatcher.getNeedReconnect(site) == 1) {
+                            Site s = dispatcher.getCurrentSiteByName(site);
+                            dispatcher.activeCloseConnection(s);
+                            for (String topic : dispatcher.getAllTopicsBySite(site)) {
+                                System.out.println("Daemon need reconnect: " + topic);
+                                // reconnect every info.resubTimeout ms
+                                if (System.currentTimeMillis() - AbstractClient.lastExceptionTopicTimeMap.get(topic) <= AbstractClient.resubTimeout)
+                                    continue;
+
+                                dispatcher.tryReconnect(topic);
+                            }
+                        } else {
+                            Site s = dispatcher.getSiteByName(site);
+                            dispatcher.activeCloseConnection(s);
+                            for (String topic : dispatcher.getAllTopicsBySite(site)) {
+                                // reconnect every info.resubTimeout ms
+                                if (System.currentTimeMillis() - AbstractClient.lastExceptionTopicTimeMap.get(topic) <= AbstractClient.resubTimeout)
+                                    continue;
+
+                                dispatcher.tryReconnect(topic);
+                                dispatcher.setReconnectTimestamp(site, System.currentTimeMillis());
+                            }
+                        }
+                    }
+
+                    // not need to put and get from waitReconnectTopic.
+
+                    try {
+                        // check reconnected interval time
+                        Thread.sleep(10);
+                    } catch (InterruptedException e) {
+                        break;
+                    }
                 }
             }
         }

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

@@ -32,6 +32,8 @@ interface MessageDispatcher {
 
     Site getSiteByName(String site);
 
+    Site getCurrentSiteByName(String site);
+
     void activeCloseConnection(Site site);
 
     List<String> getAllTopicsBySite(String site);

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

@@ -107,6 +107,7 @@ public void run() {
                 body = BasicEntityFactory.instance().createEntity(df, dt, in, extended);
                 if (body.isTable() && body.rows() == 0) {
                     for (String t : topic.split(",")) {
+                        AbstractClient.lastExceptionTopicTimeMap.put(topic, System.currentTimeMillis());
                         dispatcher.setNeedReconnect(t, 0);
                     }
                     assert (body.rows() == 0);
@@ -161,13 +162,16 @@ public void run() {
         } catch (Exception e) {
             e.printStackTrace();
             if (dispatcher.isClosed(topic)) {
-                log.error("check " + topic + " is unsubscribed");
+                if (!AbstractClient.ifUseBackupSite)
+                    log.error("check " + topic + " is unsubscribed");
                 return;
             } else {
+                AbstractClient.lastExceptionTopicTimeMap.put(topic, System.currentTimeMillis());
                 dispatcher.setNeedReconnect(topic, 1);
             }
         } catch (Throwable t) {
             t.printStackTrace();
+            AbstractClient.lastExceptionTopicTimeMap.put(topic, System.currentTimeMillis());
             dispatcher.setNeedReconnect(topic, 1);
         } finally {
             try {

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

@@ -18,7 +18,7 @@
 
 public class PollingClient extends AbstractClient {
     TopicPoller topicPoller = null;
-    private HashMap<List<String>, List<String>> users = new HashMap<>();
+    // private HashMap<List<String>, List<String>> users = new HashMap<>();
 
     private static final Logger log = LoggerFactory.getLogger(PollingClient.class);
 
@@ -36,19 +36,46 @@ public PollingClient(String subscribeHost, int subscribePort) throws SocketExcep
 
     @Override
     protected boolean doReconnect(Site site) {
-        try {
-            Thread.sleep(1000);
-            BlockingQueue<List<IMessage>> queue = subscribeInternal(site.host, site.port, site.tableName, site.actionName, (MessageHandler) null, site.msgId + 1, true, site.filter, site.deserializer, site.allowExistTopic, site.userName, site.passWord, site.msgAstable);
-            log.info("Successfully reconnected and subscribed " + site.host + ":" + site.port + ":" + site.tableName);
-            topicPoller.setQueue(queue);
-            return true;
-        } catch (Exception ex) {
-            log.error("Unable to subscribe table. Will try again after 1 seconds.");
-            ex.printStackTrace();
-            return false;
+        if (!AbstractClient.ifUseBackupSite) {
+            // not enable backupSite, use original logic.
+            try {
+                log.info("PollingClient doReconnect: " + site.host + ":" + site.port);
+                Thread.sleep(1000);
+                BlockingQueue<List<IMessage>> queue = subscribeInternal(site.host, site.port, site.tableName, site.actionName, (MessageHandler) null, site.msgId + 1, true, site.filter, site.deserializer, site.allowExistTopic, site.userName, site.passWord, site.msgAstable);
+                log.info("Successfully reconnected and subscribed " + site.host + ":" + site.port + ":" + site.tableName);
+                topicPoller.setQueue(queue);
+                return true;
+            } catch (Exception ex) {
+                log.error("Unable to subscribe table. Will try again after 1 seconds.");
+                ex.printStackTrace();
+                return false;
+            }
+        } else {
+            // enable backupSite, try to switch site and subscribe.
+            try {
+                log.info("PollingClient doReconnect: " + site.host + ":" + site.port);
+                Thread.sleep(1000);
+                subscribe(site.host, site.port, site.tableName, site.actionName, (MessageHandler) null, site.msgId + 1, true, site.filter, site.deserializer, site.allowExistTopic, site.userName, site.passWord, site.msgAstable, false);
+                String topicStr = site.host + ":" + site.port + "/" + site.tableName + "/" + site.actionName;
+                String curTopic = tableNameToTrueTopic.get(topicStr);
+                BlockingQueue<List<IMessage>> queue = queueManager.addQueue(curTopic);
+                queueManager.changeQueue(curTopic, lastQueue);
+
+                log.info("Successfully reconnected and subscribed " + site.host + ":" + site.port + ":" + site.tableName);
+                topicPoller.setQueue(lastQueue);
+                return true;
+            } catch (Exception ex) {
+                log.error("Unable to subscribe table. Will try again after 1 seconds.");
+                ex.printStackTrace();
+                return false;
+            }
         }
     }
 
+    protected void subscribe(String host, int port, String tableName, String actionName, MessageHandler handler, long offset, boolean reconnect, Vector filter, StreamDeserializer deserializer, boolean allowExistTopic, String userName, String password, boolean msgAsTable, boolean createSubInfo) throws IOException {
+        BlockingQueue<List<IMessage>> queue = subscribeInternal(host, port, tableName, actionName, handler, offset, reconnect, filter, deserializer, allowExistTopic, userName, password, false, createSubInfo);
+    }
+
     public TopicPoller subscribe(String host, int port, String tableName, String actionName, long offset, boolean reconnect, Vector filter, StreamDeserializer deserializer, String userName, String passWord) throws IOException {
         return subscribe(host, port, tableName, actionName, offset, reconnect, filter, deserializer, userName, passWord, false);
     }
@@ -57,7 +84,23 @@ public TopicPoller subscribe(String host, int port, String tableName, String act
         BlockingQueue<List<IMessage>> queue = subscribeInternal(host, port, tableName, actionName, (MessageHandler) null, offset, reconnect, filter, deserializer, false, userName, passWord, msgAsTable);
         List<String> tp = Arrays.asList(host, String.valueOf(port), tableName, actionName);
         List<String> usr = Arrays.asList(userName, passWord);
-        users.put(tp, usr);
+        // users.put(tp, usr);
+        topicPoller = new TopicPoller(queue);
+        return topicPoller;
+    }
+
+    public TopicPoller subscribe(String host, int port, String tableName, String actionName, long offset, boolean reconnect, Vector filter, StreamDeserializer deserializer, String userName, String passWord, boolean msgAsTable, List<String> backupSites, int resubTimeout, boolean subOnce) throws IOException {
+        if (resubTimeout < 0)
+            // resubTimeout default: 100ms
+            resubTimeout = 100;
+
+        BlockingQueue<List<IMessage>> queue = subscribeInternal(host, port, tableName, actionName, (MessageHandler) null, offset, reconnect, filter, deserializer, false, userName, passWord, msgAsTable, backupSites, resubTimeout, subOnce);
+        topicPoller = new TopicPoller(queue);
+        return topicPoller;
+    }
+
+    public TopicPoller subscribe(String host, int port, String tableName, String actionName, long offset, boolean reconnect, Vector filter, StreamDeserializer deserializer, String userName, String passWord, boolean msgAsTable, List<String> backupSites) throws IOException {
+        BlockingQueue<List<IMessage>> queue = subscribeInternal(host, port, tableName, actionName, (MessageHandler) null, offset, reconnect, filter, deserializer, false, userName, passWord, msgAsTable, backupSites, 100, false);
         topicPoller = new TopicPoller(queue);
         return topicPoller;
     }
@@ -120,47 +163,59 @@ public void unsubscribe(String host, int port, String tableName) throws IOExcept
 
     @Override
     protected void unsubscribeInternal(String host, int port, String tableName, String actionName) throws IOException {
-        DBConnection dbConn = new DBConnection();
-        List<String> tp = Arrays.asList(host, String.valueOf(port), tableName, actionName);
-        List<String> usr = users.get(tp);
-        String user = usr.get(0);
-        String pwd = usr.get(1);
-        if (!user.equals(""))
-            dbConn.connect(host, port, user, pwd);
-        else
-            dbConn.connect(host, port);
-        try {
-            String localIP = this.listeningHost;
-            if(localIP.equals(""))
-                localIP = dbConn.getLocalAddress().getHostAddress();
-            List<Entity> params = new ArrayList<Entity>();
-            params.add(new BasicString(localIP));
-            params.add(new BasicInt(this.listeningPort));
-            params.add(new BasicString(tableName));
-            params.add(new BasicString(actionName));
-
-            dbConn.run("stopPublishTable", params);
-            String topic = null;
-            String fullTableName = host + ":" + port + "/" + tableName + "/" + actionName;
-            synchronized (tableNameToTrueTopic) {
-                topic = tableNameToTrueTopic.get(fullTableName);
+        synchronized (this) {
+            DBConnection dbConn = new DBConnection();
+            if (!currentSiteIndexMap.isEmpty()) {
+                String topic = tableNameToTrueTopic.get( host + ":" + port + "/" + tableName + "/" + actionName);
+                Integer currentSiteIndex = currentSiteIndexMap.get(topic);
+                Site[] sites = trueTopicToSites.get(topic);
+                host = sites[currentSiteIndex].host;
+                port = sites[currentSiteIndex].port;
             }
-            synchronized (trueTopicToSites) {
+
+            List<String> tp = Arrays.asList(host, String.valueOf(port), tableName, actionName);
+            List<String> usr = users.get(tp);
+            String user = usr.get(0);
+            String pwd = usr.get(1);
+            if (!user.equals(""))
+                dbConn.connect(host, port, user, pwd);
+            else
+                dbConn.connect(host, port);
+            try {
+                String localIP = this.listeningHost;
+                if(localIP.equals(""))
+                    localIP = dbConn.getLocalAddress().getHostAddress();
+                List<Entity> params = new ArrayList<Entity>();
+                params.add(new BasicString(localIP));
+                params.add(new BasicInt(this.listeningPort));
+                params.add(new BasicString(tableName));
+                params.add(new BasicString(actionName));
+
+                dbConn.run("stopPublishTable", params);
+                String topic = null;
+                String fullTableName = host + ":" + port + "/" + tableName + "/" + actionName;
+                topic = tableNameToTrueTopic.get(fullTableName);
+
                 Site[] sites = trueTopicToSites.get(topic);
                 if (sites == null || sites.length == 0)
                     ;
                 for (int i = 0; i < sites.length; i++)
                     sites[i].closed = true;
-            }
-            synchronized (queueManager) {
+
                 queueManager.removeQueue(topic);
+
+                // init backupSites related params.
+                if (AbstractClient.ifUseBackupSite) {
+                    AbstractClient.ifUseBackupSite = false;
+                    AbstractClient.subOnce = false;
+                    AbstractClient.resubTimeout = 100;
+                }
+                log.info("Successfully unsubscribed table " + fullTableName);
+            } catch (Exception ex) {
+                throw ex;
+            } finally {
+                dbConn.close();
             }
-            log.info("Successfully unsubscribed table " + fullTableName);
-        } catch (Exception ex) {
-            throw ex;
-        } finally {
-            dbConn.close();
         }
-        return;
     }
 }

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

@@ -16,6 +16,14 @@ public synchronized BlockingQueue<List<IMessage>> addQueue(String topic) {
         throw new RuntimeException("Topic " + topic + " already subscribed");
     }
 
+    public synchronized void changeQueue(String topic, BlockingQueue<List<IMessage>> q) {
+        if (queueMap.containsKey(topic)) {
+            queueMap.put(topic, q);
+        } else {
+            throw new RuntimeException("Topic " + topic + " doesn't subscribed");
+        }
+    }
+
     public synchronized BlockingQueue<List<IMessage>> getQueue(String topic) {
         BlockingQueue<List<IMessage>> q = queueMap.get(topic);
         return q;