6
\$\begingroup\$

I'm working on an async socket server, would really appreciate any critique and some advices regarding the place where i process received messages.


My interfaces are as follows:

IAsyncClient

public delegate void ConnectedHandler(IAsyncClient a);

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

public delegate void ClientMessageSubmittedHandler(IAsyncClient a, bool close = false);

public delegate void ClientReceivingStarted();

public delegate void ClientErrorHandler(string errorMessage);

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

    IClientChainsContainer ClientChainsContainer { get; set; }

    event ConnectedHandler Connected;
    event ClientMessageReceivedHandler MessageReceived;
    event ClientMessageSubmittedHandler MessageSubmitted;
    event ClientReceivingStarted ReceivingStarted;
    event ClientErrorHandler Error;

    void InvokeConnected(IAsyncClient a);
    void InvokeMessageReceived(IAsyncClient a, List<byte> msg);
    void InvokeMessageSubmitted(IAsyncClient a, bool close = false);
    void InvokeReceivingStarted();
    void InvokeError(string errorMessage);

    Task StartClient();
    void StartReceiving();
    void SetId(Guid clientId);
    Task<bool> Send(IProcessable message, bool close = false);
    Task<bool> SendSomeCommand();
    Task<bool> SendAlarm();
}

IAsyncSocketListener

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

public delegate void MessageSubmittedHandler(Guid id, bool close);

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

    IServerChainsContainer ServerChainsContainer { get; set; }

    void StartListening();
    bool IsConnected(Guid id);
    Task<bool> Send(Guid id, IProcessable msg, bool close = false);
    Task SendToAll(IProcessable msg, bool close = false);
    Task SendToAllExcept(List<Guid> exludedClientIds, IProcessable msg, bool close = false);
}

IStateObject

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

IProcessable

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

IClientChainContainer

public interface IClientChainsContainer
{
    IChainOfResponsibility<IProcessable> SelfCommandFailsChain { get; }
    IChainOfResponsibility<IProcessable> IncomingCommandsChain { get; }
}

IServerChainContainer

public interface IServerChainsContainer
{
    IChainOfResponsibility<IProcessable> IncomingCommandsChain { get; }
}

Then things related to communication and implementation.
StateObject

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

    private const int BufferSizeInternal = 1024;
    //private const int BufferSizeInternal = 8;

    public StateObject(Socket listener)
    {
        Listener = listener;
        Close = false;
        Reset();
    }

    public StateObject(Socket listener, Guid id)
    {
        Id = id;
        Listener = listener;
        Close = false;
        Reset();
    }

    public Guid Id { get; }
    public bool Close { get; set; }
    public int BufferSize => BufferSizeInternal;
    public byte[] Buffer { get; } = new byte[BufferSizeInternal];
    public byte[] PrefixBuffer { get; private set; } = new byte[sizeof (int)];
    public Socket Listener { 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;
    }
}

