Changed the ReadTcp method to use while loop instead of Rx

Author: Ahmad Noman Musleh

src/Notebook.Sample/notebook.ipynb

@@ -0,0 +1,54 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "dotnet_interactive": {
+     "language": "csharp"
+    }
+   },
+   "outputs": [
+    {
+     "data": {
+      "text/html": [
+       "<div><div></div><div></div><div><strong>Installed Packages</strong><ul><li><span>Spotware.OpenAPI.Net, 1.3.5</span></li></ul></div></div>"
+      ]
+     },
+     "metadata": {},
+     "output_type": "display_data"
+    }
+   ],
+   "source": [
+    "#r \"nuget: Spotware.OpenAPI.Net\""
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "dotnet_interactive": {
+     "language": "csharp"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "using OpenAPI.Net;\n",
+    "using OpenAPI.Net.Auth;\n",
+    "using OpenAPI.Net.Helpers;"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": ".NET (C#)",
+   "language": "C#",
+   "name": ".net-csharp"
+  },
+  "language_info": {
+   "name": "C#"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

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.5</Version>
+    <Version>1.3.6-rc0</Version>
     <Platforms>AnyCPU</Platforms>
     <Company>Spotware</Company>
     <Authors>Spotware</Authors>

src/OpenAPI.Net/OpenClient.cs

@@ -181,10 +181,7 @@ private async Task ConnectTcp()
 
             await _sslStream.AuthenticateAsClientAsync(Host).ConfigureAwait(false);
 
-            _listenerDisposable = Observable.DoWhile(Observable.FromAsync(ReadTcp), () => !IsDisposed)
-                .Where(message => message != null)
-                .ObserveOn(TaskPoolScheduler.Default)
-                .Subscribe(OnNext);
+            _ = ReadTcp();
         }
 
         /// <summary>
@@ -349,51 +346,54 @@ public void Dispose()
         /// <summary>
         /// This method will read the TCP stream for incoming messages
         /// </summary>
-        /// <returns>Task<ProtoMessage></returns>
-        private async Task<ProtoMessage> ReadTcp()
+        /// <returns>Task</returns>
+        private async Task ReadTcp()
         {
-            try
+            while (!IsDisposed)
             {
-                var lengthArray = new byte[sizeof(int)];
+                try
+                {
+                    var lengthArray = new byte[sizeof(int)];
 
-                var readBytes = 0;
+                    var readBytes = 0;
 
-                do
-                {
-                    var count = lengthArray.Length - readBytes;
+                    do
+                    {
+                        var count = lengthArray.Length - readBytes;
 
-                    readBytes += await _sslStream.ReadAsync(lengthArray, readBytes, count).ConfigureAwait(false);
-                }
-                while (readBytes < lengthArray.Length);
+                        readBytes += await _sslStream.ReadAsync(lengthArray, readBytes, count).ConfigureAwait(false);
+                    }
+                    while (readBytes < lengthArray.Length);
 
-                Array.Reverse(lengthArray);
+                    Array.Reverse(lengthArray);
 
-                var length = BitConverter.ToInt32(lengthArray, 0);
+                    var length = BitConverter.ToInt32(lengthArray, 0);
 
-                if (length <= 0) return null;
+                    if (length <= 0) continue;
 
-                var data = new byte[length];
+                    var data = new byte[length];
 
-                readBytes = 0;
+                    readBytes = 0;
 
-                do
-                {
-                    var count = data.Length - readBytes;
+                    do
+                    {
+                        var count = data.Length - readBytes;
 
-                    readBytes += await _sslStream.ReadAsync(data, readBytes, count).ConfigureAwait(false);
-                }
-                while (readBytes < length);
+                        readBytes += await _sslStream.ReadAsync(data, readBytes, count).ConfigureAwait(false);
+                    }
+                    while (readBytes < length);
 
-                return ProtoMessage.Parser.ParseFrom(data);
-            }
-            catch (Exception ex)
-            {
-                var readException = new ReadException(ex);
+                    var message = ProtoMessage.Parser.ParseFrom(data);
 
-                OnError(readException);
-            }
+                    OnNext(message);
+                }
+                catch (Exception ex)
+                {
+                    var readException = new ReadException(ex);
 
-            return null;
+                    OnError(readException);
+                }
+            }
         }
 
         /// <summary>