From c850c8d14157eb469e70348bfc3ef4d25a31b320 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Tue, 4 Feb 2025 06:11:00 +0000 Subject: [PATCH] Add service use queue group option --- src/NATS.Client.Services/NatsSvcConfig.cs | 8 +- src/NATS.Client.Services/NatsSvcServer.cs | 4 +- .../ServicesTests.cs | 96 +++++++++++++++++++ 3 files changed, 105 insertions(+), 3 deletions(-) diff --git a/src/NATS.Client.Services/NatsSvcConfig.cs b/src/NATS.Client.Services/NatsSvcConfig.cs index 20cc88e2a..fd9181c25 100644 --- a/src/NATS.Client.Services/NatsSvcConfig.cs +++ b/src/NATS.Client.Services/NatsSvcConfig.cs @@ -54,10 +54,16 @@ public NatsSvcConfig(string name, string version) public Dictionary? Metadata { get; init; } /// - /// Queue group name. (default: "q") + /// Queue group name. (default: "q" unless is set to false) /// public string QueueGroup { get; init; } = "q"; + /// + /// Use a queue group when creating service subscriptions. (default: true) + /// Queue group name is defined using property. + /// + public bool UseQueueGroup { get; init; } = true; + /// /// Stats handler. JSON object returned by this handler will be included in /// the service stats data property. diff --git a/src/NATS.Client.Services/NatsSvcServer.cs b/src/NATS.Client.Services/NatsSvcServer.cs index 1b30c21c8..e796967f5 100644 --- a/src/NATS.Client.Services/NatsSvcServer.cs +++ b/src/NATS.Client.Services/NatsSvcServer.cs @@ -97,7 +97,7 @@ public async ValueTask StopAsync(CancellationToken cancellationToken = default) /// public ValueTask AddEndpointAsync(Func, ValueTask> handler, string? name = default, string? subject = default, string? queueGroup = default, IDictionary? metadata = default, INatsDeserialize? serializer = default, CancellationToken cancellationToken = default) { - queueGroup ??= _config.QueueGroup; + queueGroup ??= _config.UseQueueGroup ? _config.QueueGroup : null; return AddEndpointInternalAsync(handler, name, subject, queueGroup, metadata, serializer, cancellationToken); } @@ -331,7 +331,7 @@ public ValueTask AddEndpointAsync(Func, ValueTask> handler, str { subject ??= name; var epSubject = subject != null ? $"{GroupName}{_dot}{subject}" : null; - queueGroup ??= QueueGroup ?? _server._config.QueueGroup; + queueGroup ??= QueueGroup ?? (_server._config.UseQueueGroup ? _server._config.QueueGroup : null); serializer ??= _server._nats.Opts.SerializerRegistry.GetDeserializer(); return _server.AddEndpointInternalAsync(handler, name, epSubject, queueGroup, metadata, serializer, cancellationToken); } diff --git a/tests/NATS.Client.Services.Tests/ServicesTests.cs b/tests/NATS.Client.Services.Tests/ServicesTests.cs index 6f39a2809..fa2095dd8 100644 --- a/tests/NATS.Client.Services.Tests/ServicesTests.cs +++ b/tests/NATS.Client.Services.Tests/ServicesTests.cs @@ -383,4 +383,100 @@ public async Task Service_ids_unique() Assert.Equal(ids.Count, uniqueIds.Count); } + + [Fact] + public async Task Service_without_queue_group() + { + await using var server = await NatsServer.StartAsync(); + var (nats1, proxy) = server.CreateProxiedClientConnection(); + await using var nats = nats1; + var svc = new NatsSvcContext(nats); + + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + var cancellationToken = cts.Token; + + // Default with a queue group + { + await using var s1 = await svc.AddServiceAsync( + new NatsSvcConfig("s1", "1.0.0"), // default is { UseQueueGroup = true }, + cancellationToken: cancellationToken); + + await using var s2 = await svc.AddServiceAsync( + new NatsSvcConfig("s2", "1.0.0"), // default is { UseQueueGroup = true }, + cancellationToken: cancellationToken); + + await s1.AddEndpointAsync( + name: "ep1", + handler: async m => + { + await m.ReplyAsync("x", cancellationToken: cancellationToken); + }, + cancellationToken: cancellationToken); + + await s2.AddEndpointAsync( + name: "ep1", + handler: async m => + { + await m.ReplyAsync("x", cancellationToken: cancellationToken); + }, + cancellationToken: cancellationToken); + + await Retry.Until("subscribed", () => proxy.Frames.Count(f => f.Message.StartsWith("SUB ep1")) == 2); + + var sub = proxy.Frames.First(f => f.Message.StartsWith("SUB ep1")).Message; + Assert.Matches(@"SUB ep1 q \d+", sub); + + var count = 0; + await foreach (var msg in nats.RequestManyAsync("ep1", "x", cancellationToken: cancellationToken)) + { + count++; + } + + // Only one reply because of the queue group + Assert.Equal(1, count); + } + + await proxy.FlushFramesAsync(nats); + + // Without any queue group + { + await using var s1 = await svc.AddServiceAsync( + new NatsSvcConfig("s1", "1.0.0") { UseQueueGroup = false }, + cancellationToken: cancellationToken); + + await using var s2 = await svc.AddServiceAsync( + new NatsSvcConfig("s2", "1.0.0") { UseQueueGroup = false }, + cancellationToken: cancellationToken); + + await s1.AddEndpointAsync( + name: "ep1", + handler: async m => + { + await m.ReplyAsync("x", cancellationToken: cancellationToken); + }, + cancellationToken: cancellationToken); + + await s2.AddEndpointAsync( + name: "ep1", + handler: async m => + { + await m.ReplyAsync("x", cancellationToken: cancellationToken); + }, + cancellationToken: cancellationToken); + + await Retry.Until("subscribed", () => proxy.Frames.Count(f => f.Message.StartsWith("SUB ep1")) == 2); + + var sub = proxy.Frames.First(f => f.Message.StartsWith("SUB ep1")).Message; + Assert.Matches(@"SUB ep1 \d+", sub); + + var count = 0; + await foreach (var msg in nats.RequestManyAsync("ep1", "x", cancellationToken: cancellationToken)) + { + count++; + } + + // Two replies because of the absence of any queue groups + Assert.Equal(2, count); + } + } }