feat: add close reason tracking to NatsClient
This commit is contained in:
@@ -62,6 +62,12 @@ public sealed class NatsClient : IDisposable
|
|||||||
public long InBytes;
|
public long InBytes;
|
||||||
public long OutBytes;
|
public long OutBytes;
|
||||||
|
|
||||||
|
// Close reason tracking
|
||||||
|
private int _closeReason; // stores ClosedState as int for atomics
|
||||||
|
private int _skipFlushOnClose;
|
||||||
|
public ClosedState CloseReason => (ClosedState)Volatile.Read(ref _closeReason);
|
||||||
|
public bool ShouldSkipFlush => Volatile.Read(ref _skipFlushOnClose) != 0;
|
||||||
|
|
||||||
// PING keepalive state
|
// PING keepalive state
|
||||||
private int _pingsOut;
|
private int _pingsOut;
|
||||||
private long _lastIn;
|
private long _lastIn;
|
||||||
@@ -139,13 +145,24 @@ public sealed class NatsClient : IDisposable
|
|||||||
catch (OperationCanceledException)
|
catch (OperationCanceledException)
|
||||||
{
|
{
|
||||||
_logger.LogDebug("Client {ClientId} operation cancelled", Id);
|
_logger.LogDebug("Client {ClientId} operation cancelled", Id);
|
||||||
|
MarkClosed(ClosedState.ServerShutdown);
|
||||||
|
}
|
||||||
|
catch (IOException)
|
||||||
|
{
|
||||||
|
MarkClosed(ClosedState.ReadError);
|
||||||
|
}
|
||||||
|
catch (SocketException)
|
||||||
|
{
|
||||||
|
MarkClosed(ClosedState.ReadError);
|
||||||
}
|
}
|
||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
_logger.LogDebug(ex, "Client {ClientId} connection error", Id);
|
_logger.LogDebug(ex, "Client {ClientId} connection error", Id);
|
||||||
|
MarkClosed(ClosedState.ReadError);
|
||||||
}
|
}
|
||||||
finally
|
finally
|
||||||
{
|
{
|
||||||
|
MarkClosed(ClosedState.ClientClosed);
|
||||||
try { _socket.Shutdown(SocketShutdown.Both); }
|
try { _socket.Shutdown(SocketShutdown.Both); }
|
||||||
catch (SocketException) { }
|
catch (SocketException) { }
|
||||||
catch (ObjectDisposedException) { }
|
catch (ObjectDisposedException) { }
|
||||||
@@ -522,6 +539,30 @@ public sealed class NatsClient : IDisposable
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Marks this connection as closed with the given reason.
|
||||||
|
/// Sets skip-flush flag for error-related reasons.
|
||||||
|
/// Thread-safe — only the first call sets the reason.
|
||||||
|
/// </summary>
|
||||||
|
public void MarkClosed(ClosedState reason)
|
||||||
|
{
|
||||||
|
if (Interlocked.CompareExchange(ref _closeReason, (int)reason, 0) != 0)
|
||||||
|
return;
|
||||||
|
|
||||||
|
switch (reason)
|
||||||
|
{
|
||||||
|
case ClosedState.ReadError:
|
||||||
|
case ClosedState.WriteError:
|
||||||
|
case ClosedState.SlowConsumerPendingBytes:
|
||||||
|
case ClosedState.SlowConsumerWriteDeadline:
|
||||||
|
case ClosedState.TLSHandshakeError:
|
||||||
|
Volatile.Write(ref _skipFlushOnClose, 1);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
_logger.LogDebug("Client {ClientId} connection closed: {CloseReason}", Id, reason);
|
||||||
|
}
|
||||||
|
|
||||||
public void RemoveAllSubscriptions(SubList subList)
|
public void RemoveAllSubscriptions(SubList subList)
|
||||||
{
|
{
|
||||||
foreach (var sub in _subs.Values)
|
foreach (var sub in _subs.Values)
|
||||||
|
|||||||
Reference in New Issue
Block a user