Handle various protocol errors and socket exceptions

src/NATS.Client.Core/Commands/CommandWriter.cs

@@ -1,5 +1,6 @@
 using System.Buffers;
 using System.IO.Pipelines;
+using System.Net.Sockets;
 using System.Runtime.CompilerServices;
 using System.Threading.Channels;
 using Microsoft.Extensions.Logging;
@@ -593,6 +594,21 @@ private static async Task ReaderLoopAsync(
         {
             // Expected during shutdown
         }
+        catch (SocketException e)
+        {
+            logger.LogWarning(NatsLogEvents.Buffer, e, "Socket error in send buffer reader loop");
+            try
+            {
+                // We signal the connection to disconnect, which will trigger a reconnect
+                // in the connection loop.  This is necessary because the connection may
+                // be half-open, and we can't rely on the reader loop to detect that.
+                connection.SignalDisconnected(e);
+            }
+            catch (Exception e1)
+            {
+                logger.LogWarning(NatsLogEvents.Buffer, e1, "Error when signaling disconnect");
+            }
+        }
         catch (Exception e)
         {
             logger.LogError(NatsLogEvents.Buffer, e, "Unexpected error in send buffer reader loop");

src/NATS.Client.Core/NatsConnection.cs

@@ -396,8 +396,9 @@ private async ValueTask SetupReaderWriterAsync(bool reconnect)
 
         try
         {
-            // wait for INFO
-            await waitForInfoSignal.Task.ConfigureAwait(false);
+            // Wait for an INFO message from server. If we land on a dead socket and server response
+            // can't be received, this will throw a timeout exception and we will retry the connection.
+            await waitForInfoSignal.Task.WaitAsync(Opts.RequestTimeout).ConfigureAwait(false);
 
             // check to see if we should upgrade to TLS
             if (_socket is TcpConnection tcpConnection)

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

@@ -308,6 +308,8 @@ protected override async ValueTask ReceiveInternalAsync(
                     {
                         _notificationChannel?.Notify(new NatsJSProtocolNotification("Unhandled protocol message", headers.Code, headers.MessageText));
                         _logger.LogWarning(NatsJSLogEvents.ProtocolMessage, "Unhandled protocol message: {Code} {Description}", headers.Code, headers.MessageText);
+                        _userMsgs.Writer.TryComplete(new NatsJSProtocolException(headers.Code, headers.Message, headers.MessageText));
+                        EndSubscription(NatsSubEndReason.JetStreamError);
                     }
                 }
                 else

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

@@ -218,6 +218,8 @@ protected override async ValueTask ReceiveInternalAsync(
                     {
                         _notificationChannel?.Notify(new NatsJSProtocolNotification("Unhandled protocol message", headers.Code, headers.MessageText));
                         _logger.LogWarning(NatsJSLogEvents.ProtocolMessage, "Unhandled protocol message: {Code} {Description}", headers.Code, headers.MessageText);
+                        _userMsgs.Writer.TryComplete(new NatsJSProtocolException(headers.Code, headers.Message, headers.MessageText));
+                        EndSubscription(NatsSubEndReason.JetStreamError);
                     }
                 }
                 else

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

@@ -254,6 +254,8 @@ protected override async ValueTask ReceiveInternalAsync(
                     else
                     {
                         _logger.LogWarning(NatsJSLogEvents.ProtocolMessage, "Unhandled protocol message: {Code} {Description}", headers.Code, headers.MessageText);
+                        _userMsgs.Writer.TryComplete(new NatsJSProtocolException(headers.Code, headers.Message, headers.MessageText));
+                        EndSubscription(NatsSubEndReason.JetStreamError);
                     }
                 }
                 else