Skip to content

Commit 9b3f91f

Browse files
dponomarevTotktonada
dponomarev
authored and
Totktonada
committed
Extract read/write operations into util classes
Create TarantoolPacket class that describes tarantool binary protocol message. Move logic of sending bytes, receiving binary messages (packets), creating packet and connecting to a tarantool instance from TarantoolBase and TarantoolClientImpl to ProtoUtils class.
1 parent 4439b35 commit 9b3f91f

13 files changed

+596
-252
lines changed

Diff for: ‎src/it/java/org/tarantool/TestTarantoolClient.java

+5-3
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package org.tarantool;
22

3+
import org.tarantool.protocol.TarantoolPacket;
4+
35
import java.io.IOException;
46
import java.net.InetSocketAddress;
57
import java.nio.ByteBuffer;
@@ -15,7 +17,6 @@
1517

1618

1719
public class TestTarantoolClient {
18-
1920
public static final int TESTER_SPACE_ID = 513;
2021

2122
/*
@@ -81,8 +82,9 @@ protected void reconnect(int retry, Throwable lastError) {
8182
}
8283

8384
@Override
84-
protected void complete(long code, CompletableFuture<?> q) {
85-
super.complete(code, q);
85+
protected void complete(TarantoolPacket packet, CompletableFuture<?> q) {
86+
super.complete(packet, q);
87+
long code = packet.getCode();
8688
if (code != 0) {
8789
System.out.println(code);
8890
}

Diff for: ‎src/main/java/org/tarantool/CountInputStream.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -3,5 +3,5 @@
33
import java.io.InputStream;
44

55
public abstract class CountInputStream extends InputStream {
6-
abstract long getBytesRead();
6+
public abstract long getBytesRead();
77
}

Diff for: ‎src/main/java/org/tarantool/JDBCBridge.java

+8-7
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import java.util.Map;
99

1010
import org.tarantool.jdbc.SQLResultSet;
11+
import org.tarantool.protocol.TarantoolPacket;
1112

1213
public class JDBCBridge {
1314
public static final JDBCBridge EMPTY = new JDBCBridge(Collections.<TarantoolBase.SQLMetaData>emptyList(), Collections.<List<Object>>emptyList());
@@ -16,8 +17,8 @@ public class JDBCBridge {
1617
final Map<String,Integer> columnsByName;
1718
final List<List<Object>> rows;
1819

19-
protected JDBCBridge(TarantoolConnection connection) {
20-
this(connection.getSQLMetadata(),connection.getSQLData());
20+
protected JDBCBridge(TarantoolPacket pack) {
21+
this(SqlProtoUtils.getSQLMetadata(pack), SqlProtoUtils.getSQLData(pack));
2122
}
2223

2324
protected JDBCBridge(List<TarantoolBase.SQLMetaData> sqlMetadata, List<List<Object>> rows) {
@@ -30,8 +31,8 @@ protected JDBCBridge(List<TarantoolBase.SQLMetaData> sqlMetadata, List<List<Obje
3031
}
3132

3233
public static JDBCBridge query(TarantoolConnection connection, String sql, Object ... params) {
33-
connection.sql(sql, params);
34-
return new JDBCBridge(connection);
34+
TarantoolPacket pack = connection.sql(sql, params);
35+
return new JDBCBridge(pack);
3536
}
3637

3738
public static int update(TarantoolConnection connection, String sql, Object ... params) {
@@ -47,10 +48,10 @@ public static JDBCBridge mock(List<String> fields, List<List<Object>> values) {
4748
}
4849

4950
public static Object execute(TarantoolConnection connection, String sql, Object ... params) {
50-
connection.sql(sql, params);
51-
Long rowCount = connection.getSqlRowCount();
51+
TarantoolPacket pack = connection.sql(sql, params);
52+
Long rowCount = SqlProtoUtils.getSqlRowCount(pack);
5253
if(rowCount == null) {
53-
return new SQLResultSet(new JDBCBridge(connection));
54+
return new SQLResultSet(new JDBCBridge(pack));
5455
}
5556
return rowCount.intValue();
5657
}

Diff for: ‎src/main/java/org/tarantool/SqlProtoUtils.java

+47
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
package org.tarantool;
2+
3+
import org.tarantool.protocol.TarantoolPacket;
4+
5+
import java.util.ArrayList;
6+
import java.util.LinkedHashMap;
7+
import java.util.List;
8+
import java.util.Map;
9+
10+
public abstract class SqlProtoUtils {
11+
public static List<Map<String, Object>> readSqlResult(TarantoolPacket pack) {
12+
List<List<?>> data = (List<List<?>>) pack.getBody().get(Key.DATA.getId());
13+
14+
List<Map<String, Object>> values = new ArrayList<Map<String, Object>>(data.size());
15+
List<TarantoolBase.SQLMetaData> metaData = getSQLMetadata(pack);
16+
LinkedHashMap<String, Object> value = new LinkedHashMap<String, Object>();
17+
for (List row : data) {
18+
for (int i = 0; i < row.size(); i++) {
19+
value.put(metaData.get(i).getName(), row.get(i));
20+
}
21+
values.add(value);
22+
}
23+
return values;
24+
}
25+
26+
public static List<List<Object>> getSQLData(TarantoolPacket pack) {
27+
return (List<List<Object>>) pack.getBody().get(Key.DATA.getId());
28+
}
29+
30+
public static List<TarantoolBase.SQLMetaData> getSQLMetadata(TarantoolPacket pack) {
31+
List<Map<Integer, Object>> meta = (List<Map<Integer, Object>>) pack.getBody().get(Key.SQL_METADATA.getId());
32+
List<TarantoolBase.SQLMetaData> values = new ArrayList<TarantoolBase.SQLMetaData>(meta.size());
33+
for (Map<Integer, Object> c : meta) {
34+
values.add(new TarantoolBase.SQLMetaData((String) c.get(Key.SQL_FIELD_NAME.getId())));
35+
}
36+
return values;
37+
}
38+
39+
public static Long getSqlRowCount(TarantoolPacket pack) {
40+
Map<Key, Object> info = (Map<Key, Object>) pack.getBody().get(Key.SQL_INFO.getId());
41+
Number rowCount;
42+
if (info != null && (rowCount = ((Number) info.get(Key.SQL_ROW_COUNT.getId()))) != null) {
43+
return rowCount.longValue();
44+
}
45+
return null;
46+
}
47+
}

Diff for: ‎src/main/java/org/tarantool/TarantoolBase.java

+5-169
Original file line numberDiff line numberDiff line change
@@ -1,139 +1,36 @@
11
package org.tarantool;
22

3-
import java.io.DataInputStream;
4-
import java.io.DataOutputStream;
3+
import org.tarantool.protocol.ProtoUtils;
4+
import org.tarantool.protocol.TarantoolGreeting;
5+
56
import java.io.IOException;
6-
import java.io.OutputStream;
77
import java.net.Socket;
8-
import java.nio.ByteBuffer;
98
import java.nio.channels.SocketChannel;
10-
import java.security.MessageDigest;
11-
import java.security.NoSuchAlgorithmException;
12-
import java.util.ArrayList;
13-
import java.util.EnumMap;
14-
import java.util.LinkedHashMap;
159
import java.util.List;
16-
import java.util.Map;
1710
import java.util.concurrent.atomic.AtomicLong;
1811

1912
public abstract class TarantoolBase<Result> extends AbstractTarantoolOps<Integer, List<?>, Object, Result> {
20-
protected static final String WELCOME = "Tarantool ";
2113
protected String serverVersion;
2214
/**
2315
* Connection state
2416
*/
25-
protected String salt;
2617
protected MsgPackLite msgPackLite = MsgPackLite.INSTANCE;
2718
protected AtomicLong syncId = new AtomicLong();
2819
protected int initialRequestSize = 4096;
29-
/**
30-
* Read properties
31-
*/
32-
protected DataInputStream is;
33-
protected CountInputStream cis;
34-
protected Map<Integer, Object> headers;
35-
protected Map<Integer, Object> body;
3620

3721
public TarantoolBase() {
3822
}
3923

4024
public TarantoolBase(String username, String password, Socket socket) {
4125
super();
4226
try {
43-
cis = new CountInputStreamImpl(socket.getInputStream());
44-
is = new DataInputStream(cis);
45-
byte[] bytes = new byte[64];
46-
is.readFully(bytes);
47-
String firstLine = new String(bytes);
48-
if (!firstLine.startsWith(WELCOME)) {
49-
closeStreams();
50-
close();
51-
throw new CommunicationException("Welcome message should starts with tarantool but starts with '" + firstLine + "'", new IllegalStateException("Invalid welcome packet"));
52-
}
53-
serverVersion = firstLine.substring(WELCOME.length());
54-
is.readFully(bytes);
55-
this.salt = new String(bytes);
56-
if (username != null && password != null) {
57-
ByteBuffer authPacket = createAuthPacket(username, password);
58-
OutputStream os = socket.getOutputStream();
59-
os.write(authPacket.array(), 0, authPacket.remaining());
60-
os.flush();
61-
readPacket();
62-
Long code = (Long) headers.get(Key.CODE.getId());
63-
if (code != 0) {
64-
closeStreams();
65-
throw serverError(code, body.get(Key.ERROR.getId()));
66-
}
67-
}
27+
TarantoolGreeting greeting = ProtoUtils.connect(socket, username, password);
28+
this.serverVersion = greeting.getServerVersion();
6829
} catch (IOException e) {
69-
closeStreams();
7030
throw new CommunicationException("Couldn't connect to tarantool", e);
7131
}
7232
}
7333

74-
75-
protected ByteBuffer createAuthPacket(String username, final String password) throws IOException {
76-
final MessageDigest sha1;
77-
try {
78-
sha1 = MessageDigest.getInstance("SHA-1");
79-
} catch (NoSuchAlgorithmException e) {
80-
throw new IllegalStateException(e);
81-
}
82-
List auth = new ArrayList(2);
83-
auth.add("chap-sha1");
84-
85-
byte[] p = sha1.digest(password.getBytes());
86-
87-
sha1.reset();
88-
byte[] p2 = sha1.digest(p);
89-
90-
sha1.reset();
91-
sha1.update(Base64.decode(salt), 0, 20);
92-
sha1.update(p2);
93-
byte[] scramble = sha1.digest();
94-
for (int i = 0, e = 20; i < e; i++) {
95-
p[i] ^= scramble[i];
96-
}
97-
auth.add(p);
98-
return createPacket(Code.AUTH, 0L, null, Key.USER_NAME, username, Key.TUPLE, auth);
99-
}
100-
101-
protected ByteBuffer createPacket(Code code, Long syncId, Long schemaId, Object... args) throws IOException {
102-
TarantoolClientImpl.ByteArrayOutputStream bos = new TarantoolClientImpl.ByteArrayOutputStream(initialRequestSize);
103-
bos.write(new byte[5]);
104-
DataOutputStream ds = new DataOutputStream(bos);
105-
Map<Key, Object> header = new EnumMap<Key, Object>(Key.class);
106-
Map<Key, Object> body = new EnumMap<Key, Object>(Key.class);
107-
header.put(Key.CODE, code);
108-
header.put(Key.SYNC, syncId);
109-
if (schemaId != null) {
110-
header.put(Key.SCHEMA_ID, schemaId);
111-
}
112-
if (args != null) {
113-
for (int i = 0, e = args.length; i < e; i += 2) {
114-
Object value = args[i + 1];
115-
body.put((Key) args[i], value);
116-
}
117-
}
118-
msgPackLite.pack(header, ds);
119-
msgPackLite.pack(body, ds);
120-
ds.flush();
121-
ByteBuffer buffer = bos.toByteBuffer();
122-
buffer.put(0, (byte) 0xce);
123-
buffer.putInt(1, bos.size() - 5);
124-
return buffer;
125-
}
126-
127-
protected void readPacket() throws IOException {
128-
int size = ((Number) msgPackLite.unpack(is)).intValue();
129-
long mark = cis.getBytesRead();
130-
headers = (Map<Integer, Object>) msgPackLite.unpack(is);
131-
if (cis.getBytesRead() - mark < size) {
132-
body = (Map<Integer, Object>) msgPackLite.unpack(is);
133-
}
134-
is.skipBytes((int) (cis.getBytesRead() - mark - size));
135-
}
136-
13734
protected static class SQLMetaData {
13835
protected String name;
13936

@@ -153,56 +50,10 @@ public String toString() {
15350
}
15451
}
15552

156-
protected List<SQLMetaData> getSQLMetadata() {
157-
List<Map<Integer, Object>> meta = (List<Map<Integer, Object>>) body.get(Key.SQL_METADATA.getId());
158-
List<SQLMetaData> values = new ArrayList<SQLMetaData>(meta.size());
159-
for(Map<Integer,Object> c:meta ) {
160-
values.add(new SQLMetaData((String) c.get(Key.SQL_FIELD_NAME.getId())));
161-
}
162-
return values;
163-
}
164-
165-
protected List<List<Object>> getSQLData() {
166-
return (List<List<Object>>) body.get(Key.DATA.getId());
167-
}
168-
169-
protected List<Map<String, Object>> readSqlResult(List<List<?>> data) {
170-
List<Map<String, Object>> values = new ArrayList<Map<String, Object>>(data.size());
171-
List<SQLMetaData> metaData = getSQLMetadata();
172-
LinkedHashMap<String, Object> value = new LinkedHashMap<String, Object>();
173-
for (List row : data) {
174-
for (int i = 0; i < row.size(); i++) {
175-
value.put(metaData.get(i).getName(), row.get(i));
176-
}
177-
values.add(value);
178-
}
179-
return values;
180-
}
181-
182-
protected Long getSqlRowCount() {
183-
Map<Key, Object> info = (Map<Key, Object>) body.get(Key.SQL_INFO.getId());
184-
Number rowCount;
185-
if (info != null && (rowCount = ((Number) info.get(Key.SQL_ROW_COUNT.getId()))) != null) {
186-
return rowCount.longValue();
187-
}
188-
return null;
189-
}
190-
191-
19253
protected TarantoolException serverError(long code, Object error) {
19354
return new TarantoolException(code, error instanceof String ? (String) error : new String((byte[]) error));
19455
}
19556

196-
protected class ByteArrayOutputStream extends java.io.ByteArrayOutputStream {
197-
public ByteArrayOutputStream(int size) {
198-
super(size);
199-
}
200-
201-
ByteBuffer toByteBuffer() {
202-
return ByteBuffer.wrap(buf, 0, count);
203-
}
204-
}
205-
20657
protected void closeChannel(SocketChannel channel) {
20758
if (channel != null) {
20859
try {
@@ -213,21 +64,6 @@ protected void closeChannel(SocketChannel channel) {
21364
}
21465
}
21566

216-
protected void closeStreams() {
217-
if (is != null) {
218-
try {
219-
is.close();
220-
} catch (IOException ignored) {
221-
}
222-
}
223-
if (cis != null) {
224-
try {
225-
cis.close();
226-
} catch (IOException ignored) {
227-
}
228-
}
229-
}
230-
23167
protected void validateArgs(Object[] args) {
23268
if (args != null) {
23369
for (int i = 0; i < args.length; i += 2) {

0 commit comments

Comments
 (0)