TcpClientTransport.cs

이 파일의 문서화 페이지로 가기

using System.IO;
using System.Net.Sockets;
using Dreamine.Communication.Abstractions.Enums;
using Dreamine.Communication.Abstractions.Interfaces;
using Dreamine.Communication.Abstractions.Models;
using Dreamine.Communication.Core.Framing;
using Dreamine.Communication.Core.Protocols;
using Dreamine.Communication.Sockets.Options;
 
namespace Dreamine.Communication.Sockets.Clients;

public sealed class TcpClientTransport : IMessageTransport
{
    private readonly TcpClientTransportOptions _options;
    private readonly IMessageProtocolAdapter _protocolAdapter;
    private readonly IMessageFrameCodec _frameCodec;

    private TcpClient? _client;
    private CancellationTokenSource? _receiveLoopCts;
    private Task? _receiveLoopTask;
    private int _state = (int)ConnectionState.Disconnected;

    public TcpClientTransport(TcpClientTransportOptions options)
        : this(
            options,
            new DreamineEnvelopeProtocolAdapter(),
            new LengthPrefixedMessageFrameCodec())
    {
    }

    public TcpClientTransport(
        TcpClientTransportOptions options,
        IMessageProtocolAdapter protocolAdapter,
        IMessageFrameCodec frameCodec)
    {
        _options = options ?? throw new ArgumentNullException(nameof(options));
        _protocolAdapter = protocolAdapter ?? throw new ArgumentNullException(nameof(protocolAdapter));
        _frameCodec = frameCodec ?? throw new ArgumentNullException(nameof(frameCodec));
 
        ValidateOptions(_options);
    }

    public ConnectionState State => (ConnectionState)Volatile.Read(ref _state);

    public TransportKind Kind => TransportKind.Tcp;

    public event EventHandler<MessageEnvelope>? MessageReceived;

    public async Task ConnectAsync(CancellationToken cancellationToken = default)
    {
        if (State is ConnectionState.Connected or ConnectionState.Connecting)
        {
            return;
        }
 
        SetState(ConnectionState.Connecting);
 
        try
        {
            CleanupClient();
 
            _client = new TcpClient
            {
                ReceiveBufferSize = _options.ReceiveBufferSize,
                SendBufferSize = _options.SendBufferSize
            };
 
            using var timeoutCts = new CancellationTokenSource(_options.ConnectTimeoutMs);
            using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
                cancellationToken,
                timeoutCts.Token);
 
            await _client.ConnectAsync(_options.Host, _options.Port, linkedCts.Token)
                .ConfigureAwait(false);
 
            SetState(ConnectionState.Connected);
 
            _receiveLoopCts = new CancellationTokenSource();
            _receiveLoopTask = Task.Run(
                () => ReceiveLoopAsync(_receiveLoopCts.Token),
                _receiveLoopCts.Token);
        }
        catch
        {
            SetState(ConnectionState.Faulted);
            CleanupClient();
 
            throw;
        }
    }

    public async Task DisconnectAsync(CancellationToken cancellationToken = default)
    {
        if (State == ConnectionState.Disconnected)
        {
            return;
        }
 
        SetState(ConnectionState.Disconnecting);
 
        _receiveLoopCts?.Cancel();
 
        CleanupClient();
 
        if (_receiveLoopTask is not null)
        {
            try
            {
                await _receiveLoopTask.ConfigureAwait(false);
            }
            catch (OperationCanceledException)
            {
            }
            catch (ObjectDisposedException)
            {
            }
            catch (SocketException)
            {
            }
            catch (IOException)
            {
            }
        }
 
        _receiveLoopCts?.Dispose();
        _receiveLoopCts = null;
        _receiveLoopTask = null;
 
        cancellationToken.ThrowIfCancellationRequested();
 
        SetState(ConnectionState.Disconnected);
    }

    public async Task SendAsync(
        MessageEnvelope message,
        CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(message);
 
        if (_client is null || State != ConnectionState.Connected)
        {
            throw new InvalidOperationException("TCP client is not connected.");
        }
 
        try
        {
            var stream = _client.GetStream();
            var payload = _protocolAdapter.Encode(message);
 
            await _frameCodec.WriteFrameAsync(stream, payload, cancellationToken)
                .ConfigureAwait(false);
        }
        catch
        {
            SetState(ConnectionState.Faulted);
            CleanupClient();
 
            throw;
        }
    }

    public async ValueTask DisposeAsync()
    {
        await DisconnectAsync().ConfigureAwait(false);
    }

    private async Task ReceiveLoopAsync(CancellationToken cancellationToken)
    {
        if (_client is null)
        {
            return;
        }
 
        try
        {
            var stream = _client.GetStream();
 
            while (!cancellationToken.IsCancellationRequested &&
                   State == ConnectionState.Connected)
            {
                var payload = await _frameCodec.ReadFrameAsync(stream, cancellationToken)
                    .ConfigureAwait(false);
 
                if (payload is null)
                {
                    break;
                }
 
                var message = _protocolAdapter.Decode(payload);
                MessageReceived?.Invoke(this, message);
            }
 
            if (!cancellationToken.IsCancellationRequested &&
                State == ConnectionState.Connected)
            {
                SetState(ConnectionState.Disconnected);
                CleanupClient();
            }
        }
        catch (OperationCanceledException)
        {
        }
        catch (ObjectDisposedException)
        {
        }
        catch (SocketException)
        {
            if (!cancellationToken.IsCancellationRequested)
            {
                SetState(ConnectionState.Faulted);
                CleanupClient();
            }
        }
        catch (IOException)
        {
            if (!cancellationToken.IsCancellationRequested)
            {
                SetState(ConnectionState.Faulted);
                CleanupClient();
            }
        }
        catch
        {
            if (!cancellationToken.IsCancellationRequested)
            {
                SetState(ConnectionState.Faulted);
                CleanupClient();
            }
        }
    }

    private void CleanupClient()
    {
        try
        {
            _client?.Close();
        }
        catch
        {
        }
 
        try
        {
            _client?.Dispose();
        }
        catch
        {
        }
 
        _client = null;
    }

    private void SetState(ConnectionState state)
    {
        Interlocked.Exchange(ref _state, (int)state);
    }

    private static void ValidateOptions(TcpClientTransportOptions options)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(options.Host);
 
        if (options.Port <= 0 || options.Port > 65535)
        {
            throw new ArgumentOutOfRangeException(nameof(options.Port));
        }
 
        if (options.ReceiveBufferSize <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(options.ReceiveBufferSize));
        }
 
        if (options.SendBufferSize <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(options.SendBufferSize));
        }
 
        if (options.ConnectTimeoutMs <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(options.ConnectTimeoutMs));
        }
    }
}