Repository: nats-io/nats.go
Pull request #1378 (2023-08-25 15:17:40)

[ADDED] UpdatesOnly() option for object store watchers
- Added `UpdatesOnly()` option to object store watchers
- Modify KV watcher behavior for `UpdatesOnly()` so that it does not send `nil` update after creating the watcher


Patch:
diff --git a/jetstream/kv.go b/jetstream/kv.go
index 9d7eae7..3546f01 100644
--- a/jetstream/kv.go
+++ b/jetstream/kv.go
@@ -856,12 +856,18 @@ func (kv *kvs) Watch(ctx context.Context, keys string, opts ...WatchOpt) (KeyWat
 	sub.SetClosedHandler(func(_ string) {
 		close(w.updates)
 	})
-	initialPending, err := sub.InitialConsumerPending()
 	// If there were no pending messages at the time of the creation
 	// of the consumer, send the marker.
-	if err == nil && initialPending == 0 {
+	// Skip if UpdatesOnly() is set, since there will never be updates initially.
+	if !o.updatesOnly {
+		initialPending, err := sub.InitialConsumerPending()
+		if err == nil && initialPending == 0 {
+			w.initDone = true
+			w.updates <- nil
+		}
+	} else {
+		// if UpdatesOnly was used, mark initialization as complete
 		w.initDone = true
-		w.updates <- nil
 	}
 	w.sub = sub
 	return w, nil
diff --git a/kv.go b/kv.go
index 7d2ab15..17c9e2f 100644
--- a/kv.go
+++ b/kv.go
@@ -941,6 +941,7 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 			w.updates <- entry
 		}
 		// Check if done and initial values.
+		// Skip if UpdatesOnly() is set, since there will never be updates initially.
 		if !w.initDone {
 			w.received++
 			// We set this on the first trip through..
@@ -980,9 +981,15 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 	sub.mu.Lock()
 	// If there were no pending messages at the time of the creation
 	// of the consumer, send the marker.
-	if sub.jsi != nil && sub.jsi.pending == 0 {
+	// Skip if UpdatesOnly() is set, since there will never be updates initially.
+	if !o.updatesOnly {
+		if sub.jsi != nil && sub.jsi.pending == 0 {
+			w.initDone = true
+			w.updates <- nil
+		}
+	} else {
+		// if UpdatesOnly was used, mark initialization as complete
 		w.initDone = true
-		w.updates <- nil
 	}
 	// Set us up to close when the waitForMessages func returns.
 	sub.pDone = func(_ string) {
diff --git a/object.go b/object.go
index b12af5c..5aafdfe 100644
--- a/object.go
+++ b/object.go
@@ -1056,6 +1056,8 @@ func (obs *obs) Watch(opts ...WatchOpt) (ObjectWatcher, error) {
 			w.updates <- &info
 		}
 
+		// if UpdatesOnly is set, no not send nil to the channel
+		// as it would always be triggered after initializing the watcher
 		if !initDoneMarker && meta.NumPending == 0 {
 			initDoneMarker = true
 			w.updates <- nil
@@ -1064,9 +1066,17 @@ func (obs *obs) Watch(opts ...WatchOpt) (ObjectWatcher, error) {
 
 	allMeta := fmt.Sprintf(objAllMetaPreTmpl, obs.name)
 	_, err := obs.js.GetLastMsg(obs.stream, allMeta)
-	if err == ErrMsgNotFound {
+	// if there are no messages on the stream and we are not watching
+	// updates only, send nil to the channel to indicate that the initial
+	// watch is done
+	if !o.updatesOnly {
+		if errors.Is(err, ErrMsgNotFound) {
+			initDoneMarker = true
+			w.updates <- nil
+		}
+	} else {
+		// if UpdatesOnly was used, mark initialization as complete
 		initDoneMarker = true
-		w.updates <- nil
 	}
 
 	// Used ordered consumer to deliver results.
@@ -1074,6 +1084,9 @@ func (obs *obs) Watch(opts ...WatchOpt) (ObjectWatcher, error) {
 	if !o.includeHistory {
 		subOpts = append(subOpts, DeliverLastPerSubject())
 	}
+	if o.updatesOnly {
+		subOpts = append(subOpts, DeliverNew())
+	}
 	sub, err := obs.js.Subscribe(allMeta, update, subOpts...)
 	if err != nil {
 		return nil, err


Test patch:
diff --git a/jetstream/test/kv_test.go b/jetstream/test/kv_test.go
index 764182d..6fdf839 100644
--- a/jetstream/test/kv_test.go
+++ b/jetstream/test/kv_test.go
@@ -365,11 +365,8 @@ func TestKeyValueWatch(t *testing.T) {
 		watcher, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
 		expectOk(t, err)
 		defer watcher.Stop()
-		expectInitDone := expectInitDoneF(t, watcher)
 		expectUpdate := expectUpdateF(t, watcher)
 		expectDelete := expectDeleteF(t, watcher)
-		// make sure we do not get any values here
-		expectInitDone()
 
 		// now update some keys and expect updates
 		_, err = kv.Put(ctx, "name", []byte("pp"))
@@ -397,11 +394,8 @@ func TestKeyValueWatch(t *testing.T) {
 		watcher, err = kv.Watch(ctx, "t.*", jetstream.UpdatesOnly())
 		expectOk(t, err)
 		defer watcher.Stop()
-		expectInitDone = expectInitDoneF(t, watcher)
 		expectUpdate = expectUpdateF(t, watcher)
 
-		expectInitDone()
-
 		// update some keys and expect updates
 		_, err = kv.Put(ctx, "t.name", []byte("pp"))
 		expectOk(t, err)
diff --git a/test/kv_test.go b/test/kv_test.go
index 56e547a..026891a 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -330,11 +330,8 @@ func TestKeyValueWatch(t *testing.T) {
 		watcher, err := kv.WatchAll(nats.UpdatesOnly())
 		expectOk(t, err)
 		defer watcher.Stop()
-		expectInitDone := expectInitDoneF(t, watcher)
 		expectUpdate := expectUpdateF(t, watcher)
 		expectDelete := expectDeleteF(t, watcher)
-		// make sure we do not get any values here
-		expectInitDone()
 
 		// now update some keys and expect updates
 		kv.Put("name", []byte("pp"))
@@ -356,11 +353,8 @@ func TestKeyValueWatch(t *testing.T) {
 		watcher, err = kv.Watch("t.*", nats.UpdatesOnly())
 		expectOk(t, err)
 		defer watcher.Stop()
-		expectInitDone = expectInitDoneF(t, watcher)
 		expectUpdate = expectUpdateF(t, watcher)
 
-		expectInitDone()
-
 		// update some keys and expect updates
 		kv.Put("t.name", []byte("pp"))
 		expectUpdate("t.name", "pp", 11)
diff --git a/test/object_test.go b/test/object_test.go
index 550015c..8cb9d98 100644
--- a/test/object_test.go
+++ b/test/object_test.go
@@ -474,86 +474,136 @@ func TestObjectMetadata(t *testing.T) {
 }
 
 func TestObjectWatch(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	defer shutdownJSServerAndRemoveStorage(t, s)
-
-	nc, js := jsClient(t, s)
-	defer nc.Close()
-
-	obs, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "WATCH-TEST"})
-	expectOk(t, err)
-
-	watcher, err := obs.Watch()
-	expectOk(t, err)
-	defer watcher.Stop()
-
-	expectUpdate := func(name string) {
-		t.Helper()
-		select {
-		case info := <-watcher.Updates():
-			if false && info.Name != name { // TODO what is supposed to happen here?
-				t.Fatalf("Expected update for %q, but got %+v", name, info)
+	expectUpdateF := func(t *testing.T, watcher nats.ObjectWatcher) func(name string) {
+		return func(name string) {
+			t.Helper()
+			select {
+			case info := <-watcher.Updates():
+				if false && info.Name != name { // TODO what is supposed to happen here?
+					t.Fatalf("Expected update for %q, but got %+v", name, info)
+				}
+			case <-time.After(time.Second):
+				t.Fatalf("Did not receive an update like expected")
 			}
-		case <-time.After(time.Second):
-			t.Fatalf("Did not receive an update like expected")
 		}
 	}
 
-	expectNoMoreUpdates := func() {
-		t.Helper()
-		select {
-		case info := <-watcher.Updates():
-			t.Fatalf("Got an unexpected update: %+v", info)
-		case <-time.After(100 * time.Millisecond):
+	expectNoMoreUpdatesF := func(t *testing.T, watcher nats.ObjectWatcher) func() {
+		return func() {
+			t.Helper()
+			select {
+			case info := <-watcher.Updates():
+				t.Fatalf("Got an unexpected update: %+v", info)
+			case <-time.After(100 * time.Millisecond):
+			}
 		}
 	}
 
-	expectInitDone := func() {
-		t.Helper()
-		select {
-		case info := <-watcher.Updates():
-			if info != nil {
-				t.Fatalf("Did not get expected: %+v", info)
+	expectInitDoneF := func(t *testing.T, watcher nats.ObjectWatcher) func() {
+		return func() {
+			t.Helper()
+			select {
+			case info := <-watcher.Updates():
+				if info != nil {
+					t.Fatalf("Did not get expected: %+v", info)
+				}
+			case <-time.After(time.Second):
+				t.Fatalf("Did not receive a init done like expected")
 			}
-		case <-time.After(time.Second):
-			t.Fatalf("Did not receive a init done like expected")
 		}
 	}
 
-	// We should get a marker that is nil when all initital values are delivered.
-	expectInitDone()
+	t.Run("default watcher", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
 
-	_, err = obs.PutString("A", "AAA")
-	expectOk(t, err)
-	_, err = obs.PutString("B", "BBB")
-	expectOk(t, err)
+		nc, js := jsClient(t, s)
+		defer nc.Close()
 
-	// Initial Values.
-	expectUpdate("A")
-	expectUpdate("B")
-	expectNoMoreUpdates()
+		obs, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "WATCH-TEST"})
+		expectOk(t, err)
 
-	// Delete
-	err = obs.Delete("A")
-	expectOk(t, err)
+		watcher, err := obs.Watch()
+		expectOk(t, err)
+		defer watcher.Stop()
 
-	expectUpdate("A")
-	expectNoMoreUpdates()
+		expectUpdate := expectUpdateF(t, watcher)
+		expectNoMoreUpdates := expectNoMoreUpdatesF(t, watcher)
+		expectInitDone := expectInitDoneF(t, watcher)
 
-	// New
-	_, err = obs.PutString("C", "CCC")
-	expectOk(t, err)
+		// We should get a marker that is nil when all initial values are delivered.
+		expectInitDone()
 
-	// Update Meta
-	deletedInfo, err := obs.GetInfo("A", nats.GetObjectInfoShowDeleted())
-	expectOk(t, err)
-	if !deletedInfo.Deleted {
-		t.Fatalf("Expected object to be deleted.")
-	}
-	meta := &deletedInfo.ObjectMeta
-	meta.Description = "Making a change."
-	err = obs.UpdateMeta("A", meta)
-	expectErr(t, err, nats.ErrUpdateMetaDeleted)
+		_, err = obs.PutString("A", "AAA")
+		expectOk(t, err)
+		_, err = obs.PutString("B", "BBB")
+		expectOk(t, err)
+
+		// Initial Values.
+		expectUpdate("A")
+		expectUpdate("B")
+		expectNoMoreUpdates()
+
+		// Delete
+		err = obs.Delete("A")
+		expectOk(t, err)
+
+		expectUpdate("A")
+		expectNoMoreUpdates()
+
+		// New
+		_, err = obs.PutString("C", "CCC")
+		expectOk(t, err)
+
+		// Update Meta
+		deletedInfo, err := obs.GetInfo("A", nats.GetObjectInfoShowDeleted())
+		expectOk(t, err)
+		if !deletedInfo.Deleted {
+			t.Fatalf("Expected object to be deleted.")
+		}
+		meta := &deletedInfo.ObjectMeta
+		meta.Description = "Making a change."
+		err = obs.UpdateMeta("A", meta)
+		expectErr(t, err, nats.ErrUpdateMetaDeleted)
+	})
+
+	t.Run("watcher with update", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		obs, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "WATCH-TEST"})
+		expectOk(t, err)
+
+		_, err = obs.PutString("A", "AAA")
+		expectOk(t, err)
+		_, err = obs.PutString("B", "BBB")
+		expectOk(t, err)
+
+		watcher, err := obs.Watch(nats.UpdatesOnly())
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		expectUpdate := expectUpdateF(t, watcher)
+		expectNoMoreUpdates := expectNoMoreUpdatesF(t, watcher)
+
+		// when listening for updates only, we should not receive anything when watcher is started
+		expectNoMoreUpdates()
+
+		// Delete
+		err = obs.Delete("A")
+		expectOk(t, err)
+
+		expectUpdate("A")
+		expectNoMoreUpdates()
+
+		// New
+		_, err = obs.PutString("C", "CCC")
+		expectOk(t, err)
+		expectUpdate("C")
+	})
 }
 
 func TestObjectLinks(t *testing.T) {