Repository: EventStore/EventStoreDB-Client-Java (java)

Issue:
Support "CaughtUp" gRPC control message for subscriptions when a subscription becomes live
We tried to update to eventstore 23.10.0 and encountered an issue.

With the introduction of the new CaughtUp control message, there is no longer a checkpoint emitted at this point. Which breaks our usecase as we used this to emulate the CaughtUp. 
Instead of this a warning is logged
`2023-11-09T12:19:03.418576641Z [WARN] .d.AbstractRegularSubscription: Confirmed subscription a16d6684-1ddb-4bf7-8dac-5acec7af9146 received non-{event,checkpoint} variant`

So support for the new messages would be nice and would allow updating to the new lts eventstore

Related: https://github.com/EventStore/EventStore/pull/3899

Pull request description:
Support new subscription messages.
Added: Support new subscription messages.

Fixes #253 

Patch:
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java b/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java
index b073f80..0ad3194 100644
--- a/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java
@@ -88,6 +88,16 @@ abstract class AbstractRegularSubscription {
                         return;
                     }
 
+                    if (_confirmed && readResp.hasCaughtUp()) {
+                        listener.onCaughtUp(_subscription);
+                        return;
+                    }
+
+                    if (_confirmed && readResp.hasFellBehind()) {
+                        listener.onFellBehind(_subscription);
+                        return;
+                    }
+
                     if (_confirmed && !readResp.hasEvent()) {
                         logger.warn(
                                 String.format("Confirmed subscription %s received non-{event,checkpoint} variant",
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/SubscriptionListener.java b/db-client-java/src/main/java/com/eventstore/dbclient/SubscriptionListener.java
index 5cca9db..b31a182 100644
--- a/db-client-java/src/main/java/com/eventstore/dbclient/SubscriptionListener.java
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/SubscriptionListener.java
@@ -23,4 +23,17 @@ public abstract class SubscriptionListener {
      * @param subscription handle to the subscription.
      */
     public void onConfirmation(Subscription subscription) {}
+
+    /**
+     * Called when the subscription has reached the head of the stream.
+     * @param subscription handle to the subscription.
+     */
+    public void onCaughtUp(Subscription subscription) {}
+
+    /**
+     * Called when the subscription has fallen behind, meaning it's no longer keeping up with the
+     * stream's pace.
+     * @param subscription handle to the subscription.
+     */
+    public void onFellBehind(Subscription subscription) {}
 }
diff --git a/db-client-java/src/main/proto/streams.proto b/db-client-java/src/main/proto/streams.proto
index 6b6d41f..ac599fb 100644
--- a/db-client-java/src/main/proto/streams.proto
+++ b/db-client-java/src/main/proto/streams.proto
@@ -4,6 +4,7 @@ option java_package = "com.eventstore.dbclient.proto.streams";
 
 import "shared.proto";
 import "status.proto";
+import "google/protobuf/duration.proto";
 import "google/protobuf/empty.proto";
 import "google/protobuf/timestamp.proto";
 
@@ -98,8 +99,14 @@ message ReadResp {
 		uint64 first_stream_position = 5;
 		uint64 last_stream_position = 6;
 		AllStreamPosition last_all_stream_position = 7;
+		CaughtUp caught_up = 8;
+		FellBehind fell_behind = 9;
 	}
 
+	message CaughtUp {}
+
+	message FellBehind {}
+
 	message ReadEvent {
 		RecordedEvent event = 1;
 		RecordedEvent link = 2;
@@ -214,7 +221,10 @@ message BatchAppendReq {
 			google.protobuf.Empty any = 4;
 			google.protobuf.Empty stream_exists = 5;
 		}
-		google.protobuf.Timestamp deadline = 6;
+		oneof deadline_option {
+			google.protobuf.Timestamp deadline_21_10_0 = 6;
+			google.protobuf.Duration deadline = 7;
+		}
 	}
 
 	message ProposedMessage {


Test patch:
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/streams/SubscriptionTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/streams/SubscriptionTests.java
index 6e34083..ea5fc52 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/streams/SubscriptionTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/streams/SubscriptionTests.java
@@ -2,11 +2,13 @@ package com.eventstore.dbclient.streams;
 
 import com.eventstore.dbclient.*;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Optional;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -152,4 +154,30 @@ public interface SubscriptionTests extends ConnectionAware {
 
         sub.stop();
     }
+
+    @Test
+    @Timeout(value = 5, unit = TimeUnit.MINUTES)
+    default void testCaughtUpMessageIsReceived() throws Throwable {
+        EventStoreDBClient client = getDefaultClient();
+        Optional<ServerVersion> version = client.getServerVersion().get();
+
+        Assumptions.assumeTrue(version.isPresent());
+        Assumptions.assumeFalse(version.get().isLessThan(23, 10, 0));
+
+        String streamName = generateName();
+
+        ArrayList<EventData> events = serializeBazEvents(generateBazEvent(20));
+        client.appendToStream(streamName, events.iterator()).get();
+        final CountDownLatch caughtUp = new CountDownLatch(1);
+
+        Subscription subscription = client.subscribeToStream(streamName, new SubscriptionListener() {
+            @Override
+            public void onCaughtUp(Subscription subscription) {
+                caughtUp.countDown();
+            }
+        }, SubscribeToStreamOptions.get().fromStart()).get();
+
+        caughtUp.await();
+        subscription.stop();
+    }
 }