SocketClient

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 static readonly object SyncRoot = new object();

    private AutoResetEvent _receivingStarted = new AutoResetEvent(false);

    public IClientChainsContainer ClientChainsContainer { get; set; }

    private bool _close;
    private bool _isInstantiated;
    private bool _isReconnectInProgress;

    private Socket _listener;

    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 ClientMessageSubmittedHandler MessageSubmitted;
    public event ClientReceivingStarted ReceivingStarted;
    public event ClientErrorHandler Error;
    public bool IsConnected
    {
        get
        {
            var poll = _listener.Poll(1000, SelectMode.SelectRead);
            var available = _listener.Available == 0;
            return !(poll && available) && IsInstantiated && !IsReconnectInProgress;
        }
    }

    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 StartClient()
    {
        //Console.WriteLine("Start Client Thread Id: " + Thread.CurrentThread.ManagedThreadId);
        try
        {
            //var retryTask = RetryUtil.Retry(InstantiateClient, 2);
            var retryTask = _retryPolicy.ExecuteAsync(InstantiateClient);
            await retryTask;
        }
        //this catch block works after all retries failed
        catch (SocketException ex)
        {
            HandleStartClientException(ex);
        }
        catch (ObjectDisposedException ex)
        {
            HandleStartClientException(ex);
        }
        catch (AggregateException ex)
        {
            HandleStartClientException(ex);
        }
        catch (Exception ex)
        {
            HandleStartClientException(ex);
        }
    }

    private void HandleStartClientException(Exception ex)
    {
        var message = "StartClient: Could not instantiate client. " + ex.Message;
        Logger.Log.Error(message);
        Error?.Invoke(message);
        DieOrRestart();
    }

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

            try
            {
                Console.WriteLine("Instantiating client");
                var ip = IPAddress.Parse(_host);
                var endpoint = new IPEndPoint(ip, _port);

                var connectedHandler = Connected;

                _listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
                var task = ConnectClientTask(_listener, endpoint);

                await task;

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

                //we don't need to wait for that
                StartReceiving();

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

                await Subscribe();
                connectedHandler?.Invoke(this);
            }
            //this catch block works during retries
            catch (SocketException ex)
            {
                IsInstantiated = false;
                throw;
            }
            catch (ObjectDisposedException ex)
            {
                IsInstantiated = false;
                throw;
            }
            catch (AggregateException ex)
            {
                //polly dies on aggregate exceptions
                IsInstantiated = false;
                throw ex.InnerException;
            }
            finally
            {
                IsReconnectInProgress = false;
            }
        }
    }

    private void Close()
    {
        try
        {
            if (!IsConnected)
            {
                return;
            }

            _listener.Shutdown(SocketShutdown.Both);
            _listener.Close();
        }
        catch (SocketException ex)
        {
            // TODO: handle possible exception
        }
        catch (ObjectDisposedException ex)
        {
            // TODO: handle possible exception
        }
    }

    public void Dispose()
    {
        _receivingStarted.Dispose();
        Close();
    }

    private async Task ConnectClientTask(Socket client, IPEndPoint endpoint)
    {
        await
            Task.Factory.FromAsync((cb, s) => client.BeginConnect(endpoint, cb, null),
                client.EndConnect, null);
    }

    #endregion

    #region Service Methods

    private async void DieOrRestart()
    {
        var isHardMode = false;

        if (isHardMode)
        {
            Console.WriteLine("Hi, I'm dead");
        }

        else
        {
            Console.WriteLine("Soft mode non-exit, processing reconnect");
            await StartClient();
        }
    }

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

    internal void NotifyException(string message)
    {
        var exMessage = "An exception was encountered. " + message;
        Logger.Log.Error(exMessage);
        Error?.Invoke(exMessage);
    }

    internal void NotifyException(Exception ex)
    {
        var exMessage = "An exception was encountered. " + ex.Message;
        Logger.Log.Error(exMessage);
        Error?.Invoke(exMessage);
    }

    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);
    }

    #endregion

    #region Data Receiving

    //fire and forget style task
    public async void StartReceiving()
    {
        if (!IsConnected)
        {
            NotifyException("Receiving not available at the moment. Retrying");

            if (!IsReconnectInProgress)
            {
                await StartClient();
            }

            return;
        }

        try
        {
            Console.WriteLine("Starting receiving pipeline");
            IStateObject state = new StateObject(_listener);

            var sizeReceivingTask = SizeReceivingTask(state);

            _receivingStarted.Set();

            var sizeBytesRead = await sizeReceivingTask;

            await SizeReceivedHandler(sizeBytesRead, state);
        }
        catch (SocketException ex)
        {
            IsInstantiated = false;
            Close();
            await StartClient();
        }
        catch (ObjectDisposedException ex)
        {
            IsInstantiated = false;
            Close();
            await StartClient();
        }
        catch (Exception ex)
        {
            //generic catch
            IsInstantiated = false;
            Close();
            await StartClient();
        }
    }

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

                if (length == 0)
                {
                    var sizeReceivingTask = SizeReceivingTask(state);

                    var sizeBytesRead = await sizeReceivingTask;
                    bytesRead = sizeBytesRead;
                    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)
    {
        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;
            await MessageReceivedHandler(messageBytesRead, state);
        }

        else
        {
            //full message received
            var cmd = state.ResultBuffer.ToArray().ParseCommand();
            ClientChainsContainer.IncomingCommandsChain.Handle(cmd, ClientId);

            MessageReceived?.Invoke(this, state.ResultBuffer);

            state.Reset();

            var sizeReceivingTask = SizeReceivingTask(state);

            var sizeBytesRead = await sizeReceivingTask;
            await SizeReceivedHandler(sizeBytesRead, state);
        }
    }

    private async Task<int> SizeReceivingTask(IStateObject state)
    {
        return await Task.Factory.FromAsync(
            (cb, s) =>
                state.Listener.BeginReceive(state.PrefixBuffer, 0, sizeof(int), SocketFlags.None, cb, s),
            ias => state.Listener.EndReceive(ias)
            , null);
    }

    private async Task<int> MessageReceivingTask(IStateObject state, int expectedChunkSize)
    {
        return await Task.Factory.FromAsync(
            (cb, s) =>
                state.Listener.BeginReceive(state.Buffer, 0, expectedChunkSize, SocketFlags.None, cb, s),
            ias => state.Listener.EndReceive(ias)
            , null);
    }

    #endregion

    #region Data Sending

    public async Task<bool> SendSomeCommand()
    {
        var command = new SomeCommand();

        return await Send(command);
    }

    public async Task<bool> Send(IProcessable message, bool close = false)
    {
        if (!IsConnected)
        {
            NotifyException("Sending not available at the moment. Reconnecting");

            if (!IsReconnectInProgress)
            {
                await StartClient();
            }

            return false;
        }

        try
        {
            var retryTask = _retryPolicy.ExecuteAsync(async () =>
            {
                await InstantiateClient();

                var messageBytes = WrapMessage(message.ToByteArray());
                _close = close;

                var sendingTask = SendTask(_listener, messageBytes);

                await sendingTask;

                MessageSubmitted?.Invoke(this, _close);
            });

            await retryTask;

            return true;
        }
        catch (SocketException ex)
        {
            return HandleSendException(message, ex);
        }
        catch (ObjectDisposedException ex)
        {
            return HandleSendException(message, ex);
        }
        catch (AggregateException ex)
        {
            return HandleSendException(message, ex);
        }
        catch (Exception ex)
        {
            return HandleSendException(message, ex);
        }
    }

    private bool HandleSendException(IProcessable message, Exception ex)
    {
        var exMessage = "Send: Error sending message. " + ex.Message;
        Logger.Log.Error(exMessage);
        IsInstantiated = false;
        Close();
        ClientChainsContainer.SelfCommandFailsChain.Handle(message, ClientId);
        DieOrRestart();
        return false;
    }

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

    private async Task<bool> Subscribe()
    {
        var cmd = new SubscribeCommand { FingerPrint = "Hello" };

        return await Send(cmd);
    }

    private async Task SendTask(Socket listener, byte[] message)
    {
        await
            Task.Factory.FromAsync(
                (cb, s) => listener.BeginSend(message, 0, message.Length, SocketFlags.None, cb, null),
                ias => listener.EndSend(ias), null);
    }

    #endregion

    #region Event Invocators

    public void InvokeMessageReceived(IAsyncClient a, List<byte> msg)
    {
        MessageReceived?.Invoke(a, msg);
    }
    //

    public void InvokeConnected(IAsyncClient a)
    {
        Connected?.Invoke(a);
    }

    public void InvokeMessageSubmitted(IAsyncClient a, bool close)
    {
        MessageSubmitted?.Invoke(a, close);
    }

    public void InvokeReceivingStarted()
    {
        ReceivingStarted?.Invoke();
    }

    public void InvokeError(string errorMessage)
    {
        Logger.Log.Error("Error event invoked with message: " + errorMessage);
        Error?.Invoke(errorMessage);
    }

    #endregion
}

