diff --git a/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs b/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs index 2880a76d1..94726c950 100644 --- a/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs +++ b/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs @@ -178,6 +178,17 @@ public async ValueTask DisposeAsync() _msgChannel.Writer.TryComplete(); + // Dispose the current subscription. Both background loops have completed + // by now, so no new subscription can be created. Without this the sub's + // ConnectionOpened handler stays attached to the connection, rooting the + // sub for the connection's lifetime (memory leak). + var sub = _sub; + if (sub != null) + { + await sub.UnsubscribeAsync(); + await sub.DisposeAsync(); + } + using var cts = CancellationTokenSource.CreateLinkedTokenSource(_cancellationToken); cts.CancelAfter(_opts.CleanupTimeout); try diff --git a/tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs b/tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs index 023618421..2fd5c31cf 100644 --- a/tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs +++ b/tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs @@ -1,4 +1,5 @@ using System.Diagnostics; +using System.Reflection; using System.Security.Cryptography; using System.Text; using NATS.Client.Core.Tests; @@ -663,4 +664,40 @@ public async Task Metadata_field_types_match_spec() #pragma warning restore IDE0007 Assert.Equal(1U, chunks); } + + [Fact] + public async Task Get_does_not_leak_connection_opened_handlers() + { + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + var cancellationToken = cts.Token; + + await using var server = await NatsServerProcess.StartAsync(); + await using var nats = new NatsConnection(new NatsOpts { Url = server.Url }); + var js = new NatsJSContext(nats); + var ob = new NatsObjContext(js); + + var store = await ob.CreateObjectStoreAsync(new NatsObjConfig("b1"), cancellationToken); + await store.PutAsync("k1", new byte[] { 1, 2, 3 }, cancellationToken: cancellationToken); + + // The ordered push consumer created per GetAsync registers a ConnectionOpened + // handler that must be removed when the consumer is disposed. If it is not, the + // handler (and the subscription it roots) leaks for the connection's lifetime. + var baseline = ConnectionOpenedHandlerCount(nats); + + for (var i = 0; i < 10; i++) + { + var bytes = await store.GetBytesAsync("k1", cancellationToken); + Assert.Equal(new byte[] { 1, 2, 3 }, bytes); + } + + Assert.Equal(baseline, ConnectionOpenedHandlerCount(nats)); + } + + private static int ConnectionOpenedHandlerCount(NatsConnection nats) + { + var field = typeof(NatsConnection).GetField("ConnectionOpened", BindingFlags.Instance | BindingFlags.NonPublic); + Assert.NotNull(field); + var handler = (Delegate?)field.GetValue(nats); + return handler?.GetInvocationList().Length ?? 0; + } }