AJ-636: add new logic for streaming backup;

chengyitian · chengyitian · commit 1cb5d028c103 · 2024-04-17T16:56:10.000+08:00
diff --git a/src/com/xxdb/streaming/client/AbstractClient.java b/src/com/xxdb/streaming/client/AbstractClient.java
@@ -37,7 +37,12 @@ public abstract class AbstractClient implements MessageDispatcher {
     protected HashMap<List<String>, List<String>> users = new HashMap<>();
     protected boolean isClose_ = false;
     protected LinkedBlockingQueue<DBConnection> connList = new LinkedBlockingQueue<>();
+    protected static boolean ifUseBackupSite;
+    protected String lastBackupSiteTopic = "";
     protected Map<String, Integer> currentSiteIndexMap = new ConcurrentHashMap<>();
+    protected static Map<String, Long> lastExceptionTopicTimeMap = new ConcurrentHashMap<>();
+    protected static Integer resubTimeout;
+    protected static boolean subOnce;
 
     private Daemon daemon = null;
 
@@ -166,6 +171,17 @@ public Site getSiteByName(String site) {
         return null;
     }
 
+    public Site getCurrentSiteByName(String site) {
+        List<String> topics = this.getAllTopicsBySite(site);
+        if (topics.size() > 0) {
+            Site[] sites = trueTopicToSites.get(topics.get(0));
+            Integer currentSiteIndex = currentSiteIndexMap.get(topics.get(0));
+            return sites[currentSiteIndex];
+        }
+
+        return null;
+    }
+
     public Map<String, StreamDeserializer> getSubInfos(){
         return subInfos_;
     }
@@ -227,32 +243,65 @@ public boolean tryReconnect(String topic) {
                 }
 
                 boolean reconnected = false;
-                for (int i = 0; i < sites.length; i ++) {
-                    Integer currentSiteIndex = currentSiteIndexMap.get(topic);
-                    if (currentSiteIndexMap.get(topic) != -1)
-                        i = currentSiteIndex;
-                    Site site = sites[i];
-                    boolean siteReconnected = false;
-                    for (int attempt = 0; attempt < 2; attempt++) {
-                        // try twice every site
-                        if (doReconnect(site)) {
-                            siteReconnected = true;
-                            // if site current connect successfully, break
+                Integer currentSiteIndex = currentSiteIndexMap.get(topic);
+                if (currentSiteIndex != null && currentSiteIndex != -1) {
+                    int totalSites = sites.length;
+                    // set successfulSiteIndex init value to -1.
+                    int successfulSiteIndex = -1;
+
+                    // Starting from currentSiteIndex, go around in a circle until you return to the position just before it (circular looping).
+                    for (int offset = 0; offset < totalSites; offset++) {
+                        // Implement wrapping around using modulo operation
+                        int i = (currentSiteIndex + offset) % totalSites;
+
+                        Site site = sites[i];
+                        boolean siteReconnected = false;
+
+                        for (int attempt = 0; attempt < 1; attempt++) {
+                            // try twice for every site.
+                            if (doReconnect(site)) {
+                                siteReconnected = true;
+                                // if site reconnect successfully, break.
+                                break;
+                            }
+                        }
+
+                        if (siteReconnected) {
+                            reconnected = true;
+                            successfulSiteIndex = i;
+                            // if current site reconnect successfully, no continue try other sites, break.
                             break;
                         }
                     }
-                    if (siteReconnected) {
-                        reconnected = true;
-                        currentSiteIndexMap.put(topic, i);
-                        // if current site connect successfully, no tryReconnect other site, break
-                        break;
+
+                    // Determine whether to delete the original currentSiteIndex node based on the subOnce parameter.
+                    if (subOnce && reconnected) {
+                        List<Site> siteList = new ArrayList<>(Arrays.asList(sites));
+                        // Remove the original currentSiteIndex node from the list.
+                        siteList.remove((int) currentSiteIndex);
+                        // update sites
+                        sites = siteList.toArray(new Site[0]);
+
+                        // Calculate the index of the newly successful connection node after a successful deletion.
+                        if (successfulSiteIndex > currentSiteIndex) {
+                            // If the successfully connected node is after the deleted node, reduce the index by 1.
+                            successfulSiteIndex -= 1;
+                        }
+                        // update currentSiteIndexMap to new successfully connected site's index;
+                        currentSiteIndexMap.put(topic, successfulSiteIndex);
+                    } else if (reconnected) {
+                        // not delete site, but update successfulSiteIndex.
+                        currentSiteIndexMap.put(topic, successfulSiteIndex);
                     }
                 }
 
+                log.info("Successfully switched to node: " + sites[currentSiteIndexMap.get(topic)].host + ":" + sites[currentSiteIndexMap.get(topic)].port);
+
                 if (!reconnected) {
                     waitReconnectTopic.add(topic);
                     return false;
                 } else {
+                    reconnectTable.remove(topic.substring(0, topic.indexOf("/")));
                     waitReconnectTopic.remove(topic);
                     return true;
                 }
@@ -438,13 +487,14 @@ protected BlockingQueue<List<IMessage>> subscribeInternal(String host, int port,
                                                               long offset, boolean reconnect, Vector filter,  StreamDeserializer deserializer,
                                                               boolean allowExistTopic, String userName, String passWord, boolean msgAsTable)
             throws IOException, RuntimeException {
-        return subscribeInternal(host, port, tableName, actionName, handler, offset, reconnect, filter, deserializer, allowExistTopic, userName, passWord, msgAsTable, null);
+        return subscribeInternal(host, port, tableName, actionName, handler, offset, reconnect, filter, deserializer, allowExistTopic, userName, passWord, msgAsTable, null, 100, false);
     }
 
     protected BlockingQueue<List<IMessage>> subscribeInternal(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, List<String> backupSites) throws IOException, RuntimeException {
+                                                              boolean allowExistTopic, String userName, String passWord, boolean msgAsTable,
+                                                              List<String> backupSites, int resubTimeout, boolean subOnce) throws IOException, RuntimeException {
         Entity re;
         String topic = "";
         DBConnection dbConn;
@@ -455,6 +505,9 @@ protected BlockingQueue<List<IMessage>> subscribeInternal(String host, int port,
 
         List<Site> parsedBackupSites = new ArrayList<>();
         if (Objects.nonNull(backupSites) && !backupSites.isEmpty()) {
+            AbstractClient.resubTimeout = resubTimeout;
+            AbstractClient.subOnce = subOnce;
+            AbstractClient.ifUseBackupSite = true;
             // prepare backupSites
             for (int i = 0; i < backupSites.size() + 1; i++) {
                 if (i == 0) {
@@ -494,6 +547,7 @@ protected BlockingQueue<List<IMessage>> subscribeInternal(String host, int port,
                     params.add(new BasicString(actionName));
                     re = dbConn.run("getSubscriptionTopic", params);
                     topic = ((BasicAnyVector) re).getEntity(0).getString();
+                    lastBackupSiteTopic = topic;
                     params.clear();
 
                     // set current site index
@@ -598,7 +652,20 @@ protected BlockingQueue<List<IMessage>> subscribeInternal(String host, int port,
 
                 re = dbConn.run("publishTable", params);
                 connList.add(dbConn);
-                if (re instanceof BasicAnyVector) {
+                if (ifUseBackupSite) {
+                    synchronized (subInfos_){
+                        subInfos_.put(topic, deserializer);
+                    }
+                    synchronized (tableNameToTrueTopic) {
+                        tableNameToTrueTopic.put(host + ":" + port + "/" + tableName + "/" + actionName, topic);
+                    }
+                    synchronized (HATopicToTrueTopic) {
+                        HATopicToTrueTopic.put(topic, topic);
+                    }
+                    synchronized (trueTopicToSites) {
+                        trueTopicToSites.put(topic, trueTopicToSites.get(lastBackupSiteTopic));
+                    }
+                } else if (re instanceof BasicAnyVector) {
                     BasicStringVector HASiteStrings = (BasicStringVector) (((BasicAnyVector) re).getEntity(1));
                     int HASiteNum = HASiteStrings.rows();
                     Site[] sites = new Site[HASiteNum];
diff --git a/src/com/xxdb/streaming/client/Daemon.java b/src/com/xxdb/streaming/client/Daemon.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,65 @@ 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) {
+                            System.out.println("flag site: " + site);
+                            Site s = dispatcher.getCurrentSiteByName(site);
+                            dispatcher.activeCloseConnection(s);
+                            String lastTopic = "";
+                            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;
+
+                                log.info("flag1 try to reconnect topic " + topic);
+                                dispatcher.tryReconnect(topic);
+                                // lastTopic = topic;
+                            }
+                            // dispatcher.setNeedReconnect(lastTopic, 2);
+                        } else {
+                            // try reconnect after 3 second when reconnecting stat
+                            // long ts = dispatcher.getReconnectTimestamp(site);
+                            // if (System.currentTimeMillis() >= ts + 3000) {
+                            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;
+
+                                log.info("flag2 try to reconnect topic " + topic);
+                                dispatcher.tryReconnect(topic);
+                                // }
+                                dispatcher.setReconnectTimestamp(site, System.currentTimeMillis());
+                            }
+                        }
+                    }
+
+                    Set<String> waitReconnectTopic = dispatcher.getAllReconnectTopic();
+                    synchronized (waitReconnectTopic) {
+                        for (String topic : waitReconnectTopic) {
+                            // reconnect every info.resubTimeout ms
+                            if (System.currentTimeMillis() - AbstractClient.lastExceptionTopicTimeMap.get(topic) <= AbstractClient.resubTimeout)
+                                continue;
+                            dispatcher.tryReconnect(topic);
+                        }
+                    }
+
+                    try {
+                        // check reconnected interval time
+                        Thread.sleep(10);
+                    } catch (InterruptedException e) {
+                        break;
+                    }
                 }
             }
         }
diff --git a/src/com/xxdb/streaming/client/MessageDispatcher.java b/src/com/xxdb/streaming/client/MessageDispatcher.java
@@ -32,6 +32,8 @@ interface MessageDispatcher {
 
     Site getSiteByName(String site);
 
+    Site getCurrentSiteByName(String site);
+
     void activeCloseConnection(Site site);
 
     List<String> getAllTopicsBySite(String site);
diff --git a/src/com/xxdb/streaming/client/MessageParser.java b/src/com/xxdb/streaming/client/MessageParser.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 {
diff --git a/src/com/xxdb/streaming/client/PollingClient.java b/src/com/xxdb/streaming/client/PollingClient.java
@@ -37,6 +37,7 @@ public PollingClient(String subscribeHost, int subscribePort) throws SocketExcep
     @Override
     protected boolean doReconnect(Site site) {
         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);
@@ -62,8 +63,12 @@ public TopicPoller subscribe(String host, int port, String tableName, String act
         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);
+    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);
         List<String> tp = Arrays.asList(host, String.valueOf(port), tableName, actionName);
         List<String> usr = Arrays.asList(userName, passWord);
         users.put(tp, usr);
diff --git a/src/com/xxdb/streaming/client/ThreadPooledClient.java b/src/com/xxdb/streaming/client/ThreadPooledClient.java
@@ -119,6 +119,7 @@ public void run() {
     }
 
     protected boolean doReconnect(Site site) {
+        log.info("ThreadPooledClient doReconnect: " + site.host + ":" + site.port);
         threadPool.shutdownNow();
         try {
             Thread.sleep(1000);
@@ -146,8 +147,12 @@ public void subscribe(String host, int port, String tableName, String actionName
         }
     }
 
-    public 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, List<String> backupSites) throws IOException {
-        BlockingQueue<List<IMessage>> queue = subscribeInternal(host, port, tableName, actionName, handler, offset, reconnect, filter, deserializer, allowExistTopic, userName, passWord, msgAsTable, backupSites);
+    public 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, 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, handler, offset, reconnect, filter, deserializer, allowExistTopic, userName, passWord, msgAsTable, backupSites, resubTimeout, subOnce);
         String topicStr = host + ":" + port + "/" + tableName + "/" + actionName;
         List<String> usr = Arrays.asList(userName, passWord);
         synchronized (queueHandlers) {
diff --git a/src/com/xxdb/streaming/client/ThreadedClient.java b/src/com/xxdb/streaming/client/ThreadedClient.java
