diff --git a/src/NATS.Client.KeyValueStore/NatsKVContext.cs b/src/NATS.Client.KeyValueStore/NatsKVContext.cs index 646f03b01..3d94cb48d 100644 --- a/src/NATS.Client.KeyValueStore/NatsKVContext.cs +++ b/src/NATS.Client.KeyValueStore/NatsKVContext.cs @@ -19,15 +19,30 @@ public enum NatsKVStorageType /// public class NatsKVContext : INatsKVContext { - private const string KvStreamNamePrefix = "KV_"; + internal const string KvStreamNamePrefix = "KV_"; private static readonly int KvStreamNamePrefixLen = KvStreamNamePrefix.Length; private static readonly Regex ValidBucketRegex = new(pattern: @"\A[a-zA-Z0-9_-]+\z", RegexOptions.Compiled); + private readonly NatsKVOpts _opts; /// /// Create a new Key Value Store context /// /// JetStream context - public NatsKVContext(INatsJSContext context) => JetStreamContext = context; + /// Context options + public NatsKVContext(INatsJSContext context, NatsKVOpts opts) + { + JetStreamContext = context; + _opts = opts; + } + + /// + /// Create a new Key Value Store context + /// + /// JetStream context + public NatsKVContext(INatsJSContext context) + : this(context, NatsKVOpts.Default) + { + } /// public INatsJSContext JetStreamContext { get; } @@ -45,7 +60,7 @@ public async ValueTask CreateStoreAsync(NatsKVConfig config, Cance var stream = await JetStreamContext.CreateStreamAsync(streamConfig, cancellationToken); - return new NatsKVStore(config.Bucket, JetStreamContext, stream); + return new NatsKVStore(config.Bucket, JetStreamContext, stream, _opts); } /// @@ -61,7 +76,7 @@ public async ValueTask GetStoreAsync(string bucket, CancellationTo } // TODO: KV mirror - return new NatsKVStore(bucket, JetStreamContext, stream); + return new NatsKVStore(bucket, JetStreamContext, stream, _opts); } /// @@ -73,7 +88,7 @@ public async ValueTask UpdateStoreAsync(NatsKVConfig config, Cance var stream = await JetStreamContext.UpdateStreamAsync(streamConfig, cancellationToken); - return new NatsKVStore(config.Bucket, JetStreamContext, stream); + return new NatsKVStore(config.Bucket, JetStreamContext, stream, _opts); } /// @@ -85,7 +100,7 @@ public async ValueTask CreateOrUpdateStoreAsync(NatsKVConfig confi var stream = await JetStreamContext.CreateOrUpdateStreamAsync(streamConfig, cancellationToken); - return new NatsKVStore(config.Bucket, JetStreamContext, stream); + return new NatsKVStore(config.Bucket, JetStreamContext, stream, _opts); } /// @@ -251,3 +266,10 @@ private static StreamConfig CreateStreamConfig(NatsKVConfig config) return streamConfig; } } + +public class NatsKVOpts +{ + public static readonly NatsKVOpts Default = new(); + + public bool UseDirectGetApiWithKeysInSubject { get; init; } +} diff --git a/src/NATS.Client.KeyValueStore/NatsKVStore.cs b/src/NATS.Client.KeyValueStore/NatsKVStore.cs index 16e7869da..8f49dbb6a 100644 --- a/src/NATS.Client.KeyValueStore/NatsKVStore.cs +++ b/src/NATS.Client.KeyValueStore/NatsKVStore.cs @@ -56,14 +56,18 @@ public class NatsKVStore : INatsKVStore private static readonly NatsKVException KeyCannotStartOrEndWithPeriodException = new("Key cannot start or end with a period"); private static readonly NatsKVException KeyContainsInvalidCharactersException = new("Key contains invalid characters"); private readonly INatsJSStream _stream; + private readonly NatsKVOpts _opts; private readonly string _kvBucket; + private readonly string _streamName; - internal NatsKVStore(string bucket, INatsJSContext context, INatsJSStream stream) + internal NatsKVStore(string bucket, INatsJSContext context, INatsJSStream stream, NatsKVOpts opts) { Bucket = bucket; JetStreamContext = context; _stream = stream; + _opts = opts; _kvBucket = $"$KV.{Bucket}."; + _streamName = NatsKVContext.KvStreamNamePrefix + Bucket; } /// @@ -333,7 +337,19 @@ public async ValueTask>> TryGetEntryAsync(string ke if (_stream.Info.Config.AllowDirect) { - var direct = await _stream.GetDirectAsync(request, serializer, cancellationToken); + NatsMsg direct; + if (_opts.UseDirectGetApiWithKeysInSubject) + { + direct = await JetStreamContext.Connection.RequestAsync( + subject: $"{JetStreamContext.Opts.Prefix}.DIRECT.GET.{_streamName}.{keySubject}", + data: null, + replySerializer: serializer, + cancellationToken: cancellationToken); + } + else + { + direct = await _stream.GetDirectAsync(request, serializer, cancellationToken); + } if (direct is { Headers: { } headers } msg) { diff --git a/tests/NATS.Client.KeyValueStore.Tests/DirectGetTest.cs b/tests/NATS.Client.KeyValueStore.Tests/DirectGetTest.cs new file mode 100644 index 000000000..7e03139c9 --- /dev/null +++ b/tests/NATS.Client.KeyValueStore.Tests/DirectGetTest.cs @@ -0,0 +1,63 @@ +using NATS.Client.Core.Tests; + +namespace NATS.Client.KeyValueStore.Tests; + +public class DirectGetTest(ITestOutputHelper output) +{ + [Fact] + public async Task API_subject_test() + { + await using var server = await NatsServer.StartJSAsync(); + var (nats1, proxy) = server.CreateProxiedClientConnection(); + await using var nats = nats1; + + var js = new NatsJSContext(nats); + + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var cancellationToken = cts.Token; + + // default + { + var kv = new NatsKVContext(js); + + var store = await kv.CreateStoreAsync(new NatsKVConfig("b1"), cancellationToken); + await store.PutAsync("x", 1, cancellationToken: cancellationToken); + await store.PutAsync("x", 2, cancellationToken: cancellationToken); + + await proxy.FlushFramesAsync(nats); + + var entry = await store.GetEntryAsync("x", cancellationToken: cancellationToken); + Assert.Equal(2, entry.Value); + + var proto = proxy.ClientFrames[0].Message; + Assert.StartsWith("PUB $JS.API.DIRECT.GET.KV_b1 _INBOX.", proto); + Assert.EndsWith("""␍␊{"last_by_subj":"$KV.b1.x"}""", proto); + foreach (var proxyFrame in proxy.ClientFrames) + { + output.WriteLine(proxyFrame.Message); + } + } + + // key in api subject + { + var kv = new NatsKVContext(js, new NatsKVOpts { UseDirectGetApiWithKeysInSubject = true }); + + var store = await kv.CreateStoreAsync(new NatsKVConfig("b1"), cancellationToken); + await store.PutAsync("x", 1, cancellationToken: cancellationToken); + await store.PutAsync("x", 2, cancellationToken: cancellationToken); + + await proxy.FlushFramesAsync(nats); + + var entry = await store.GetEntryAsync("x", cancellationToken: cancellationToken); + Assert.Equal(2, entry.Value); + + var proto = proxy.ClientFrames[0].Message; + Assert.StartsWith("PUB $JS.API.DIRECT.GET.KV_b1.$KV.b1.x _INBOX.", proto); + Assert.EndsWith(""" 0␍␊""", proto); + foreach (var proxyFrame in proxy.ClientFrames) + { + output.WriteLine(proxyFrame.Message); + } + } + } +}