feat: add an topic event enumerable (#575)

Introduces an interface ITopicEvent which covers both normal
messages (TopicMessage) and system events (TopicSystemEvent). We
refactor the internals of the stream to be over ITopicEvents, so
that we can filter to just TopicMessage (as is implemented
currently) or pass through all events in an event enumerable.

To enumerate over all events in the stream, call GetAllEventsAsyncEnumerable,
or WithCancellationForAllEvents.

This largely parallels the existing enumerator over messages only. See the
integration tests for an example usage.

src/Momento.Sdk/Internal/ScsTopicClient.cs

@@ -142,7 +142,7 @@ private async Task<TopicSubscribeResponse> SendSubscribe(string cacheName, strin
         }
 
         var response = new TopicSubscribeResponse.Subscription(
-            cancellationToken => subscriptionWrapper.GetNextRelevantMessageFromGrpcStreamAsync(cancellationToken),
+            cancellationToken => subscriptionWrapper.GetNextEventFromGrpcStreamAsync(cancellationToken),
             subscriptionWrapper.Dispose);
         return _logger.LogTraceTopicRequestSuccess(RequestTypeTopicSubscribe, cacheName, topicName,
             response);
@@ -198,7 +198,7 @@ public async Task Subscribe()
             _subscribed = true;
         }
 
