fix: NatsJSConsumer never disposing receive activities

src/NATS.Client.Core/INatsMsg.cs

@@ -0,0 +1,12 @@
+namespace NATS.Client.Core;
+
+/// <summary>
+/// Shared contract for all NATS message types.
+/// </summary>
+public interface INatsMsg
+{
+    /// <summary>
+    /// Headers of the user message if set.
+    /// </summary>
+    NatsHeaders? Headers { get; }
+}

src/NATS.Client.Core/Internal/ActivityEndingMsgReader.cs

@@ -6,21 +6,22 @@ namespace NATS.Client.Core.Internal;
 
 // ActivityEndingMsgReader servers 2 purposes
 // 1. End activity for OpenTelemetry
-// 2. Keep the INatsSub<T> from being garbage collected as long as calls interacting
+// 2. Keep the NatsSubBase from being garbage collected as long as calls interacting
 //    with the _inner channel are being made
 // To achieve (1):
 // Calls that result in a read from the _inner channel should msg.Headers?.Activity?.Dispose()
 // To achieve (2):
 // Synchronous calls should call GC.KeepAlive(_sub); immediately before returning
 // Asynchronous calls should allocate a GCHandle.Alloc(_sub) at the start of the method,
 // and then free it in a try/finally block
-internal sealed class ActivityEndingMsgReader<T> : ChannelReader<NatsMsg<T>>
+internal sealed class ActivityEndingMsgReader<T> : ChannelReader<T>
+    where T : struct, INatsMsg
 {
-    private readonly ChannelReader<NatsMsg<T>> _inner;
+    private readonly ChannelReader<T> _inner;
 
-    private readonly INatsSub<T> _sub;
+    private readonly NatsSubBase _sub;
 
-    public ActivityEndingMsgReader(ChannelReader<NatsMsg<T>> inner, INatsSub<T> sub)
+    public ActivityEndingMsgReader(ChannelReader<T> inner, NatsSubBase sub)
     {
         _inner = inner;
         _sub = sub;
@@ -64,7 +65,7 @@ public override Task Completion
 
     /// <inheritdoc/>
     [MethodImpl(MethodImplOptions.AggressiveInlining)]
-    public override bool TryRead(out NatsMsg<T> item)
+    public override bool TryRead(out T item)
     {
         if (!_inner.TryRead(out item))
             return false;
@@ -88,7 +89,7 @@ public override async ValueTask<bool> WaitToReadAsync(CancellationToken cancella
         }
     }
 
-    public override async ValueTask<NatsMsg<T>> ReadAsync(CancellationToken cancellationToken = default)
+    public override async ValueTask<T> ReadAsync(CancellationToken cancellationToken = default)
     {
         var handle = GCHandle.Alloc(_sub);
         try
@@ -103,16 +104,16 @@ public override async ValueTask<NatsMsg<T>> ReadAsync(CancellationToken cancella
         }
     }
 
-    public override bool TryPeek(out NatsMsg<T> item)
+    public override bool TryPeek(out T item)
     {
         GC.KeepAlive(_sub);
         return _inner.TryPeek(out item);
     }
 
 #if NETSTANDARD2_0
-    public async IAsyncEnumerable<NatsMsg<T>> ReadAllAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
+    public async IAsyncEnumerable<T> ReadAllAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
 #else
-    public override async IAsyncEnumerable<NatsMsg<T>> ReadAllAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
+    public override async IAsyncEnumerable<T> ReadAllAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
 #endif
     {
         var handle = GCHandle.Alloc(_sub);

src/NATS.Client.Core/NatsMsg.cs

@@ -35,7 +35,7 @@ public enum NatsMsgFlags : byte
 /// </para>
 /// </remarks>
 /// <typeparam name="T">Data type of the payload</typeparam>
-public interface INatsMsg<T>
+public interface INatsMsg<T> : INatsMsg
 {
     /// <summary>The destination subject to publish to.</summary>
     string Subject { get; init; }
@@ -46,9 +46,6 @@ public interface INatsMsg<T>
     /// <summary>Message size in bytes.</summary>
     int Size { get; init; }
 
-    /// <summary>Pass additional information using name-value pairs.</summary>
-    NatsHeaders? Headers { get; init; }
-
     /// <summary>Serializable data object.</summary>
     T? Data { get; init; }
 

src/NATS.Client.Core/NatsSub.cs

@@ -23,7 +23,7 @@ public NatsSub(
             connection.GetBoundedChannelOpts(opts?.ChannelOpts),
             msg => Connection.OnMessageDropped(this, _msgs?.Reader.Count ?? 0, msg));
 
-        Msgs = new ActivityEndingMsgReader<T>(_msgs.Reader, this);
+        Msgs = new ActivityEndingMsgReader<NatsMsg<T>>(_msgs.Reader, this);
 
         Serializer = serializer;
     }

src/NATS.Client.JetStream/Internal/NatsJSConsume.cs

@@ -136,7 +136,7 @@ public NatsJSConsume(
         // sufficiently large value to avoid blocking socket reads in the
         // NATS connection).
         _userMsgs = Channel.CreateBounded<NatsJSMsg<TMsg>>(1000);
-        Msgs = _userMsgs.Reader;
+        Msgs = new ActivityEndingMsgReader<NatsJSMsg<TMsg>>(_userMsgs.Reader, this);
 
         // Capacity as 1 is enough here since it's used for signaling only.
         _pullRequests = Channel.CreateBounded<PullRequest>(1);

src/NATS.Client.JetStream/Internal/NatsJSFetch.cs

@@ -76,7 +76,7 @@ public NatsJSFetch(
         // sufficiently large value to avoid blocking socket reads in the
         // NATS connection).
         _userMsgs = Channel.CreateBounded<NatsJSMsg<TMsg>>(1000);
-        Msgs = _userMsgs.Reader;
+        Msgs = new ActivityEndingMsgReader<NatsJSMsg<TMsg>>(_userMsgs.Reader, this);
 
         if (_debug)
         {

src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs

@@ -4,6 +4,7 @@
 using Microsoft.Extensions.Logging;
 using NATS.Client.Core;
 using NATS.Client.Core.Commands;
+using NATS.Client.Core.Internal;
 using NATS.Client.JetStream.Models;
 
 namespace NATS.Client.JetStream.Internal;
@@ -97,7 +98,7 @@ public NatsJSOrderedConsume(
         _userMsgs = Channel.CreateBounded<NatsJSMsg<TMsg>>(
             Connection.GetBoundedChannelOpts(opts?.ChannelOpts),
             msg => Connection.OnMessageDropped(this, _userMsgs?.Reader.Count ?? 0, msg.Msg));
-        Msgs = _userMsgs.Reader;
+        Msgs = new ActivityEndingMsgReader<NatsJSMsg<TMsg>>(_userMsgs.Reader, this);
 
         // Pull request channel is set as unbounded because we don't want to drop
         // them and minimize potential lock contention.

src/NATS.Client.JetStream/NatsJSMsg.cs

@@ -27,7 +27,7 @@ namespace NATS.Client.JetStream;
 /// </para>
 /// </remarks>
 /// <typeparam name="T">Data type of the payload</typeparam>
-public interface INatsJSMsg<out T>
+public interface INatsJSMsg<out T> : INatsMsg
 {
     /// <summary>
     /// Subject of the user message.
@@ -45,11 +45,6 @@ public interface INatsJSMsg<out T>
     /// </remarks>
     int Size { get; }
 
-    /// <summary>
-    /// Headers of the user message if set.
-    /// </summary>
-    NatsHeaders? Headers { get; }
-
     /// <summary>
     /// Deserialized user data.
     /// </summary>