Non blocking client 3

.github/workflows/pr.yml

@@ -5,21 +5,6 @@ on:
     branches: [ main ]
 
 jobs:
-  commit:
-    name: Commit Message Validation
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v2
-        with:
-          fetch-depth: 0
-      - run: git show-ref
-      - uses: actions-rs/install@v0.1
-        with:
-          crate: git-cz
-          version: latest
-      - name: Validate commit messages
-        run: git-cz check ${{ github.event.pull_request.base.sha }}..${{ github.event.pull_request.head.sha }}
-
   test:
     name: Test
     runs-on: ubuntu-latest
@@ -38,7 +23,7 @@ jobs:
         uses: gradle/actions/setup-gradle@v3
 
       - name: Start the cluster
-        run: ./scripts/quick_start.sh
+        run: docker compose -f ci/docker-compose.yml up -d
 
       - name: Run test
         run: ./gradlew test

build.gradle.kts

@@ -2,7 +2,7 @@ import com.adarshr.gradle.testlogger.TestLoggerExtension
 import com.adarshr.gradle.testlogger.theme.ThemeType
 
 group = "cloud.xline"
-version = "1.0-SNAPSHOT"
+version = "0.1.0-SNAPSHOT"
 
 buildscript {
     repositories {
@@ -27,6 +27,7 @@ subprojects {
     apply(plugin = "java-library")
     apply(plugin = "org.gradle.test-retry")
     apply(plugin = "com.adarshr.test-logger")
+    apply(plugin = "maven-publish")
 
     tasks {
         named<JavaCompile>("compileJava") {
@@ -47,8 +48,26 @@ subprojects {
         }
     }
 
-    extensions.getByType<TestLoggerExtension>().apply {
+    configure<PublishingExtension> {
+        publications {
+            create<MavenPublication>("maven") {
+                groupId = rootProject.group.toString()
+                version = rootProject.version.toString()
+
+                from(components["java"])
+
+            }
+        }
+    }
+
+    configure<JavaPluginExtension> {
+        withSourcesJar()
+        withJavadocJar()
+    }
+
+    configure<TestLoggerExtension> {
         theme = ThemeType.MOCHA_PARALLEL
         showStandardStreams = false
     }
+
 }

ci/docker-compose.yml

@@ -0,0 +1,71 @@
+version: '3.9'
+
+networks:
+  xline_network:
+    driver: bridge
+    ipam:
+      driver: default
+      config:
+        - subnet: "172.18.0.0/16"
+
+services:
+  node1:
+    image: ghcr.io/xline-kv/xline:latest
+    networks:
+      xline_network:
+        ipv4_address: 172.18.0.2
+    volumes:
+      - .:/mnt
+    ports:
+      - "2379:2379"
+    environment:
+      RUST_LOG: curp=debug,xline=debug
+    command: >
+      xline
+      --name node1
+      --members node1=http://172.18.0.2:2379,node2=http://172.18.0.3:2379,node3=http://172.18.0.4:2379
+      --storage-engine rocksdb
+      --data-dir /usr/local/xline/data-dir
+      --auth-public-key /mnt/public.pem
+      --auth-private-key /mnt/private.pem
+      --is-leader
+
+  node2:
+    image: ghcr.io/xline-kv/xline:latest
+    networks:
+      xline_network:
+        ipv4_address: 172.18.0.3
+    volumes:
+      - .:/mnt
+    ports:
+      - "2380:2379"
+    environment:
+      RUST_LOG: curp=debug,xline=debug
+    command: >
+      xline
+      --name node2
+      --members node1=http://172.18.0.2:2379,node2=http://172.18.0.3:2379,node3=http://172.18.0.4:2379
+      --storage-engine rocksdb
+      --data-dir /usr/local/xline/data-dir
+      --auth-public-key /mnt/public.pem
+      --auth-private-key /mnt/private.pem
+
+  node3:
+    image: ghcr.io/xline-kv/xline:latest
+    networks:
+      xline_network:
+        ipv4_address: 172.18.0.4
+    volumes:
+      - .:/mnt
+    ports:
+      - "2381:2379"
+    environment:
+      RUST_LOG: curp=debug,xline=debug
+    command: >
+      xline
+      --name node3
+      --members node1=http://172.18.0.2:2379,node2=http://172.18.0.3:2379,node3=http://172.18.0.4:2379
+      --storage-engine rocksdb
+      --data-dir /usr/local/xline/data-dir
+      --auth-public-key /mnt/public.pem
+      --auth-private-key /mnt/private.pem

ci/private.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCnAxxSXJYWZCKr
+6f6j0HRUwkhX/0+GXjEclWoLA5+KZAuWMSu8bz6X+IScv4vNwORlGSWOnrz+8mb2
+I0F6teVZWfWFqsnyWk7IxM+h9yTg7aY/8685YfWTL7fpWq1/3Fniz4QbsYFuzB1V
+gaZ5fD2CSYIKzSD+qVSlXF25JDFHV7b2OdHrX0UKZOTWY/VE//STt+PJKdX9R3pl
+kGwAzJIkkcAZy0vhvqT3ASTgXchNeN8wGYYb3YirkqIsQB5Xcs1R1W+yz+IrVa6/
+0WMcyE6qtJPZ0lviyT0nHV/pZjXuD4B0aja/1fk/HmXDPMjpK1BuCBTStM/KlcrA
+oAxo+YDhAgMBAAECggEAIyJhY+Y8YMuCC753JkklH+ubQn/gX/kSxduc6mJBvuBb
+G6aOd97DQT8zzrHxHEDXC3ml0AIO6mdeR6uVC9aWQBzPrOYIA+cBqfTVZVJTvMnh
+7pQ6KY01F1izjPDZjQtzEWbseNL30rI3/ZP/zJDZc745EEKlDU3cE8mBogA+Ka6w
+GLozT9qQf8knBrtzxH6SvrZpfaRlP95is82b4IuPhqYdG7dVYFTALE1MyVrCbS4Y
+KytjNLgwp1bIQtWrzMebBGoiU+DvDcRY8zvOfFupDwpYCt3p1aU5wyYYdr74esV7
+jjqHj89Ua65JHJ3XnMAaMc4dHM2FsGqMsOv/DDKInQKBgQDawckQEekx0QuP3eJP
+GWdZ87oc+FVjDe3bYhAnCf/yXRJoqcs5vr1m1yCXFfsjbQFYHWXR9AUtNn5HCwOZ
+zoT1Mv96fXBVGQORgzvlUWS43uKpfIPDVv2I6ZcKSIQAGOgcWYvmBDhYqPHgmx3o
+VSrNGWtLdyw3rD1J6O+1RwtbiwKBgQDDchmY59EXBiTvlyT3Qjl0vZFMHa+TElbh
+ikNtYltbUHtamOXZzpdk/KA7X2dYi0QpVfbbpfP/ly5lYvgZwl8h90Obopru+ACM
+ndlKBfNQYArmWY6bJ2CwF7j1aTCCHZuVuX6/pzFVStRcssn15uoVaIyKd/MhJzLF
+S3ertQkSwwKBgAniMYRhWsjeaghQ/RWXzzyYL3N5oNn92h5MWvB4mjDIFbnW2hC8
+1m/cDmPlIVijZyklAuGuhcFaMfBhxgLf+s/dQv+0xSuDGs8rP7yHpeZYY6NGtelQ
+d9oEu8dCKXybo3kMbq6wyB7xWyRLvdkuZ+WmXVumgb/uL0K0nIfzMscrAoGAeA1e
+K845YSslBQaSbk7/e/X1iguyDWT2eRO01zvTYgPNwZipl2CPHjkPM2km0fy5oaps
+N/94IUd7+EsSmsAKL5LytGbtRFyR+c376rw8+OIFz/iy4BsQCRqJQjWa1lHZf96x
+PIg2hW2xhD9OTv3IS94sdeG4NmUdipMQryhEqoECgYEAkvXOg66IAVTrO6qgoyl5
+42oufa/QE+qOAYoQEpmx3SZx6tMkycfAQqUHYcXhW1HNjyGbbg/sl13yddnPQqig
++ObtQNSIqGZWCc/HIqM//pPI3MHPhWARMOmAbk0I1mT0QKhuFfSugV2xb1Dj/Rvf
+0VdB8txY+5Wz6zP1F2g46gM=
+-----END PRIVATE KEY-----

ci/public.pem

@@ -0,0 +1,9 @@
+-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEApwMcUlyWFmQiq+n+o9B0
+VMJIV/9Phl4xHJVqCwOfimQLljErvG8+l/iEnL+LzcDkZRkljp68/vJm9iNBerXl
+WVn1harJ8lpOyMTPofck4O2mP/OvOWH1ky+36Vqtf9xZ4s+EG7GBbswdVYGmeXw9
+gkmCCs0g/qlUpVxduSQxR1e29jnR619FCmTk1mP1RP/0k7fjySnV/Ud6ZZBsAMyS
+JJHAGctL4b6k9wEk4F3ITXjfMBmGG92Iq5KiLEAeV3LNUdVvss/iK1Wuv9FjHMhO
+qrST2dJb4sk9Jx1f6WY17g+AdGo2v9X5Px5lwzzI6StQbggU0rTPypXKwKAMaPmA
+4QIDAQAB
+-----END PUBLIC KEY-----

jxline-core/src/main/java/cloud/xline/jxline/Txn.java

@@ -17,8 +17,8 @@
 package cloud.xline.jxline;
 
 import cloud.xline.jxline.kv.TxnResponse;
-import io.etcd.jetcd.op.Cmp;
-import io.etcd.jetcd.op.Op;
+import cloud.xline.jxline.op.Cmp;
+import cloud.xline.jxline.op.Op;
 
 import java.util.concurrent.CompletableFuture;
 

jxline-core/src/main/java/cloud/xline/jxline/exceptions/CurpException.java

@@ -105,4 +105,9 @@ public static CurpException toCurpException(Status status, @Nullable Metadata tr
         CurpError curpError = trailers.get(STATUS_DETAILS_KEY);
         return new CurpException(curpError);
     }
+
+    @Override
+    public String toString() {
+        return "CurpError(" + error.toString().replace("\n", "") + ")";
+    }
 }

jxline-core/src/main/java/cloud/xline/jxline/impl/KVImpl.java

@@ -4,6 +4,7 @@
 import cloud.xline.jxline.ProtocolClient;
 import cloud.xline.jxline.Txn;
 import cloud.xline.jxline.kv.*;
+import cloud.xline.jxline.op.TxnImpl;
 import cloud.xline.jxline.support.Requests;
 import com.xline.protobuf.Command;
 import io.etcd.jetcd.ByteSequence;
@@ -16,7 +17,7 @@
 
 import static java.util.Objects.requireNonNull;
 
-public class KVImpl extends Impl implements KV {
+class KVImpl extends Impl implements KV {
 
     private final ProtocolClient protocolClient;
 
@@ -37,7 +38,7 @@ public CompletableFuture<PutResponse> put(
         requireNonNull(value, "value should not be null");
         requireNonNull(option, "option should not be null");
         Command cmd =
-                Requests.mapPutRequest(key, value, option, this.connectionManager().getNamespace());
+                Requests.mapPutCommand(key, value, option, this.connectionManager().getNamespace());
         return protocolClient.propose(
                 cmd,
                 true,
@@ -55,7 +56,7 @@ public CompletableFuture<GetResponse> get(ByteSequence key, GetOption option) {
         requireNonNull(key, "key should not be null");
         requireNonNull(option, "option should not be null");
         Command cmd =
-                Requests.mapRangeRequest(key, option, this.connectionManager().getNamespace());
+                Requests.mapRangeCommand(key, option, this.connectionManager().getNamespace());
         return protocolClient.propose(
                 cmd,
                 true,
@@ -73,7 +74,7 @@ public CompletableFuture<DeleteResponse> delete(ByteSequence key, DeleteOption o
         requireNonNull(key, "key should not be null");
         requireNonNull(option, "option should not be null");
         Command cmd =
-                Requests.mapDeleteRequest(key, option, this.connectionManager().getNamespace());
+                Requests.mapDeleteCommand(key, option, this.connectionManager().getNamespace());
         return protocolClient.propose(
                 cmd,
                 true,
@@ -94,6 +95,14 @@ public CompletableFuture<CompactResponse> compact(long revision, CompactOption o
 
     @Override
     public Txn txn() {
-        return null;
+        return TxnImpl.newTxn(
+                this.connectionManager().getNamespace(),
+                cmd ->
+                        protocolClient.propose(
+                                cmd,
+                                true,
+                                (sr, asr) ->
+                                        new TxnResponse(
+                                                sr, asr, this.connectionManager().getNamespace())));
     }
 }

jxline-core/src/main/java/cloud/xline/jxline/impl/ProtocolClientImpl.java

@@ -191,6 +191,10 @@ CommandResponse fastRound(ProposeId id, Command cmd) {
                 throw XlineException.toXlineException(e);
             } catch (ExecutionException e) {
                 Throwable cause = e.getCause();
+                // extract the most inner exception
+                while (cause instanceof ExecutionException) {
+                    cause = cause.getCause();
+                }
                 if (!(cause instanceof CurpException)) {
                     throw XlineException.toXlineException(cause);
                 }
@@ -368,8 +372,8 @@ void checkUpdate(FetchClusterResponse res) {
                 if (res.hasLeaderId() && this.term < res.getTerm()) {
                     this.term = res.getTerm();
                     this.leaderId = res.getLeaderId();
-                    logger().info("client term updates to " + this.term);
-                    logger().info("client leader id updates to " + this.leaderId);
+                    logger().info("client term updates to {}", this.term);
+                    logger().info("client leader id updates to {}", this.leaderId);
                 }
                 if (res.getClusterVersion() == this.clusterVersion) {
                     return;

jxline-core/src/main/java/cloud/xline/jxline/op/Cmp.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2016-2021 The jetcd authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cloud.xline.jxline.op;
+
+import cloud.xline.jxline.Txn;
+import com.google.protobuf.ByteString;
+import com.xline.protobuf.Compare;
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.support.Util;
+
+
+/** The compare predicate in {@link Txn}. */
+public class Cmp {
+
+    public enum Op {
+        EQUAL,
+        GREATER,
+        LESS,
+        NOT_EQUAL
+    }
+
+    private final ByteString key;
+    private final Op op;
+    private final CmpTarget<?> target;
+
+    public Cmp(ByteSequence key, Op compareOp, CmpTarget<?> target) {
+        this.key = ByteString.copyFrom(key.getBytes());
+        this.op = compareOp;
+        this.target = target;
+    }
+
+    Compare toCompare(ByteSequence namespace) {
+        Compare.Builder compareBuilder =
+                Compare.newBuilder().setKey(Util.prefixNamespace(this.key, namespace));
+        switch (this.op) {
+            case EQUAL:
+                compareBuilder.setResult(Compare.CompareResult.EQUAL);
+                break;
+            case GREATER:
+                compareBuilder.setResult(Compare.CompareResult.GREATER);
+                break;
+            case LESS:
+                compareBuilder.setResult(Compare.CompareResult.LESS);
+                break;
+            case NOT_EQUAL:
+                compareBuilder.setResult(Compare.CompareResult.NOT_EQUAL);
+                break;
+            default:
+                throw new IllegalArgumentException("Unexpected compare type (" + this.op + ")");
+        }
+
+        Compare.CompareTarget target = this.target.getTarget();
+        Object value = this.target.getTargetValue();
+
+        compareBuilder.setTarget(target);
+        switch (target) {
+            case VERSION:
+                compareBuilder.setVersion((Long) value);
+                break;
+            case VALUE:
+                compareBuilder.setValue((ByteString) value);
+                break;
+            case MOD:
+                compareBuilder.setModRevision((Long) value);
+                break;
+            case CREATE:
+                compareBuilder.setCreateRevision((Long) value);
+                break;
+            default:
+                throw new IllegalArgumentException("Unexpected target type (" + target + ")");
+        }
+
+        return compareBuilder.build();
+    }
+}