-        public async ValueTask<TopicMessage?> GetNextRelevantMessageFromGrpcStreamAsync(
+        public async ValueTask<ITopicEvent?> GetNextEventFromGrpcStreamAsync(
             CancellationToken cancellationToken)
         {
             while (!cancellationToken.IsCancellationRequested)
@@ -251,10 +251,10 @@ public async Task Subscribe()
                         {
                             case _TopicValue.KindOneofCase.Text:
                                 _logger.LogTraceTopicMessageReceived("text", _cacheName, _topicName);
-                                return new TopicMessage.Text(message.Item.Value, checked((long)message.Item.TopicSequenceNumber), message.Item.PublisherId == "" ? null : message.Item.PublisherId);
+                                return new TopicMessage.Text(message.Item.Value, checked((long)_lastSequenceNumber), message.Item.PublisherId == "" ? null : message.Item.PublisherId);
                             case _TopicValue.KindOneofCase.Binary:
                                 _logger.LogTraceTopicMessageReceived("binary", _cacheName, _topicName);
-                                return new TopicMessage.Binary(message.Item.Value, checked((long)message.Item.TopicSequenceNumber), message.Item.PublisherId == "" ? null : message.Item.PublisherId);
+                                return new TopicMessage.Binary(message.Item.Value, checked((long)_lastSequenceNumber), message.Item.PublisherId == "" ? null : message.Item.PublisherId);
                             case _TopicValue.KindOneofCase.None:
                             default:
                                 _logger.LogTraceTopicMessageReceived("unknown", _cacheName, _topicName);
@@ -266,10 +266,11 @@ public async Task Subscribe()
                         _logger.LogTraceTopicDiscontinuityReceived(_cacheName, _topicName,
                             message.Discontinuity.LastTopicSequence, message.Discontinuity.NewTopicSequence);
                         _lastSequenceNumber = message.Discontinuity.NewTopicSequence;
-                        break;
+                        return new TopicSystemEvent.Discontinuity(checked((long)message.Discontinuity.LastTopicSequence),
+                            checked((long)message.Discontinuity.NewTopicSequence));
                     case _SubscriptionItem.KindOneofCase.Heartbeat:
                         _logger.LogTraceTopicMessageReceived("heartbeat", _cacheName, _topicName);
-                        break;
+                        return new TopicSystemEvent.Heartbeat();
                     case _SubscriptionItem.KindOneofCase.None:
                         _logger.LogTraceTopicMessageReceived("none", _cacheName, _topicName);
                         break;

src/Momento.Sdk/Responses/Topic/ITopicEvent.cs

@@ -0,0 +1,12 @@
+namespace Momento.Sdk.Responses;
+
+/// <summary>
+/// Represents an event that can be published to a topic.
+/// 
+/// This includes not just topic mesasges, but also system events
+/// such as discontinuities and heartbeats.
+/// </summary>
+public interface ITopicEvent
+{
+
+}

src/Momento.Sdk/Responses/Topic/TopicMessage.cs

@@ -1,5 +1,6 @@
 using Momento.Protos.CacheClient.Pubsub;
 using Momento.Sdk.Exceptions;
+using Momento.Sdk.Internal.ExtensionMethods;
 
 namespace Momento.Sdk.Responses;
 
@@ -31,7 +32,7 @@ namespace Momento.Sdk.Responses;
 /// }
 /// </code>
 /// </summary>
-public abstract class TopicMessage
+public abstract class TopicMessage : ITopicEvent
 {
     /// <summary>
     /// A topic message containing a text value.
@@ -63,6 +64,12 @@ public Text(_TopicValue topicValue, long topicSequenceNumber, string? tokenId =
         /// This can be used to securely identify the sender of a message.
         /// </summary>
         public string? TokenId { get; }
+
+        /// <inheritdoc />
+        public override string ToString()
+        {
+            return $"{base.ToString()}: Value: \"{this.Value.Truncate()}\" SequenceNumber: {this.TopicSequenceNumber} TokenId: \"{this.TokenId}\"";
+        }
     }
 
     /// <summary>
@@ -95,6 +102,12 @@ public Binary(_TopicValue topicValue, long topicSequenceNumber, string? tokenId
         /// This can be used to securely identify the sender of a message.
         /// </summary>
         public string? TokenId { get; }
+
+        /// <inheritdoc />
+        public override string ToString()
+        {
+            return $"{base.ToString()}: Value: \"{Value.ToPrettyHexString().Truncate()}\" SequenceNumber: {this.TopicSequenceNumber} TokenId: \"{this.TokenId}\"";
+        }
     }
 
     /// <include file="../../docs.xml" path='docs/class[@name="Error"]/description/*' />

src/Momento.Sdk/Responses/Topic/TopicSubscribeResponse.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Net;
 using System.Threading;
 using System.Threading.Tasks;
 using Momento.Sdk.Exceptions;
@@ -43,28 +44,60 @@ public abstract class TopicSubscribeResponse
     /// </summary>
     public class Subscription : TopicSubscribeResponse, IDisposable, IAsyncEnumerable<TopicMessage?>
     {
-        private readonly Func<CancellationToken, ValueTask<TopicMessage?>> _moveNextFunction;
+        private readonly Func<CancellationToken, ValueTask<ITopicEvent?>> _moveNextFunction;
         private CancellationTokenSource _subscriptionCancellationToken = new();
         private readonly Action _disposalAction;
 
         /// <summary>
         /// Constructs a Subscription with a wrapped topic iterator and an action to dispose of it.
         /// </summary>
-        public Subscription(Func<CancellationToken, ValueTask<TopicMessage?>> moveNextFunction, Action disposalAction)
+        public Subscription(Func<CancellationToken, ValueTask<ITopicEvent?>> moveNextFunction, Action disposalAction)
         {
             _moveNextFunction = moveNextFunction;
             _disposalAction = disposalAction;
         }
 
         /// <summary>
-        /// Gets the enumerator for this topic. This subscription represents a single view on a topic, so multiple
+        /// Gets the message enumerator for this topic. Includes text and binary messages, but excludes system events.
+        /// </summary>
+        public IAsyncEnumerable<TopicMessage?> WithCancellation(CancellationToken cancellationToken)
+        {
+            return new AsyncEnumerableWrapper<TopicMessage?>(GetAsyncEnumerator(cancellationToken));
+        }
+
+        /// <summary>
+        /// Gets the event enumerator with cancellation for all topic events, including system events.
+        /// </summary>
+        public IAsyncEnumerable<ITopicEvent?> WithCancellationForAllEvents(CancellationToken cancellationToken)
+        {
+            return new AsyncEnumerableWrapper<ITopicEvent?>(GetAllEventsAsyncEnumerator(cancellationToken));
+        }
+
+        /// <summary>
+        /// Gets the message enumerator for this topic. Includes text and binary messages, but excludes system events.
+        ///
+        /// This subscription represents a single view on a topic, so multiple
         /// enumerators will interfere with each other.
         /// </summary>
+        /// <param name="cancellationToken">A cancellation token to cancel the enumeration.</param>
+        /// <returns>An async enumerator for the topic messages.</returns>
         public IAsyncEnumerator<TopicMessage?> GetAsyncEnumerator(CancellationToken cancellationToken = default)
         {
             return new TopicMessageEnumerator(_moveNextFunction, _subscriptionCancellationToken.Token, cancellationToken);
         }
 
+        /// <summary>
+        /// Gets an enumerator for all events on this topic, including text and binary messages, and system events.
+        ///
+        /// This subscription represents a single view on a topic, so multiple
+        /// enumerators will interfere with each other.
+        /// </summary>
+        /// <param name="cancellationToken">A cancellation token to cancel the enumeration.</param>
+        /// <returns>An async enumerator for all events on the topic.</returns>
+        IAsyncEnumerator<ITopicEvent?> GetAllEventsAsyncEnumerator(CancellationToken cancellationToken)
+        {
+            return new AllTopicEventsEnumerator(_moveNextFunction, _subscriptionCancellationToken.Token, cancellationToken);
+        }
 
         /// <summary>
         /// Unsubscribe from this topic.
@@ -78,12 +111,12 @@ public void Dispose()
 
     private class TopicMessageEnumerator : IAsyncEnumerator<TopicMessage?>
     {
-        private readonly Func<CancellationToken, ValueTask<TopicMessage?>> _moveNextFunction;
+        private readonly Func<CancellationToken, ValueTask<ITopicEvent?>> _moveNextFunction;
         private readonly CancellationToken _subscriptionCancellationToken;
         private readonly CancellationToken _enumeratorCancellationToken;
 
         public TopicMessageEnumerator(
-            Func<CancellationToken, ValueTask<TopicMessage?>> moveNextFunction,
+            Func<CancellationToken, ValueTask<ITopicEvent?>> moveNextFunction,
             CancellationToken subscriptionCancellationToken,
             CancellationToken enumeratorCancellationToken)
         {
@@ -94,6 +127,63 @@ public TopicMessageEnumerator(
 
         public TopicMessage? Current { get; private set; }
 
+        public async ValueTask<bool> MoveNextAsync()
+        {
+            // We iterate over the stream until we get a TopicMessage, an error, or the stream is closed.
+            // We skip over system events like heartbeats and discontinuities.
+            while (true)
+            {
+                if (_subscriptionCancellationToken.IsCancellationRequested || _enumeratorCancellationToken.IsCancellationRequested)
+                {
+                    Current = null;
+                    return false;
+                }
+
+                var nextEvent = await _moveNextFunction.Invoke(_enumeratorCancellationToken);
+                switch (nextEvent)
+                {
+                    case TopicMessage.Text:
+                    case TopicMessage.Binary:
+                        Current = (TopicMessage)nextEvent;
+                        return true;
+                    case TopicMessage.Error:
+                        Current = (TopicMessage)nextEvent;
+                        return false;
+                    // This enumerator excludes system events from the stream
+                    case TopicSystemEvent.Discontinuity:
+                    case TopicSystemEvent.Heartbeat:
+                        continue;
+                    default:
+                        Current = null;
+                        return false;
+                }
+            }
+        }
+
+        public ValueTask DisposeAsync()
+        {
+            return new ValueTask();
+        }
+    }
+
+    private class AllTopicEventsEnumerator : IAsyncEnumerator<ITopicEvent?>
+    {
+        private readonly Func<CancellationToken, ValueTask<ITopicEvent?>> _moveNextFunction;
+        private readonly CancellationToken _subscriptionCancellationToken;
+        private readonly CancellationToken _enumeratorCancellationToken;
+
+        public AllTopicEventsEnumerator(
+            Func<CancellationToken, ValueTask<ITopicEvent?>> moveNextFunction,
+            CancellationToken subscriptionCancellationToken,
+            CancellationToken enumeratorCancellationToken)
+        {
+            _moveNextFunction = moveNextFunction;
+            _subscriptionCancellationToken = subscriptionCancellationToken;
+            _enumeratorCancellationToken = enumeratorCancellationToken;
+        }
+
+        public ITopicEvent? Current { get; private set; }
+
         public async ValueTask<bool> MoveNextAsync()
         {
             if (_subscriptionCancellationToken.IsCancellationRequested || _enumeratorCancellationToken.IsCancellationRequested)
@@ -102,15 +192,18 @@ public async ValueTask<bool> MoveNextAsync()
                 return false;
             }
 
-            var nextMessage = await _moveNextFunction.Invoke(_enumeratorCancellationToken);
-            switch (nextMessage)
+            var nextEvent = await _moveNextFunction.Invoke(_enumeratorCancellationToken);
+
+            switch (nextEvent)
             {
                 case TopicMessage.Text:
                 case TopicMessage.Binary:
-                    Current = nextMessage;
+                case TopicSystemEvent.Discontinuity:
+                case TopicSystemEvent.Heartbeat:
+                    Current = nextEvent;
                     return true;
                 case TopicMessage.Error:
-                    Current = nextMessage;
+                    Current = nextEvent;
                     return false;
                 default:
                     Current = null;
@@ -124,6 +217,23 @@ public ValueTask DisposeAsync()
         }
     }
 
+    // Helper class to wrap async enumerators into async enumerable
+    // This is necessary to support multiple enumerators on the same subscription.
+    private class AsyncEnumerableWrapper<T> : IAsyncEnumerable<T>
+    {
+        private readonly IAsyncEnumerator<T> _asyncEnumerator;
+
+        public AsyncEnumerableWrapper(IAsyncEnumerator<T> asyncEnumerator)
+        {
+            _asyncEnumerator = asyncEnumerator;
+        }
+
+        public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+        {
+            return _asyncEnumerator;
+        }
+    }
+
     /// <include file="../../docs.xml" path='docs/class[@name="Error"]/description/*' />
     public class Error : TopicSubscribeResponse, IError
     {

src/Momento.Sdk/Responses/Topic/TopicSystemEvent.cs

@@ -0,0 +1,56 @@
+namespace Momento.Sdk.Responses;
+
+/// <summary>
+/// Represents a system event that can be published to a topic.
+/// </summary>
+public abstract class TopicSystemEvent : ITopicEvent
+{
+    /// <summary>
+    /// Represents a heartbeat event.
+    /// </summary>
+    public class Heartbeat : TopicSystemEvent
+    {
+        /// <summary>
+        /// Constructs a new heartbeat event.
+        /// </summary>
+        public Heartbeat()
+        {
+
+        }
+
+        /// <inheritdoc/>
+        public override string ToString() => base.ToString() ?? "Heartbeat";
+    }
+
+    /// <summary>
+    /// Represents a discontinuity event.
+    /// </summary>
+    public class Discontinuity : TopicSystemEvent
+    {
+        /// <summary>
+        /// Constructs a new discontinuity event.
+        /// </summary>
+        /// <param name="lastKnownSequenceNumber">The last known sequence number before the discontinuity.</param>
+        /// <param name="sequenceNumber">The sequence number of the discontinuity.</param>
+        public Discontinuity(long lastKnownSequenceNumber, long sequenceNumber)
+        {
+            LastKnownSequenceNumber = lastKnownSequenceNumber;
+            SequenceNumber = sequenceNumber;
+        }
+
+        /// <summary>
+        /// The last known sequence number before the discontinuity.
+        /// </summary>
+        public long LastKnownSequenceNumber { get; }
+        /// <summary>
+        /// The sequence number of the discontinuity.
+        /// </summary>
+        public long SequenceNumber { get; }
+
+        /// <inheritdoc/>
+        public override string ToString()
+        {
+            return $"{base.ToString()}: LastKnownSequenceNumber: {LastKnownSequenceNumber} SequenceNumber: {SequenceNumber}";
+        }
+    }
+}