Skip to content

Commit 63439e3

Browse files
mweberUKLmweberUKL
committed
added support for the cluster API
1 parent 62c6b91 commit 63439e3

7 files changed

Lines changed: 288 additions & 104 deletions

File tree

‎src/main/java/eu/antidotedb/client/AntidoteClient.java‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,15 @@
11
package eu.antidotedb.client;
22

3+
import com.google.protobuf.ByteString;
4+
import eu.antidotedb.antidotepb.AntidotePB;
35
import eu.antidotedb.antidotepb.AntidotePB.ApbCommitResp;
46
import eu.antidotedb.client.messages.AntidoteRequest;
57
import eu.antidotedb.client.messages.AntidoteResponse;
68
import eu.antidotedb.client.transformer.TransformerFactory;
79

10+
import java.io.IOException;
811
import java.net.InetSocketAddress;
12+
import java.net.Socket;
913
import java.util.Arrays;
1014
import java.util.Collections;
1115
import java.util.List;
@@ -185,5 +189,43 @@ public NoTransaction noTransaction() {
185189
}
186190

187191

192+
public static boolean createDC(InetSocketAddress managerNode, List<String> nodeNames) throws IOException {
193+
try (Socket s = new Socket()) {
194+
s.connect(managerNode);
195+
AntidotePB.ApbCreateDC createDCMsg = AntidotePB.ApbCreateDC.newBuilder()
196+
.addAllNodes(nodeNames)
197+
.build();
198+
SocketSender socketSender = new SocketSender(s);
199+
AntidotePB.ApbOperationResp createDCResp = socketSender.handle(createDCMsg).accept(new AntidoteResponse.MsgOperationResp.Extractor());
200+
return createDCResp.getSuccess();
201+
}
202+
}
203+
204+
public static ByteString getConnectionDescriptor(InetSocketAddress managerNode) throws IOException {
205+
try (Socket s = new Socket()) {
206+
s.connect(managerNode);
207+
AntidotePB.ApbGetConnectionDescriptor apbGetConnectionDescriptor = AntidotePB.ApbGetConnectionDescriptor.newBuilder()
208+
.build();
209+
SocketSender socketSender = new SocketSender(s);
210+
AntidotePB.ApbGetConnectionDescriptorResponse connectionDescriptor = socketSender.handle(apbGetConnectionDescriptor).accept(new AntidoteResponse.MsgGetConnectionDescriptorResponse.Extractor());
211+
if (connectionDescriptor.getSuccess()) {
212+
return connectionDescriptor.getConDesc();
213+
} else {
214+
throw new IOException("Error getting connection descriptor of node: Error code " + connectionDescriptor.getErrorcode());
215+
}
216+
}
217+
}
218+
219+
public static boolean connectToDCs(InetSocketAddress managerNode, List<ByteString> descriptors) throws IOException {
220+
try (Socket s = new Socket()) {
221+
s.connect(managerNode);
222+
AntidotePB.ApbConnectToDcs apbConnectToDcs = AntidotePB.ApbConnectToDcs.newBuilder()
223+
.addAllDescriptors(descriptors)
224+
.build();
225+
SocketSender socketSender = new SocketSender(s);
226+
AntidotePB.ApbOperationResp createDCResp = socketSender.handle(apbConnectToDcs).accept(new AntidoteResponse.MsgOperationResp.Extractor());
227+
return createDCResp.getSuccess();
228+
}
229+
}
188230
}
189231

‎src/main/java/eu/antidotedb/client/ApbCoder.java‎

Lines changed: 38 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,15 @@ public static AntidoteRequest<?> decodeRequest(InputStream stream) throws IOExce
5757
case ApbStaticReadObjects:
5858
AntidotePB.ApbStaticReadObjects apbStaticReadObjects = AntidotePB.ApbStaticReadObjects.parseFrom(data);
5959
return AntidoteRequest.of(apbStaticReadObjects);
60+
case ApbCreateDC:
61+
AntidotePB.ApbCreateDC apbCreateDC = AntidotePB.ApbCreateDC.parseFrom(data);
62+
return AntidoteRequest.of(apbCreateDC);
63+
case ApbConnectToDCs:
64+
AntidotePB.ApbConnectToDcs apbConnectToDcs = AntidotePB.ApbConnectToDcs.parseFrom(data);
65+
return AntidoteRequest.of(apbConnectToDcs);
66+
case ApbGetConnectionDescriptor:
67+
AntidotePB.ApbGetConnectionDescriptor apbGetConnectionDescriptor = AntidotePB.ApbGetConnectionDescriptor.parseFrom(data);
68+
return AntidoteRequest.of(apbGetConnectionDescriptor);
6069
default:
6170
throw new RuntimeException("Unexpected request message code: " + msgCode);
6271
}
@@ -81,9 +90,6 @@ public static AntidoteResponse decodeResponse(InputStream stream) throws IOExcep
8190
// dataInputStream.readFully(data, 0, size - 1);
8291

