4
\$\begingroup\$

I'm working on a flexible socket framework and trying to polish it so it's actually something usable in further development. Would really appreciate any critique, also I’m really interested whether I'm using cancellation tokens correctly on client (will use them on server later as well). This is an improved version of the previous implementation.


My interfaces:

IAsyncClient

public delegate void ConnectedHandler(IAsyncClient a);

public delegate void ClientReceivingStarted();

public delegate void ClientMessageReceivedHandler(IAsyncClient a, List<byte> msg);

public delegate void ClientMessageFailedHandler(IAsyncClient a, List<byte> msg);

public delegate void ClientMessageSubmittedHandler(IAsyncClient a);

public delegate void ClientErrorHandler(string message, Exception ex);

public delegate void ClientNoLicenseHandler(string additionalInfo);

public interface IAsyncClient : IDisposable
{
    bool IsConnected { get; }

    event ConnectedHandler Connected;
    event ClientReceivingStarted ReceivingStarted;
    event ClientMessageReceivedHandler MessageReceived;
    event ClientMessageFailedHandler MessageFailed;
    event ClientMessageSubmittedHandler MessageSubmitted;
    event ClientErrorHandler Error;
    event ClientNoLicenseHandler NoLicense;

    Task<bool> ConnectClientAsync();
    void CloseClient();
    void SetId(Guid clientId);
    Task<bool> Send(IProcessable message);
    Task<bool> SendAlarm();
}

IAsyncSocketListener

public delegate void MessageReceivedHandler(Guid id, List<byte> msg);

public delegate void MessageSubmittedHandler(Guid id);

public interface IAsyncSocketListener : IDisposable
{
    event MessageReceivedHandler MessageReceived;
    event MessageSubmittedHandler MessageSubmitted;

    Task StartServer();
    void StopServer();
    bool IsConnected(Guid id);
    Task<bool> Send(Guid id, IProcessable msg);
    Task SendToAll(IProcessable msg);
    Task SendToAllExcept(List<Guid> exludedClientIds, IProcessable msg);
}

IProcessable

public interface IProcessable
{
    byte[] ToByteArray();
}

ICommandWrrapper

public interface ICommandWrapper
{
    Guid ClientId { get; set; }

    IProcessable Command { get; set; }
}

IStateObject

public interface IStateObject
{
    Guid Id { get; }
    int BufferSize { get; }
    byte[] Buffer { get; }
    byte[] PrefixBuffer { get; }
    TcpClient Socket { get; }
    List<byte> ResultBuffer { get; }
    int ExpectedLength { get; set; }
    void Allocate(int length);
    void Append(byte[] chunk, int receive);
    void Reset();
}

This basically represents the API exposed by the libraries.

How for the implementation:


StreamSecurityClient

public sealed class StreamSecurityClient : IAsyncClient
{
    #region Fields and Props

    private readonly string _host;
    private readonly ushort _port;

    private readonly string _login;
    private readonly string _pass;

    private readonly bool _isHardMode;

    private readonly Key _key;

    private readonly LockObjectProvider _lockProvider = new LockObjectProvider();
    private readonly object _syncRoot = new object();
    private CancellationTokenSource _connectionTokenSource = new CancellationTokenSource();
    private CancellationTokenSource _receivingTokenSource = new CancellationTokenSource();
    private CancellationTokenSource _sendingTokenSource = new CancellationTokenSource();

    public CancellationTokenSource ConnectionTokenSource
    {
        get
        {
            lock (_lockProvider.GetLock(nameof(ConnectionTokenSource)))
            {
                return _connectionTokenSource;
            }
        }
        set
        {
            lock (_lockProvider.GetLock(nameof(ConnectionTokenSource)))
            {
                _connectionTokenSource = value;
            }
        }
    }

    public CancellationTokenSource ReceivingTokenSource
    {
        get
        {
            lock (_lockProvider.GetLock(nameof(ReceivingTokenSource)))
            {
                return _receivingTokenSource;
            }
        }
        set
        {
            lock (_lockProvider.GetLock(nameof(ReceivingTokenSource)))
            {
                _receivingTokenSource = value;
            }
        }
    }

