AJ-647: override EventClient unsubscribeInternal;

Author: chengyitian

src/com/xxdb/streaming/cep/EventClient.java

@@ -2,6 +2,8 @@
 
 import com.xxdb.DBConnection;
 import com.xxdb.comm.ErrorCodeInfo;
+import com.xxdb.data.BasicInt;
+import com.xxdb.data.BasicString;
 import com.xxdb.data.Entity;
 import com.xxdb.data.Utils;
 import com.xxdb.streaming.client.AbstractClient;
@@ -12,6 +14,7 @@
 import java.io.IOException;
 import java.net.SocketException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 
@@ -83,6 +86,52 @@ public void unsubscribe(String host, int port, String tableName, String actionNa
         unsubscribeInternal(host, port, tableName, actionName);
     }
 
+    @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 (trueTopicToSites) {
+                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);
+            }
+            log.info("Successfully unsubscribed table " + fullTableName);
+        } catch (Exception ex) {
+            throw ex;
+        } finally {
+            dbConn.close();
+        }
+        return;
+    }
+
 
     @Override
     protected boolean doReconnect(Site site) {

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

@@ -178,7 +178,7 @@ public class Site {
         long msgId;
         boolean reconnect;
         Vector filter = null;
-        boolean closed = false;
+        public boolean closed = false;
         boolean allowExistTopic = false;
         StreamDeserializer deserializer;
         String userName = "";

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

@@ -4,7 +4,7 @@
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 
-class QueueManager {
+public class QueueManager {
     private HashMap<String, BlockingQueue<List<IMessage>>> queueMap = new HashMap<String, BlockingQueue<List<IMessage>>>();
 
     public synchronized BlockingQueue<List<IMessage>> addQueue(String topic) {