Added new SendMessage method overload without payload type

Author: Ahmad Noman Musleh

src/OpenAPI.Net/Helpers/MessageFactory.cs

@@ -40,7 +40,7 @@ public static ProtoMessage GetMessage<T>(this T message, ProtoOAPayloadType payl
         public static IMessage GetMessage(ProtoMessage protoMessage)
         {
             var payload = protoMessage.Payload;
-
+            
             return protoMessage.PayloadType switch
             {
                 (int)ProtoOAPayloadType.ProtoOaErrorRes => ProtoOAErrorRes.Parser.ParseFrom(payload),
@@ -89,7 +89,14 @@ public static IMessage GetMessage(ProtoMessage protoMessage)
             };
         }
 
-        private static ProtoMessage GetMessage(uint payloadType, ByteString payload, string clientMessageId = null)
+        /// <summary>
+        /// Returns a ProtoMessage based on your provided parameters
+        /// </summary>
+        /// <param name="payloadType">The message payloadType as unint</param>
+        /// <param name="payload">The message payload as a ByteString</param>
+        /// <param name="clientMessageId">The client message ID for ProtoMessage</param>
+        /// <returns>ProtoMessage</returns>
+        public static ProtoMessage GetMessage(uint payloadType, ByteString payload, string clientMessageId = null)
         {
             var message = new ProtoMessage
             {

src/OpenAPI.Net/Helpers/MessagePayloadTypeExtension.cs

@@ -0,0 +1,32 @@
+using Google.Protobuf;
+using System;
+using System.Reflection;
+
+namespace OpenAPI.Net.Helpers
+{
+    public static class MessagePayloadTypeExtension
+    {
+        /// <summary>
+        /// This method returns the payload type of a message
+        /// </summary>
+        /// <typeparam name="T">IMessage</typeparam>
+        /// <param name="message">The message</param>
+        /// <returns>uint (Payload Type)</returns>
+        /// <exception cref="InvalidOperationException"></exception>
+        public static uint GetPayloadType<T>(this T message) where T : IMessage
+        {
+            PropertyInfo property;
+
+            try
+            {
+                property = message.GetType().GetProperty("PayloadType");
+            }
+            catch (Exception ex) when (ex is AmbiguousMatchException || ex is ArgumentNullException)
+            {
+                throw new InvalidOperationException($"Couldn't get the PayloadType of the message {message}", ex);
+            }
+
+            return (uint)property.GetValue(message);
+        }
+    }
+}

src/OpenAPI.Net/OpenAPI.Net.csproj

@@ -9,7 +9,7 @@
     <PackageTags>cTrader, Open API, Spotware</PackageTags>
     <Description>A .NET RX library for Spotware Open API</Description>
     <PackageId>Spotware.OpenAPI.Net</PackageId>
-    <Version>1.3.6-rc0</Version>
+    <Version>1.3.6-rc1</Version>
     <Platforms>AnyCPU</Platforms>
     <Company>Spotware</Company>
     <Authors>Spotware</Authors>

src/OpenAPI.Net/OpenClient.cs

@@ -26,7 +26,7 @@ public sealed class OpenClient : IDisposable, IObservable<IMessage>
 
         private readonly ConcurrentDictionary<int, IObserver<IMessage>> _observers = new ConcurrentDictionary<int, IObserver<IMessage>>();
 
-        private readonly CancellationTokenSource _messagesCancellationTokenSource = new CancellationTokenSource();
+        private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
 
         private readonly TimeSpan _requestDelay;
 
@@ -36,8 +36,6 @@ public sealed class OpenClient : IDisposable, IObservable<IMessage>
 
         private SslStream _sslStream;
 
-        private IDisposable _listenerDisposable;
-
         private IDisposable _heartbeatDisposable;
 
         private IDisposable _webSocketDisconnectionHappenedDisposable;
@@ -129,7 +127,7 @@ public async Task Connect()
                 await ConnectTcp();
             }
 
-            _ = StartSendingMessages(_messagesCancellationTokenSource.Token);
+            _ = StartSendingMessages(_cancellationTokenSource.Token);
 
             _heartbeatDisposable = Observable.Interval(_heartbeatInerval).DoWhile(() => !IsDisposed)
                 .Subscribe(x => SendHeartbeat());
@@ -180,7 +178,7 @@ private async Task ConnectTcp()
 
             await _sslStream.AuthenticateAsClientAsync(Host).ConfigureAwait(false);
 
-            _ = ReadTcp();
+            _ = ReadTcp(_cancellationTokenSource.Token);
         }
 
         /// <summary>
@@ -197,6 +195,24 @@ public IDisposable Subscribe(IObserver<IMessage> observer)
             return Disposable.Create(() => OnObserverDispose(observer));
         }
 
+        /// <summary>
+        /// This method will insert your message on messages queue, it will not send the message instantly
+        /// By using this overload of SendMessage method you avoid passing the message payload type
+        /// and it gets the payload from message itself
+        /// </summary>
+        /// <typeparam name="T">Message Type</typeparam>
+        /// <param name="message">Message</param>
+        /// <param name="clientMsgId">The client message ID (optional)</param>
+        /// <exception cref="InvalidOperationException">If getting message payload type fails</exception>
+        /// <returns>Task</returns>
+        public async Task SendMessage<T>(T message, string clientMsgId = null) where T :
+            IMessage
+        {
+            var protoMessage = MessageFactory.GetMessage(message.GetPayloadType(), message.ToByteString(), clientMsgId);
+            
+            await SendMessage(protoMessage);
+        }
+
         /// <summary>
         /// This method will insert your message on messages queue, it will not send the message instantly
         /// </summary>
@@ -316,9 +332,8 @@ public void Dispose()
             IsDisposed = true;
 
             _heartbeatDisposable?.Dispose();
-            _listenerDisposable?.Dispose();
 
-            _messagesCancellationTokenSource.Cancel();
+            _cancellationTokenSource.Cancel();
 
             _messagesChannel.Writer.TryComplete();
 
@@ -346,7 +361,7 @@ public void Dispose()
         /// This method will read the TCP stream for incoming messages
         /// </summary>
         /// <returns>Task</returns>
-        private async Task ReadTcp()
+        private async Task ReadTcp(CancellationToken cancellationToken)
         {
             while (!IsDisposed)
             {
@@ -360,7 +375,7 @@ private async Task ReadTcp()
                     {
                         var count = lengthArray.Length - readBytes;
 
-                        readBytes += await _sslStream.ReadAsync(lengthArray, readBytes, count).ConfigureAwait(false);
+                        readBytes += await _sslStream.ReadAsync(lengthArray, readBytes, count, cancellationToken).ConfigureAwait(false);
                     }
                     while (readBytes < lengthArray.Length);
 
@@ -378,14 +393,17 @@ private async Task ReadTcp()
                     {
                         var count = data.Length - readBytes;
 
-                        readBytes += await _sslStream.ReadAsync(data, readBytes, count).ConfigureAwait(false);
+                        readBytes += await _sslStream.ReadAsync(data, readBytes, count, cancellationToken).ConfigureAwait(false);
                     }
                     while (readBytes < length);
 
                     var message = ProtoMessage.Parser.ParseFrom(data);
 
                     OnNext(message);
                 }
+                catch (TaskCanceledException)
+                {
+                }
                 catch (Exception ex)
                 {
                     var readException = new ReadException(ex);