diff --git a/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs b/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs index ad2812eeb..1eaaf4c5b 100644 --- a/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs +++ b/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs @@ -259,6 +259,17 @@ private async Task CommandLoop() if (_opts.IgnoreDeletes && operation is NatsKVOperation.Del or NatsKVOperation.Purge) { + // Check in case all entries are deleted and we want to terminate + // the watcher loop on no-data. + if (msg.Metadata?.NumPending == 0 && _opts.OnNoData != null) + { + if (await _opts.OnNoData(_cancellationToken)) + { + _entryChannel.Writer.Complete(); + return; + } + } + continue; } diff --git a/tests/NATS.Client.KeyValueStore.Tests/NatsKVWatcherTest.cs b/tests/NATS.Client.KeyValueStore.Tests/NatsKVWatcherTest.cs index 615a5cb50..9b6c3df9a 100644 --- a/tests/NATS.Client.KeyValueStore.Tests/NatsKVWatcherTest.cs +++ b/tests/NATS.Client.KeyValueStore.Tests/NatsKVWatcherTest.cs @@ -619,4 +619,93 @@ await Assert.ThrowsAsync(async () => }); } } + + [Fact] + public async Task ReadAfterDelete() + { + await using var server = await NatsServer.StartJSAsync(); + await using var nats = await server.CreateClientConnectionAsync(); + + var js = new NatsJSContext(nats); + var kv = new NatsKVContext(js); + + var store = await kv.CreateStoreAsync("b1"); + + // Read all entries, should be empty. + List> results = new(); + await foreach (var entry in store.WatchAsync(">", opts: new NatsKVWatchOpts + { + OnNoData = (_) => ValueTask.FromResult(true), + })) + { + results.Add(entry); + } + + // Should be no results here. + Assert.False(results.Any()); + + // Add k1 + await store.PutAsync("k1", "v1"); + + // Check if there, should be true + var result1 = await store.TryGetEntryAsync("k1"); + Assert.True(result1.Success); + + // Remove k1 + await store.DeleteAsync("k1"); + + // Check if there, should be false + var result = await store.TryGetEntryAsync("k1"); + Assert.False(result.Success); + + // Read all entries. + results.Clear(); + await foreach (var entry in store.WatchAsync(">", opts: new NatsKVWatchOpts + { + OnNoData = (_) => ValueTask.FromResult(true), + })) + { + results.Add(entry); + if (entry.Delta == 0) + { + break; + } + } + + // Should be 1 entry, which is the deleted OP + Assert.Single(results); + Assert.Equal(NatsKVOperation.Del, results[0].Operation); + + // Watch and ignore deletes.... Really OnNoData should execute as we're excluding deletes and there should be no entries coming back, but this just times out. + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + var cancellationToken = cts.Token; + + results.Clear(); + + try + { + await foreach (var entry in store.WatchAsync( + ">", + opts: new NatsKVWatchOpts + { + IgnoreDeletes = true, + OnNoData = (_) => ValueTask.FromResult(true), + }, + cancellationToken: cancellationToken)) + { + results.Add(entry); + if (entry.Delta == 0) + { + break; + } + } + } + catch (TaskCanceledException exCancelled) + { + Assert.Fail("Task was cancelled waiting for OnNoData"); + } + + // Should be no results here. + Assert.False(results.Any()); + } }