diff --git a/src/NATS.Client.KeyValueStore/INatsKVStore.cs b/src/NATS.Client.KeyValueStore/INatsKVStore.cs index 959bc6cf7..58b8a48ee 100644 --- a/src/NATS.Client.KeyValueStore/INatsKVStore.cs +++ b/src/NATS.Client.KeyValueStore/INatsKVStore.cs @@ -91,6 +91,7 @@ public interface INatsKVStore /// A used to cancel the API call. /// Serialized value type /// An asynchronous enumerable which can be used in await foreach loops + /// There was a conflict in options, e.g. IncludeHistory and UpdatesOnly are only valid when ResumeAtRevision is not set. IAsyncEnumerable> WatchAsync(IEnumerable keys, INatsDeserialize? serializer = default, NatsKVWatchOpts? opts = default, CancellationToken cancellationToken = default); /// @@ -112,6 +113,7 @@ public interface INatsKVStore /// A used to cancel the API call. /// Serialized value type /// An async enumerable of entries to be used in an await foreach + /// There was a conflict in options, e.g. IncludeHistory and UpdatesOnly are only valid when ResumeAtRevision is not set. IAsyncEnumerable> HistoryAsync(string key, INatsDeserialize? serializer = default, NatsKVWatchOpts? opts = default, CancellationToken cancellationToken = default); /// @@ -134,5 +136,6 @@ public interface INatsKVStore /// Watch options /// A used to cancel the API call. /// An async enumerable of keys to be used in an await foreach + /// There was a conflict in options, e.g. IncludeHistory and UpdatesOnly are only valid when ResumeAtRevision is not set. IAsyncEnumerable GetKeysAsync(NatsKVWatchOpts? opts = default, CancellationToken cancellationToken = default); } diff --git a/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs b/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs index cf95b136f..281780f38 100644 --- a/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs +++ b/src/NATS.Client.KeyValueStore/Internal/NatsKVWatcher.cs @@ -404,11 +404,17 @@ private async ValueTask CreatePushConsumer(string origin) config.HeadersOnly = true; } + // Resume from a specific revision ? if (sequence > 0) { config.DeliverPolicy = ConsumerConfigDeliverPolicy.ByStartSequence; config.OptStartSeq = sequence + 1; } + else if (_opts.ResumeAtRevision > 0) + { + config.DeliverPolicy = ConsumerConfigDeliverPolicy.ByStartSequence; + config.OptStartSeq = _opts.ResumeAtRevision; + } var consumer = await _context.CreateOrUpdateConsumerAsync( _stream, diff --git a/src/NATS.Client.KeyValueStore/NatsKVOpts.cs b/src/NATS.Client.KeyValueStore/NatsKVOpts.cs index 88483e229..6a915cfc5 100644 --- a/src/NATS.Client.KeyValueStore/NatsKVOpts.cs +++ b/src/NATS.Client.KeyValueStore/NatsKVOpts.cs @@ -25,11 +25,19 @@ public record NatsKVWatchOpts /// /// Include history of the entries /// + /// + /// Setting this will cause the watcher to throw + /// if the values for and/or are set. + /// public bool IncludeHistory { get; init; } = false; /// /// Only retrieve updates, not current values /// + /// + /// Setting this will cause the watcher to throw + /// if the values for and/or are set. + /// public bool UpdatesOnly { get; init; } = false; /// @@ -41,6 +49,33 @@ public record NatsKVWatchOpts /// Async function called when the enumerator reaches the end of data. Return True to break the async enumeration, False to allow the enumeration to continue. /// public Func>? OnNoData { get; init; } + + /// + /// The revision to start from, if set to 0 (default) this will be ignored. + /// + /// Setting this to a non-zero value will cause the watcher to throw + /// if the values for and/or are set. + /// + /// + public ulong ResumeAtRevision { get; init; } + + internal void ThrowIfInvalid() + { + if (ResumeAtRevision > 0) + { + if (IncludeHistory || UpdatesOnly) + { + throw new InvalidOperationException("IncludeHistory and UpdatesOnly are only valid when ResumeAtRevision is not set."); + } + } + else + { + if (IncludeHistory && UpdatesOnly) + { + throw new InvalidOperationException("IncludeHistory and UpdatesOnly are mutually exclusive."); + } + } + } } public record NatsKVDeleteOpts diff --git a/src/NATS.Client.KeyValueStore/NatsKVStore.cs b/src/NATS.Client.KeyValueStore/NatsKVStore.cs index ba8e26185..2c80ddc62 100644 --- a/src/NATS.Client.KeyValueStore/NatsKVStore.cs +++ b/src/NATS.Client.KeyValueStore/NatsKVStore.cs @@ -340,6 +340,7 @@ public async IAsyncEnumerable> WatchAsync(IEnumerable } } + /// public async IAsyncEnumerable> HistoryAsync(string key, INatsDeserialize? serializer = default, NatsKVWatchOpts? opts = default, [EnumeratorCancellation] CancellationToken cancellationToken = default) { try @@ -382,6 +383,7 @@ public async ValueTask GetStatusAsync(CancellationToken cancellati public IAsyncEnumerable> WatchAsync(INatsDeserialize? serializer = default, NatsKVWatchOpts? opts = default, CancellationToken cancellationToken = default) => WatchAsync([">"], serializer, opts, cancellationToken); + /// public async ValueTask PurgeDeletesAsync(NatsKVPurgeOpts? opts = default, CancellationToken cancellationToken = default) { opts ??= NatsKVPurgeOpts.Default; @@ -459,6 +461,8 @@ internal async ValueTask> WatchInternalAsync(IEnumerable(); + opts.ThrowIfInvalid(); + var watcher = new NatsKVWatcher( context: _context, bucket: Bucket, diff --git a/tests/NATS.Client.KeyValueStore.Tests/NatsKVWatcherTest.cs b/tests/NATS.Client.KeyValueStore.Tests/NatsKVWatcherTest.cs index 3ade0edc1..a66ca1f5f 100644 --- a/tests/NATS.Client.KeyValueStore.Tests/NatsKVWatcherTest.cs +++ b/tests/NATS.Client.KeyValueStore.Tests/NatsKVWatcherTest.cs @@ -460,4 +460,163 @@ public async Task Watch_with_multiple_filter_on_old_server() Assert.Equal(10094, exception.Error.ErrCode); Assert.Equal("consumer delivery policy is deliver last per subject, but optional filter subject is not set", exception.Error.Description); } + + // Test that watch can resume from a specific revision + [Fact] + public async Task Watch_resume_at_revision() + { + await using var server = NatsServer.StartJS(); + await using var nats = server.CreateClientConnection(); + + const string bucket = "Watch_resume_at_revision"; + var config = new NatsKVConfig(bucket) { History = 10 }; + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var cancellationToken = cts.Token; + + var js = new NatsJSContext(nats); + var kv = new NatsKVContext(js); + var store = await kv.CreateStoreAsync(config, cancellationToken: cancellationToken); + + await store.PutAsync("k1", 1, cancellationToken: cancellationToken); + await store.PutAsync("k2", 2, cancellationToken: cancellationToken); + var revK3 = await store.PutAsync("k3", 3, cancellationToken: cancellationToken); + await store.PutAsync("k4", 3, cancellationToken: cancellationToken); + + // Watch all + var watchOps = new NatsKVWatchOpts() { MetaOnly = true, }; + var watchAll = store.WatchAsync(opts: watchOps, cancellationToken: cancellationToken); + + // Expect to see k1, k2, k3 and k4 + var allEntries = new List<(ulong Revision, string key)>(); + await foreach (var key in watchAll) + { + allEntries.Add((key.Revision, key.Key)); + if (key.Delta == 0) + { + break; + } + } + + // Expects k1, k2, k3 and k4 + allEntries.Should().HaveCount(4); + + // Watch from the revision of k3 + var watchOpsFromRevK3 = watchOps with { ResumeAtRevision = revK3, }; + + var watchFromRevision = store.WatchAsync(opts: watchOpsFromRevK3, cancellationToken: cancellationToken); + + // Expect to see k2 and k3, and k4 + var fromRevisionEntries = new List<(ulong Revision, string key)>(); + await foreach (var key in watchFromRevision) + { + fromRevisionEntries.Add((key.Revision, key.Key)); + if (key.Delta == 0) + { + break; + } + } + + // Expects k2, k3 and k4 + fromRevisionEntries.Should().HaveCount(2); + + // Watch from none existing revision + var noData = false; + var watchOpsNoneExisting = watchOps with + { + ResumeAtRevision = 9999, + OnNoData = (_) => + { + noData = true; + return ValueTask.FromResult(true); + }, + }; + + var watchFromNoneExistingRevision = + store.WatchAsync(opts: watchOpsNoneExisting, cancellationToken: cancellationToken); + + // Expect to see no data + await foreach (var key in watchFromNoneExistingRevision) + { + // We should not see any entries, if we get here something is wrong + Assert.Fail("Should not return any entries, and OnNoData should have been called to bail out"); + } + + noData.Should().BeTrue(); + } + + [Fact] + public async Task Validate_watch_options() + { + await using var server = NatsServer.StartJS(); + await using var nats = server.CreateClientConnection(); + + const string bucket = nameof(Validate_watch_options); + var config = new NatsKVConfig(bucket) { History = 10 }; + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var cancellationToken = cts.Token; + var js = new NatsJSContext(nats); + var kv = new NatsKVContext(js); + var store = await kv.CreateStoreAsync(config, cancellationToken: cancellationToken); + + for (var i = 0; i < 10; i++) + { + await store.PutAsync("x", i, cancellationToken: cancellationToken); + } + + // Valid options + foreach (var opts in new[] + { + new NatsKVWatchOpts { IncludeHistory = false, UpdatesOnly = false, ResumeAtRevision = 5 }, + new NatsKVWatchOpts { IncludeHistory = true, UpdatesOnly = false, ResumeAtRevision = 0 }, + new NatsKVWatchOpts { IncludeHistory = false, UpdatesOnly = true, ResumeAtRevision = 0 }, + }) + { + var count = 0; + var cts2 = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + + if (opts.UpdatesOnly) + { + cts2.Cancel(); + count++; + } + + try + { + await foreach (var entry in store.WatchAsync([">"], opts: opts, cancellationToken: cts2.Token)) + { + count++; + _output.WriteLine($"entry: {entry.Key} ({entry.Revision}): {entry.Value}"); + if (entry.Value == 9) + break; + } + } + catch (TaskCanceledException) + { + } + catch (OperationCanceledException) + { + } + + count.Should().BeGreaterThan(0); + } + + // Invalid options + foreach (var opts in new[] + { + new NatsKVWatchOpts { IncludeHistory = true, UpdatesOnly = false, ResumeAtRevision = 5 }, + new NatsKVWatchOpts { IncludeHistory = true, UpdatesOnly = true, ResumeAtRevision = 5 }, + new NatsKVWatchOpts { IncludeHistory = false, UpdatesOnly = true, ResumeAtRevision = 5 }, + new NatsKVWatchOpts { IncludeHistory = true, UpdatesOnly = true, ResumeAtRevision = 0 }, + }) + { + await Assert.ThrowsAsync(async () => + { + await foreach (var entry in store.WatchAsync([">"], opts: opts, cancellationToken: cancellationToken)) + { + } + }); + } + } }