No Responders and Prepare for Release (#193)

* Handle no-responders * prep for release Signed-off-by: Colin Sullivan <[email protected]>
nats-io · Oct 13, 2020 · b5c310d · b5c310d
1 parent 901c8fe
commit b5c310d
Show file tree

Hide file tree

Showing 9 changed files with 62 additions and 26 deletions.
diff --git a/doc/DoxyFile.STAN.Client b/doc/DoxyFile.STAN.Client
@@ -38,7 +38,7 @@ PROJECT_NAME           = "NATS Streaming .NET Client"
 # could be handy for archiving the generated documentation or if some version
 # control system is used.
 
-PROJECT_NUMBER         = 0.2.1
+PROJECT_NUMBER         = 0.3.0
 
 # Using the PROJECT_BRIEF tag one can provide an optional one line description
 # for a project that appears at the top of each page and should give viewer a

diff --git a/src/STAN.Client/STAN.Client.csproj b/src/STAN.Client/STAN.Client.csproj
@@ -26,9 +26,9 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Google.Protobuf" Version="3.10.0" />
-    <PackageReference Include="Google.Protobuf.Tools" Version="3.10.0" PrivateAssets="All" />
-    <PackageReference Include="NATS.Client" Version="0.10.0" />
+    <PackageReference Include="Google.Protobuf" Version="3.13.0" />
+    <PackageReference Include="Google.Protobuf.Tools" Version="3.13.0" PrivateAssets="All" />
+    <PackageReference Include="NATS.Client" Version="0.11.0" />
   </ItemGroup>
 
 </Project>
diff --git a/src/STAN.Client/StanConnection.cs b/src/STAN.Client/StanConnection.cs
@@ -149,7 +149,11 @@ public class Connection : IStanConnection, IDisposable
         private ISubscription pingSubscription = null;
 
         private readonly object pingLock = new object();
+
+#pragma warning disable 0618
         private readonly NUID pubNUID = new NUID();
+#pragma warning restore 0618
+
         private Timer pingTimer;
         private readonly byte[] pingBytes;
         private readonly string pingRequests;
@@ -246,6 +250,8 @@ internal Connection(string stanClusterID, string clientID, StanOptions options)
             }
             catch (NATSTimeoutException)
             {
+                // This also covers no-responders as the NATSNoRespondersException
+                // is a subclass of NATSTimeoutException
                 protoUnsubscribe();
                 if (ncOwned)
                 {
@@ -324,7 +330,7 @@ internal Connection(string stanClusterID, string clientID, StanOptions options)
                     pingMaxOut = response.PingMaxOut;
                     pingBytes = ProtocolSerializer.createPing(connID);
 
-                    pingTimer = new Timer(pingServer, null, pingInterval, Timeout.Infinite);
+                    pingTimer = new Timer(pingServer, null, pingInterval, pingInterval);
                 }
             }
             if (unsubPing)
@@ -353,21 +359,13 @@ private void pingServer(object state)
                     return;
                 }
 
-                // disable timer, will never be called.
-                pingTimer.Change(Timeout.Infinite, Timeout.Infinite);
-
                 pingOut++;
                 conn = nc;
 
                 if (pingOut > pingMaxOut)
                 {
                     lostConnection = true;
                 }
-                else
-                {
-                    //reactivate timer.
-                    pingTimer.Change(pingInterval, Timeout.Infinite);
-                }
             }
 
             if (lostConnection)
@@ -379,6 +377,7 @@ private void pingServer(object state)
             try
             {
                 conn.Publish(pingRequests, pingInbox, pingBytes);
+                conn.FlushBuffer();
             }
             catch (Exception ex)
             when (
@@ -461,10 +460,18 @@ private void cleanupOnClose(Exception ex)
             }
         }
 
-        private void processPingResponse(object sender, MsgHandlerEventArgs e)
+        private bool IsNoResponseMsg(Msg m)
+        {
+            return m != null && m.Data.Length == 0 && m.HasHeaders && "503".Equals(m.Header["Status"]);
+        }
+
+        private void processPingResponse(object sender, MsgHandlerEventArgs args)
         {
-            // No data means OK (no need to unmarshall)
-            var data = e.Message.Data;
+            if (IsNoResponseMsg(args.Message))
+                return;
+
+            // No data can be a valid message
+            var data = args.Message.Data;
             if (data?.Length > 0)
             {
                 var pingResp = new PingResponse();
@@ -482,6 +489,7 @@ private void processPingResponse(object sender, MsgHandlerEventArgs e)
                 if (err?.Length > 0)
                 {
                     closeDueToPing(new StanException(err));
+                    return;
                 }
             }
 
@@ -501,7 +509,10 @@ private void processHeartBeat(object sender, MsgHandlerEventArgs args)
             }
 
             if (lnc != null)
