refactor: move ExpectedRevision to StreamState (#313)

* refactor: move ExpectedRevision to StreamState.

---------

Co-authored-by: YoEight <yo.eight@gmail.com>

Author: w1am

db-client-java/src/main/java/io/kurrent/dbclient/AppendToStream.java

@@ -42,7 +42,7 @@ public CompletableFuture<WriteResult> execute() {
 
     private CompletableFuture<WriteResult> append(ManagedChannel channel, List<EventData> events) {
         CompletableFuture<WriteResult> result = new CompletableFuture<>();
-        StreamsOuterClass.AppendReq.Options.Builder options = this.options.getExpectedRevision().applyOnWire(StreamsOuterClass.AppendReq.Options.newBuilder()
+        StreamsOuterClass.AppendReq.Options.Builder options = this.options.getStreamState().applyOnWire(StreamsOuterClass.AppendReq.Options.newBuilder()
                 .setStreamIdentifier(Shared.StreamIdentifier.newBuilder()
                         .setStreamName(ByteString.copyFromUtf8(streamName))
                         .build()));
@@ -58,33 +58,33 @@ private CompletableFuture<WriteResult> append(ManagedChannel channel, List<Event
                     logPosition = new Position(p.getCommitPosition(), p.getPreparePosition());
                 }
 
-                ExpectedRevision nextExpectedRevision = success.hasNoStream() ? ExpectedRevision.noStream()
-                        : ExpectedRevision.expectedRevision(success.getCurrentRevision());
+                StreamState nextExpectedRevision = success.hasNoStream() ? StreamState.noStream()
+                        : StreamState.streamRevision(success.getCurrentRevision());
 
                 return new WriteResult(nextExpectedRevision, logPosition);
             }
             if (resp.hasWrongExpectedVersion()) {
                 StreamsOuterClass.AppendResp.WrongExpectedVersion wev = resp.getWrongExpectedVersion();
 
-                ExpectedRevision expectedRevision;
+                StreamState expectedRevision;
                 if (wev.getExpectedRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.ExpectedRevisionOptionCase.EXPECTED_ANY) {
-                    expectedRevision = ExpectedRevision.any();
+                    expectedRevision = StreamState.any();
                 } else if (wev.getExpectedRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.ExpectedRevisionOptionCase.EXPECTED_STREAM_EXISTS) {
-                    expectedRevision = ExpectedRevision.streamExists();
+                    expectedRevision = StreamState.streamExists();
                 } else {
-                    expectedRevision = ExpectedRevision.expectedRevision(wev.getExpectedRevision());
+                    expectedRevision = StreamState.streamRevision(wev.getExpectedRevision());
                 }
 
-                ExpectedRevision currentRevision;
+                StreamState streamState;
                 if (wev.getCurrentRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.CurrentRevisionOptionCase.CURRENT_NO_STREAM) {
-                    currentRevision = ExpectedRevision.noStream();
+                    streamState = StreamState.noStream();
                 } else {
-                    currentRevision = ExpectedRevision.expectedRevision(wev.getCurrentRevision());
+                    streamState = StreamState.streamRevision(wev.getCurrentRevision());
                 }
 
                 String streamName = options.getStreamIdentifier().getStreamName().toStringUtf8();
 
-                throw new WrongExpectedVersionException(streamName, expectedRevision, currentRevision);
+                throw new WrongExpectedVersionException(streamName, expectedRevision, streamState);
             }
 
             throw new IllegalStateException("AppendResponse has neither Success or WrongExpectedVersion variants");

db-client-java/src/main/java/io/kurrent/dbclient/AppendToStreamOptions.java

@@ -3,7 +3,7 @@
 /**
  * Options of the append stream request.
  */
-public class AppendToStreamOptions extends OptionsWithExpectedRevisionBase<AppendToStreamOptions> {
+public class AppendToStreamOptions extends OptionsWithStreamStateBase<AppendToStreamOptions> {
     private AppendToStreamOptions() {
     }
 

db-client-java/src/main/java/io/kurrent/dbclient/DeleteStream.java

@@ -4,7 +4,6 @@
 import io.kurrent.dbclient.proto.streams.StreamsGrpc;
 import io.kurrent.dbclient.proto.streams.StreamsOuterClass;
 import com.google.protobuf.ByteString;
-import io.grpc.Metadata;
 
 import java.util.concurrent.CompletableFuture;
 
@@ -28,7 +27,7 @@ public CompletableFuture<DeleteResult> execute() {
 
             if (this.softDelete) {
                 StreamsOuterClass.DeleteReq req = StreamsOuterClass.DeleteReq.newBuilder()
-                        .setOptions(this.options.getExpectedRevision().applyOnWire(StreamsOuterClass.DeleteReq.Options.newBuilder()
+                        .setOptions(this.options.getStreamState().applyOnWire(StreamsOuterClass.DeleteReq.Options.newBuilder()
                                 .setStreamIdentifier(Shared.StreamIdentifier.newBuilder()
                                         .setStreamName(ByteString.copyFromUtf8(streamName))
                                         .build())))
@@ -45,7 +44,7 @@ public CompletableFuture<DeleteResult> execute() {
             }
 
             StreamsOuterClass.TombstoneReq req = StreamsOuterClass.TombstoneReq.newBuilder()
-                    .setOptions(this.options.getExpectedRevision().applyOnWire(StreamsOuterClass.TombstoneReq.Options.newBuilder()
+                    .setOptions(this.options.getStreamState().applyOnWire(StreamsOuterClass.TombstoneReq.Options.newBuilder()
                             .setStreamIdentifier(Shared.StreamIdentifier.newBuilder()
                                     .setStreamName(ByteString.copyFromUtf8(streamName))
                                     .build())))

db-client-java/src/main/java/io/kurrent/dbclient/DeleteStreamOptions.java

@@ -3,7 +3,7 @@
 /**
  * Options of the delete or tombstone stream request.
  */
-public class DeleteStreamOptions extends OptionsWithExpectedRevisionBase<DeleteStreamOptions> {
+public class DeleteStreamOptions extends OptionsWithStreamStateBase<DeleteStreamOptions> {
     DeleteStreamOptions() {}
 
     /**

db-client-java/src/main/java/io/kurrent/dbclient/GrpcUtils.java

@@ -61,8 +61,8 @@ public void onError(Throwable t) {
                             }
 
                             try {
-                                ExpectedRevision expected = ExpectedRevision.fromRawLong(Long.parseLong(expectedStr));
-                                ExpectedRevision actual = ExpectedRevision.fromRawLong(Long.parseLong(actualStr));
+                                StreamState expected = StreamState.fromRawLong(Long.parseLong(expectedStr));
+                                StreamState actual = StreamState.fromRawLong(Long.parseLong(actualStr));
 
                                 dest.completeExceptionally(new WrongExpectedVersionException(streamName, expected, actual));
                                 return;

db-client-java/src/main/java/io/kurrent/dbclient/OptionsWithExpectedRevisionBase.java

@@ -0,0 +1,36 @@
+package io.kurrent.dbclient;
+
+class OptionsWithStreamStateBase<T> extends OptionsBase<T> {
+    private StreamState streamState;
+
+    protected OptionsWithStreamStateBase() {
+        this.streamState = StreamState.any();
+    }
+
+    StreamState getStreamState() {
+        return this.streamState;
+    }
+
+    /**
+     * Asks the server to check that the stream receiving is at the expected state.
+
+     * @param state - expected revision.
+     * @return updated options.
+     */
+    @SuppressWarnings("unchecked")
+    public T streamState(StreamState state) {
+        this.streamState = state;
+        return (T) this;
+    }
+
+
+    /**
+     * Asks the server to check that the stream receiving is at the given expected revision.
+
+     * @param revision - expected revision.
+     * @return updated options.
+     */
+    public T streamRevision(long revision) {
+        return streamState(StreamState.streamRevision(revision));
+    }
+}

db-client-java/src/main/java/io/kurrent/dbclient/OptionsWithStreamStateBase.java

@@ -14,66 +14,66 @@
  * guarantee it. Any other <i>ExpectedRevision</i> instances are meant for optimistic concurrency checks.
  * </p>
  */
-public abstract class ExpectedRevision {
+public abstract class StreamState {
     /**
      * This writes should not conflict with anything and should always succeed.
      */
-    public static ExpectedRevision any() {
-        return new AnyExpectedRevision();
+    public static StreamState any() {
+        return new AnyStreamState();
     }
 
     /**
      * The stream being written to should not yet exist. If it does exist, treats that as a concurrency problem.
      */
-    public static ExpectedRevision noStream() {
-        return new NoStreamExpectedRevision();
+    public static StreamState noStream() {
+        return new NoStreamState();
     }
 
     /**
      * The stream should exist. If it or a metadata stream does not exist, treats that as a concurrency problem.
      */
-    public static ExpectedRevision streamExists() {
-        return new StreamExistsExpectedRevision();
+    public static StreamState streamExists() {
+        return new StreamExistsState();
     }
 
     /**
      * States that the last event written to the stream should have an event revision matching your expected value.
      */
-    public static ExpectedRevision expectedRevision(long revision) {
-        return new SpecificExpectedRevision(revision);
+    public static StreamState streamRevision(long revision) {
+        return new StreamRevisionStreamState(revision);
     }
 
-    public static ExpectedRevision fromRawLong(long revision) {
+    public static StreamState fromRawLong(long revision) {
         if (revision == -1)
-            return ExpectedRevision.noStream();
+            return StreamState.noStream();
         if (revision == -2)
-            return ExpectedRevision.any();
+            return StreamState.any();
         if (revision == -4)
-            return ExpectedRevision.streamExists();
+            return StreamState.streamExists();
 
         if (revision < 0)
-            throw new RuntimeException(String.format("Invalid expected revision long representation '%s'", revision));
+            throw new RuntimeException(String.format("Invalid stream revision long representation '%s'", revision));
 
-        return ExpectedRevision.expectedRevision(revision);
+        return StreamState.streamRevision(revision);
     }
 
-    ExpectedRevision() {}
+    StreamState() {}
 
     abstract StreamsOuterClass.AppendReq.Options.Builder applyOnWire(StreamsOuterClass.AppendReq.Options.Builder options);
     abstract StreamsOuterClass.DeleteReq.Options.Builder applyOnWire(StreamsOuterClass.DeleteReq.Options.Builder options);
     abstract StreamsOuterClass.TombstoneReq.Options.Builder applyOnWire(StreamsOuterClass.TombstoneReq.Options.Builder options);
 
     public long toRawLong() {
-        if (this instanceof  NoStreamExpectedRevision)
+        if (this instanceof NoStreamState)
             return -1;
 
-        if (this instanceof AnyExpectedRevision)
+        if (this instanceof AnyStreamState)
             return -2;
 
-        if (this instanceof StreamExistsExpectedRevision)
+        if (this instanceof StreamExistsState)
             return -4;
 
-        SpecificExpectedRevision revision = (SpecificExpectedRevision) this;
+        StreamRevisionStreamState revision = (StreamRevisionStreamState) this;
 
         return revision.version;
     }
@@ -89,7 +89,7 @@ public int hashCode() {
         return Objects.hash(getClass());
     }
 
-    public static class NoStreamExpectedRevision extends ExpectedRevision {
+    public static class NoStreamState extends StreamState {
         @Override
         public StreamsOuterClass.AppendReq.Options.Builder applyOnWire(StreamsOuterClass.AppendReq.Options.Builder options) {
             return options.setNoStream(Shared.Empty.getDefaultInstance());
@@ -107,11 +107,11 @@ public StreamsOuterClass.TombstoneReq.Options.Builder applyOnWire(StreamsOuterCl
 
         @Override
         public String toString() {
-            return "ExpectedNoStream";
+            return "NoStreamState";
         }
     }
 
-    public static class AnyExpectedRevision extends ExpectedRevision {
+    public static class AnyStreamState extends StreamState {
         @Override
         public StreamsOuterClass.AppendReq.Options.Builder applyOnWire(StreamsOuterClass.AppendReq.Options.Builder options) {
             return options.setAny(Shared.Empty.getDefaultInstance());
@@ -129,11 +129,11 @@ public StreamsOuterClass.TombstoneReq.Options.Builder applyOnWire(StreamsOuterCl
 
         @Override
         public String toString() {
-            return "ExpectedAny";
+            return "AnyStreamState";
         }
     }
 
-    public static class StreamExistsExpectedRevision extends ExpectedRevision {
+    public static class StreamExistsState extends StreamState {
         @Override
         public StreamsOuterClass.AppendReq.Options.Builder applyOnWire(StreamsOuterClass.AppendReq.Options.Builder options) {
             return options.setStreamExists(Shared.Empty.getDefaultInstance());
@@ -151,22 +151,22 @@ public StreamsOuterClass.TombstoneReq.Options.Builder applyOnWire(StreamsOuterCl
 
         @Override
         public String toString() {
-            return "ExpectedStreamExists";
+            return "StreamExistsState";
         }
     }
 
-    public static class SpecificExpectedRevision extends ExpectedRevision {
+    public static class StreamRevisionStreamState extends StreamState {
         final long version;
 
-        SpecificExpectedRevision(long version) {
+        StreamRevisionStreamState(long version) {
             this.version = version;
         }
 
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
-            SpecificExpectedRevision that = (SpecificExpectedRevision) o;
+            StreamRevisionStreamState that = (StreamRevisionStreamState) o;
             return version == that.version;
         }
 

…o/kurrent/dbclient/ExpectedRevision.java …ava/io/kurrent/dbclient/StreamState.javadb-client-java/src/main/java/io/kurrent/dbclient/ExpectedRevision.java renamed to db-client-java/src/main/java/io/kurrent/dbclient/StreamState.java db-client-java/src/main/java/io/kurrent/dbclient/ExpectedRevision.java renamed to db-client-java/src/main/java/io/kurrent/dbclient/StreamState.java

@@ -6,18 +6,18 @@
  * Returned after writing to a stream.
  */
 public class WriteResult {
-    private final ExpectedRevision nextExpectedRevision;
+    private final StreamState nextExpectedRevision;
     private final Position logPosition;
 
-    WriteResult(ExpectedRevision nextExpectedRevision, Position logPosition) {
+    WriteResult(StreamState nextExpectedRevision, Position logPosition) {
         this.nextExpectedRevision = nextExpectedRevision;
         this.logPosition = logPosition;
     }
 
     /**
      * Next expected version of the stream.
      */
-    public ExpectedRevision getNextExpectedRevision() {
+    public StreamState getNextExpectedRevision() {
         return nextExpectedRevision;
     }