feat: [DEVEX-304] supports new caught up and fell behind events.

kurrentdb/subscription_event.go

@@ -1,5 +1,9 @@
 package kurrentdb
 
+import (
+	"time"
+)
+
 // SubscriptionEvent used to handle catch-up subscription notifications raised throughout its lifecycle.
 type SubscriptionEvent struct {
 	// When KurrentDB sends an event to the subscription.
@@ -9,9 +13,21 @@ type SubscriptionEvent struct {
 	// When a checkpoint was created.
 	CheckPointReached *Position
 	// When an event is caught up
-	CaughtUp *Subscription
+	CaughtUp *CaughtUp
 	// When an event has fallen behind
-	FellBehind *Subscription
+	FellBehind *FellBehind
+}
+
+type CaughtUp struct {
+	Date           time.Time
+	Position       *Position
+	StreamRevision *uint64
+}
+
+type FellBehind struct {
+	Date           time.Time
+	Position       *Position
+	StreamRevision *uint64
 }
 
 // PersistentSubscriptionEvent used to handle persistent subscription notifications raised throughout its lifecycle.

kurrentdb/subscriptions.go

@@ -97,14 +97,51 @@ func (sub *Subscription) Recv() *SubscriptionEvent {
 			}
 		case *api.ReadResp_CaughtUp_:
 			{
+				var caughtUp *CaughtUp = nil
+				wire := result.GetCaughtUp()
+
+				if wire != nil && wire.GetTimestamp() != nil {
+					caughtUp = new(CaughtUp)
+					caughtUp.Date = wire.GetTimestamp().AsTime()
+
+					if wire.StreamRevision != nil {
+						caughtUp.StreamRevision = new(uint64)
+						*caughtUp.StreamRevision = uint64(wire.GetStreamRevision())
+					} else {
+						caughtUp.Position = new(Position)
+						*caughtUp.Position = Position{
+							Commit:  wire.GetPosition().CommitPosition,
+							Prepare: wire.GetPosition().PreparePosition,
+						}
+					}
+				}
+
 				return &SubscriptionEvent{
-					CaughtUp: sub,
+					CaughtUp: caughtUp,
 				}
 			}
 		case *api.ReadResp_FellBehind_:
 			{
+				var fellbehind *FellBehind = nil
+				wire := result.GetFellBehind()
+
+				if wire != nil && wire.GetTimestamp() != nil {
+					fellbehind = new(FellBehind)
+					fellbehind.Date = wire.GetTimestamp().AsTime()
+
+					if wire.StreamRevision != nil {
+						fellbehind.StreamRevision = new(uint64)
+						*fellbehind.StreamRevision = uint64(wire.GetStreamRevision())
+					} else {
+						fellbehind.Position = new(Position)
+						*fellbehind.Position = Position{
+							Commit:  wire.GetPosition().CommitPosition,
+							Prepare: wire.GetPosition().PreparePosition,
+						}
+					}
+				}
 				return &SubscriptionEvent{
-					FellBehind: sub,
+					FellBehind: fellbehind,
 				}
 			}
 		}

protos/streams.proto

@@ -104,9 +104,37 @@ message ReadResp {
 		FellBehind fell_behind = 9;
 	}
 
-	message CaughtUp {}
+	// The $all or stream subscription has caught up and become live.
+	message CaughtUp {
+		// Current time in the server when the subscription caught up
+		google.protobuf.Timestamp timestamp = 1;
+
+		// Checkpoint for resuming a stream subscription.
+		// For stream subscriptions it is populated unless the stream is empty.
+		// For $all subscriptions it is not populated.
+		optional int64 stream_revision = 2;
+
+		// Checkpoint for resuming a $all subscription.
+		// For stream subscriptions it is not populated.
+		// For $all subscriptions it is populated unless the database is empty.
+		optional Position position = 3;
+	}
+
+	// The $all or stream subscription has fallen back into catchup mode and is no longer live.
+	message FellBehind {
+		// Current time in the server when the subscription fell behind
+		google.protobuf.Timestamp timestamp = 1;
+
+		// Checkpoint for resuming a stream subscription.
+		// For stream subscriptions it is populated unless the stream is empty.
+		// For $all subscriptions it is not populated.
+		optional int64 stream_revision = 2;
 
-	message FellBehind {}
+		// Checkpoint for resuming a $all subscription.
+		// For stream subscriptions it is not populated.
+		// For $all subscriptions it is populated unless the database is empty.
+		optional Position position = 3;
+	}
 
 	message ReadEvent {
 		RecordedEvent event = 1;
@@ -133,7 +161,16 @@ message ReadResp {
 	message Checkpoint {
 		uint64 commit_position = 1;
 		uint64 prepare_position = 2;
+
+		// Current time in the server when the checkpoint was reached
+		google.protobuf.Timestamp timestamp = 3;
+	}
+
+	message Position {
+		uint64 commit_position = 1;
+		uint64 prepare_position = 2;
 	}
+
 	message StreamNotFound {
 		event_store.client.StreamIdentifier stream_identifier = 1;
 	}
@@ -314,4 +351,4 @@ message TombstoneResp {
 		uint64 commit_position = 1;
 		uint64 prepare_position = 2;
 	}
-}
+}