    public CancellationTokenSource SendingTokenSource
    {
        get
        {
            lock (_lockProvider.GetLock(nameof(SendingTokenSource)))
            {
                return _sendingTokenSource;
            }
        }
        set
        {
            lock (_lockProvider.GetLock(nameof(SendingTokenSource)))
            {
                _sendingTokenSource = value;
            }
        }
    }

    private AutoResetEvent _receivingStarted = new AutoResetEvent(false);

    private bool _isInstantiated;
    private bool _isReconnectInProgress;

    private TcpClient _client;

    private readonly Policy _retryPolicy = Policy.Handle<Exception>()
        .WaitAndRetryAsync(2, retryAttempt => TimeSpan.FromSeconds(1));

    private bool IsInstantiated
    {
        get
        {
            lock (_syncRoot)
            {
                return _isInstantiated;
            }
        }
        set
        {
            lock (_syncRoot)
            {
                _isInstantiated = value;
            }
        }
    }

    private bool IsReconnectInProgress
    {
        get
        {
            lock (_syncRoot)
            {
                return _isReconnectInProgress;
            }
        }
        set
        {
            lock (_syncRoot)
            {
                _isReconnectInProgress = value;
            }
        }
    }

    public event ConnectedHandler Connected;
    public event ClientMessageReceivedHandler MessageReceived;
    public event ClientMessageFailedHandler MessageFailed;
    public event ClientMessageSubmittedHandler MessageSubmitted;
    public event ClientReceivingStarted ReceivingStarted;
    public event ClientErrorHandler Error;
    public event ClientNoLicenseHandler NoLicense;

    public bool IsConnected
    {
        get
        {
            try
            {
                var poll = _client.Client.Poll(1000, SelectMode.SelectRead);
                var available = _client.Available == 0;
                return !(poll && available) && IsInstantiated && !IsReconnectInProgress;
            }
            catch (Exception ex)
            {
                return false;
            }
        }
    }

    internal Guid ClientId { get; set; }

    #endregion

    #region Instantiation and Cleanup

    public StreamSecurityClient(Key key)
    {
        _key = key;

        _host = ConfigurationManager.AppSettings["Host"];
        _port = Convert.ToUInt16(ConfigurationManager.AppSettings["Port"]);
        _isHardMode = Convert.ToBoolean(ConfigurationManager.AppSettings["IsHardMode"]);
        _login = key.Client;
        _pass = key.Pass;
    }

    public StreamSecurityClient(string host, ushort port, Key key)
    {
        _key = key;

        _host = host;
        _port = port;
        _login = key.Client;
        _pass = key.Pass;
    }

    public async Task<bool> ConnectClientAsync()
    {
        return await WrapTryCatch(async () =>
            {
                return await _retryPolicy.ExecuteAsync(InstantiateClient);
            },
            ex =>
            {
                NotifyError("Could not connect.", ex);
                return false;
            });
    }

    private async Task<bool> InstantiateClient()
    {
        if (!IsInstantiated || !IsConnected)
        {
            _receivingStarted = new AutoResetEvent(false);
            IsReconnectInProgress = true;

            return await WrapTryCatch(async () =>
                {
                    Console.WriteLine("Instantiating client");
                    var ip = IPAddress.Parse(_host);
                    var endpoint = new IPEndPoint(ip, _port);

                    CleanseTokens();

                    ConnectionTokenSource = new CancellationTokenSource();
                    ReceivingTokenSource = new CancellationTokenSource();
                    SendingTokenSource = new CancellationTokenSource();

                    _client = _host == "127.0.0.1" ? new TcpClient() : new TcpClient(endpoint);

                    await ConnectClientTask(_client, endpoint);

                    Console.WriteLine("Connected");
                    IsInstantiated = true;
                    IsReconnectInProgress = false;

                    StartReceivingAsync();
                    StartHeartbeatAsync();

                    _receivingStarted.WaitOne();
                    ReceivingStarted?.Invoke();

                    await Subscribe();

                    Connected?.Invoke(this);
                    return true;
                },
                ex =>
                {
                    CloseClient();
                },
                () =>
                {
                    IsReconnectInProgress = false;
                }
            );
        }

        return true;
    }

