Skip to content

Commit

Permalink
Make sure queue group is written to wire (#414)
Browse files Browse the repository at this point in the history
* Make sure queue group is written to wire

* Test fix
  • Loading branch information
mtmk authored Feb 27, 2024
1 parent c56cbdd commit da0ebcb
Show file tree
Hide file tree
Showing 2 changed files with 27 additions and 1 deletion.
2 changes: 1 addition & 1 deletion src/NATS.Client.Core/Commands/ProtocolWriter.cs
Original file line number Diff line number Diff line change
Expand Up @@ -114,7 +114,7 @@ public void WriteSubscribe(IBufferWriter<byte> writer, int sid, string subject,

if (queueGroup != null)
{
written = _subjectEncoding.GetBytes(subject, span);
written = _subjectEncoding.GetBytes(queueGroup, span);
span[written] = (byte)' ';
size += written + 1;
span = span[(written + 1)..];
Expand Down
26 changes: 26 additions & 0 deletions tests/NATS.Client.Core.Tests/ProtocolTest.cs
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,32 @@ await Retry.Until(
proxy.Dispose();
}

[Fact]
public async Task Subscription_queue_group()
{
await using var server = NatsServer.Start();
var (nats, proxy) = server.CreateProxiedClientConnection();

await using var sub1 = await nats.SubscribeCoreAsync<int>("foo", queueGroup: "group1");
await using var sub2 = await nats.SubscribeCoreAsync<int>("foo", queueGroup: "group2");

await Retry.Until(
"frames collected",
() => proxy.ClientFrames.Count(f => f.Message.StartsWith("SUB foo")) == 2);

var frames = proxy.ClientFrames.Select(f => f.Message).ToList();

foreach (var frame in frames)
{
_output.WriteLine($"frame: {frame}");
}

Assert.StartsWith("SUB foo group1 ", frames[0]);
Assert.StartsWith("SUB foo group2 ", frames[1]);

await nats.DisposeAsync();
}

[Fact]
public async Task Publish_empty_message_for_notifications()
{
Expand Down

0 comments on commit da0ebcb

Please sign in to comment.