Rename TarantoolBinaryPackage => TarantoolBinaryPacket; fix TarantoolClusterClient.java:158

pda4code · pda4code · commit 6e63dfab7b51 · 2019-03-12T01:17:39.000+03:00
diff --git a/src/it/java/org/tarantool/TestTarantoolClient.java b/src/it/java/org/tarantool/TestTarantoolClient.java
@@ -1,6 +1,6 @@
 package org.tarantool;
 
-import org.tarantool.server.TarantoolBinaryPackage;
+import org.tarantool.server.TarantoolBinaryPacket;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -71,7 +71,7 @@ protected void reconnect(int retry, Throwable lastError) {
         }
 
         @Override
-        protected void complete(TarantoolBinaryPackage pack, CompletableFuture<?> q) {
+        protected void complete(TarantoolBinaryPacket pack, CompletableFuture<?> q) {
             super.complete(pack, q);
             Long code = pack.getCode();
             if (code != 0) {
diff --git a/src/main/java/org/tarantool/JDBCBridge.java b/src/main/java/org/tarantool/JDBCBridge.java
@@ -8,7 +8,7 @@
 import java.util.Map;
 
 import org.tarantool.jdbc.SQLResultSet;
-import org.tarantool.server.TarantoolBinaryPackage;
+import org.tarantool.server.TarantoolBinaryPacket;
 
 public class JDBCBridge {
     public static final JDBCBridge EMPTY = new JDBCBridge(Collections.<TarantoolBase.SQLMetaData>emptyList(), Collections.<List<Object>>emptyList());
@@ -17,7 +17,7 @@ public class JDBCBridge {
     final Map<String,Integer> columnsByName;
     final List<List<Object>> rows;
 
-    protected JDBCBridge(TarantoolBinaryPackage pack) {
+    protected JDBCBridge(TarantoolBinaryPacket pack) {
         this(SqlProtoUtils.getSQLMetadata(pack), SqlProtoUtils.getSQLData(pack));
     }
 
@@ -31,7 +31,7 @@ protected JDBCBridge(List<TarantoolBase.SQLMetaData> sqlMetadata, List<List<Obje
     }
 
     public static JDBCBridge query(TarantoolConnection connection, String sql, Object ... params) {
-        TarantoolBinaryPackage pack = connection.sql(sql, params);
+        TarantoolBinaryPacket pack = connection.sql(sql, params);
         return new JDBCBridge(pack);
     }
 
@@ -48,7 +48,7 @@ public static JDBCBridge mock(List<String> fields, List<List<Object>> values)  {
     }
 
     public static Object execute(TarantoolConnection connection, String sql, Object ... params) {
-        TarantoolBinaryPackage pack = connection.sql(sql, params);
+        TarantoolBinaryPacket pack = connection.sql(sql, params);
         Long rowCount = SqlProtoUtils.getSqlRowCount(pack);
         if(rowCount == null) {
             return new SQLResultSet(new JDBCBridge(pack));
diff --git a/src/main/java/org/tarantool/SqlProtoUtils.java b/src/main/java/org/tarantool/SqlProtoUtils.java
@@ -1,17 +1,16 @@
 package org.tarantool;
 
-import org.tarantool.server.TarantoolBinaryPackage;
+import org.tarantool.server.TarantoolBinaryPacket;
 
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.CompletableFuture;
 
 public abstract class SqlProtoUtils {
 
 
-    public static List<Map<String, Object>> readSqlResult(TarantoolBinaryPackage pack) {
+    public static List<Map<String, Object>> readSqlResult(TarantoolBinaryPacket pack) {
         List<List<?>> data = (List<List<?>>) pack.getBody().get(Key.DATA.getId());
 
         List<Map<String, Object>> values = new ArrayList<Map<String, Object>>(data.size());
@@ -26,12 +25,12 @@ public static List<Map<String, Object>> readSqlResult(TarantoolBinaryPackage pac
         return values;
     }
 
-    public static List<List<Object>> getSQLData(TarantoolBinaryPackage pack) {
+    public static List<List<Object>> getSQLData(TarantoolBinaryPacket pack) {
         return (List<List<Object>>) pack.getBody().get(Key.DATA.getId());
     }
 
 
-    public static List<TarantoolBase.SQLMetaData> getSQLMetadata(TarantoolBinaryPackage pack) {
+    public static List<TarantoolBase.SQLMetaData> getSQLMetadata(TarantoolBinaryPacket pack) {
         List<Map<Integer, Object>> meta = (List<Map<Integer, Object>>) pack.getBody().get(Key.SQL_METADATA.getId());
         List<TarantoolBase.SQLMetaData> values = new ArrayList<TarantoolBase.SQLMetaData>(meta.size());
         for (Map<Integer, Object> c : meta) {
@@ -40,7 +39,7 @@ public static List<TarantoolBase.SQLMetaData> getSQLMetadata(TarantoolBinaryPack
         return values;
     }
 
-    public static Long getSqlRowCount(TarantoolBinaryPackage pack) {
+    public static Long getSqlRowCount(TarantoolBinaryPacket pack) {
         Map<Key, Object> info = (Map<Key, Object>) pack.getBody().get(Key.SQL_INFO.getId());
         Number rowCount;
         if (info != null && (rowCount = ((Number) info.get(Key.SQL_ROW_COUNT.getId()))) != null) {
diff --git a/src/main/java/org/tarantool/TarantoolClientImpl.java b/src/main/java/org/tarantool/TarantoolClientImpl.java
@@ -1,7 +1,7 @@
 package org.tarantool;
 
 import org.tarantool.server.BinaryProtoUtils;
-import org.tarantool.server.TarantoolBinaryPackage;
+import org.tarantool.server.TarantoolBinaryPacket;
 import org.tarantool.server.TarantoolInstanceConnection;
 
 import java.io.IOException;
@@ -376,7 +376,7 @@ protected void readThread() {
         try {
             while (!Thread.currentThread().isInterrupted()) {
                 try {
-                    TarantoolBinaryPackage pack = readFromInstance();
+                    TarantoolBinaryPacket pack = readFromInstance();
 
                     CompletableFuture<?> future = getFuture(pack);
 
@@ -393,12 +393,13 @@ protected void readThread() {
         }
     }
 
-    protected CompletableFuture<?> getFuture(TarantoolBinaryPackage pack) {
+    protected CompletableFuture<?> getFuture(TarantoolBinaryPacket pack) {
         return futures.remove(pack.getSync());
     }
 
-    protected TarantoolBinaryPackage readFromInstance() throws IOException, InterruptedException {
-        return BinaryProtoUtils.readPacket(currConnection.getReadChannel());
+    protected TarantoolBinaryPacket readFromInstance() throws IOException, InterruptedException {
+//        return BinaryProtoUtils.readPacket(currConnection.getReadChannel());
+        return currConnection.readPacket();
     }
 
     protected void writeThread() {
@@ -437,15 +438,16 @@ protected void writeThread() {
 
     protected void sendToInstance(ByteBuffer writerBuffer) throws IOException {
 //        communicationProvider.writeBuffer(writerBuffer);
-        BinaryProtoUtils.writeFully(currConnection.getChannel(), writerBuffer);
+//        BinaryProtoUtils.writeFully(currConnection.getChannel(), writerBuffer);
+        currConnection.writeBuffer(writerBuffer);
     }
 
 
     protected void fail(CompletableFuture<?> q, Exception e) {
         q.completeExceptionally(e);
     }
 
-    protected void complete(TarantoolBinaryPackage pack, CompletableFuture<?> q) {
+    protected void complete(TarantoolBinaryPacket pack, CompletableFuture<?> q) {
         if (q != null) {
             long code = pack.getCode();
             if (code == 0) {
@@ -463,7 +465,7 @@ protected void complete(TarantoolBinaryPackage pack, CompletableFuture<?> q) {
         }
     }
 
-    protected void completeSql(CompletableFuture<?> q, TarantoolBinaryPackage pack) {
+    protected void completeSql(CompletableFuture<?> q, TarantoolBinaryPacket pack) {
         Long rowCount = SqlProtoUtils.getSqlRowCount(pack);
         if (rowCount!=null) {
             ((CompletableFuture) q).complete(rowCount);
diff --git a/src/main/java/org/tarantool/TarantoolClusterClient.java b/src/main/java/org/tarantool/TarantoolClusterClient.java
@@ -155,21 +155,19 @@ protected void connect(InstanceConnectionProvider communicationProvider) throws
     }
 
     @Override
-    protected TarantoolBinaryPackage readFromInstance() throws IOException, InterruptedException {
+    protected TarantoolBinaryPacket readFromInstance() throws IOException, InterruptedException {
 
         readSelector.select();
 
         SelectionKey selectedKey = readSelector.selectedKeys().iterator().next();
 
         TarantoolInstanceConnection connection = (TarantoolInstanceConnection) selectedKey.attachment();
-        ReadableByteChannel readChannel = connection
-                .getReadChannel();
 
-        return BinaryProtoUtils.readPacket(readChannel);
+        return connection.readPacket();
     }
 
     @Override
-    protected CompletableFuture<?> getFuture(TarantoolBinaryPackage pack) {
+    protected CompletableFuture<?> getFuture(TarantoolBinaryPacket pack) {
         Long sync = pack.getSync();
         if (!futuresSentToOldConnection.isEmpty()) {
             CompletableFuture<?> oldConnectionFuture = futuresSentToOldConnection.remove(sync);
diff --git a/src/main/java/org/tarantool/TarantoolConnection.java b/src/main/java/org/tarantool/TarantoolConnection.java
@@ -25,18 +25,18 @@ public TarantoolConnection(String username, String password, Socket socket) thro
 
     @Override
     protected List<?> exec(Code code, Object... args) {
-        TarantoolBinaryPackage responsePackage = writeAndRead(code, args);
+        TarantoolBinaryPacket responsePackage = writeAndRead(code, args);
         return (List) responsePackage.getBody().get(Key.DATA.getId());
     }
 
-    protected TarantoolBinaryPackage writeAndRead(Code code, Object... args) {
+    protected TarantoolBinaryPacket writeAndRead(Code code, Object... args) {
         try {
             ByteBuffer packet = BinaryProtoUtils.createPacket(code, syncId.incrementAndGet(), null, args);
 
             out.write(packet.array(), 0, packet.remaining());
             out.flush();
 
-            TarantoolBinaryPackage responsePackage = BinaryProtoUtils.readPacket(in);
+            TarantoolBinaryPacket responsePackage = BinaryProtoUtils.readPacket(in);
 
             Map<Integer, Object> headers = responsePackage.getHeaders();
             Map<Integer, Object> body = responsePackage.getBody();
@@ -75,17 +75,17 @@ public void close() {
 
     @Override
     public Long update(String sql, Object... bind) {
-        TarantoolBinaryPackage pack = sql(sql, bind);
+        TarantoolBinaryPacket pack = sql(sql, bind);
         return SqlProtoUtils.getSqlRowCount(pack);
     }
 
     @Override
     public List<Map<String, Object>> query(String sql, Object... bind) {
-        TarantoolBinaryPackage pack = sql(sql, bind);
+        TarantoolBinaryPacket pack = sql(sql, bind);
         return SqlProtoUtils.readSqlResult(pack);
     }
 
-    protected TarantoolBinaryPackage sql(String sql, Object[] bind) {
+    protected TarantoolBinaryPacket sql(String sql, Object[] bind) {
         return writeAndRead(Code.EXECUTE, Key.SQL_TEXT, sql, Key.SQL_BIND, bind);
     }
 
diff --git a/src/main/java/org/tarantool/server/BinaryProtoUtils.java b/src/main/java/org/tarantool/server/BinaryProtoUtils.java
@@ -1,7 +1,6 @@
 package org.tarantool.server;
 
 import org.tarantool.Base64;
-import org.tarantool.ByteBufferInputStream;
 import org.tarantool.Code;
 import org.tarantool.CommunicationException;
 import org.tarantool.CountInputStream;
@@ -28,7 +27,7 @@ public abstract class BinaryProtoUtils {
     private final static int DEFAULT_INITIAL_REQUEST_SIZE = 4096;
     public static final String WELCOME = "Tarantool ";
 
-    public static TarantoolBinaryPackage readPacket(InputStream inputStream) throws IOException {
+    public static TarantoolBinaryPacket readPacket(InputStream inputStream) throws IOException {
         int size = inputStream.read();
 
         CountInputStreamImpl msgStream = new CountInputStreamImpl(inputStream);
@@ -40,7 +39,7 @@ public static TarantoolBinaryPackage readPacket(InputStream inputStream) throws
             body = (Map<Integer, Object>) getMsgPackLite().unpack(msgStream);
         }
 
-        return new TarantoolBinaryPackage(headers, body);
+        return new TarantoolBinaryPacket(headers, body);
     }
 
     /**
@@ -78,7 +77,7 @@ public static TarantoolInstanceConnectionMeta connect(Socket socket, String user
             os.write(authPacket.array(), 0, authPacket.remaining());
             os.flush();
 
-            TarantoolBinaryPackage responsePackage = readPacket(socket.getInputStream());
+            TarantoolBinaryPacket responsePackage = readPacket(socket.getInputStream());
             Long code = (Long) responsePackage.getHeaders().get(Key.CODE.getId());
             if (code != 0) {
                 Object error = responsePackage.getBody().get(Key.ERROR.getId());
@@ -118,7 +117,7 @@ public static TarantoolInstanceConnectionMeta connect(SocketChannel channel, Str
         if (username != null && password != null) {
             ByteBuffer authPacket = createAuthPacket(username, password, salt);
             writeFully(channel, authPacket);
-            TarantoolBinaryPackage authResponse = readPacket(((ReadableByteChannel) channel));
+            TarantoolBinaryPacket authResponse = readPacket(((ReadableByteChannel) channel));
             Long code = (Long) authResponse.getHeaders().get(Key.CODE.getId());
             if (code != 0) {
                 Object error = authResponse.getBody().get(Key.ERROR.getId());
@@ -154,7 +153,7 @@ public static void writeFully(SocketChannel channel, ByteBuffer buffer) throws I
      * @throws
      * @throws java.nio.channels.NonReadableChannelException – If this channel was not opened for reading
      */
-    public static TarantoolBinaryPackage readPacket(ReadableByteChannel bufferReader)
+    public static TarantoolBinaryPacket readPacket(ReadableByteChannel bufferReader)
             throws CommunicationException, IOException {
 
         ByteBuffer buffer = ByteBuffer.allocate(LENGTH_OF_SIZE_MESSAGE);
@@ -189,11 +188,11 @@ public static TarantoolBinaryPackage readPacket(ReadableByteChannel bufferReader
             body = (Map<Integer, Object>) unpackedBody;
         }
 
-        return new TarantoolBinaryPackage(headers, body);
+        return new TarantoolBinaryPacket(headers, body);
     }
 
     @Deprecated
-    private static TarantoolBinaryPackage readPacket(CountInputStream inputStream) throws IOException {
+    private static TarantoolBinaryPacket readPacket(CountInputStream inputStream) throws IOException {
         int size = ((Number) getMsgPackLite().unpack(inputStream)).intValue();
 
         long mark = inputStream.getBytesRead();
@@ -219,7 +218,7 @@ private static TarantoolBinaryPackage readPacket(CountInputStream inputStream) t
             body = (Map<Integer, Object>) unpackedBody;
         }
 
-        return new TarantoolBinaryPackage(headers, body);
+        return new TarantoolBinaryPacket(headers, body);
     }
 
 
diff --git a/src/main/java/org/tarantool/server/TarantoolBinaryPacket.java b/src/main/java/org/tarantool/server/TarantoolBinaryPacket.java
@@ -5,16 +5,16 @@
 import java.util.Collections;
 import java.util.Map;
 
-public class TarantoolBinaryPackage {
+public class TarantoolBinaryPacket {
     private final Map<Integer, Object> headers;
     private final Map<Integer, Object> body;
 
-    public TarantoolBinaryPackage(Map<Integer, Object> headers, Map<Integer, Object> body) {
+    public TarantoolBinaryPacket(Map<Integer, Object> headers, Map<Integer, Object> body) {
         this.headers = headers;
         this.body = body;
     }
 
-    public TarantoolBinaryPackage(Map<Integer, Object> headers) {
+    public TarantoolBinaryPacket(Map<Integer, Object> headers) {
         this.headers = headers;
         body = null;
     }
diff --git a/src/main/java/org/tarantool/server/TarantoolInstanceConnection.java b/src/main/java/org/tarantool/server/TarantoolInstanceConnection.java
@@ -4,7 +4,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.nio.channels.ReadableByteChannel;
+import java.nio.ByteBuffer;
 import java.nio.channels.SocketChannel;
 
 public class TarantoolInstanceConnection implements Closeable {
@@ -74,13 +74,17 @@ public SocketChannel getChannel() {
         return channel;
     }
 
-    public ReadableByteChannel getReadChannel() {
-        return readChannel;
+    public void writeBuffer(ByteBuffer writerBuffer) throws IOException {
+        BinaryProtoUtils.writeFully(getChannel(), writerBuffer);
+    }
+
+    public TarantoolBinaryPacket readPacket() throws IOException {
+        return BinaryProtoUtils.readPacket(readChannel);
     }
 
     private void closeConnection() {
         try {
-            readChannel.close();//also closes the channel
+            readChannel.close();//also closes this.channel
         } catch (IOException ignored) {
 
         }
diff --git a/src/test/java/org/tarantool/ClientReconnectClusterIT.java b/src/test/java/org/tarantool/ClientReconnectClusterIT.java
diff --git a/src/test/java/org/tarantool/jdbc/JdbcExceptionHandlingTest.java b/src/test/java/org/tarantool/jdbc/JdbcExceptionHandlingTest.java