    private void CleanseTokens()
    {
        ConnectionTokenSource?.Dispose();
        ReceivingTokenSource?.Dispose();
        SendingTokenSource?.Dispose();

        ConnectionTokenSource = null;
        ReceivingTokenSource = null;
        SendingTokenSource = null;
    }

    public void CloseClient()
    {
        try
        {
            lock (_syncRoot)
            {
                if (IsInstantiated || IsReconnectInProgress)
                {
                    Console.WriteLine("Closing gracefully");

                    ConnectionTokenSource.Cancel();
                    ReceivingTokenSource.Cancel();
                    SendingTokenSource.Cancel();

                    _client?.Close();
                    _client?.Dispose();
                    _client = null;

                    IsInstantiated = false;
                }
            }
        }
        catch (SocketException ex)
        {
            // TODO: handle possible exception
        }
        catch (ObjectDisposedException ex)
        {
            // TODO: handle possible exception
        }
    }

    public void Dispose()
    {
        _receivingStarted.Dispose();
        CloseClient();
        CleanseTokens();
    }

    private async Task ConnectClientTask(TcpClient client, IPEndPoint endpoint)
    {
        ConnectionTokenSource.Token.ThrowIfCancellationRequested();
        await client.ConnectAsync(endpoint.Address, endpoint.Port);//why no cancellation?
    }

    #endregion

    #region Service Methods

    public void SetId(Guid clientId)
    {
        ClientId = clientId;
    }

    internal void NotifyError(string message)
    {
        NotifyError(message, new Exception(message));
    }

    internal void NotifyError(string message, Exception ex)
    {
        var exMessage = message + " " + ex.Message;
        Logger.Log.Error(exMessage);
        Error?.Invoke(message, ex);
    }

    private byte[] WrapMessage(byte[] message)
    {
        // Get the length prefix for the message
        var lengthPrefix = BitConverter.GetBytes(message.Length);

        // Concatenate the length prefix and the message
        var resultingArr = new byte[lengthPrefix.Length + message.Length];
        lengthPrefix.CopyTo(resultingArr, 0);
        message.CopyTo(resultingArr, lengthPrefix.Length);

        return resultingArr;
    }

    private byte[] WrapKeepaliveMessage()
    {
        return BitConverter.GetBytes(0);
    }

    private async Task<T> WrapTryCatch<T>(Func<Task<T>> tryAction, Func<Exception, T> catchAction, Action finallyAction = null)
    {
        try
        {
            return await tryAction();
        }
        catch (SocketException ex)
        {
            return catchAction(ex);
        }
        catch (ObjectDisposedException ex)
        {
            return catchAction(ex);
        }
        catch (OperationCanceledException ex)
        {
            return catchAction(ex);
        }
        catch (AggregateException ex)
        {
            return catchAction(ex);
        }
        catch (Exception ex)
        {
            return catchAction(ex);
        }
        finally
        {
            finallyAction?.Invoke();
        }
    }

    private async Task<T> WrapTryCatch<T>(Func<Task<T>> tryAction, Action<Exception> catchAction, Action finallyAction = null)
    {
        try
        {
            return await tryAction();
        }
        catch (SocketException ex)
        {
            catchAction(ex);
            throw;
        }
        catch (ObjectDisposedException ex)
        {
            catchAction(ex);
            throw;
        }
        catch (OperationCanceledException ex)
        {
            catchAction(ex);
            throw;
        }
        catch (AggregateException ex)
        {
            catchAction(ex);
            throw;
        }
        catch (Exception ex)
        {
            catchAction(ex);
            throw;
        }
        finally
        {
            finallyAction?.Invoke();
        }
    }