+            {
                 lnc.Publish(args.Message.Reply, null);
+                lnc.FlushBuffer();
+            }
         }
 
         internal PublishAck removeAck(string guid)
@@ -569,7 +580,9 @@ internal void processMsg(object sender, MsgHandlerEventArgs args)
 
         static public string newGUID()
         {
+#pragma warning disable 0618
             return NUID.NextGlobal;
+#pragma warning restore 0618
         }
 
         public void Publish(string subject, byte[] data)
@@ -799,6 +812,12 @@ public void Close()
                     // and gracefully close the streaming connection.  The streaming server
                     // will cleanup this client on its own.
                 }
+                catch (NATSTimeoutException)
+                {
+                    // Also covers NATSNoRespondersException.  Either way, 
+                    // Assume that there 's no streaming server available and
+                    // close the connection below if required.
+                }
 
                 if (reply != null)
                 {

diff --git a/src/Samples/RxSample/RxSample.csproj b/src/Samples/RxSample/RxSample.csproj
@@ -13,4 +13,7 @@
     <ProjectReference Include="..\..\STAN.Client\STAN.Client.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <PackageReference Include="NATS.Client" Version="0.11.0" />
+  </ItemGroup>
 </Project>
diff --git a/src/Samples/StanPub/StanPub.csproj b/src/Samples/StanPub/StanPub.csproj
@@ -12,4 +12,7 @@
     <ProjectReference Include="..\..\STAN.Client\STAN.Client.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <PackageReference Include="NATS.Client" Version="0.11.0" />
+  </ItemGroup>
 </Project>
diff --git a/src/Samples/StanSub/StanSub.csproj b/src/Samples/StanSub/StanSub.csproj
@@ -12,4 +12,7 @@
     <ProjectReference Include="..\..\STAN.Client\STAN.Client.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <PackageReference Include="NATS.Client" Version="0.11.0" />
+  </ItemGroup>
 </Project>
diff --git a/src/Tests/IntegrationTests/IntegrationTests.csproj b/src/Tests/IntegrationTests/IntegrationTests.csproj
@@ -9,9 +9,12 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.4.0" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.8.0-preview-20200921-01" />
     <PackageReference Include="xunit" Version="2.4.1" />
-    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.1" />
+    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.3"><IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+<PrivateAssets>all</PrivateAssets>
+</PackageReference>
+    <PackageReference Include="NATS.Client" Version="0.11.0" />
   </ItemGroup>
 
   <ItemGroup>

diff --git a/src/Tests/IntegrationTests/PingTests.cs b/src/Tests/IntegrationTests/PingTests.cs
@@ -62,6 +62,11 @@ public void TestPingsNatsConnGone()
                 {
                     nc.SubscribeAsync(StanConsts.DefaultDiscoverPrefix + "." + Context.ClusterId + ".pings", (obj, args) =>
                     {
+                        if (args.Message.Data.Length == 0)
+                        {
+                            return;
+                        }
+
                         count++;
                         if (count > StanConsts.DefaultPingMaxOut)
                         {
@@ -99,7 +104,7 @@ public void TestPingStreamingServerGone()
                     AutoResetEvent ev = new AutoResetEvent(false);
 
                     var so = Context.GetStanTestOptions(Context.Server1);
-                    so.PingInterval = 200;
+                    so.PingInterval = 100;
                     so.PingMaxOutstanding = 3;
                     so.ConnectionLostEventHandler = (obj, args) =>
                     {
@@ -108,8 +113,9 @@ public void TestPingStreamingServerGone()
 
                     using (var sc = Context.GetStanConnection(so))
                     {
+                        sc.NATSConnection.FlushBuffer();
                         nss.Shutdown();
-                        Assert.True(ev.WaitOne(20000));
+                        Assert.True(ev.WaitOne(2000));
                     }
                 }
             }
@@ -124,10 +130,6 @@ public void TestPingCloseUnlockPubCalls()
                 using (var nss = Context.StartStreamingServerWithExternal(Context.Server1))
                 {
                     var ev = new AutoResetEvent(false);
-                    //var so = StanOptions.GetDefaultOptions();
-                    //so.PingInterval = 50;
-                    //so.PingMaxOutstanding = 10;
-                    //so.PubAckWait = 100;
 
                     using (var sc = Context.GetStanConnection(Context.Server1))
                     {

diff --git a/src/Tests/UnitTests/UnitTests.csproj b/src/Tests/UnitTests/UnitTests.csproj
@@ -9,9 +9,12 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.4.0" />
     <PackageReference Include="xunit" Version="2.4.1" />
-    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.1" />
+    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.3"><IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+<PrivateAssets>all</PrivateAssets>
+</PackageReference>
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.7.1" />
+    <PackageReference Include="NATS.Client" Version="0.11.0" />
   </ItemGroup>
 
   <ItemGroup>