8392
switch (msgCode) {
84-
case 0:
85-
AntidotePB.ApbErrorResp apbErrorResp = AntidotePB.ApbErrorResp.parseFrom(data);
86-
return AntidoteResponse.of(apbErrorResp);
8793
case 111:
8894
AntidotePB.ApbOperationResp apbOperationResp = AntidotePB.ApbOperationResp.parseFrom(data);
8995
return AntidoteResponse.of(apbOperationResp);
@@ -99,6 +105,9 @@ public static AntidoteResponse decodeResponse(InputStream stream) throws IOExcep
99105
case 128:
100106
AntidotePB.ApbStaticReadObjectsResp apbStaticReadObjectsResp = AntidotePB.ApbStaticReadObjectsResp.parseFrom(data);
101107
return AntidoteResponse.of(apbStaticReadObjectsResp);
108+
case ApbGetConnectionDescriptorResponse:
109+
AntidotePB.ApbGetConnectionDescriptorResponse apbGetConnectionDescriptorResponse = AntidotePB.ApbGetConnectionDescriptorResponse.parseFrom(data);
110+
return AntidoteResponse.of(apbGetConnectionDescriptorResponse);
102111
default:
103112
throw new RuntimeException("Unexpected message code: " + msgCode);
104113
}
@@ -143,11 +152,19 @@ public static void encodeRequest(AntidotePB.ApbStaticUpdateObjects op, OutputStr
143152
encode(122, op, stream);
144153
}
145154

155+
public static void encodeRequest(AntidotePB.ApbCreateDC op, OutputStream stream) {
156+
encode(ApbCreateDC, op, stream);
157+
}
158+
159+
public static void encodeRequest(AntidotePB.ApbConnectToDcs op, OutputStream stream) {
160+
encode(ApbConnectToDCs, op, stream);
161+
}
146162

147-
public static void encodeResponse(AntidotePB.ApbErrorResp op, OutputStream stream) {
148-
encode(0, op, stream);
163+
public static void encodeRequest(AntidotePB.ApbGetConnectionDescriptor op, OutputStream stream) {
164+
encode(ApbGetConnectionDescriptor, op, stream);
149165
}
150166

167+
151168
public static void encodeResponse(AntidotePB.ApbOperationResp op, OutputStream stream) {
152169
encode(111, op, stream);
153170
}
@@ -168,6 +185,10 @@ public static void encodeResponse(AntidotePB.ApbStaticReadObjectsResp op, Output
168185
encode(128, op, stream);
169186
}
170187

188+
public static void encodeResponse(AntidotePB.ApbGetConnectionDescriptorResponse op, OutputStream stream) {
189+
encode(ApbGetConnectionDescriptorResponse, op, stream);
190+
}
191+
171192
private static void encode(int msgCode, GeneratedMessageV3 msg, OutputStream stream) {
172193
int serializedSize = msg.getSerializedSize();
173194
ByteBuffer buffer = ByteBuffer.allocate(5);
@@ -228,17 +249,23 @@ public Void handle(AntidotePB.ApbStaticUpdateObjects op) {
228249
encodeRequest(op, stream);
229250
return null;
230251
}
231-
});
232-
}
233252

