AJ-681: fix issue about backupSite issues;

Author: chengyitian

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

@@ -258,11 +258,14 @@ public boolean tryReconnect(String topic) {
                     int successfulSiteIndex = -1;
 
                     // Starting from currentSiteIndex, go around in a circle until you return to the position just before it (circular looping).
+                    Site lastSite = null;
                     for (int offset = 0; offset < totalSites; offset++) {
                         // Implement wrapping around using modulo operation
                         int i = (currentSiteIndex + offset) % totalSites;
 
                         Site site = sites[i];
+                        if (offset == 0)
+                            lastSite = site;
                         boolean siteReconnected = false;
 
                         for (int attempt = 0; attempt < 2; attempt++) {
@@ -286,7 +289,9 @@ public boolean tryReconnect(String topic) {
                     if (subOnce && reconnected) {
                         List<Site> siteList = new ArrayList<>(Arrays.asList(sites));
                         // Remove the original currentSiteIndex node from the list.
-                        siteList.remove((int) currentSiteIndex);
+                        if (!(siteList.get(successfulSiteIndex).host.equals(lastSite.host) && siteList.get(successfulSiteIndex).port == lastSite.port))
+                            siteList.remove((int) currentSiteIndex);
+
                         // update sites
                         sites = siteList.toArray(new Site[0]);
                         trueTopicToSites.put(topic, sites);

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

@@ -154,24 +154,18 @@ else if(batchSize != -1 && throttle != -1 || batchSize != -1 && secondThrottle !
 
     @Override
     protected boolean doReconnect(Site site) {
-        synchronized (this) {
-            log.info("ThreadedClient doReconnect: " + site.host + ":" + site.port);
+        if (!AbstractClient.ifUseBackupSite) {
+            // not enable backupSite, use original logic
             String topicStr = site.host + ":" + site.port + "/" + site.tableName + "/" + site.actionName;
             Thread handlerLopper = null;
-            if (!handlerLoppers.containsKey(topicStr)) {
-                if (!AbstractClient.ifUseBackupSite) {
+            synchronized (handlerLoppers) {
+                if (!handlerLoppers.containsKey(topicStr))
                     throw new RuntimeException("Subscribe thread is not started");
-                }
-            } else {
                 handlerLopper = handlerLoppers.get(topicStr);
-                handlerLopper.interrupt();
             }
-
+            handlerLopper.interrupt();
             try {
-//                System.out.println("doReconnect 尝试切换节点：" + site.host + ":" + site.port);
-//                System.out.println("site msg id: " +site.msgId);
-                subscribe(site.host, site.port, site.tableName, site.actionName, site.handler, site.msgId + 1, true, site.filter, site.deserializer, site.allowExistTopic, site.userName, site.passWord, false);
-                // System.out.println("doReconnect 尝试切换节点成功：" + site.host + ":" + site.port);
+                subscribe(site.host, site.port, site.tableName, site.actionName, site.handler, site.msgId + 1, true, site.filter, site.deserializer, site.allowExistTopic, site.userName, site.passWord);
                 Date d = new Date();
                 DateFormat df = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
                 log.info(df.format(d) + " Successfully reconnected and subscribed " + site.host + ":" + site.port + "/" + site.tableName + "/" + site.actionName);
@@ -183,6 +177,29 @@ protected boolean doReconnect(Site site) {
                 ex.printStackTrace();
                 return false;
             }
+        } else {
+            // enable backupSite, try to switch site and subscribe.
+            synchronized (this) {
+                log.info("ThreadedClient doReconnect: " + site.host + ":" + site.port);
+                try {
+                    System.out.println("doReconnect 尝试切换节点：" + site.host + ":" + site.port);
+                    // System.out.println("site msg id: " +site.msgId);
+                    subscribe(site.host, site.port, site.tableName, site.actionName, site.handler, site.msgId + 1, true, site.filter, site.deserializer, site.allowExistTopic, site.userName, site.passWord, false);
+                    System.out.println("doReconnect 尝试切换节点成功：" + site.host + ":" + site.port);
+                    // System.out.println("切换后 handlerLoppers: " + handlerLoppers.get(topicStr).getName());
+                    // System.out.println("切换成功后，handlerLoppers size: " + handlerLoppers.size());
+                    Date d = new Date();
+                    DateFormat df = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
+                    log.info(df.format(d) + " Successfully reconnected and subscribed " + site.host + ":" + site.port + "/" + site.tableName + "/" + site.actionName);
+                    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.host + ":" + site.port + "/" + site.tableName + "/" + site.actionName);
+                    ex.printStackTrace();
+                    return false;
+                }
+            }
         }
     }
 
@@ -198,16 +215,11 @@ 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 createSubInfo) throws IOException {
+    /**
+     * This internal subscribe method only use for when enable backupSite, try to switch site and subscribe.
+     */
+    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 createSubInfo) throws IOException {
         BlockingQueue<List<IMessage>> queue = subscribeInternal(host, port, tableName, actionName, handler, offset, reconnect, filter, deserializer, allowExistTopic, userName, password, false, createSubInfo);
-        HandlerLopper handlerLopper = new HandlerLopper(queue, handler);
-        handlerLopper.start();
-        String topicStr = host + ":" + port + "/" + tableName + "/" + actionName;
-        List<String> usr = Arrays.asList(userName, password);
-        synchronized (handlerLoppers) {
-            handlerLoppers.put(topicStr, handlerLopper);
-            // users.put(topicStr, usr);
-        }
     }
 
     public void subscribe(String host, int port, String tableName, String actionName, MessageHandler handler, long offset, boolean reconnect, Vector filter, StreamDeserializer deserializer, boolean allowExistTopic, int batchSize, int throttle, String userName, String password) throws IOException {