    private async Task WrapTryCatch(Func<Task> tryAction, Action<Exception> catchAction, bool rethrow = false, Action finallyAction = null)
    {
        try
        {
            await tryAction();
        }
        catch (SocketException ex)
        {
            catchAction(ex);
            if (rethrow)
            {
                throw;
            }
        }
        catch (ObjectDisposedException ex)
        {
            catchAction(ex);
            if (rethrow)
            {
                throw;
            }
        }
        catch (OperationCanceledException ex)
        {
            catchAction(ex);
            if (rethrow)
            {
                throw;
            }
        }
        catch (AggregateException ex)
        {
            catchAction(ex);
            if (rethrow)
            {
                throw;
            }
        }
        catch (Exception ex)
        {
            catchAction(ex);
            if (rethrow)
            {
                throw;
            }
        }
        finally
        {
            finallyAction?.Invoke();
        }
    }

    #endregion

    #region Data Receiving

    //fire and forget style task
    private async Task StartReceivingAsync()
    {
        await WrapTryCatch(async () =>
            {
                Console.WriteLine("Starting receiving pipeline");
                IStateObject state = new StateObject(_client);

                _receivingStarted.Set();

                var bytesRead = await SizeReceivingTask(state);
                await SizeReceivedHandler(bytesRead, state);
            },
            ex =>
            {
                NotifyError("Client receiving pipeline could not be started", ex);
                CloseClient();
            });
    }

    public async Task StartHeartbeatAsync()
    {
        await Task.Run(() =>
        {
            while (true)
            {
                if (!IsConnected)
                {
                    NotifyError("Heartbeat: no connection.");
                    CloseClient();
                    break;
                }
                Thread.Sleep(1000);
            }
        });
    }

    private async Task SizeReceivedHandler(int bytesRead, IStateObject state)
    {
        while (true)
        {
            ReceivingTokenSource.Token.ThrowIfCancellationRequested();

            if (bytesRead > 0)
            {
                var length = BitConverter.ToInt32(state.PrefixBuffer, 0);

                if (length == 0)
                {
                    bytesRead = await SizeReceivingTask(state);
                    continue;
                }

                state.Allocate(length);
                state.ExpectedLength = length;

                var messageBytesRead = await MessageReceivingTask(state, state.Buffer.Length);
                await MessageReceivedHandler(messageBytesRead, state);
            }
            break;
        }
    }

    private async Task MessageReceivedHandler(int bytesRead, IStateObject state)
    {
        while (true)
        {
            ReceivingTokenSource.Token.ThrowIfCancellationRequested();

            if (bytesRead > 0)
            {
                //size received
                state.Append(state.Buffer, bytesRead);
            }

            if (state.ResultBuffer.Count < state.ExpectedLength)
            {
                var nextChunk = Math.Min(state.BufferSize, state.ExpectedLength - state.ResultBuffer.Count);
                var messageReceivingTask = MessageReceivingTask(state, nextChunk);
                var messageBytesRead = await messageReceivingTask;
                bytesRead = messageBytesRead;
                continue;
            }

            else
            {
                //full message received
                MessageReceived?.Invoke(this, state.ResultBuffer);

                state.Reset();

                var sizeBytesRead = await SizeReceivingTask(state);
                await SizeReceivedHandler(sizeBytesRead, state);
            }
            break;
        }
    }

    private async Task<int> SizeReceivingTask(IStateObject state)
    {
        ReceivingTokenSource.Token.ThrowIfCancellationRequested();
        return
            await state.Socket.GetStream().ReadAsync(state.PrefixBuffer, 0, sizeof(int), ReceivingTokenSource.Token);
    }

    private async Task<int> MessageReceivingTask(IStateObject state, int expectedChunkSize)
    {
        ReceivingTokenSource.Token.ThrowIfCancellationRequested();
        return
            await state.Socket.GetStream().ReadAsync(state.Buffer, 0, expectedChunkSize, ReceivingTokenSource.Token);
    }