234-
public static void encodeResponse(AntidoteResponse response, OutputStream stream) {
235-
response.accept(new AntidoteResponse.Handler<Void>() {
236253
@Override
237-
public Void handle(AntidotePB.ApbErrorResp op) {
238-
encodeResponse(op, stream);
254+
public Void handle(AntidotePB.ApbCreateDC op) {
255+
encodeRequest(op, stream);
256+
return null;
257+
}
258+
259+
@Override
260+
public Void handle(AntidotePB.ApbConnectToDcs op) {
261+
encodeRequest(op, stream);
239262
return null;
240263
}
264+
});
265+
}
241266

267+
public static void encodeResponse(AntidoteResponse response, OutputStream stream) {
268+
response.accept(new AntidoteResponse.Handler<Void>() {
242269
@Override
243270
public Void handle(AntidotePB.ApbOperationResp op) {
244271
encodeResponse(op, stream);

‎src/main/java/eu/antidotedb/client/MessageCodes.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,5 +36,10 @@ public class MessageCodes {
3636
* The Constant ApbAbortTransaction.
3737
*/
3838
public static final int ApbAbortTransaction = 120;
39+
40+
public static final int ApbCreateDC = 129;
41+
public static final int ApbConnectToDCs = 130;
42+
public static final int ApbGetConnectionDescriptor = 131;
43+
public static final int ApbGetConnectionDescriptorResponse = 132;
3944
}
4045

‎src/main/java/eu/antidotedb/client/SocketSender.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,4 +92,34 @@ public AntidoteResponse handle(AntidotePB.ApbStaticUpdateObjects op) {
9292
throw new AntidoteSocketException(e);
9393
}
9494
}
95+
96+
@Override
97+
public AntidoteResponse handle(AntidotePB.ApbCreateDC op) {
98+
try {
99+
ApbCoder.encodeRequest(op, socket.getOutputStream());
100+
return ApbCoder.decodeResponse(socket.getInputStream());
101+
} catch (IOException e) {
102+
throw new AntidoteSocketException(e);
103+
}
104+
}
105+
106+
@Override
107+
public AntidoteResponse handle(AntidotePB.ApbConnectToDcs op) {
108+
try {
109+
ApbCoder.encodeRequest(op, socket.getOutputStream());
110+
return ApbCoder.decodeResponse(socket.getInputStream());
111+
} catch (IOException e) {
112+
throw new AntidoteSocketException(e);
113+
}
114+
}
115+
116+
@Override
117+
public AntidoteResponse handle(AntidotePB.ApbGetConnectionDescriptor op) {
118+
try {
119+
ApbCoder.encodeRequest(op, socket.getOutputStream());
120+
return ApbCoder.decodeResponse(socket.getInputStream());
121+
} catch (IOException e) {
122+
throw new AntidoteSocketException(e);
123+
}
124+
}
95125
}

‎src/main/java/eu/antidotedb/client/messages/AntidoteRequest.java‎

Lines changed: 116 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@
1111
* {@link #of(AntidotePB.ApbStaticReadObjects)} and {@link #of(AntidotePB.ApbStaticUpdateObjects)}.
1212
*/
1313
public abstract class AntidoteRequest<Response> extends AntidoteMessage {
14+
15+
1416
/**
1517
* A transformer for all possible cases of Antidote requests.
1618
* <p>
@@ -20,14 +22,14 @@ public abstract class AntidoteRequest<Response> extends AntidoteMessage {
2022
* @see AntidoteRequest#accept(Handler)
2123
*/
2224
public interface Handler<V> {
25+
2326
default V handle(AntidotePB.ApbReadObjects op) {
2427
throw new ExtractionError("Unexpected message: " + op);
2528
}
2629

2730
default V handle(AntidotePB.ApbUpdateObjects op) {
2831
throw new ExtractionError("Unexpected message: " + op);
2932
}
30-
3133
default V handle(AntidotePB.ApbStartTransaction op) {
3234
throw new ExtractionError("Unexpected message: " + op);
3335
}
@@ -47,8 +49,19 @@ default V handle(AntidotePB.ApbStaticReadObjects op) {
4749
default V handle(AntidotePB.ApbStaticUpdateObjects op) {
4850
throw new ExtractionError("Unexpected message: " + op);
4951
}
50-
}
5152

53+
default V handle(AntidotePB.ApbCreateDC op) {
54+
throw new ExtractionError("Unexpected message: " + op);
55+
}
56+
57+
default V handle(AntidotePB.ApbConnectToDcs op) {
58+
throw new ExtractionError("Unexpected message: " + op);
59+
}
60+
61+
default V handle(AntidotePB.ApbGetConnectionDescriptor op) {
62+
throw new ExtractionError("Unexpected message: " + op);
63+
}
64+
}
5265
/**
5366
* An exception thrown by response extractors when response type does not match.
5467
*
@@ -64,12 +77,11 @@ public static class ExtractionError extends RuntimeException {
6477
public ExtractionError(String message) {
6578
super(message);
6679
}
67-
}
6880

81+
}
6982
public static MsgReadObjects of(AntidotePB.ApbReadObjects op) {
7083
return new MsgReadObjects(op);
7184
}
72-
7385
public static MsgUpdateObjects of(AntidotePB.ApbUpdateObjects op) {
7486
return new MsgUpdateObjects(op);
7587
}
@@ -94,6 +106,18 @@ public static MsgStaticUpdateObjects of(AntidotePB.ApbStaticUpdateObjects op) {
94106
return new MsgStaticUpdateObjects(op);
95107
}
96108

109+
public static MsgCreateDC of(AntidotePB.ApbCreateDC op) {
110+
return new MsgCreateDC(op);
111+
}
112+
113+
public static MsgConnectToDCs of(AntidotePB.ApbConnectToDcs op) {
114+
return new MsgConnectToDCs(op);
115+
}
116+
117+
public static MsgGetConnectionDescriptor of(AntidotePB.ApbGetConnectionDescriptor op) {
118+
return new MsgGetConnectionDescriptor(op);
119+
}
120+
97121
/**
98122
* Handle the request using the given request transformer.
99123
*
@@ -344,4 +368,92 @@ public String toString() {
344368
'}';
345369
}
346370
}
371+
372+
private static class MsgCreateDC extends AntidoteRequest<AntidotePB.ApbOperationResp> {
373+
private AntidotePB.ApbCreateDC op;
374+
375+
public static class Extractor implements Handler<AntidotePB.ApbCreateDC> {
376+
377+
378+
@Override
379+
public AntidotePB.ApbCreateDC handle(AntidotePB.ApbCreateDC op) {
380+
return op;
381+
}
382+
}
383+
384+
private MsgCreateDC(AntidotePB.ApbCreateDC op) {
385+
this.op = op;
386+
}
387+
388+
@Override
389+
public <V> V accept(Handler<V> handler) {
390+
return handler.handle(op);
391+
}
392+
393+
@Override
394+
public AntidoteResponse.Handler<AntidotePB.ApbOperationResp> readResponseExtractor() {
395+
return new AntidoteResponse.MsgOperationResp.Extractor();
396+
}
397+
398+
@Override
399+
public String toString() {
400+
return "MsgCreateDC{" +
401+
"op=" + op +
402+
'}';
403+
}
404+
}
405+
406+
private static class MsgConnectToDCs extends AntidoteRequest<AntidotePB.ApbOperationResp> {
407+
private final AntidotePB.ApbConnectToDcs op;
408+
409+
public static class Extractor implements Handler<AntidotePB.ApbConnectToDcs> {
410+
411+
412+
@Override
413+
public AntidotePB.ApbConnectToDcs handle(AntidotePB.ApbConnectToDcs op) {
414+
return op;
415+
}
416+
}
417+
418+
private MsgConnectToDCs(AntidotePB.ApbConnectToDcs apbConnectToDcs) {
419+
this.op = apbConnectToDcs;
420+
}
421+
422+
@Override
423+
public <V> V accept(Handler<V> handler) {
424+
return handler.handle(op);
425+
}
426+
427+
@Override
428+
public AntidoteResponse.Handler<AntidotePB.ApbOperationResp> readResponseExtractor() {
429+
return new AntidoteResponse.MsgOperationResp.Extractor();
430+
}
431+
}
432+
433+
private static class MsgGetConnectionDescriptor extends AntidoteRequest<AntidotePB.ApbGetConnectionDescriptorResponse> {
434+
private final AntidotePB.ApbGetConnectionDescriptor op;
435+
436+
public static class Extractor implements Handler<AntidotePB.ApbGetConnectionDescriptor> {
437+
438+
439+
@Override
440+
public AntidotePB.ApbGetConnectionDescriptor handle(AntidotePB.ApbGetConnectionDescriptor op) {
441+
return op;
442+
}
443+
}
444+
445+
private MsgGetConnectionDescriptor(AntidotePB.ApbGetConnectionDescriptor op) {
446+
this.op = op;
447+
}
448+
449+
@Override
450+
public <V> V accept(Handler<V> handler) {
451+
return handler.handle(op);
452+
}
453+
454+
@Override
455+
public AntidoteResponse.Handler<AntidotePB.ApbGetConnectionDescriptorResponse> readResponseExtractor() {
456+
return new AntidoteResponse.MsgGetConnectionDescriptorResponse.Extractor();
457+
}
458+
}
347459
}

0 commit comments

Comments
 (0)