Update ThreadedClientsubscribeReverseTest.java

Author: weisun123

test/com/xxdb/streaming/reverse/ThreadedClientsubscribeReverseTest.java

@@ -105,11 +105,9 @@ public void doEvent(IMessage msg) {
                 String script = String.format("insert into Receive values(%d,%s,%f)", Integer.parseInt(msg.getEntity(0).getString()), msg.getEntity(1).getString(), Double.valueOf(msg.getEntity(2).toString()));
                 conn.run(script);
                 System.out.println(msg.getEntity(0).getString());
-                Thread.sleep(5);
+                //Thread.sleep(5);
             } catch (IOException e) {
                 e.printStackTrace();
-            } catch (InterruptedException e) {
-                throw new RuntimeException(e);
             }
         }
     };
@@ -3423,6 +3421,7 @@ public void test_ThreadClient_subscribe_backupSites_server_disconnect_backupSite
         conn3.connect(HOST,port_list[1],"admin","123456");
         conn3.run(script1);
         conn3.run(script2);
+        conn3.run("n=100000;t=table(1..n as tag,timestamp(1..n) as ts,take(100.0,n) as data);" + "Trades.append!(t)");
         controller_conn.run("try{stopDataNode('"+HOST+":"+port_list[2]+"')}catch(ex){}");
         System.out.println(port_list[2]+"节点断掉啦---------------------------------------------------");
         Thread.sleep(10000);
@@ -3459,7 +3458,7 @@ public void test_ThreadClient_subscribe_backupSites_server_disconnect_backupSite
         conn2.run(script2);
         Vector filter1 = (Vector) conn.run("1..100000");
         List<String> backupSites = new ArrayList<>(Collections.singleton(HOST+":"+port_list[2]));
-        client.subscribe(HOST,port_list[1],"Trades","subTread1",MessageHandler_handler, -1,true,filter1, (StreamDeserializer) null,true,1000, 1,"admin","123456",backupSites,10,false);
+        client.subscribe(HOST,port_list[1],"Trades","subTread1",MessageHandler_handler, 0,true,filter1, (StreamDeserializer) null,true,1000, 1,"admin","123456",backupSites,10,true);
         System.out.println("Successful subscribe");
         conn1.run("n=100000;t=table(1..n as tag,timestamp(1..n) as ts,take(100.0,n) as data);" + "Trades.append!(t)");
         conn2.run("n=100000;t=table(1..n as tag,timestamp(1..n) as ts,take(100.0,n) as data);" + "Trades.append!(t)");
@@ -3474,14 +3473,21 @@ public void test_ThreadClient_subscribe_backupSites_server_disconnect_backupSite
         conn3.run(script2);
         conn3.run("n=100000;t=table(1..n as tag,timestamp(1..n) as ts,take(100.0,n) as data);" + "Trades.append!(t)");
         controller_conn.run("try{stopDataNode('"+HOST+":"+port_list[2]+"')}catch(ex){}");
+        //System.out.println(port_list[2]+"节点断掉啦---------------------------------------------------");
         Thread.sleep(10000);
         controller_conn.run("try{startDataNode('"+HOST+":"+port_list[2]+"')}catch(ex){}");
-        Thread.sleep(10000);
 
+        Thread.sleep(2000);
+        DBConnection conn4 = new DBConnection();
+        conn4.connect(HOST,port_list[2],"admin","123456");
+        conn4.run(script1);
+        conn4.run(script2);
+        conn4.run("n=100000;t=table(1..n as tag,timestamp(1..n) as ts,take(100.0,n) as data);" + "Trades.append!(t)");
+        Thread.sleep(6000);
         BasicTable row_num = (BasicTable)conn.run("select count(*) from Receive");
         System.out.println(row_num.getColumn(0).get(0));
         assertEquals("100000",row_num.getColumn(0).get(0).getString());
-        //client.unsubscribe(HOST,port_list[1],"Trades","subTread1");
+       // client.unsubscribe(HOST,port_list[1],"Trades","subTread1");
     }
 
     @Test(timeout = 180000)
@@ -3517,7 +3523,7 @@ public void doEvent(IMessage msg) {
         }
     };
     @Test(timeout = 180000)