SocketServer

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

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

    private readonly IDictionary<Guid, IStateObject> _clients = new Dictionary<Guid, IStateObject>();

    public event MessageReceivedHandler MessageReceived;
    public event MessageSubmittedHandler MessageSubmitted;

    public IServerChainsContainer ServerChainsContainer { get; set; }

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

    #endregion

    #region Instantiation and Cleanup

    public async void StartListening()
    {
        //var host = Dns.GetHostEntry(string.Empty);
        var ip = IPAddress.Loopback;
        var endpoint = new IPEndPoint(ip, _port);

        try
        {
            using (var listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            {
                listener.Bind(endpoint);
                listener.Listen(_limit);

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

                    IStateObject state;

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

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

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

    private async Task<Socket> AcceptClientTask(Socket listener)
    {
        return await
            Task.Factory.FromAsync((cb, s) => listener.BeginAccept(cb, listener),
                ias => listener.EndAccept(ias), null);
    }

    public void Dispose()
    {
        foreach (var id in _clients.Keys)
        {
            Close(id);
        }
    }

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

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

        try
        {
            state.Listener.Shutdown(SocketShutdown.Both);
            state.Listener.Close();
        }
        catch (SocketException)
        {
            // TODO:
        }
        finally
        {
            lock (_clients)
            {
                _clients.Remove(state.Id);
                Console.WriteLine("Client disconnected with Id {0}", state.Id);
            }
        }
    }

    #endregion

    #region Service Methods

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

        var poll = state.Listener.Poll(1000, SelectMode.SelectRead);
        var available = state.Listener.Available == 0;

        return !(poll && available);
    }

    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 (_clients)
        {
            IStateObject state;
            return _clients.TryGetValue(id, out state) ? state : null;
        }
    }

    #endregion

    #region Data Receiving

    private async void StartReceiving(IStateObject state)
    {
        try
        {
            var sizeBytesRead = await SizeReceivingTask(state);

            await SizeReceivedHandler(sizeBytesRead, 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)
                {
                    var sizeReceivingTask = SizeReceivingTask(state);

                    var sizeBytesRead = await sizeReceivingTask;
                    bytesRead = sizeBytesRead;
                    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)
    {
        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;
            await MessageReceivedHandler(messageBytesRead, state);
        }

        else
        {
            var cmd = state.ResultBuffer.ToArray().ParseCommand();
            ServerChainsContainer.IncomingCommandsChain.Handle(cmd, state.Id);

            MessageReceived?.Invoke(state.Id, state.ResultBuffer);

            state.Reset();

            StartReceiving(state);
        }
    }

    private async Task<int> SizeReceivingTask(IStateObject state)
    {
        return await Task.Factory.FromAsync(
            (cb, s) =>
                state.Listener.BeginReceive(state.PrefixBuffer, 0, sizeof(int), SocketFlags.None, cb, s),
            ias => state.Listener.EndReceive(ias)
            , null);
    }

    private async Task<int> MessageReceivingTask(IStateObject state, int expectedChunkSize)
    {
        return await Task.Factory.FromAsync(
            (cb, s) =>
                state.Listener.BeginReceive(state.Buffer, 0, expectedChunkSize, SocketFlags.None, cb, s),
            ias => state.Listener.EndReceive(ias)
            , null);
    }

    #endregion

    #region Data Sending

    public async Task<bool> Send(Guid id, IProcessable msg, bool close = false)
    {
        var state = GetClient(id);

        return await SendInternal(msg, state, close);
    }

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

        lock (_clients)
        {
            var states = _clients.Values;

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

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

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

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

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

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

    private async Task<bool> SendInternal(IProcessable msg, IStateObject state, bool close = false)
    {
        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());

            state.Close = close;

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

    private async Task SendTask(Socket listener, byte[] message)
    {
        await
            Task.Factory.FromAsync(
                (cb, s) => listener.BeginSend(message, 0, message.Length, SocketFlags.None, cb, null),
                ias => listener.EndSend(ias), null);
    }

    #endregion
}

Aside from the communication part which might not be ideal, have a look at data receiving - here after the full message is received I call ClientChainsContainer.IncomingCommandsChain.Handle(cmd, ClientId);
or
ServerChainsContainer.IncomingCommandsChain.Handle(cmd, state.Id); on client or server respectively. Now for the implementation of these containers and handling - the idea and most sourcecode are taken from here.
IChainOfResponsibility

public interface IChainOfResponsibility<T>
{
    Guid AddHandler(Expression<Func<T, bool>> condition, Action<T, object> handlerAction, string name = "");
    IHandler<T> AddHandlerFluent();
    void RemoveFromChain(Guid handlerGuid);
    void BuildChainInOrder();
    void Handle(T message, object state);
}

IHandler

public interface IHandler<T>
{
    void SetSuccessor(IHandler<T> handler);
    void HandleRequest(T o, object state);
    void SetSpecification(ISpecification<T> specification);

    IHandler<T> And(Func<T, bool> condition);
    IHandler<T> Or(Func<T, bool> condition);
    IHandler<T> Do(Action<T, object> act);
    IHandler<T> WithName(string name);
    IHandler<T> AsFallback();
}

ISpecification

public interface ISpecification<T>
{
    Expression<Func<T, bool>> Expression { get; set; }

    bool IsSatisfiedBy(T o);
}

Approver

public class Approver<T> : IHandler<T>
{
    private Action<T, object> _action;
    private bool _isFallback;
    private string _name;
    private ISpecification<T> _specification;
    private IHandler<T> _successor;

    internal Approver(Action<T, object> action, bool isFallback, string name)
    {
        _name = name;
        _action = action;
        _isFallback = isFallback;
    }

    internal Approver()
    {
        _isFallback = false;
    }

    public void SetSuccessor(IHandler<T> handler)
    {
        _successor = handler;
    }

    public void HandleRequest(T o, object state)
    {
        if (CanHandle(o))
        {
            //o.Process();
            Console.WriteLine("{0}: Request handled by {1}.  ", o, _name);
            _action.Invoke(o, state);
            Console.WriteLine("****************************************");

            return;
        }

        _successor?.HandleRequest(o, state);
    }

    public void SetSpecification(ISpecification<T> specification)
    {
        _specification = specification;
    }

    public bool CanHandle(T o)
    {
        if (_specification == null)
        {
            return false;
        }

        if (_isFallback)
        {
            return _specification.IsSatisfiedBy(o);
        }

        if (o != null)
        {
            return _specification.IsSatisfiedBy(o);
        }

        return false;
    }

    public IHandler<T> And(Func<T, bool> condition)
    {
        if (_specification == null)
        {
            _specification = new Specification<T>(arg => condition(arg));
        }

        else
        {
            _specification.Expression = _specification.Expression.And(arg => condition(arg));
        }

        return this;
    }

    public IHandler<T> Or(Func<T, bool> condition)
    {
        if (_specification == null)
        {
            _specification = new Specification<T>(arg => condition(arg));
        }

        else
        {
            _specification.Expression = _specification.Expression.Or(arg => condition(arg));
        }
        return this;
    }

    public IHandler<T> Do(Action<T, object> action)
    {
        _action = action;
        return this;
    }

    public IHandler<T> WithName(string name)
    {
        _name = name;
        return this;
    }

    public IHandler<T> AsFallback()
    {
        _isFallback = true;
        return this;
    }
}

ChainOfResponsibility

public class ChainOfResponsibility<T> : IChainOfResponsibility<T>
{
    private readonly IDictionary<Guid, IHandler<T>> _handlersChain = new Dictionary<Guid, IHandler<T>>();

    public Guid AddHandler(Expression<Func<T, bool>> condition, Action<T, object> handlerAction, string name = "")
    {
        return AddHandler(condition, handlerAction, false, name);
    }

    public void RemoveFromChain(Guid handlerGuid)
    {
        if (_handlersChain.ContainsKey(handlerGuid))
        {
            _handlersChain.Remove(handlerGuid);
        }
    }

    public void BuildChainInOrder()
    {
        var lastItem = _handlersChain.Last();

        for (var i = 0; i < _handlersChain.Count; i++)
        {
            var currentPair = _handlersChain.ElementAt(i);
            if (currentPair.Key != lastItem.Key)
            {
                var nextPair = _handlersChain.ElementAt(i + 1);
                currentPair.Value.SetSuccessor(nextPair.Value);
            }
        }
    }

    public void Handle(T message, object state)
    {
        _handlersChain.Values.FirstOrDefault()?.HandleRequest(message, state);
    }

    public IHandler<T> AddHandlerFluent()
    {
        var handler = new Approver<T>();
        var guid = Guid.NewGuid();
        _handlersChain.Add(new KeyValuePair<Guid, IHandler<T>>(guid, handler));
        return handler;
    }

    private Guid AddHandler(Expression<Func<T, bool>> condition, Action<T, object> handlerAction, bool isDefault, string name = "")
    {
        var guid = Guid.NewGuid();

        var handler = new Approver<T>(handlerAction, isDefault, name);
        var spec = new Specification<T>(condition);
        handler.SetSpecification(spec);

        _handlersChain.Add(new KeyValuePair<Guid, IHandler<T>>(guid, handler));

        return guid;
    }
}

Specification

public class Specification<T> : ISpecification<T>
{
    public Expression<Func<T, bool>> Expression { get; set; }

    public Specification(Expression<Func<T, bool>> expression)
    {
        if (expression == null)
        {
            throw new ArgumentNullException();
        }

        Expression = expression;
    }

    public bool IsSatisfiedBy(T o)
    {
        if (Expression == null)
        {
            return false;
        }

        var func = Expression.Compile();
        return func(o);
    }
}

PredicateBuilder

public static class PredicateBuilder
{
    public static Expression<Func<T, bool>> True<T>() { return f => true; }
    public static Expression<Func<T, bool>> False<T>() { return f => false; }

    public static Expression<Func<T, bool>> Or<T>(this Expression<Func<T, bool>> expr1,
                                                        Expression<Func<T, bool>> expr2)
    {
        var invokedExpr = Expression.Invoke(expr2, expr1.Parameters);
        return Expression.Lambda<Func<T, bool>>
              (Expression.OrElse(expr1.Body, invokedExpr), expr1.Parameters);
    }

    public static Expression<Func<T, bool>> And<T>(this Expression<Func<T, bool>> expr1,
                                                         Expression<Func<T, bool>> expr2)
    {
        var invokedExpr = Expression.Invoke(expr2, expr1.Parameters);
        return Expression.Lambda<Func<T, bool>>
              (Expression.AndAlso(expr1.Body, invokedExpr), expr1.Parameters);
    }
}

With all these i can configure my client in flexible manner to process different commands the way i want, this is how example container looks like:

public class WinClientChainsContainer : IClientChainsContainer
{
    public IChainOfResponsibility<IProcessable> SelfCommandFailsChain { get; }

    public IChainOfResponsibility<IProcessable> IncomingCommandsChain { get; }

    public WinClientChainsContainer(IAsyncClient client)
    {
        Console.WriteLine("Building chains");
        Logger.InitLogger();
        Logger.Log.Info("Building chains");

        SelfCommandFailsChain = new ChainOfResponsibility<IProcessable>();

        SelfCommandFailsChain.AddHandlerFluent()
            .And(x => x is SubscribeCommand)
            .WithName("Sub sending failed handler")
            .Do((processable, state) => ProcessSubscriptionFailed(processable, client));

        SelfCommandFailsChain.AddHandlerFluent()
            .And(x => x is SomeCommand)
            .WithName("Some sending failed handler")
            .Do((processable, state) => ProcessSomeSendingFailed(processable, client));

        SelfCommandFailsChain.AddHandlerFluent()
            .And(x => true)
            .WithName("Default Fallback Handler")
            .Do((processable, state) => client.InvokeError("Self command handler not found"))
            .AsFallback();

        SelfCommandFailsChain.BuildChainInOrder();

        //

        IncomingCommandsChain = new ChainOfResponsibility<IProcessable>();

        IncomingCommandsChain.AddHandlerFluent()
            .And(x => x is SubscriptionProcessedCommand)
            .And(x => ((SubscriptionProcessedCommand)x).IsApproved)
            .WithName("Sub Command Handler")
            .Do((processable, state) => ProcessSubscriptionCommand(processable, client));

        IncomingCommandsChain.AddHandlerFluent()
            .And(x => x is SubscriptionProcessedCommand)
            .And(x => !((SubscriptionProcessedCommand)x).IsApproved)
            .WithName("Sub Denied Command Handler")
            .Do((processable, state) => ProcessSubscriptionDeniedCommand(processable, client));

        IncomingCommandsChain.AddHandlerFluent()
            .And(x => x is SomeCommand)
            .WithName("SubSome Command Handler")
            .Do((processable, state) => ProcessSomeCommand(processable, client));

        IncomingCommandsChain.AddHandlerFluent()
            .And(x => true)
            .WithName("Default Fallback Handler")
            .Do((processable, state) => client.InvokeError("Received command could not be handled"))
            .AsFallback();

        IncomingCommandsChain.BuildChainInOrder();
    }

    private void ProcessSubscriptionCommand(IProcessable cmd, IAsyncClient client)
    {
        var subCmd = cmd as SubscriptionProcessedCommand;

        if (subCmd?.ClientId != null)
        {
            client.SetId(subCmd.ClientId.Value);
        }
    }

    private void ProcessSomeCommand(IProcessable cmd, IAsyncClient client)
    {
        Console.WriteLine("I'm dead");
    }

    private void ProcessSubscriptionFailed(IProcessable cmd, IAsyncClient client)
    {
        Console.WriteLine("Subscription Sending Failed");
    }

    private void ProcessSomeSendingFailed(IProcessable cmd, IAsyncClient client)
    {
        Console.WriteLine("Some Sending Failed");
    }
}

This is pretty much it, so to recap - any critique is welcome, but two main question are whether the network interaction is done correctly and is there any way to handle incoming data in flexible way? Right now this chain of responsibility "framework" does the trick but seems to be a little bit like an overkill.

\$\endgroup\$
2
\$\begingroup\$

That's more code than I can handle, so I'll just cover some of it.

IAsyncClient exposes too much methods and is way too complex.

IClientChainsContainer ClientChainsContainer { get; set; }

It is my understanding that "client" class should handle the connection. It's job is to keep connection open and send messages to the server. Why does it expose IClientChainsContainer then? Shouldn't some other component, that is not part of the client, be responsible for processing/parsing messages?

event ClientErrorHandler Error;

I think returning Exception instead of a string would be more useful. Also if the outside code needs to be notified about the error, shouldn't you just throw the exception (or let existing exception through)? That's way more common approach.

void InvokeConnected(IAsyncClient a);
void InvokeMessageReceived(IAsyncClient a, List<byte> msg);
void InvokeMessageSubmitted(IAsyncClient a, bool close = false);
void InvokeReceivingStarted();
void InvokeError(string errorMessage);

Why are those methods public? That does not feel right. Anyone from anywhere can call any of those methods in any order. That's sounds like a disaster to me.

Task StartClient();
void StartReceiving();

So in which order should I call those to get things running? Should I call StartClient and then StartReceiving ? Should I await StartClient? You should expose a single method, say, Task<bool> ConnectAsync(), so I can call it to connect to the server and be done with it. Straightforward and simple. Everyting else should be an implementation detail. If StartReceiving reads a single message, then you should change it's signature to Task<byte[]> ReceiveAsync().

void SetId(Guid clientId);

Mutable ID is almost always a bad idea. I would remove this method form the interface and pass id directly to constructor. If you choose to keep this method, might as well make it into a property.

Task<bool> SendSomeCommand();
Task<bool> SendAlarm();

"SomeCommand" ? What is "some command"? You should use meaningful names. And you should not use domain-specific methods in otherwise domain-agnostic API. I would refactor those into extension methods.

That being said, here is the API that looks alright to me:

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

    Task<bool> ConnectAsync();
    Task<bool> DisconnectAsync();

    //close what? connection? something else? needs a better name
    Task<bool> SendAsync(byte[] message, bool close = false);
    Task<byte[]> ReceiveAsync();

    event ConnectedHandler Connected;
    event ClientMessageSubmittedHandler MessageSubmitted;
    event ClientReceivingStarted ReceivingStarted;
    event ClientMessageReceivedHandler MessageReceived;
    event ClientErrorHandler Error;
}

Regarding implementation:

private readonly string _pass;

Hopefully this one is encrypted. Passwords lying around in plain text is a big security risk.

private static readonly object SyncRoot = new object();

Why is it static? Do you really want multiple instances of your client to fight over the same lock? Looks weird.

Your implementation is not thread-safe. That's risky. Synchronizing a single class even with all async's and await's flying around should be way easier, than synchronizing all the other places that are going to use IAsyncClient. However it is definitely possible to design an application in such a way, that thread safety is not going to be an issue.


P.S. You might want to check out Microsoft's TPL.Dataflow library. It does what your IChainOfResponsibility does, but probably better.

\$\endgroup\$

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.