feat: add multi stream append support (#330)

Author: YoEight

.github/workflows/tests.yml

@@ -20,7 +20,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        test: [Streams, PersistentSubscriptions, Telemetry]
+        test: [Streams, PersistentSubscriptions, Telemetry, MultiStreamAppend]
 
     runs-on: ubuntu-latest
     steps:
@@ -102,7 +102,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        test: [Streams, PersistentSubscriptions]
+        test: [Streams, PersistentSubscriptions, MultiStreamAppend]
 
     runs-on: ubuntu-latest
     steps:

src/main/java/io/kurrent/dbclient/AppendStreamFailure.java

@@ -0,0 +1,36 @@
+package io.kurrent.dbclient;
+
+public class AppendStreamFailure {
+    private final io.kurrentdb.protocol.streams.v2.AppendStreamFailure inner;
+
+    AppendStreamFailure(io.kurrentdb.protocol.streams.v2.AppendStreamFailure inner) {
+        this.inner = inner;
+    }
+
+    public String getStreamName() {
+        return this.inner.getStream();
+    }
+
+    public void visit(MultiAppendStreamErrorVisitor visitor) {
+        if (this.inner.getErrorCase() == io.kurrentdb.protocol.streams.v2.AppendStreamFailure.ErrorCase.STREAM_REVISION_CONFLICT) {
+            visitor.onWrongExpectedRevision(this.inner.getStreamRevisionConflict().getStreamRevision());
+            return;
+        }
+
+        if (this.inner.getErrorCase() == io.kurrentdb.protocol.streams.v2.AppendStreamFailure.ErrorCase.ACCESS_DENIED) {
+            visitor.onAccessDenied(this.inner.getAccessDenied());
+        }
+
+        if (this.inner.getErrorCase() == io.kurrentdb.protocol.streams.v2.AppendStreamFailure.ErrorCase.STREAM_DELETED) {
+            visitor.onStreamDeleted();
+            return;
+        }
+
+       if (this.inner.getErrorCase() == io.kurrentdb.protocol.streams.v2.AppendStreamFailure.ErrorCase.TRANSACTION_MAX_SIZE_EXCEEDED) {
+            visitor.onTransactionMaxSizeExceeded(this.inner.getTransactionMaxSizeExceeded().getMaxSize());
+            return;
+       }
+
+       throw new IllegalArgumentException("Append failure does not match any known error type");
+    }
+}

src/main/java/io/kurrent/dbclient/AppendStreamRequest.java

@@ -0,0 +1,27 @@
+package io.kurrent.dbclient;
+
+import java.util.Iterator;
+
+public class AppendStreamRequest {
+    private final String streamName;
+    private final Iterator<EventData> events;
+    private final StreamState expectedState;
+
+    public AppendStreamRequest(String streamName, Iterator<EventData> events, StreamState expectedState) {
+        this.streamName = streamName;
+        this.events = events;
+        this.expectedState = expectedState;
+    }
+
+    public String getStreamName() {
+        return streamName;
+    }
+
+    public Iterator<EventData> getEvents() {
+        return events;
+    }
+
+    public StreamState getExpectedState() {
+        return expectedState;
+    }
+}

src/main/java/io/kurrent/dbclient/AppendStreamSuccess.java

@@ -0,0 +1,21 @@
+package io.kurrent.dbclient;
+
+public class AppendStreamSuccess {
+    private final io.kurrentdb.protocol.streams.v2.AppendStreamSuccess inner;
+
+    AppendStreamSuccess(io.kurrentdb.protocol.streams.v2.AppendStreamSuccess inner) {
+        this.inner = inner;
+    }
+
+    public String getStreamName() {
+        return this.inner.getStream();
+    }
+
+    public long getStreamRevision() {
+        return this.inner.getStreamRevision();
+    }
+
+    public long getPosition() {
+        return this.inner.getPosition();
+    }
+}

src/main/java/io/kurrent/dbclient/ConnectionSettingsBuilder.java

@@ -17,7 +17,7 @@
 public class ConnectionSettingsBuilder {
     private static final Logger logger = LoggerFactory.getLogger(ConnectionSettingsBuilder.class);
     private static final Set<String> SUPPORTED_PROTOCOLS = new HashSet<>(Arrays.asList(
-            "esdb", "esdb+discover", "kurrent", "kurrent+discover", "kdb", "kdb+discover", "kurrentdb", "kurrentdb+discover"
+            "esdb", "esdb+discover", "kurrentdb", "kurrent+discover", "kdb", "kdb+discover", "kurrentdb", "kurrentdb+discover"
     ));
 
     private boolean _dnsDiscover = false;

src/main/java/io/kurrent/dbclient/ContentTypeMapper.java

@@ -0,0 +1,20 @@
+package io.kurrent.dbclient;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class ContentTypeMapper {
+    private static final Map<String, String> CONTENT_TYPE_MAP;
+
+    static {
+        Map<String, String> map = new HashMap<>();
+        map.put("application/json", "Json");
+        map.put("application/octet-stream", "Binary");
+        CONTENT_TYPE_MAP = Collections.unmodifiableMap(map);
+    }
+
+    public static String toSchemaDataFormat(String contentType) {
+        return CONTENT_TYPE_MAP.getOrDefault(contentType, contentType);
+    }
+}

src/main/java/io/kurrent/dbclient/FeatureFlags.java

@@ -7,5 +7,6 @@ class FeatureFlags {
     public final static int PERSISTENT_SUBSCRIPTION_RESTART_SUBSYSTEM = 8;
     public final static int PERSISTENT_SUBSCRIPTION_GET_INFO = 16;
     public final static int PERSISTENT_SUBSCRIPTION_TO_ALL = 32;
+    public final static int MULTI_STREAM_APPEND = 64;
     public final static int PERSISTENT_SUBSCRIPTION_MANAGEMENT = PERSISTENT_SUBSCRIPTION_LIST | PERSISTENT_SUBSCRIPTION_REPLAY | PERSISTENT_SUBSCRIPTION_GET_INFO | PERSISTENT_SUBSCRIPTION_RESTART_SUBSYSTEM;
 }

src/main/java/io/kurrent/dbclient/GossipClient.java

@@ -21,7 +21,7 @@ class GossipClient {
 
     public GossipClient(KurrentDBClientSettings settings, ManagedChannel channel) {
         _channel = channel;
-        _stub = GrpcUtils.configureStub(GossipGrpc.newStub(_channel), settings, new GossipOption(), (long)settings.getGossipTimeout());
+        _stub = GrpcUtils.configureStub(GossipGrpc.newStub(_channel), settings, new GossipOption(), settings.getGossipTimeout());
     }
 
     public void shutdown() {

src/main/java/io/kurrent/dbclient/GrpcUtils.java

@@ -113,10 +113,14 @@ static public StreamsOuterClass.ReadReq.Options.StreamOptions toStreamOptions(St
     }
 
     static public <S extends AbstractAsyncStub<S>, O> S configureStub(S stub, KurrentDBClientSettings settings, OptionsBase<O> options) {
-        return configureStub(stub, settings, options, null);
+        return configureStub(stub, settings, options, null, true);
     }
 
-    static public <S extends AbstractAsyncStub<S>, O> S configureStub(S stub, KurrentDBClientSettings settings, OptionsBase<O> options, Long forceDeadlineInMs) {
+    static public <S extends AbstractAsyncStub<S>, O> S configureStub(S stub, KurrentDBClientSettings settings, OptionsBase<O> options, long forceDeadlineInMs) {
+        return configureStub(stub, settings, options, forceDeadlineInMs, true);
+    }
+
+    static public <S extends AbstractAsyncStub<S>, O> S configureStub(S stub, KurrentDBClientSettings settings, OptionsBase<O> options, Long forceDeadlineInMs, boolean forwardRequiresLeader) {
         S finalStub = stub;
         ConnectionMetadata metadata = new ConnectionMetadata();
 
@@ -146,7 +150,7 @@ static public <S extends AbstractAsyncStub<S>, O> S configureStub(S stub, Kurren
             metadata.authenticated(credentials);
         }
 
-        if (options.isLeaderRequired() || settings.getNodePreference() == NodePreference.LEADER) {
+        if (forwardRequiresLeader && (options.isLeaderRequired() || settings.getNodePreference() == NodePreference.LEADER)) {
             metadata.requiresLeader();
         }
 

src/main/java/io/kurrent/dbclient/KurrentDBClient.java

@@ -75,6 +75,10 @@ public CompletableFuture<WriteResult> appendToStream(String streamName, AppendTo
         return new AppendToStream(this.getGrpcClient(), streamName, events, options).execute();
     }
 
+    public CompletableFuture<MultiAppendWriteResult> multiAppend(AppendToStreamOptions options, Iterator<AppendStreamRequest> requests) {
+        return new MultiStreamAppend(this.getGrpcClient(), requests).execute();
+    }
+
     /**
      * Sets a stream's metadata.
      * @param streamName stream's name.