AJ-817: rename param from resubTimeout to resubscribeTimeout

Author: HakukuMatata

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

@@ -41,7 +41,7 @@ public abstract class AbstractClient implements MessageDispatcher {
     protected String lastBackupSiteTopic = "";
     protected Map<String, Integer> currentSiteIndexMap = new ConcurrentHashMap<>();
     protected static Map<String, Long> lastExceptionTopicTimeMap = new ConcurrentHashMap<>();
-    protected static Integer resubTimeout;
+    protected static Integer resubscribeInterval;
     protected static boolean subOnce;
     protected BlockingQueue<List<IMessage>> lastQueue;
     protected String lastSuccessSubscribeTopic = "";
@@ -521,7 +521,7 @@ 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, int resubTimeout, boolean subOnce, boolean createSubInfo) throws IOException, RuntimeException {
+                                                              List<String> backupSites, int resubscribeInterval, boolean subOnce, boolean createSubInfo) throws IOException, RuntimeException {
         Entity re;
         String topic = "";
         DBConnection dbConn = null;
@@ -543,7 +543,7 @@ protected BlockingQueue<List<IMessage>> subscribeInternal(String host, int port,
                 }
             }
 
-            AbstractClient.resubTimeout = resubTimeout;
+            AbstractClient.resubscribeInterval = resubscribeInterval;
             AbstractClient.subOnce = subOnce;
             AbstractClient.ifUseBackupSite = true;
 
@@ -771,8 +771,8 @@ 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, int resubTimeout, boolean subOnce) throws IOException, RuntimeException {
-        return subscribeInternal(host, port, tableName, actionName, handler, offset, reconnect, filter, deserializer, allowExistTopic, userName, passWord, msgAsTable, backupSites, resubTimeout, subOnce, true);
+                                                              List<String> backupSites, int resubscribeInterval, boolean subOnce) throws IOException, RuntimeException {
+        return subscribeInternal(host, port, tableName, actionName, handler, offset, reconnect, filter, deserializer, allowExistTopic, userName, passWord, msgAsTable, backupSites, resubscribeInterval, subOnce, true);
     }
 
     protected BlockingQueue<List<IMessage>> subscribeInternal(String host, int port,

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

@@ -165,8 +165,8 @@ public void run() {
                             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)
+                                // reconnect every info.resubscribeTimeout ms
+                                if (System.currentTimeMillis() - AbstractClient.lastExceptionTopicTimeMap.get(topic) <= AbstractClient.resubscribeInterval)
                                     continue;
 
                                 dispatcher.tryReconnect(topic);
@@ -175,8 +175,8 @@ public void run() {
                             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)
+                                // reconnect every info.resubscribeTimeout ms
+                                if (System.currentTimeMillis() - AbstractClient.lastExceptionTopicTimeMap.get(topic) <= AbstractClient.resubscribeInterval)
                                     continue;
 
                                 dispatcher.tryReconnect(topic);

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

@@ -89,12 +89,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, int resubTimeout, boolean subOnce) throws IOException {
-        if (resubTimeout < 0)
-            // resubTimeout default: 100ms
-            resubTimeout = 100;
+    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 resubscribeTimeout, boolean subOnce) throws IOException {
+        if (resubscribeTimeout < 0)
+            // resubscribeTimeout default: 100ms
+            resubscribeTimeout = 100;
 
-        BlockingQueue<List<IMessage>> queue = subscribeInternal(host, port, tableName, actionName, (MessageHandler) null, offset, reconnect, filter, deserializer, false, userName, passWord, msgAsTable, backupSites, resubTimeout, subOnce);
+        BlockingQueue<List<IMessage>> queue = subscribeInternal(host, port, tableName, actionName, (MessageHandler) null, offset, reconnect, filter, deserializer, false, userName, passWord, msgAsTable, backupSites, resubscribeTimeout, subOnce);
         topicPoller = new TopicPoller(queue);
         return topicPoller;
     }
@@ -208,7 +208,7 @@ protected void unsubscribeInternal(String host, int port, String tableName, Stri
                 if (AbstractClient.ifUseBackupSite) {
                     AbstractClient.ifUseBackupSite = false;
                     AbstractClient.subOnce = false;
-                    AbstractClient.resubTimeout = 100;
+                    AbstractClient.resubscribeInterval = 100;
                 }
                 log.info("Successfully unsubscribed table " + fullTableName);
             } catch (Exception ex) {

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

@@ -193,12 +193,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, int resubTimeout, boolean subOnce) throws IOException {
-        if (resubTimeout < 0)
-            // resubTimeout default: 100ms
-            resubTimeout = 100;
+    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 resubscribeTimeout, boolean subOnce) throws IOException {
+        if (resubscribeTimeout < 0)
+            // resubscribeTimeout default: 100ms
+            resubscribeTimeout = 100;
 
-        BlockingQueue<List<IMessage>> queue = subscribeInternal(host, port, tableName, actionName, handler, offset, reconnect, filter, deserializer, allowExistTopic, userName, passWord, msgAsTable, backupSites, resubTimeout, subOnce);
+        BlockingQueue<List<IMessage>> queue = subscribeInternal(host, port, tableName, actionName, handler, offset, reconnect, filter, deserializer, allowExistTopic, userName, passWord, msgAsTable, backupSites, resubscribeTimeout, subOnce);
         String topicStr = host + ":" + port + "/" + tableName + "/" + actionName;
         synchronized (queueHandlers) {
             queueHandlers.put(tableNameToTrueTopic.get(topicStr), new QueueHandlerBinder(queue, handler));
@@ -375,7 +375,7 @@ protected void unsubscribeInternal(String host, int port, String tableName, Stri
                     if (AbstractClient.ifUseBackupSite) {
                         AbstractClient.ifUseBackupSite = false;
                         AbstractClient.subOnce = false;
-                        AbstractClient.resubTimeout = 100;
+                        AbstractClient.resubscribeInterval = 100;
                     }
                     log.info("Successfully unsubscribed table " + fullTableName);
                 } catch (Exception ex) {

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

@@ -237,15 +237,15 @@ 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, int batchSize, int throttle, String userName, String password, List<String> backupSites, int resubTimeout, boolean subOnce) throws IOException {
+    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, List<String> backupSites, int resubscribeTimeout, boolean subOnce) throws IOException {
         if(batchSize<=0)
             throw new IllegalArgumentException("BatchSize must be greater than zero");
         if(throttle<0)
             throw new IllegalArgumentException("Throttle must be greater than or equal to zero");
-        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, false, backupSites, resubTimeout, subOnce);
+        if (resubscribeTimeout < 0)
+            // resubscribeTimeout default: 100ms
+            resubscribeTimeout = 100;
+        BlockingQueue<List<IMessage>> queue = subscribeInternal(host, port, tableName, actionName, handler, offset, reconnect, filter, deserializer, allowExistTopic, userName, password, false, backupSites, resubscribeTimeout, subOnce);
         HandlerLopper handlerLopper = new HandlerLopper(queue, handler, batchSize, throttle == 0 ? -1 : throttle);
         handlerLopper.start();
         String topicStr = host + ":" + port + "/" + tableName + "/" + actionName;
@@ -538,7 +538,7 @@ protected void unsubscribeInternal(String host, int port, String tableName, Stri
                     if (AbstractClient.ifUseBackupSite) {
                         AbstractClient.ifUseBackupSite = false;
                         AbstractClient.subOnce = false;
-                        AbstractClient.resubTimeout = 100;
+                        AbstractClient.resubscribeInterval = 100;
                     }
                     log.info("Successfully unsubscribed table " + fullTableName);
                 } catch (Exception ex) {