[DEV-1038] fix: add missing no stream case to expected revision error handling (#363)

* fix: missing noStream case to expected revision error handling

Handle EXPECTED_NO_STREAM case in AppendToStream error response parsing. Previously this case fell through to the else branch, incorrectly returning a streamRevision state instead of noStream.

Author: w1am

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

@@ -7,6 +7,7 @@
 import io.grpc.ManagedChannel;
 import io.grpc.stub.StreamObserver;
 
+
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -69,6 +70,8 @@ private CompletableFuture<WriteResult> append(ManagedChannel channel, List<Event
                     expectedRevision = StreamState.any();
                 } else if (wev.getExpectedRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.ExpectedRevisionOptionCase.EXPECTED_STREAM_EXISTS) {
                     expectedRevision = StreamState.streamExists();
+                } else if (wev.getExpectedRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.ExpectedRevisionOptionCase.EXPECTED_NO_STREAM) {
+                    expectedRevision = StreamState.noStream();
                 } else {
                     expectedRevision = StreamState.streamRevision(wev.getExpectedRevision());
                 }

src/test/java/io/kurrent/dbclient/streams/AppendTests.java

@@ -12,41 +12,97 @@ public interface AppendTests extends ConnectionAware {
     default void testAppendSingleEventNoStream() throws Throwable {
         KurrentDBClient client = getDatabase().defaultClient();
         String streamName = generateName();
-        String eventType = "TestEvent";
         UUID eventId = UUID.randomUUID();
         Foo foo = new Foo();
         byte[] fooBytes = mapper.writeValueAsBytes(foo);
 
-        EventData event = EventData.builderAsJson(eventType, fooBytes)
+        EventData event = EventData.builderAsJson("TestEvent", fooBytes)
                 .metadataAsBytes(fooBytes)
                 .eventId(eventId)
                 .build();
 
         WriteResult appendResult = client.appendToStream(
-                streamName,
-                AppendToStreamOptions.get().streamState(StreamState.noStream()),
-                event
-        ).get();
+                streamName, AppendToStreamOptions.get().streamState(StreamState.noStream()), event).get();
 
         Assertions.assertEquals(StreamState.streamRevision(0), appendResult.getNextExpectedRevision());
 
-        ReadResult result = client.readStream(
-                streamName,
-                ReadStreamOptions.get().fromEnd().backwards().maxCount(1)
-        ).get();
-
-        Assertions.assertEquals(1, result.getEvents().size());
+        ReadResult result = client.readStream(streamName, ReadStreamOptions.get().fromEnd().backwards().maxCount(1)).get();
         RecordedEvent first = result.getEvents().get(0).getEvent();
         ObjectNode userMetadata = mapper.readValue(first.getUserMetadata(), ObjectNode.class);
 
         Assertions.assertAll(
                 () -> Assertions.assertEquals(streamName, first.getStreamId()),
-                () -> Assertions.assertEquals(eventType, first.getEventType()),
+                () -> Assertions.assertEquals("TestEvent", first.getEventType()),
                 () -> Assertions.assertEquals(eventId.toString(), first.getEventId().toString()),
                 () -> Assertions.assertEquals(foo, mapper.readValue(first.getEventData(), Foo.class)),
                 () -> Assertions.assertEquals(foo, mapper.readValue(first.getUserMetadata(), Foo.class)),
                 () -> Assertions.assertFalse(userMetadata.has(ClientTelemetryConstants.Metadata.TRACE_ID)),
                 () -> Assertions.assertFalse(userMetadata.has(ClientTelemetryConstants.Metadata.SPAN_ID))
         );
     }
+
+    @Test
+    default void testAppendMultipleEventsAtOnce() throws Throwable {
+        KurrentDBClient client = getDatabase().defaultClient();
+        String streamName = generateName();
+        int eventCount = 5;
+
+        WriteResult result = client.appendToStream(streamName,
+                AppendToStreamOptions.get().streamState(StreamState.noStream()),
+                generateEvents(eventCount, "TestEvent").iterator()).get();
+
+        Assertions.assertEquals(StreamState.streamRevision(eventCount - 1), result.getNextExpectedRevision());
+        Assertions.assertEquals(eventCount, client.readStream(streamName, ReadStreamOptions.get()).get().getEvents().size());
+    }
+
+    @Test
+    default void testStreamStateOptimisticConcurrency() throws Throwable {
+        KurrentDBClient client = getDatabase().defaultClient();
+
+        String anyStream = generateName();
+        appendEvent(client, anyStream, StreamState.any());
+        appendEvent(client, anyStream, StreamState.any());
+        Assertions.assertEquals(2, client.readStream(anyStream, ReadStreamOptions.get()).get().getEvents().size());
+
+        String existsStream = generateName();
+        assertWrongExpectedVersion(client, existsStream, StreamState.streamExists(), StreamState.streamExists(), StreamState.noStream());
+        appendEvent(client, existsStream, StreamState.noStream());
+        appendEvent(client, existsStream, StreamState.streamExists());
+
+        String noStream = generateName();
+        appendEvent(client, noStream, StreamState.noStream());
+        assertWrongExpectedVersion(client, noStream, StreamState.noStream(), StreamState.noStream(), StreamState.streamRevision(0));
+
+        String revStream = generateName();
+        appendEvent(client, revStream, StreamState.noStream());
+        appendEvent(client, revStream, StreamState.streamRevision(0));
+        assertWrongExpectedVersion(client, revStream, StreamState.streamRevision(99), StreamState.streamRevision(99), StreamState.streamRevision(1));
+    }
+
+    default EventData createTestEvent() throws Exception {
+        return EventData.builderAsJson("TestEvent", mapper.writeValueAsBytes(new Foo()))
+                .eventId(UUID.randomUUID())
+                .build();
+    }
+
+    default void appendEvent(KurrentDBClient client, String streamName, StreamState state) throws Exception {
+        client.appendToStream(streamName, AppendToStreamOptions.get().streamState(state), createTestEvent()).get();
+    }
+
+    default void assertWrongExpectedVersion(KurrentDBClient client, String streamName, StreamState state, StreamState expectedState, StreamState actualState) {
+        WrongExpectedVersionException ex = Assertions.assertThrows(WrongExpectedVersionException.class, () -> {
+            try {
+                appendEvent(client, streamName, state);
+            } catch (java.util.concurrent.ExecutionException e) {
+                if (e.getCause() != null) {
+                    throw e.getCause();
+                }
+                throw e;
+            }
+        });
+        Assertions.assertEquals(streamName, ex.getStreamName());
+        Assertions.assertEquals(expectedState, ex.getExpectedState());
+        Assertions.assertEquals(actualState, ex.getActualState());
+    }
+
 }