    #endregion

    #region Data Sending

    public async Task<bool> Send(IProcessable message)
    {
        return await WrapTryCatch(async () =>
            {
                SendingTokenSource.Token.ThrowIfCancellationRequested();

                var messageBytes = WrapMessage(message.ToByteArray());
                var retryTask = _retryPolicy.ExecuteAsync(async () =>
                {
                    await SendTask(_client, messageBytes);
                    MessageSubmitted?.Invoke(this);
                });

                await retryTask;
                return true;
            },
            ex =>
            {
                NotifyError("Message sending failed.", ex);
                CloseClient();
                return false;
            });
    }

    public async Task<bool> SendAlarm()
    {
        //should prevent multiple sending attempts
        return await Send(new RaiseAlarmCommand());
    }

    private async Task<bool> Subscribe()
    {
        return await Send(new SubscribeCommand { FingerPrint = "Hello" });
    }

    private async Task SendTask(TcpClient client, byte[] message)
    {
        SendingTokenSource.Token.ThrowIfCancellationRequested();
        await client.GetStream().WriteAsync(message, 0, message.Length, SendingTokenSource.Token);
    }

    #endregion
}

StreamSecurityServer

public sealed class StreamSecurityServer : IAsyncSocketListener
{
    #region Fields and Props

    private readonly ushort _port;
    private readonly int _limit;
    private readonly object _syncRoot = new object();

    private readonly LockObjectProvider _lockProvider = new LockObjectProvider();
    private TcpListener _listener;
    private IDictionary<Guid, IStateObject> _clients;

    public event MessageReceivedHandler MessageReceived;
    public event MessageSubmittedHandler MessageSubmitted;

    public StreamSecurityServer(ushort port, int limit)
    {
        _port = port;
        _limit = limit;
    }

    #endregion

    #region Instantiation and Cleanup

    public async Task StartServer()
    {
        var ip = IPAddress.Loopback;
        var endpoint = new IPEndPoint(ip, _port);

        try
        {
            lock (_lockProvider.GetLock(nameof(_clients)))
            {
                _clients = new ConcurrentDictionary<Guid, IStateObject>();
            }

            _listener = new TcpListener(endpoint.Address, endpoint.Port);
            _listener.Start();

            while (true)
            {
                var incClient = await AcceptClientTask(_listener);

                IStateObject state;

                lock (_lockProvider.GetLock(nameof(_clients)))
                {
                    var id = Guid.NewGuid();

                    state = new StateObject(incClient, id);
                    _clients.Add(id, state);
                    Console.WriteLine("Client connected. New client Id: " + id);
                }

                StartReceiving(state);
            }
        }
        catch (SocketException ex)
        {
            // TODO:
        }
        catch (ObjectDisposedException ex)
        {
            //TODO:
        }
    }

    public void StopServer()
    {
        //todo tokens here
        _listener.Stop();
        foreach (var id in _clients.Keys)
        {
            Close(id);
        }

        lock (_lockProvider.GetLock(nameof(_clients)))
        {
            _clients = null;
        }
    }

    private async Task<TcpClient> AcceptClientTask(TcpListener listener)
    {
        return await listener.AcceptTcpClientAsync();
    }

    public void Dispose()
    {
        StopServer();
    }

    public void Close(Guid id)
    {
        var state = GetClient(id);

        if (state == null)
        {
            throw new Exception("Client does not exist.");
        }

        try
        {
            state.Socket.Close();
            state.Socket.Dispose();
        }
        catch (SocketException)
        {
            // TODO:
        }
        finally
        {
            lock (_lockProvider.GetLock(nameof(_clients)))
            {
                _clients.Remove(state.Id);
                Console.WriteLine("Client disconnected with Id {0}", state.Id);
                state = null;
            }
        }
    }

    #endregion

    #region Service Methods

