AJ-676: fix issue about streaming backupSites;

Author: chengyitian

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

@@ -16,6 +16,14 @@ public synchronized BlockingQueue<List<IMessage>> addQueue(String topic) {
         throw new RuntimeException("Topic " + topic + " already subscribed");
     }
 
+    public synchronized BlockingQueue<List<IMessage>> changeQueue(String topic, BlockingQueue<List<IMessage>> q) {
+        if (queueMap.containsKey(topic)) {
+            queueMap.put(topic, q);
+            return q;
+        }
+        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;

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

@@ -186,6 +186,10 @@ protected boolean doReconnect(Site site) {
                     // 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);
+                    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);
                     // System.out.println("切换后 handlerLoppers: " + handlerLoppers.get(topicStr).getName());
                     // System.out.println("切换成功后，handlerLoppers size: " + handlerLoppers.size());
                     Date d = new Date();