fix: affectRows

src/client/src/TSDBJNIConnector.c

@@ -344,7 +344,7 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_getResultSetImp(
   STscObj *pObj = pSql->pTscObj;
 
   if (tscIsUpdateQuery(pSql)) {
-    taos_free_result(pSql);  // free result here
+    // taos_free_result(pSql);  // free result here
     jniTrace("jobj:%p, conn:%p, no resultset, %p", jobj, pObj, (void *)tres);
     return 0;
   } else {

src/connector/jdbc/pom.xml

@@ -62,6 +62,12 @@
 			<artifactId>commons-lang3</artifactId>
 			<version>${commons-lang3.version}</version>
 		</dependency>
+		<dependency>
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+			<version>4.8.2</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<build>
@@ -98,6 +104,14 @@
 					<showDeprecation>true</showDeprecation>
 				</configuration>
 			</plugin>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-surefire-plugin</artifactId>
+				<version>2.12.4</version>
+				<configuration>
+					<testFailureIgnore>true</testFailureIgnore>
+				</configuration>
+			</plugin>
 		</plugins>
 	</build>
 

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

@@ -51,8 +51,6 @@ public class TSDBJNIConnector {
 
     /**
      * Returns the status of last result set in current connection
-     *
-     * @return
      */
     public boolean isResultsetClosed() {
         return this.isResultsetClosed;
@@ -112,7 +110,7 @@ public class TSDBJNIConnector {
      *
      * @throws SQLException
      */
-    public int executeQuery(String sql) throws SQLException {
+    public long executeQuery(String sql) throws SQLException {
         if (!this.isResultsetClosed) {
             freeResultSet(taosResultSetPointer);
         }
@@ -127,7 +125,6 @@ public class TSDBJNIConnector {
         }
         int code = this.getErrCode(pSql);
 
-        affectedRows = code;
         if (code < 0) {
             affectedRows = -1;
             if (code == TSDBConstants.JNI_TDENGINE_ERROR) {
@@ -146,7 +143,7 @@ public class TSDBJNIConnector {
         if (taosResultSetPointer != TSDBConstants.JNI_NULL_POINTER) {
             isResultsetClosed = false;
         }
-        return code;
+        return pSql;
     }
 
     private native long executeQueryImp(byte[] sqlBytes, long connection);
@@ -199,8 +196,6 @@ public class TSDBJNIConnector {
     /**
      * Close the open result set which is associated to the current connection. If the result set is already
      * closed, return 0 for success.
-     *
-     * @return
      */
     public int freeResultSet() {
         int resCode = TSDBConstants.JNI_SUCCESS;
@@ -217,7 +212,7 @@ public class TSDBJNIConnector {
     /**
      * Get affected rows count
      */
-    public int getAffectedRows(Long pSql) {
+    public int getAffectedRows(long pSql) {
         int affectedRows = this.affectedRows;
         if (affectedRows < 0) {
             affectedRows = this.getAffectedRowsImp(this.taos, pSql);
@@ -225,7 +220,7 @@ public class TSDBJNIConnector {
         return affectedRows;
     }
 
-    private native int getAffectedRowsImp(long connection, Long pSql);
+    private native int getAffectedRowsImp(long connection, long pSql);
 
     /**
      * Get schema metadata

src/connector/jdbc/src/test/java/com/taosdata/jdbc/TSDBAsyncSubscribeTest.java

@@ -0,0 +1,96 @@
+package com.taosdata.jdbc;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Properties;
+
+import static org.junit.Assert.assertTrue;
+
+public class TSDBAsyncSubscribeTest {
+    Connection connection = null;
+    Statement statement = null;
+    String dbName = "test";
+    String tName = "t0";
+    String host = "localhost";
+    String topic = "test";
+    long subscribId = 0;
+
+//    @Before
+    public void createDatabase() throws SQLException {
+        try {
+            Class.forName("com.taosdata.jdbc.TSDBDriver");
+        } catch (ClassNotFoundException e) {
+            return;
+        }
+        Properties properties = new Properties();
+        properties.setProperty(TSDBDriver.PROPERTY_KEY_HOST, host);
+        connection = DriverManager.getConnection("jdbc:TAOS://" + host + ":0/" + "?user=root&password=taosdata"
+                , properties);
+
+        statement = connection.createStatement();
+        statement.executeUpdate("create database if not exists " + dbName);
+        statement.executeUpdate("create table if not exists " + dbName + "." + tName + " (ts timestamp, k int, v int)");
+        long ts = System.currentTimeMillis();
+        for (int i = 0; i < 5; i++) {
+            ts += i;
+            statement.executeUpdate("insert into \" + dbName + \".\" + tName + \" values (" + ts + ", " + (100 + i) + ", " + i + ")");
+        }
+    }
+
+//    @Test
+    public void subscribe() throws Exception {
+        TSDBSubscribe subscribe = null;
+        try {
+            String rawSql = "select * from " + dbName + "." + tName + ";";
+            System.out.println(rawSql);
+            subscribe = ((TSDBConnection) connection).createSubscribe();
+            subscribId = subscribe.subscribe(topic, rawSql, false, 1000, new CallBack("first"));
+
+            assertTrue(subscribId > 0);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+        Thread.sleep(2000);
+        subscribe.unsubscribe(subscribId, true);
+    }
+
+    private static class CallBack implements TSDBSubscribeCallBack {
+        private String name = "";
+
+        public CallBack(String name) {
+            this.name = name;
+        }
+
+        @Override
+        public void invoke(TSDBResultSet resultSet) {
+             System.out.println("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();
+                }
+                resultSet.close();
+
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+//    @After
+    public void close() throws Exception {
+        statement.executeQuery("drop database test");
+        statement.close();
+        connection.close();
+    }
+}

src/connector/jdbc/src/test/java/com/taosdata/jdbc/TSDBImportTest.java

@@ -0,0 +1,77 @@
+package com.taosdata.jdbc;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+
+public class TSDBImportTest {
+    Connection connection = null;
+    Statement statement = null;
+    String dbName = "test";
+    String tName = "t0";
+    String host = "localhost";
+
+    @Before
+    public void createDatabase() throws SQLException {
+        try {
+            Class.forName("com.taosdata.jdbc.TSDBDriver");
+        } catch (ClassNotFoundException e) {
+            return;
+        }
+        Properties properties = new Properties();
+        properties.setProperty(TSDBDriver.PROPERTY_KEY_HOST, host);
+        connection = DriverManager.getConnection("jdbc:TAOS://" + host + ":0/" + "?user=root&password=taosdata"
+                , properties);
+
+        statement = connection.createStatement();
+        statement.executeUpdate("drop database if exists " + dbName);
+        statement.executeUpdate("create database if not exists " + dbName);
+        statement.executeUpdate("create table if not exists " + dbName + "." + tName + " (ts timestamp, k int, v int)");
+
+    }
+
+    @Test
+    public void insertInto() throws Exception {
+        long ts = System.currentTimeMillis();
+
+        for (int i = 0; i < 50; i++) {
+            ts += i;
+            int row = statement.executeUpdate("insert into " + dbName + "." + tName + " values (" + ts + ", " + (100 + i) + ", " + i + ")");
+            System.out.println("insert into " + dbName + "." + tName + " values (" + ts + ", " + (100 + i) + ", " + i + ")" + "\t" + row);
+            assertEquals(1, row);
+        }
+    }
+
+    @Test
+    public void importInto() throws Exception {
+        // 避免时间重复
+        long ts = System.currentTimeMillis() + 1000;
+
+        StringBuilder sqlBuilder = new StringBuilder("insert into ").append(dbName).append(".").append(tName).append(" values ");
+
+        for (int i = 0; i < 50; i++) {
+            int a = i / 5;
+            long t = ts + a;
+            sqlBuilder.append("(").append(t).append(",").append((100 + i)).append(",").append(i).append(") ");
+        }
+        System.out.println(sqlBuilder.toString());
+        int rows = statement.executeUpdate(sqlBuilder.toString());
+        System.out.println(rows);
+        assertEquals(10, rows);
+    }
+
+    @After
+    public void close() throws Exception {
+        statement.executeQuery("drop database " + dbName);
+        statement.close();
+        connection.close();
+    }
+}

src/connector/jdbc/src/test/java/com/taosdata/jdbc/TSDBSubscribeTest.java

@@ -0,0 +1,90 @@
+package com.taosdata.jdbc;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Properties;
+
+import static org.junit.Assert.assertTrue;
+
+public class TSDBSubscribeTest {
+    Connection connection = null;
+    Statement statement = null;
+    String dbName = "test";
+    String tName = "t0";
+    String host = "localhost";
+    String topic = "test";
+
+//    @Before
+    public void createDatabase() throws SQLException {
+        try {
+            Class.forName("com.taosdata.jdbc.TSDBDriver");
+        } catch (ClassNotFoundException e) {
+            return;
+        }
+        Properties properties = new Properties();
+        properties.setProperty(TSDBDriver.PROPERTY_KEY_HOST, host);
+        connection = DriverManager.getConnection("jdbc:TAOS://" + host + ":0/" + "?user=root&password=taosdata"
+                , properties);
+
+        statement = connection.createStatement();
+        statement.executeUpdate("create database if not exists " + dbName);
+        statement.executeUpdate("create table if not exists " + dbName + "." + tName + " (ts timestamp, k int, v int)");
+        long ts = System.currentTimeMillis();
+        for (int i = 0; i < 5; i++) {
+            ts += i;
+            statement.executeUpdate("insert into \" + dbName + \".\" + tName + \" values (" + ts + ", " + (100 + i) + ", " + i + ")");
+        }
+    }
+
+//    @Test
+    public void subscribe() throws Exception {
+        TSDBSubscribe subscribe = null;
+        long subscribId = 0;
+        try {
+
+            String rawSql = "select * from " + dbName + "." + tName + ";";
+            System.out.println(rawSql);
+            subscribe = ((TSDBConnection) connection).createSubscribe();
+            subscribId = subscribe.subscribe(topic, rawSql, false, 1000);
+
+            assertTrue(subscribId > 0);
+
+            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 + "==========");
+                }
+                resSet.close();
+                a++;
+                if (a >= 3) {
+                    break;
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            if (null != subscribe && 0 != subscribId) {
+                subscribe.unsubscribe(subscribId, true);
+            }
+        }
+    }
+
+//    @After
+    public void close() throws Exception {
+        statement.executeQuery("drop database " + dbName);
+        statement.close();
+        connection.close();
+    }
+}