    public bool IsConnected(Guid id)
    {
        var state = GetClient(id);

        try
        {
            var poll = state.Socket.Client.Poll(1000, SelectMode.SelectRead);
            var available = state.Socket.Available == 0;

            return !(poll && available);
        }
        catch (Exception ex)
        {
            return false;
        }
    }

    private byte[] WrapMessage(byte[] message)
    {
        var lengthPrefix = BitConverter.GetBytes(message.Length);

        var resultingArr = new byte[lengthPrefix.Length + message.Length];
        lengthPrefix.CopyTo(resultingArr, 0);
        message.CopyTo(resultingArr, lengthPrefix.Length);

        return resultingArr;
    }

    private byte[] WrapKeepaliveMessage()
    {
        return BitConverter.GetBytes(0);
    }

    private IStateObject GetClient(Guid id)
    {
        lock (_lockProvider.GetLock(nameof(_clients)))
        {
            IStateObject state;
            return _clients.TryGetValue(id, out state) ? state : null;
        }
    }

    #endregion

    #region Data Receiving

    private async Task StartReceiving(IStateObject state)
    {
        try
        {
            var bytesRead = await SizeReceivingTask(state);
            await SizeReceivedHandler(bytesRead, state);
        }
        catch (SocketException ex)
        {
            Close(state.Id);
        }
        catch (ObjectDisposedException ex)
        {
            Close(state.Id);
        }
        catch (NullReferenceException ex)
        {
            Close(state.Id);
        }
    }

    private async Task SizeReceivedHandler(int bytesRead, IStateObject state)
    {
        while (true)
        {
            if (bytesRead > 0)
            {
                var length = BitConverter.ToInt32(state.PrefixBuffer, 0);

                if (length == 0)
                {
                    bytesRead = await SizeReceivingTask(state);
                    continue;
                }

                state.Allocate(length);
                state.ExpectedLength = length;

                var messageBytes = await MessageReceivingTask(state, state.Buffer.Length);
                await MessageReceivedHandler(messageBytes, state);
            }
            break;
        }
    }

    private async Task MessageReceivedHandler(int bytesRead, IStateObject state)
    {
        while (true)
        {
            if (bytesRead > 0)
            {
                //size received
                state.Append(state.Buffer, bytesRead);
            }

            if (state.ResultBuffer.Count < state.ExpectedLength)
            {
                var nextChunk = Math.Min(state.BufferSize, state.ExpectedLength - state.ResultBuffer.Count);
                var messageBytes = await MessageReceivingTask(state, nextChunk);
                bytesRead = messageBytes;
                continue;
            }

            else
            {
                MessageReceived?.Invoke(state.Id, state.ResultBuffer);
                state.Reset();
                await StartReceiving(state);
            }
            break;
        }
    }

    private async Task<int> SizeReceivingTask(IStateObject state)
    {
        //todo token here
        return await state.Socket.GetStream().ReadAsync(state.PrefixBuffer, 0, sizeof(int));
    }

    private async Task<int> MessageReceivingTask(IStateObject state, int expectedChunkSize)
    {
        //todo token here
        return await state.Socket.GetStream().ReadAsync(state.Buffer, 0, expectedChunkSize);
    }

    #endregion

    #region Data Sending

    public async Task<bool> Send(Guid id, IProcessable msg)
    {
        var state = GetClient(id);
        return await SendInternal(msg, state);
    }

    public async Task SendToAll(IProcessable msg)
    {
        var sendingTasks = new List<Task>();

        lock (_lockProvider.GetLock(nameof(_clients)))
        {
            var states = _clients.Values;

            foreach (var state in states)
            {
                sendingTasks.Add(SendInternal(msg, state));
            }
        }

        await Task.WhenAll(sendingTasks.ToArray());
    }

    public async Task SendToAllExcept(List<Guid> exludedClientIds, IProcessable msg)
    {
        var sendingTasks = new List<Task>();

        lock (_lockProvider.GetLock(nameof(_clients)))
        {
            var states = _clients.Where(x => !exludedClientIds.Contains(x.Key)).Select(x => x.Value);

            foreach (var state in states)
            {
                sendingTasks.Add(SendInternal(msg, state));
            }
        }

        await Task.WhenAll(sendingTasks.ToArray());
    }