-    public void test_ThreadClient_subscribe_backupSites_resubTimeout() throws Exception {
+    public void test_ThreadClient_subscribe_backupSites_resubTimeout1() throws Exception {
         DBConnection controller_conn = new DBConnection();
         controller_conn.connect(controller_host,controller_port,"admin","123456");
         controller_conn.run("try{startDataNode('"+HOST+":"+port_list[1]+"')}catch(ex){}");
@@ -3533,12 +3539,16 @@ public void test_ThreadClient_subscribe_backupSites_resubTimeout() throws Except
         conn1.connect(HOST,port_list[1],"admin","123456");
         conn1.run(script1);
         conn1.run(script2);
+        DBConnection conn2 = new DBConnection();
+        conn2.connect(HOST,port_list[2],"admin","123456");
+        conn2.run(script1);
+        conn2.run(script2);
         Vector filter1 = (Vector) conn.run("1..100000");
-        List<String> backupSites = new ArrayList<>(Collections.singleton(HOST+":"+PORT));
-        client.subscribe(HOST,port_list[1],"Trades","subTread1",MessageHandler_handler1, -1,true,filter1, (StreamDeserializer) null,true,1000, 1,"admin","123456",backupSites,10000,true);
+        List<String> backupSites = new ArrayList<>(Collections.singleton(HOST+":"+port_list[2]));
+        client.subscribe(HOST,port_list[1],"Trades","subTread1",MessageHandler_handler1, -1,true,filter1, (StreamDeserializer) null,true,1, 1,"admin","123456",backupSites,10000,true);
         System.out.println("Successful subscribe");
-        conn.run("n=100000;t=table(1..n as tag,now()+1..n as ts,rand(100.0,n) as data);" + "Trades.append!(t)");
         conn1.run("n=100000;t=table(1..n as tag,now()+1..n as ts,rand(100.0,n) as data);" + "Trades.append!(t)");
+        conn2.run("n=100000;t=table(1..n as tag,now()+1..n as ts,rand(100.0,n) as data);" + "Trades.append!(t)");
         Thread.sleep(1000);
         controller_conn.run("try{stopDataNode('"+HOST+":"+port_list[1]+"')}catch(ex){}");
         Thread.sleep(10000);
@@ -3548,6 +3558,82 @@ public void test_ThreadClient_subscribe_backupSites_resubTimeout() throws Except
         System.out.println(re.getString());
         //client.unsubscribe(HOST,port_list[1],"Trades","subTread1");
     }
+    public static MessageHandler MessageHandler_handler2 = new MessageHandler() {
+        List<Long> list = new ArrayList<>();
+
+        @Override
+        public void doEvent(IMessage msg) {
+            Long re = System.currentTimeMillis();
+           // client.
+            // String script = String.format("insert into Receive values(%d,%s,%f,%d)", Integer.parseInt(msg.getEntity(0).getString()), msg.getEntity(1).getString(), Double.valueOf(msg.getEntity(2).toString()),System.currentTimeMillis());
+            //conn.run(script);
+            //System.out.println("当前时间：" + re);
+           // list.add(re);
+            //System.out.println(list.toString());
+
+        }
+    };
+    @Test(timeout = 180000)
+    public void test_ThreadClient_subscribe_backupSites_resubTimeout() throws Exception {
+        DBConnection controller_conn = new DBConnection();
+        controller_conn.connect(controller_host,controller_port,"admin","123456");
+        controller_conn.run("try{startDataNode('"+HOST+":"+port_list[1]+"')}catch(ex){}");
+        controller_conn.run("sleep(1000)");
+        String script1 = "st1 = streamTable(1000000:0,`tag`ts`data,[INT,TIMESTAMP,DOUBLE])\n" +
+                "share(st1,`Trades)\t\n"
+                + "setStreamTableFilterColumn(objByName(`Trades),`tag)";
+        conn.run(script1);
+        String script2 = "st2 = streamTable(1000000:0,`tag`ts`data`now,[INT,TIMESTAMP,DOUBLE,TIMESTAMP])\n" +
+                "share(st2, `Receive)\t\n";
+        conn.run(script2);
+        DBConnection conn1 = new DBConnection();
+        conn1.connect(HOST,port_list[1],"admin","123456");
+        conn1.run(script1);
+        conn1.run(script2);
+        Vector filter1 = (Vector) conn.run("1..10000000");
+        List<String> backupSites = new ArrayList<>(Collections.singleton(HOST+":"+PORT));
+        client.subscribe(HOST,port_list[1],"Trades","subTread1",MessageHandler_handler2, -1,true,filter1, (StreamDeserializer) null,true,1, 1,"admin","123456",backupSites,10000,true);
+        System.out.println("Successful subscribe");
+        conn.run("n=10000000;t=table(1..n as tag,now()+1..n as ts,rand(100.0,n) as data);" + "Trades.append!(t)");
+        conn1.run("n=10000000;t=table(1..n as tag,now()+1..n as ts,rand(100.0,n) as data);" + "Trades.append!(t)");
+        System.out.println("Successful subscribe111");
+        class MyThread extends Thread {
+            @Override
+            public void run() {
+                    try {
+                        conn.run("n=1000000;t=table(1..n as tag,now()+1..n as ts,rand(100.0,n) as data);" + "Trades.append!(t)");
+                        conn1.run("n=1000000;t=table(1..n as tag,now()+1..n as ts,rand(100.0,n) as data);" + "Trades.append!(t)");
+                    } catch (Exception e) {
+                        // 捕获异常并打印错误信息
+                        System.err.println( e.getMessage());
+                    }
+            }
+        }
+        class MyThread1 extends Thread {
+            @Override
+            public void run() {
+                    try {
+                        controller_conn.run("try{stopDataNode('"+HOST+":"+port_list[1]+"')}catch(ex){}");
+                    } catch (Exception e) {
+                        // 捕获异常并打印错误信息
+                        System.err.println(e.getMessage());
+                    }
+            }
+        }
+//        MyThread thread = new MyThread();
+//        MyThread1 thread1 = new MyThread1();
+//        thread.start();
+//        Thread.sleep(5);
+//        thread1.start();
+//        thread.join();
+//        Thread.sleep(1000);
+//        Thread.sleep(5000);
+         controller_conn.run("try{startDataNode('"+HOST+":"+port_list[1]+"')}catch(ex){}");
+        //Thread.sleep(20000);
+        //BasicTable re = (BasicTable)conn.run("select tag ,now,deltas(now) from Receive where deltas(now)>0 order by  deltas(now) desc limit 1\n");
+        //System.out.println(re.getString());
+        //client.unsubscribe(HOST,port_list[1],"Trades","subTread1");
+    }
 
    // @Test(timeout = 180000)
     public void test_ThreadClient_subscribe_backupSites_server_disconnect_1() throws IOException, InterruptedException {