Repository: nats-io/nats.go
Pull request #1377 (2023-08-25 08:15:18)

[ADDED] UpdatesOnly option to KV watcher


Patch:
diff --git a/jetstream/kv.go b/jetstream/kv.go
index 82deaba..9d7eae7 100644
--- a/jetstream/kv.go
+++ b/jetstream/kv.go
@@ -164,6 +164,8 @@ type (
 		ignoreDeletes bool
 		// Include all history per subject, not just last one.
 		includeHistory bool
+		// Include only updates for keys.
+		updatesOnly bool
 		// retrieve only the meta data of the entry
 		metaOnly bool
 	}
@@ -835,6 +837,9 @@ func (kv *kvs) Watch(ctx context.Context, keys string, opts ...WatchOpt) (KeyWat
 	if !o.includeHistory {
 		subOpts = append(subOpts, nats.DeliverLastPerSubject())
 	}
+	if o.updatesOnly {
+		subOpts = append(subOpts, nats.DeliverNew())
+	}
 	if o.metaOnly {
 		subOpts = append(subOpts, nats.HeadersOnly())
 	}
diff --git a/jetstream/options.go b/jetstream/options.go
index b407a91..e1b4b5b 100644
--- a/jetstream/options.go
+++ b/jetstream/options.go
@@ -328,11 +328,25 @@ func (opt watchOptFn) configureWatcher(opts *watchOpts) error {
 // IncludeHistory instructs the key watcher to include historical values as well.
 func IncludeHistory() WatchOpt {
 	return watchOptFn(func(opts *watchOpts) error {
+		if opts.updatesOnly {
+			return fmt.Errorf("%w: include history can not be used with updates only", ErrInvalidOption)
+		}
 		opts.includeHistory = true
 		return nil
 	})
 }
 
+// UpdatesOnly instructs the key watcher to only include updates on values (without latest values when started).
+func UpdatesOnly() WatchOpt {
+	return watchOptFn(func(opts *watchOpts) error {
+		if opts.includeHistory {
+			return fmt.Errorf("%w: updates only can not be used with include history", ErrInvalidOption)
+		}
+		opts.updatesOnly = true
+		return nil
+	})
+}
+
 // IgnoreDeletes will have the key watcher not pass any deleted keys.
 func IgnoreDeletes() WatchOpt {
 	return watchOptFn(func(opts *watchOpts) error {
diff --git a/kv.go b/kv.go
index 34d585d..7d2ab15 100644
--- a/kv.go
+++ b/kv.go
@@ -123,6 +123,8 @@ type watchOpts struct {
 	ignoreDeletes bool
 	// Include all history per subject, not just last one.
 	includeHistory bool
+	// Include only updates for keys.
+	updatesOnly bool
 	// retrieve only the meta data of the entry
 	metaOnly bool
 }
@@ -136,11 +138,25 @@ func (opt watchOptFn) configureWatcher(opts *watchOpts) error {
 // IncludeHistory instructs the key watcher to include historical values as well.
 func IncludeHistory() WatchOpt {
 	return watchOptFn(func(opts *watchOpts) error {
+		if opts.updatesOnly {
+			return errors.New("nats: include history can not be used with updates only")
+		}
 		opts.includeHistory = true
 		return nil
 	})
 }
 
+// UpdatesOnly instructs the key watcher to only include updates on values (without latest values when started).
+func UpdatesOnly() WatchOpt {
+	return watchOptFn(func(opts *watchOpts) error {
+		if opts.includeHistory {
+			return errors.New("nats: updates only can not be used with include history")
+		}
+		opts.updatesOnly = true
+		return nil
+	})
+}
+
 // IgnoreDeletes will have the key watcher not pass any deleted keys.
 func IgnoreDeletes() WatchOpt {
 	return watchOptFn(func(opts *watchOpts) error {
@@ -943,6 +959,9 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 	if !o.includeHistory {
 		subOpts = append(subOpts, DeliverLastPerSubject())
 	}
+	if o.updatesOnly {
+		subOpts = append(subOpts, DeliverNew())
+	}
 	if o.metaOnly {
 		subOpts = append(subOpts, HeadersOnly())
 	}


Test patch:
diff --git a/jetstream/test/kv_test.go b/jetstream/test/kv_test.go
index ea54d48..764182d 100644
--- a/jetstream/test/kv_test.go
+++ b/jetstream/test/kv_test.go
@@ -154,99 +154,262 @@ func TestKeyValueHistory(t *testing.T) {
 }
 
 func TestKeyValueWatch(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	defer shutdownJSServerAndRemoveStorage(t, s)
-
-	nc, js := jsClient(t, s)
-	defer nc.Close()
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCH"})
-	expectOk(t, err)
-
-	watcher, err := kv.WatchAll(ctx)
-	expectOk(t, err)
-	defer watcher.Stop()
-
-	expectUpdate := func(key, value string, revision uint64) {
-		t.Helper()
-		select {
-		case v := <-watcher.Updates():
-			if v.Key() != key || string(v.Value()) != value || v.Revision() != revision {
-				t.Fatalf("Did not get expected: %+v vs %q %q %d", v, key, value, revision)
+	expectUpdateF := func(t *testing.T, watcher jetstream.KeyWatcher) func(key, value string, revision uint64) {
+		return func(key, value string, revision uint64) {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v.Key() != key || string(v.Value()) != value || v.Revision() != revision {
+					t.Fatalf("Did not get expected: %q %q %d vs %q %q %d", v.Key(), string(v.Value()), v.Revision(), key, value, revision)
+				}
+			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")
 		}
 	}
-	expectDelete := func(key string, revision uint64) {
-		t.Helper()
-		select {
-		case v := <-watcher.Updates():
-			if v.Operation() != jetstream.KeyValueDelete {
-				t.Fatalf("Expected a delete operation but got %+v", v)
-			}
-			if v.Revision() != revision {
-				t.Fatalf("Did not get expected revision: %d vs %d", revision, v.Revision())
+	expectDeleteF := func(t *testing.T, watcher jetstream.KeyWatcher) func(key string, revision uint64) {
+		return func(key string, revision uint64) {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v.Operation() != jetstream.KeyValueDelete {
+					t.Fatalf("Expected a delete operation but got %+v", v)
+				}
+				if v.Revision() != revision {
+					t.Fatalf("Did not get expected revision: %d vs %d", revision, v.Revision())
+				}
+			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")
 		}
 	}
-	expectInitDone := func() {
-		t.Helper()
-		select {
-		case v := <-watcher.Updates():
-			if v != nil {
-				t.Fatalf("Did not get expected: %+v", v)
+	expectInitDoneF := func(t *testing.T, watcher jetstream.KeyWatcher) func() {
+		return func() {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v != nil {
+					t.Fatalf("Did not get expected: %+v", v)
+				}
+			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")
 		}
 	}
 
-	// Make sure we already got an initial value marker.
-	expectInitDone()
+	t.Run("default watcher", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
 
-	_, err = kv.Create(ctx, "name", []byte("derek"))
-	expectOk(t, err)
-	expectUpdate("name", "derek", 1)
-	_, err = kv.Put(ctx, "name", []byte("rip"))
-	expectOk(t, err)
-	expectUpdate("name", "rip", 2)
-	_, err = kv.Put(ctx, "name", []byte("ik"))
-	expectOk(t, err)
-	expectUpdate("name", "ik", 3)
-	_, err = kv.Put(ctx, "age", []byte("22"))
-	expectOk(t, err)
-	expectUpdate("age", "22", 4)
-	_, err = kv.Put(ctx, "age", []byte("33"))
-	expectOk(t, err)
-	expectUpdate("age", "33", 5)
-	expectOk(t, kv.Delete(ctx, "age"))
-	expectDelete("age", 6)
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
 
-	// Stop first watcher.
-	watcher.Stop()
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCH"})
+		expectOk(t, err)
 
-	// Now try wildcard matching and make sure we only get last value when starting.
-	_, err = kv.Put(ctx, "t.name", []byte("rip"))
-	expectOk(t, err)
-	_, err = kv.Put(ctx, "t.name", []byte("ik"))
-	expectOk(t, err)
-	_, err = kv.Put(ctx, "t.age", []byte("22"))
-	expectOk(t, err)
-	_, err = kv.Put(ctx, "t.age", []byte("44"))
-	expectOk(t, err)
+		watcher, err := kv.WatchAll(ctx)
+		expectOk(t, err)
+		defer watcher.Stop()
 
-	watcher, err = kv.Watch(ctx, "t.*")
-	expectOk(t, err)
-	defer watcher.Stop()
+		expectInitDone := expectInitDoneF(t, watcher)
+		expectUpdate := expectUpdateF(t, watcher)
+		expectDelete := expectDeleteF(t, watcher)
+		// Make sure we already got an initial value marker.
+		expectInitDone()
+
+		_, err = kv.Create(ctx, "name", []byte("derek"))
+		expectOk(t, err)
+		expectUpdate("name", "derek", 1)
+		_, err = kv.Put(ctx, "name", []byte("rip"))
+		expectOk(t, err)
+		expectUpdate("name", "rip", 2)
+		_, err = kv.Put(ctx, "name", []byte("ik"))
+		expectOk(t, err)
+		expectUpdate("name", "ik", 3)
+		_, err = kv.Put(ctx, "age", []byte("22"))
+		expectOk(t, err)
+		expectUpdate("age", "22", 4)
+		_, err = kv.Put(ctx, "age", []byte("33"))
+		expectOk(t, err)
+		expectUpdate("age", "33", 5)
+		expectOk(t, kv.Delete(ctx, "age"))
+		expectDelete("age", 6)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		// Now try wildcard matching and make sure we only get last value when starting.
+		_, err = kv.Put(ctx, "t.name", []byte("rip"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.name", []byte("ik"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("22"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("44"))
+		expectOk(t, err)
+
+		watcher, err = kv.Watch(ctx, "t.*")
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		expectInitDone = expectInitDoneF(t, watcher)
+		expectUpdate = expectUpdateF(t, watcher)
+		expectUpdate("t.name", "ik", 8)
+		expectUpdate("t.age", "44", 10)
+		expectInitDone()
+	})
 
-	expectUpdate("t.name", "ik", 8)
-	expectUpdate("t.age", "44", 10)
-	expectInitDone()
+	t.Run("watcher with history included", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCH", History: 64})
+		expectOk(t, err)
+
+		_, err = kv.Create(ctx, "name", []byte("derek"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "name", []byte("rip"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "name", []byte("ik"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "age", []byte("22"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "age", []byte("33"))
+		expectOk(t, err)
+		expectOk(t, kv.Delete(ctx, "age"))
+
+		// when using IncludeHistory(), UpdatesOnly() is not allowed
+		if _, err := kv.WatchAll(ctx, jetstream.IncludeHistory(), jetstream.UpdatesOnly()); !errors.Is(err, jetstream.ErrInvalidOption) {
+			t.Fatalf("Expected %v, got %v", jetstream.ErrInvalidOption, err)
+		}
+
+		watcher, err := kv.WatchAll(ctx, jetstream.IncludeHistory())
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		expectInitDone := expectInitDoneF(t, watcher)
+		expectUpdate := expectUpdateF(t, watcher)
+		expectDelete := expectDeleteF(t, watcher)
+		expectUpdate("name", "derek", 1)
+		expectUpdate("name", "rip", 2)
+		expectUpdate("name", "ik", 3)
+		expectUpdate("age", "22", 4)
+		expectUpdate("age", "33", 5)
+		expectDelete("age", 6)
+		expectInitDone()
+		_, err = kv.Put(ctx, "name", []byte("pp"))
+		expectOk(t, err)
+		expectUpdate("name", "pp", 7)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		_, err = kv.Put(ctx, "t.name", []byte("rip"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.name", []byte("ik"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("22"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("44"))
+		expectOk(t, err)
+
+		// try wildcard watcher and make sure we get all historical values
+		watcher, err = kv.Watch(ctx, "t.*", jetstream.IncludeHistory())
+		expectOk(t, err)
+		defer watcher.Stop()
+		expectInitDone = expectInitDoneF(t, watcher)
+		expectUpdate = expectUpdateF(t, watcher)
+
+		expectUpdate("t.name", "rip", 8)
+		expectUpdate("t.name", "ik", 9)
+		expectUpdate("t.age", "22", 10)
+		expectUpdate("t.age", "44", 11)
+		expectInitDone()
+
+		_, err = kv.Put(ctx, "t.name", []byte("pp"))
+		expectOk(t, err)
+		expectUpdate("t.name", "pp", 12)
+	})
+
+	t.Run("watcher with updates only", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCH", History: 64})
+		expectOk(t, err)
+
+		_, err = kv.Create(ctx, "name", []byte("derek"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "name", []byte("rip"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "age", []byte("22"))
+		expectOk(t, err)
+
+		// when using UpdatesOnly(), IncludeHistory() is not allowed
+		if _, err := kv.WatchAll(ctx, jetstream.UpdatesOnly(), jetstream.IncludeHistory()); !errors.Is(err, jetstream.ErrInvalidOption) {
+			t.Fatalf("Expected %v, got %v", jetstream.ErrInvalidOption, err)
+		}
+
+		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"))
+		expectOk(t, err)
+		expectUpdate("name", "pp", 4)
+		_, err = kv.Put(ctx, "age", []byte("44"))
+		expectOk(t, err)
+		expectUpdate("age", "44", 5)
+		expectOk(t, kv.Delete(ctx, "age"))
+		expectDelete("age", 6)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		_, err = kv.Put(ctx, "t.name", []byte("rip"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.name", []byte("ik"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("22"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("44"))
+		expectOk(t, err)
+
+		// try wildcard watcher and make sure we do not get any values initially
+		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)
+		expectUpdate("t.name", "pp", 11)
+		_, err = kv.Put(ctx, "t.age", []byte("66"))
+		expectOk(t, err)
+		expectUpdate("t.age", "66", 12)
+	})
 }
 
 func TestKeyValueWatchContext(t *testing.T) {
diff --git a/test/kv_test.go b/test/kv_test.go
index 78f741e..56e547a 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -149,88 +149,224 @@ func TestKeyValueHistory(t *testing.T) {
 }
 
 func TestKeyValueWatch(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	defer shutdownJSServerAndRemoveStorage(t, s)
-
-	nc, js := jsClient(t, s)
-	defer nc.Close()
-
-	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "WATCH"})
-	expectOk(t, err)
-
-	watcher, err := kv.WatchAll()
-	expectOk(t, err)
-	defer watcher.Stop()
-
-	expectUpdate := func(key, value string, revision uint64) {
-		t.Helper()
-		select {
-		case v := <-watcher.Updates():
-			if v.Key() != key || string(v.Value()) != value || v.Revision() != revision {
-				t.Fatalf("Did not get expected: %+v vs %q %q %d", v, key, value, revision)
+	expectUpdateF := func(t *testing.T, watcher nats.KeyWatcher) func(key, value string, revision uint64) {
+		return func(key, value string, revision uint64) {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v.Key() != key || string(v.Value()) != value || v.Revision() != revision {
+					t.Fatalf("Did not get expected: %q %q %d vs %q %q %d", v.Key(), string(v.Value()), v.Revision(), key, value, revision)
+				}
+			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")
 		}
 	}
-	expectDelete := func(key string, revision uint64) {
-		t.Helper()
-		select {
-		case v := <-watcher.Updates():
-			if v.Operation() != nats.KeyValueDelete {
-				t.Fatalf("Expected a delete operation but got %+v", v)
-			}
-			if v.Revision() != revision {
-				t.Fatalf("Did not get expected revision: %d vs %d", revision, v.Revision())
+	expectDeleteF := func(t *testing.T, watcher nats.KeyWatcher) func(key string, revision uint64) {
+		return func(key string, revision uint64) {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v.Operation() != nats.KeyValueDelete {
+					t.Fatalf("Expected a delete operation but got %+v", v)
+				}
+				if v.Revision() != revision {
+					t.Fatalf("Did not get expected revision: %d vs %d", revision, v.Revision())
+				}
+			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")
 		}
 	}
-	expectInitDone := func() {
-		t.Helper()
-		select {
-		case v := <-watcher.Updates():
-			if v != nil {
-				t.Fatalf("Did not get expected: %+v", v)
+	expectInitDoneF := func(t *testing.T, watcher nats.KeyWatcher) func() {
+		return func() {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v != nil {
+					t.Fatalf("Did not get expected: %+v", v)
+				}
+			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")
 		}
 	}
 
-	// Make sure we already got an initial value marker.
-	expectInitDone()
+	t.Run("default watcher", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
 
-	kv.Create("name", []byte("derek"))
-	expectUpdate("name", "derek", 1)
-	kv.Put("name", []byte("rip"))
-	expectUpdate("name", "rip", 2)
-	kv.Put("name", []byte("ik"))
-	expectUpdate("name", "ik", 3)
-	kv.Put("age", []byte("22"))
-	expectUpdate("age", "22", 4)
-	kv.Put("age", []byte("33"))
-	expectUpdate("age", "33", 5)
-	kv.Delete("age")
-	expectDelete("age", 6)
+		nc, js := jsClient(t, s)
+		defer nc.Close()
 
-	// Stop first watcher.
-	watcher.Stop()
+		kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "WATCH"})
+		expectOk(t, err)
 
-	// Now try wildcard matching and make sure we only get last value when starting.
-	kv.Put("t.name", []byte("rip"))
-	kv.Put("t.name", []byte("ik"))
-	kv.Put("t.age", []byte("22"))
-	kv.Put("t.age", []byte("44"))
+		watcher, err := kv.WatchAll()
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		expectInitDone := expectInitDoneF(t, watcher)
+		expectUpdate := expectUpdateF(t, watcher)
+		expectDelete := expectDeleteF(t, watcher)
+		// Make sure we already got an initial value marker.
+		expectInitDone()
+
+		kv.Create("name", []byte("derek"))
+		expectUpdate("name", "derek", 1)
+		kv.Put("name", []byte("rip"))
+		expectUpdate("name", "rip", 2)
+		kv.Put("name", []byte("ik"))
+		expectUpdate("name", "ik", 3)
+		kv.Put("age", []byte("22"))
+		expectUpdate("age", "22", 4)
+		kv.Put("age", []byte("33"))
+		expectUpdate("age", "33", 5)
+		kv.Delete("age")
+		expectDelete("age", 6)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		// Now try wildcard matching and make sure we only get last value when starting.
+		kv.Put("t.name", []byte("rip"))
+		kv.Put("t.name", []byte("ik"))
+		kv.Put("t.age", []byte("22"))
+		kv.Put("t.age", []byte("44"))
+
+		watcher, err = kv.Watch("t.*")
+		expectOk(t, err)
+		defer watcher.Stop()
 
-	watcher, err = kv.Watch("t.*")
-	expectOk(t, err)
-	defer watcher.Stop()
+		expectInitDone = expectInitDoneF(t, watcher)
+		expectUpdate = expectUpdateF(t, watcher)
+		expectUpdate("t.name", "ik", 8)
+		expectUpdate("t.age", "44", 10)
+		expectInitDone()
+	})
+
+	t.Run("watcher with history included", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "WATCH", History: 64})
+		expectOk(t, err)
+
+		kv.Create("name", []byte("derek"))
+		kv.Put("name", []byte("rip"))
+		kv.Put("name", []byte("ik"))
+		kv.Put("age", []byte("22"))
+		kv.Put("age", []byte("33"))
+		kv.Delete("age")
+
+		// when using UpdatesOnly(), IncludeHistory() is not allowed
+		if _, err := kv.WatchAll(nats.IncludeHistory(), nats.UpdatesOnly()); !strings.Contains(err.Error(), "updates only can not be used with include history") {
+			t.Fatalf("Expected error to contain %q, got %q", "updates only can not be used with include history", err)
+		}
+
+		watcher, err := kv.WatchAll(nats.IncludeHistory())
+		expectOk(t, err)
+		defer watcher.Stop()
+		expectInitDone := expectInitDoneF(t, watcher)
+		expectUpdate := expectUpdateF(t, watcher)
+		expectDelete := expectDeleteF(t, watcher)
+		expectUpdate("name", "derek", 1)
+		expectUpdate("name", "rip", 2)
+		expectUpdate("name", "ik", 3)
+		expectUpdate("age", "22", 4)
+		expectUpdate("age", "33", 5)
+		expectDelete("age", 6)
+		expectInitDone()
+		kv.Put("name", []byte("pp"))
+		expectUpdate("name", "pp", 7)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		kv.Put("t.name", []byte("rip"))
+		kv.Put("t.name", []byte("ik"))
+		kv.Put("t.age", []byte("22"))
+		kv.Put("t.age", []byte("44"))
+
+		// try wildcard watcher and make sure we get all historical values
+		watcher, err = kv.Watch("t.*", nats.IncludeHistory())
+		expectOk(t, err)
+		defer watcher.Stop()
+		expectInitDone = expectInitDoneF(t, watcher)
+		expectUpdate = expectUpdateF(t, watcher)
+
+		expectUpdate("t.name", "rip", 8)
+		expectUpdate("t.name", "ik", 9)
+		expectUpdate("t.age", "22", 10)
+		expectUpdate("t.age", "44", 11)
+		expectInitDone()
+
+		kv.Put("t.name", []byte("pp"))
+		expectUpdate("t.name", "pp", 12)
+	})
 
-	expectUpdate("t.name", "ik", 8)
-	expectUpdate("t.age", "44", 10)
-	expectInitDone()
+	t.Run("watcher with updates only", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "WATCH", History: 64})
+		expectOk(t, err)
+
+		kv.Create("name", []byte("derek"))
+		kv.Put("name", []byte("rip"))
+		kv.Put("age", []byte("22"))
+
+		// when using UpdatesOnly(), IncludeHistory() is not allowed
+		if _, err := kv.WatchAll(nats.UpdatesOnly(), nats.IncludeHistory()); !strings.Contains(err.Error(), "include history can not be used with updates only") {
+			t.Fatalf("Expected error to contain %q, got %q", "include history can not be used with updates only", err)
+		}
+
+		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"))
+		expectUpdate("name", "pp", 4)
+		kv.Put("age", []byte("44"))
+		expectUpdate("age", "44", 5)
+		kv.Delete("age")
+		expectDelete("age", 6)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		kv.Put("t.name", []byte("rip"))
+		kv.Put("t.name", []byte("ik"))
+		kv.Put("t.age", []byte("22"))
+		kv.Put("t.age", []byte("44"))
+
+		// try wildcard watcher and make sure we do not get any values initially
+		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)
+		kv.Put("t.age", []byte("66"))
+		expectUpdate("t.age", "66", 12)
+	})
 }
 
 func TestKeyValueWatchContext(t *testing.T) {