    private async Task<bool> SendInternal(IProcessable msg, IStateObject state)
    {
        try
        {
            if (state == null)
            {
                //TODO: Notify error, cleanup
                throw new Exception("Client does not exist.");
            }

            if (!IsConnected(state.Id))
            {
                //TODO: Notify error, cleanup
                throw new Exception("Destination socket is not connected.");
            }

            var messageBytes = WrapMessage(msg.ToByteArray());

            var sendingTask = SendTask(state.Socket, messageBytes);
            await sendingTask;
            MessageSubmitted?.Invoke(state.Id);
            return true;
        }
        catch (SocketException ex)
        {
            // TODO:
            return false;
        }
        catch (ArgumentException ex)
        {
            // TODO:
            return false;
        }
    }

    private async Task SendTask(TcpClient client, byte[] message)
    {
        await client.GetStream().WriteAsync(message, 0, message.Length);
    }

    #endregion
}

StateObject

public sealed class StateObject : IStateObject
{
    /* Contains the state information. */

    private const int BufferSizeInternal = 1024;

    public StateObject(TcpClient socket)
    {
        Socket = socket;
        Reset();
    }

    public StateObject(TcpClient socket, Guid id)
    {
        Id = id;
        Socket = socket;
        Reset();
    }

    public Guid Id { get; }
    public int BufferSize => BufferSizeInternal;
    public byte[] Buffer { get; } = new byte[BufferSizeInternal];
    public byte[] PrefixBuffer { get; private set; } = new byte[sizeof(int)];
    public TcpClient Socket { get; }
    public List<byte> ResultBuffer { get; private set; }
    public int ExpectedLength { get; set; }

    public void Allocate(int length)
    {
        ResultBuffer = new List<byte>(length);
    }

    public void Append(byte[] chunk, int receive)
    {
        ResultBuffer.AddRange(chunk.Take(receive));
    }

    public void Reset()
    {
        ResultBuffer = new List<byte>();
        PrefixBuffer = new byte[sizeof(int)];
        ExpectedLength = 0;
    }
}

That's about it for the core implementation. Essentially, it's a cleaner and simplified version of my previous client/server implementation that takes all the advices into account. This is still work in progress. As stated before - any critique is welcome and the main questions are whether I'm doing the CancellationTokens stuff right and whether there are some critical implementation flaws.

\$\endgroup\$
  • 2
    \$\begingroup\$ You referenced a previous review a few times. You might want to link to it to provide additional context for reviewers. \$\endgroup\$ – RubberDuck Jun 28 '17 at 2:59
  • \$\begingroup\$ Good point, link to previous version added \$\endgroup\$ – HardLuck Jun 28 '17 at 7:08
2
\$\begingroup\$

Events

public delegate void ClientMessageReceivedHandler(IAsyncClient a, List<byte> msg);

For someone like me who's worked with real events before it's very weird to see that you use the event keyword but instead of using the EventHandler or the generic EventHandler<> you use your own delegates that have a different signature then the expected one:

public delegate void EventHandler<TEventArgs>(object sender, TEventArgs e)

Where TEventArgs would be a type derived from EventArgs that carries your parameters.

This has the advantage that if you later decide to extend it, you won't have to change each and every usage.


Essentially, it's a cleaner and simplified version

I think you simplified this too much :-)

public delegate void ClientMessageReceivedHandler(IAsyncClient a, List<byte> msg);

because a as a parameter name is never cleaner and abbreviating message to msg is also not a good style. In fact in C# we rarely use abbreviations.


public interface IProcessable
{
    byte[] ToByteArray();
}

When I see this interface I expect to find a method like Process that takes some parameters and processes them to yield a result. In this case however the interface should rather be called either IConvertable or because you name every parameter of that type msg so why not call it IMessage?

