feat: add subscribe

src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBConnection.java

@@ -84,6 +84,14 @@ public class TSDBConnection implements Connection {
 		}
 	}
 
+	public TSDBSubscribe createSubscribe() throws SQLException {
+		if (!this.connector.isClosed()) {
+			return new TSDBSubscribe(this.connector);
+		} else {
+			throw new SQLException(TSDBConstants.FixErrMsg(TSDBConstants.JNI_CONNECTION_NULL));
+		}
+	}
+
 	public PreparedStatement prepareStatement(String sql) throws SQLException {
 		if (!this.connector.isClosed()) {
 			return new TSDBPreparedStatement(this.connector, sql);

src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBJNIConnector.java

@@ -22,8 +22,8 @@ public class TSDBJNIConnector {
     static volatile Boolean isInitialized = false;
 
     static {
-        System.loadLibrary("taos");
         System.out.println("java.library.path:" + System.getProperty("java.library.path"));
+        System.loadLibrary("taos");
     }
 
     /**
@@ -261,31 +261,31 @@ public class TSDBJNIConnector {
     /**
      * Subscribe to a table in TSDB
      */
-    public long subscribe(String host, String user, String password, String database, String table, long time, int period) {
-        return subscribeImp(host, user, password, database, table, time, period);
+    public long subscribe(String topic, String sql, boolean restart, int period) {
+        return subscribeImp(this.taos, restart, topic, sql, period);
     }
 
-    private native long subscribeImp(String host, String user, String password, String database, String table, long time, int period);
+    public native long subscribeImp(long connection, boolean restart, String topic, String sql, int period);
 
     /**
      * Consume a subscribed table
      */
-    public TSDBResultSetRowData consume(long subscription) {
+    public long consume(long subscription) {
         return this.consumeImp(subscription);
     }
 
-    private native TSDBResultSetRowData consumeImp(long subscription);
+    private native long consumeImp(long subscription);
 
     /**
      * Unsubscribe a table
      *
      * @param subscription
      */
-    public void unsubscribe(long subscription) {
-        unsubscribeImp(subscription);
+    public void unsubscribe(long subscription, boolean isKeep) {
+        unsubscribeImp(subscription, isKeep);
     }
 
-    private native void unsubscribeImp(long subscription);
+    private native void unsubscribeImp(long subscription, boolean isKeep);
 
     /**
      * Validate if a <I>create table</I> sql statement is correct without actually creating that table
@@ -293,7 +293,7 @@ public class TSDBJNIConnector {
     public boolean validateCreateTableSql(String sql) {
         long connection = taos;
         int res = validateCreateTableSqlImp(connection, sql.getBytes());
-        return res != 0 ? false : true;
+        return res == 0;
     }
 
     private native int validateCreateTableSqlImp(long connection, byte[] sqlBytes);

src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBSubscribe.java

+/***************************************************************************
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ *****************************************************************************/
+package com.taosdata.jdbc;
+
+import javax.management.OperationsException;
+import java.sql.SQLException;
+import java.util.Map;
+import java.util.TimerTask;
+import java.util.concurrent.*;
+
+public class TSDBSubscribe {
+    private TSDBJNIConnector connecter = null;
+    private static ScheduledExecutorService pool;
+    private static Map<Long, TSDBTimerTask> timerTaskMap = new ConcurrentHashMap<>();
+    private static Map<Long, ScheduledFuture> scheduledMap = new ConcurrentHashMap();
+
+    private static class TimerInstance {
+        private static final ScheduledExecutorService instance = Executors.newScheduledThreadPool(1);
+    }
+
+    public static ScheduledExecutorService getTimerInstance() {
+        return TimerInstance.instance;
+    }
+
+    public TSDBSubscribe(TSDBJNIConnector connecter) throws SQLException {
+        if (null != connecter) {
+            this.connecter = connecter;
+        } else {
+            throw new SQLException(TSDBConstants.FixErrMsg(TSDBConstants.JNI_CONNECTION_NULL));
+        }
+    }
+
+    public long subscribe(String topic, String sql, boolean restart, int period) throws SQLException {
+        if (this.connecter.isClosed()) {
+            throw new SQLException(TSDBConstants.FixErrMsg(TSDBConstants.JNI_CONNECTION_NULL));
+        }
+        if (period < 1000) {
+            throw new SQLException(TSDBConstants.WrapErrMsg(TSDBConstants.INVALID_VARIABLES));
+        }
+        return this.connecter.subscribe(topic, sql, restart, period);
+    }
+
+    public long subscribe(String topic, String sql, boolean restart, int period, TSDBSubscribeCallBack callBack) throws SQLException {
+        if (this.connecter.isClosed()) {
+            throw new SQLException(TSDBConstants.FixErrMsg(TSDBConstants.JNI_CONNECTION_NULL));
+        }
+        final long subscription = this.connecter.subscribe(topic, sql, restart, period);
+        if (null != callBack) {
+            pool = getTimerInstance();
+
+            TSDBTimerTask timerTask = new TSDBTimerTask(subscription, callBack);
+
+            timerTaskMap.put(subscription, timerTask);
+
+            ScheduledFuture scheduledFuture = pool.scheduleAtFixedRate(timerTask, 1, 1000, TimeUnit.MILLISECONDS);
+            scheduledMap.put(subscription, scheduledFuture);
+        }
+        return subscription;
+    }
+
+    public TSDBResultSet consume(long subscription) throws OperationsException, SQLException {
+        if (this.connecter.isClosed()) {
+            throw new SQLException(TSDBConstants.FixErrMsg(TSDBConstants.JNI_CONNECTION_NULL));
+        }
+        if (0 == subscription) {
+            throw new OperationsException("Invalid use of consume");
+        }
+        long resultSetPointer = this.connecter.consume(subscription);
+
+        if (resultSetPointer == TSDBConstants.JNI_CONNECTION_NULL) {
+            throw new SQLException(TSDBConstants.FixErrMsg(TSDBConstants.JNI_CONNECTION_NULL));
+        } else if (resultSetPointer == TSDBConstants.JNI_NULL_POINTER) {
+            return null;
+        } else {
+            return new TSDBResultSet(this.connecter, resultSetPointer);
+        }
+    }
+
+    public void unsubscribe(long subscription, boolean isKeep) throws SQLException {
+        if (this.connecter.isClosed()) {
+            throw new SQLException(TSDBConstants.FixErrMsg(TSDBConstants.JNI_CONNECTION_NULL));
+        }
+
+        synchronized (timerTaskMap.get(subscription)) {
+            while (1 == timerTaskMap.get(subscription).getState()) {
+                try {
+                    Thread.sleep(10);
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            }
+            timerTaskMap.get(subscription).setState(2);
+            if (!timerTaskMap.isEmpty() && timerTaskMap.containsKey(subscription)) {
+                timerTaskMap.get(subscription).cancel();
+                timerTaskMap.remove(subscription);
+                scheduledMap.get(subscription).cancel(false);
+                scheduledMap.remove(subscription);
+            }
+            this.connecter.unsubscribe(subscription, isKeep);
+        }
+    }
+
+    class TSDBTimerTask extends TimerTask {
+        private long subscription;
+        private TSDBSubscribeCallBack callBack;
+        // 0: not running 1: running 2: cancel
+        private int state = 0;
+
+        public TSDBTimerTask(long subscription, TSDBSubscribeCallBack callBack) {
+            this.subscription = subscription;
+            this.callBack = callBack;
+        }
+
+        public int getState() {
+            return this.state;
+        }
+
+        public void setState(int state) {
+            this.state = state;
+        }
+
+        @Override
+        public void run() {
+            synchronized (this) {
+                if (2 == state) {
+                    return;
+                }
+
+                state = 1;
+
+                try {
+                    TSDBResultSet resultSet = consume(subscription);
+                    callBack.invoke(resultSet);
+                } catch (Exception e) {
+                    this.cancel();
+                    throw new RuntimeException(e);
+                }
+                state = 0;
+            }
+        }
+    }
+}
+

src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBSubscribeCallBack.java

+/***************************************************************************
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ *****************************************************************************/
+package com.taosdata.jdbc;
+
+public interface TSDBSubscribeCallBack {
+    void invoke(TSDBResultSet resultSet);
+}

src/connector/jdbc/src/test/java/TestAsyncTSDBSubscribe.java

+import com.taosdata.jdbc.*;
+import org.apache.commons.lang3.StringUtils;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.Properties;
+
+public class TestAsyncTSDBSubscribe {
+    public static void main(String[] args) {
+        String usage = "java -cp taos-jdbcdriver-1.0.3_dev-dist.jar com.taosdata.jdbc.TSDBSubscribe -db dbName -topic topicName " +
+                "-tname tableName -h host";
+        if (args.length < 2) {
+            System.err.println(usage);
+            return;
+        }
+
+        String dbName = "";
+        String tName = "";
+        String host = "localhost";
+        String topic = "";
+        for (int i = 0; i < args.length; i++) {
+            if ("-db".equalsIgnoreCase(args[i]) && i < args.length - 1) {
+                dbName = args[++i];
+            }
+            if ("-tname".equalsIgnoreCase(args[i]) && i < args.length - 1) {
+                tName = args[++i];
+            }
+            if ("-h".equalsIgnoreCase(args[i]) && i < args.length - 1) {
+                host = args[++i];
+            }
+            if ("-topic".equalsIgnoreCase(args[i]) && i < args.length - 1) {
+                topic = args[++i];
+            }
+        }
+        if (StringUtils.isEmpty(dbName) || StringUtils.isEmpty(tName) || StringUtils.isEmpty(topic)) {
+            System.err.println(usage);
+            return;
+        }
+
+        Connection connection = null;
+        TSDBSubscribe subscribe = null;
+        long subscribId = 0;
+        try {
+            Class.forName("com.taosdata.jdbc.TSDBDriver");
+            Properties properties = new Properties();
+            properties.setProperty(TSDBDriver.PROPERTY_KEY_HOST, host);
+            connection = DriverManager.getConnection("jdbc:TAOS://" + host + ":0/" + dbName + "?user=root&password=taosdata", properties);
+            String rawSql = "select * from " + tName + ";";
+            subscribe = ((TSDBConnection) connection).createSubscribe();
+            subscribId = subscribe.subscribe(topic, rawSql, false, 1000, new CallBack("first"));
+            long subscribId2 = subscribe.subscribe("test", rawSql, false, 1000, new CallBack("second"));
+            int a = 0;
+            Thread.sleep(2000);
+            subscribe.unsubscribe(subscribId, true);
+            System.err.println("cancel subscribe");
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    private static class CallBack implements TSDBSubscribeCallBack {
+        private String name = "";
+
+        public CallBack(String name) {
+            this.name = name;
+        }
+
+        @Override
+        public void invoke(TSDBResultSet resultSet) {
+            try {
+                while (null !=resultSet && resultSet.next()) {
+                    System.out.print("callback_" + name + ": ");
+                    for (int i = 1; i <= resultSet.getMetaData().getColumnCount(); i++) {
+                        System.out.printf(i + ": " + resultSet.getString(i) + "\t");
+                    }
+                    System.out.println();
+                }
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+    }
+}

src/connector/jdbc/src/test/java/TestPreparedStatement.java

@@ -11,8 +11,14 @@ public class TestPreparedStatement {
             Class.forName("com.taosdata.jdbc.TSDBDriver");
             Properties properties = new Properties();
             properties.setProperty(TSDBDriver.PROPERTY_KEY_HOST, "192.168.1.117");
-            Connection connection = DriverManager.getConnection("jdbc:TAOS://192.168.1.117:0/?user=root&password=taosdata", properties);
+            Connection connection = DriverManager.getConnection("jdbc:TAOS://10.211.55.3:0/log?user=root&password=taosdata", properties);
+            String createSql = "create table t (ts timestamp, speed int);";
+            Statement statement = connection.createStatement();
+            statement.executeQuery(createSql);
             String rawSql = "SELECT ts, c1 FROM (select c1, ts from db.tb1) SUB_QRY";
+            if (1 < 2) {
+                return;
+            }
 //            String[] params = new String[]{"ts", "c1"};
             PreparedStatement pstmt = (TSDBPreparedStatement) connection.prepareStatement(rawSql);
             ResultSet resSet = pstmt.executeQuery();

src/connector/jdbc/src/test/java/TestTSDBResultSetRowData.java

+import com.taosdata.jdbc.ColumnMetaData;
+import com.taosdata.jdbc.DatabaseMetaDataResultSet;
+import com.taosdata.jdbc.TSDBResultSetRowData;
+
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+public class TestTSDBResultSetRowData {
+    public static void main(String[] args) throws SQLException {
+        DatabaseMetaDataResultSet resultSet = new DatabaseMetaDataResultSet();
+        List<ColumnMetaData> columnMetaDataList = new ArrayList(1);
+        ColumnMetaData colMetaData = new ColumnMetaData();
+        colMetaData.setColIndex(0);
+        colMetaData.setColName("TABLE_TYPE");
+        colMetaData.setColSize(10);
+        colMetaData.setColType(8);
+        columnMetaDataList.add(colMetaData);
+
+        List<TSDBResultSetRowData> rowDataList = new ArrayList(2);
+        TSDBResultSetRowData rowData = new TSDBResultSetRowData(2);
+        rowData.setString(0, "TABLE");
+        rowDataList.add(rowData);
+        rowData = new TSDBResultSetRowData(2);
+        rowData.setString(0, "STABLE");
+        rowDataList.add(rowData);
+
+        resultSet.setColumnMetaDataList(columnMetaDataList);
+        resultSet.setRowDataList(rowDataList);
+
+        while (resultSet.next()) {
+            System.out.println(resultSet.getString(1));
+        }
+    }
+}

src/connector/jdbc/src/test/java/TestTSDBSubscribe.java

+import com.taosdata.jdbc.TSDBConnection;
+import com.taosdata.jdbc.TSDBDriver;
+import com.taosdata.jdbc.TSDBResultSet;
+import com.taosdata.jdbc.TSDBSubscribe;
+import org.apache.commons.lang3.StringUtils;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.Properties;
+
+public class TestTSDBSubscribe {
+    public static void main(String[] args) throws Exception {
+        String usage = "java -cp taos-jdbcdriver-1.0.3_dev-dist.jar com.taosdata.jdbc.TSDBSubscribe -db dbName " +
+                "-topic topicName -tname tableName -h host";
+        if (args.length < 2) {
+            System.err.println(usage);
+            return;
+        }
+
+        String dbName = "";
+        String tName = "";
+        String host = "localhost";
+        String topic = "";
+        for (int i = 0; i < args.length; i++) {
+            if ("-db".equalsIgnoreCase(args[i]) && i < args.length - 1) {
+                dbName = args[++i];
+            }
+            if ("-tname".equalsIgnoreCase(args[i]) && i < args.length - 1) {
+                tName = args[++i];
+            }
+            if ("-h".equalsIgnoreCase(args[i]) && i < args.length - 1) {
+                host = args[++i];
+            }
+            if ("-topic".equalsIgnoreCase(args[i]) && i < args.length - 1) {
+                topic = args[++i];
+            }
+        }
+        if (StringUtils.isEmpty(dbName) || StringUtils.isEmpty(tName) || StringUtils.isEmpty(topic)) {
+            System.err.println(usage);
+            return;
+        }
+
+        Connection connection = null;
+        TSDBSubscribe subscribe = null;
+        long subscribId = 0;
+        try {
+            Class.forName("com.taosdata.jdbc.TSDBDriver");
+            Properties properties = new Properties();
+            properties.setProperty(TSDBDriver.PROPERTY_KEY_HOST, host);
+            connection = DriverManager.getConnection("jdbc:TAOS://" + host + ":0/" + dbName + "?user=root&password=taosdata"
+                    , properties);
+            String rawSql = "select * from " + tName + ";";
+            subscribe = ((TSDBConnection) connection).createSubscribe();
+            subscribId = subscribe.subscribe(topic, rawSql, false, 1000);
+            int a = 0;
+            while (true) {
+                Thread.sleep(900);
+                TSDBResultSet resSet = subscribe.consume(subscribId);
+
+                while (resSet.next()) {
+                    for (int i = 1; i <= resSet.getMetaData().getColumnCount(); i++) {
+                        System.out.printf(i + ": " + resSet.getString(i) + "\t");
+                    }
+                    System.out.println("\n======" + a + "==========");
+                }
+
+                a++;
+                if (a >= 10) {
+                    break;
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            if (null != subscribe && 0 != subscribId) {
+                subscribe.unsubscribe(subscribId, true);
+            }
+            if (null != connection) {
+                connection.close();
+            }
+        }
+    }
+}