public interface ICommandWrapper
{
    Guid ClientId { get; set; }

    IProcessable Command { get; set; }
}

Although here it's a Command. This is very confusing and inconsistant. Either it's a command, or a message or something else. Currently it seems to be a god-type for every purpose.

I also don't understand what an ICommandWrapper is but I guess it just adds a Guid to IProcessiable. A better design would be to create an interface like IIdentifiable that types that support an Id would additionaly implement.


Consistency

Task<bool> ConnectClientAsync();
void CloseClient();
void SetId(Guid clientId);
Task<bool> Send(IProcessable message);
Task<bool> SendAlarm();

Your API isn't consitent. Some methods that return a Task have the suffix Async whereas others don't. See Does the use of the “Async” suffix in a method name depend on whether the 'async' modifier is used? for more info.

void SetId(Guid clientId);

Why can I set an id but cannot read it? If I'm setting it for the current IAsyncClient then it should rather be a property although I doubt an id should be settable at all. If this is however for something else then the method name does not clearly communicate this.


public interface IAsyncSocketListener : IDisposable
{
    event MessageReceivedHandler MessageReceived;
    event MessageSubmittedHandler MessageSubmitted;

    Task StartServer();
    void StopServer();
    bool IsConnected(Guid id);
    Task<bool> Send(Guid id, IProcessable msg);
    Task SendToAll(IProcessable msg);
    Task SendToAllExcept(List<Guid> exludedClientIds, IProcessable msg);
}

This interface has the same issues an the one above. Additionaly the purpose of the IsConnected method isn't clear. At the first glance I thought it returns a value indicating whether the server is connected but I guess it's about a client, right? If so then it should be called IsClientConnected.

    Task SendToAll(IProcessable msg);
    Task SendToAllExcept(List<Guid> exludedClientIds, IProcessable msg);

The last method has the arguments in the wrong order. For the sake of consistency the excluded ids should come last because this kind of extends what SendToAll does and adds a new parameter. They are very similar and should stay that way.


public interface IStateObject
{
    Guid Id { get; }
    int BufferSize { get; }
    byte[] Buffer { get; }
    byte[] PrefixBuffer { get; }
    TcpClient Socket { get; }
    List<byte> ResultBuffer { get; }
    int ExpectedLength { get; set; }
    void Allocate(int length);
    void Append(byte[] chunk, int receive);
    void Reset();
}

I'm wondering why this interface exposes the TcpClient. Isn't your framework built on top of it? I thought it encapsulates it and provide an alternative API. I think a direct acces to it is not a good idea. Especially that the property is called just Socket which doesn't say anthing.


Other issues

public List<byte> ResultBuffer { get; private set; }

public void Allocate(int length)
{
    ResultBuffer = new List<byte>(length);
}

You have this code in the StateObject where you can Allocate something but this does not really make sence as I can still do ResultBuffer.Clear() on the list or add items by myself and the allocation would be corrupted. You should either return an IEnumerable<byte> or give the user the full control over the property. Having both is confusing because I can modify the list in anyway but set a new one. To replace it I need to call Allocate and then the unexpected happens. The property returns a new list. You need to rethink this.

public void Append(byte[] chunk, int receive)
{
    ResultBuffer.AddRange(chunk.Take(receive));
}

Later there is another method that modifies this list. This is another operation that I could do myself.

You should not let the user modify your internal data. Provide such an API that they are not able to destroy the ResultBuffer or to modify uncontrolled any other state and make it invalid.

\$\endgroup\$
  • \$\begingroup\$ Thanks you very much, that's really valuable input. Have you been able to have a look at the cancellation token usage as well? I'm almost certain i'm re-inventing the wheel here \$\endgroup\$ – HardLuck Jun 29 '17 at 5:12
  • \$\begingroup\$ @HardLuck I'm also curious how well you did there but let's wait for someone else to comment on that part as my knowledge is rather limited in this area ;-) \$\endgroup\$ – t3chb0t Jun 29 '17 at 7:11

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.