7
\$\begingroup\$

I'm working on a simple async TCP client server application, i'd like my network code to be reviewed. I really feel that I'm doing something wrong, especially in receiving code.

Interface for client:

public delegate void ConnectedHandler(IAsyncClient a);

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

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

public delegate void ClientErrorHandler(string errorMessage);

public interface IAsyncClient : INetworkNode, IDisposable
{
    event ConnectedHandler Connected;
    event ClientMessageReceivedHandler MessageReceived;
    event ClientMessageSubmittedHandler MessageSubmitted;
    event ClientErrorHandler Error;
    void StartClient();
    bool IsConnected { get; }

    Task StartReceiving();
    void Send(IProcessable message, bool close);
}

INetworkNode:

public interface INetworkNode
{
    byte[] WrapMessage(byte[] message);

    byte[] WrapKeepaliveMessage();
}

Wrapper needed for client:

internal class PolicyWrapper
{
    internal PolicyResult InitiateRetryPolicy(int retryCount, int retyDelaySeconds, Action executionTarget,
        Action<Exception> internalExceptionHandler)
    {
        var policyResult = Policy
            .Handle<SocketException>()
            //.Or<ArgumentException>(ex => ex.ParamName == "example")
            .WaitAndRetry(retryCount,
                i => TimeSpan.FromSeconds(retyDelaySeconds),
                (exception, span) => internalExceptionHandler(exception))
            .ExecuteAndCapture(executionTarget);

        return policyResult;
    }

    public void InitiateEndlessRetryPolicy(int retyDelaySeconds, Action executionTarget,
        Action<Exception> internalExceptionHandler)
    {
        Policy
            .Handle<SocketException>()
            //.Or<ArgumentException>(ex => ex.ParamName == "example")
            .WaitAndRetryForever(i => TimeSpan.FromSeconds(retyDelaySeconds),
                (exception, span) => internalExceptionHandler(exception)).Execute(executionTarget);
    }
}

Protocol specification:

public class FramedPacketProtocol : IProtocol
{
    /// <summary>
    ///     Wraps a message. The wrapped message is ready to send to a stream.
    /// </summary>
    /// <remarks>
    ///     <para>Generates a length prefix for the message and returns the combined length prefix and message.</para>
    /// </remarks>
    /// <param name="message">The message to send.</param>
    public 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 ret = new byte[lengthPrefix.Length + message.Length];
        lengthPrefix.CopyTo(ret, 0);
        message.CopyTo(ret, lengthPrefix.Length);

        return ret;
    }

    /// <summary>
    ///     Wraps a keepalive (0-length) message. The wrapped message is ready to send to a stream.
    /// </summary>
    public byte[] WrapKeepaliveMessage()
    {
        return BitConverter.GetBytes(0);
    }        
}

StateObject:

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

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

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

    public int 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;
    }
}

The core of it all, where I'm sure I've made lots of mistakes:

public sealed class StreamSecurityClient : IAsyncClient
{
    private const ushort Port = 8080;
    private static readonly object SyncRoot = new object();
    private readonly ManualResetEvent _connected = new ManualResetEvent(false);
    private readonly IProtocol _protocol;
    private readonly ManualResetEvent _received = new ManualResetEvent(false);
    private readonly ManualResetEvent _sent = new ManualResetEvent(false);
    private bool _chainsBuilt;
    private bool _close;
    private ChainOfResponsibility<IProcessable> _incomingCommandChain;
    private bool _isInstantiated;
    private bool _isReconnectInProgress;
    private Socket _listener;
    private ChainOfResponsibility<IProcessable> _selfCommandFailsChain;

    public StreamSecurityClient(IProtocol protocol)
    {
        IsInstantiated = false;
        _protocol = protocol;
    }

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

    private bool ChainsBuilt
    {
        get
        {
            lock (SyncRoot)
            {
                return _chainsBuilt;
            }
        }
        set
        {
            lock (SyncRoot)
            {
                _chainsBuilt = 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 ClientErrorHandler Error;
    public bool IsConnected => _listener.Connected && IsInstantiated && !IsReconnectInProgress;

    public void StartClient()
    {
        if (!IsInstantiated || !IsConnected)
        {
            IsReconnectInProgress = true;

            try
            {
                BuildChains();

                var wrapper = new PolicyWrapper();

                //wrapper.InitiateEndlessRetryPolicy(5, InstantiateClient, NotifyException);
                var res = wrapper.InitiateRetryPolicy(3, 5, InstantiateClient, NotifyException);

                if (res.Outcome != OutcomeType.Successful)
                {
                    throw res.FinalException.InnerException;
                }

                IsInstantiated = true;
            }

                //this catch block works after all retries failed
            catch (SocketException ex)
            {
                Console.WriteLine("Hard mode exc after retries, dead");
            }

            catch (ObjectDisposedException ex)
            {
                Console.WriteLine("Hard mode exc after retries, dead");
            }

            finally
            {
                IsReconnectInProgress = false;
            }
        }
    }

    public void Dispose()
    {
        _connected.Dispose();
        _sent.Dispose();
        _received.Dispose();
        Close();
    }

    public byte[] WrapMessage(byte[] message)
    {
        return _protocol.WrapMessage(message);
    }

    public byte[] WrapKeepaliveMessage()
    {
        return _protocol.WrapKeepaliveMessage();
    }

    private void BuildChains()
    {
        if (!ChainsBuilt)
        {
            _selfCommandFailsChain = new ChainOfResponsibility<IProcessable>();

            _selfCommandFailsChain.AddHandler(x => x is SubscribeCommand,
                processable => PassHandling(processable, new SubscriptionFailedCommandProcessor()));

            _selfCommandFailsChain.AddHandler(x => x is DeathCommand,
                processable => PassHandling(processable, new DeathCommandProcessor()));

            _selfCommandFailsChain.BuildChainInOrder();

            //

            _incomingCommandChain = new ChainOfResponsibility<IProcessable>();

            _incomingCommandChain.AddHandler(x => x is DeathCommand,
                processable => PassHandling(processable, new DeathCommandProcessor()));

            _incomingCommandChain.BuildChainInOrder();

            //

            ChainsBuilt = true;
        }
    }

    internal void PassHandling(IProcessable command, ICommandProcessor processor)
    {
        processor.Handle(command, this);
    }

    private async void InstantiateClient()
    {
        try
        {
            //var host = Dns.GetHostEntry(string.Empty);
            var ip = IPAddress.Loopback;
            var endpoint = new IPEndPoint(ip, Port);

            var connectedHandler = Connected;

            Console.WriteLine("Trying to connect");
            _listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            var task = ConnectClientTask(_listener, endpoint);

            await task;

            task.ContinueWith(async cnt =>
            {
                if (cnt.Exception != null)
                {
                    //as this is a task we should pull the InnerException of AggregateException
                    throw cnt.Exception.InnerException;
                }

                Console.WriteLine("Connected");
                IsInstantiated = true;
                _connected.Set();
                await StartReceiving();
                Subscribe();
                connectedHandler?.Invoke(this);
            }, TaskContinuationOptions.None).Wait();
        }
            //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;
        }
    }

    private void Subscribe()
    {
        var cmd = new SubscribeCommand {FingerPrint = "Hello"};
        Send(cmd);
    }

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

    internal void NotifyException(string message)
    {
        Console.WriteLine("An exception was encountered. " + message);
    }

    internal void NotifyException(Exception ex)
    {
        Console.WriteLine("An exception was encountered. " + ex.Message);
    }

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

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

    #region StartReceiving data

    public async Task StartReceiving()
    {
        if (!IsConnected || !IsInstantiated)
        {
            NotifyException("Receiving not available at the moment");

            if (!IsReconnectInProgress)
            {
                StartClient();
            }

            return;
        }

        try
        {
            var state = new StateObject(_listener);

            var sizeReceivingTask = SizeReceivingTask(state);

            //TODO: this task restarts itself. There should be a flag to stop this process
            await
                sizeReceivingTask.ContinueWith(cnt => { SizeReceivedHandler(cnt, state); },
                    TaskContinuationOptions.None);
        }
        catch (SocketException ex)
        {
            IsInstantiated = false;
            Close();
            StartClient();
        }
        catch (ObjectDisposedException ex)
        {
            IsInstantiated = false;
            Close();
            StartClient();
        }
    }

    private async void SizeReceivedHandler(Task<int> cnt, StateObject state)
    {
        if (cnt.Exception != null)
        {
            //as this is a task we can should pull the InnerException of AggregateException
            throw cnt.Exception.InnerException;
        }

        var receive = cnt.Result;

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

            state.ExpectedLength = length;

            await MessageReceivingTask(state, state.Buffer.Length)
                .ContinueWith(task => MessageReceivedHandler(task, state));
        }
    }

    private async void MessageReceivedHandler(Task<int> cnt, StateObject state)
    {
        if (cnt.Exception != null)
        {
            //as this is a task we can should pull the InnerException of AggregateException
            throw cnt.Exception.InnerException;
        }

        var receive = cnt.Result;

        if (receive > 0)
        {
            //size received

            state.Append(state.Buffer, receive);
        }

        if (state.ResultBuffer.Count < state.ExpectedLength)
        {
            var nextChunk = Math.Min(state.BufferSize, state.ExpectedLength - state.ResultBuffer.Count);
            var messageReceivingTask = MessageReceivingTask(state, nextChunk);
            await messageReceivingTask.ContinueWith(task => MessageReceivedHandler(task, state));
        }

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

            state.Reset();

            var sizeReceivingTask = SizeReceivingTask(state);

            await sizeReceivingTask.ContinueWith(sizeTask => { SizeReceivedHandler(sizeTask, state); },
                TaskContinuationOptions.None);
        }
    }

    private async Task<int> SizeReceivingTask(StateObject state)
    {
        return await MessageReceivingTask(state, sizeof (int));
    }

    private async Task<int> MessageReceivingTask(StateObject 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 Send data

    public void SendDeathCommand()
    {
        var command = new DeathCommand();

        //this is either endless or leads to death
        StartClient();

        Send(command);
    }

    public void Send(IProcessable message, bool close = false)
    {
        if (!IsConnected || !IsInstantiated)
        {
            NotifyException("Sending not available at the moment");

            if (!IsReconnectInProgress)
            {
                StartClient();
            }

            return;
        }

        try
        {
            var wrapper = new PolicyWrapper();
            wrapper.InitiateRetryPolicy(3, 5, () =>
            {
                StartClient();

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

                var sendingTask = SendTask(_listener, messageBytes);

                sendingTask.ContinueWith(cnt =>
                {
                    MessageSubmitted?.Invoke(this, _close);

                    _sent.Set();
                });
            }, NotifyException);
        }
        catch (SocketException ex)
        {
            IsInstantiated = false;
            Close();
            _selfCommandFailsChain.Handle(message);
            StartClient();
        }
        catch (ObjectDisposedException ex)
        {
            IsInstantiated = false;
            Close();
            _selfCommandFailsChain.Handle(message);
            StartClient();
        }
        catch (AggregateException ex)
        {
            //polly dies on aggregate exceptions
            IsInstantiated = false;
            Close();
            _selfCommandFailsChain.Handle(message);
            StartClient();
            //throw ex.InnerException;
        }
    }

    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
}

Chain of responsibility:

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

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

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

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

        return guid;
    }

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

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

        foreach (var keyHandlerPair in _handlersChain)
        {
            if (keyHandlerPair.Key != lastItem.Key)
            {
                var handler = keyHandlerPair.Value;
                handler.SetSuccessor(handler);
            }
        }
    }

    public void BuildChainInOrderWithDefault(Action<T> handlerAction)
    {
        AddHandler(x => true, handlerAction, "Default Fallback Handler");

        BuildChainInOrder();
    }

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

Approver:

public class Approver<T> : IHandler<T> where T : IProcessable
{
    private readonly Action<T> _action;
    private readonly string _name;
    private ISpecification<T> _specification;
    private IHandler<T> _successor;

    public Approver(string name, Action<T> action)
    {
        _name = name;
        _action = action;
    }

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

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

        _successor?.HandleRequest(o);
    }

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

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

public interface IHandler<T> where T : IProcessable
{
    void SetSuccessor(IHandler<T> handler);
    void HandleRequest(T o);
    void SetSpecification(ISpecification<T> specification);
}

Specification:

public interface ISpecification<in T>
{
    bool IsSatisfiedBy(T o);
}

public class Specification<T> : ISpecification<T>
{
    private readonly Func<T, bool> _expression;

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

    public bool IsSatisfiedBy(T o)
    {
        return _expression(o);
    }
}

public static class SpecificationExtensions
{
    public static Specification<T> And<T>(this ISpecification<T> left, ISpecification<T> right)
    {
        if (left != null && right != null)
        {
            return new Specification<T>(o => left.IsSatisfiedBy(o) && right.IsSatisfiedBy(o));
        }
        return null;
    }

    public static Specification<T> Or<T>(this ISpecification<T> left, ISpecification<T> right)
    {
        if (left != null && right != null)
        {
            return new Specification<T>(o => left.IsSatisfiedBy(o) || right.IsSatisfiedBy(o));
        }
        return null;
    }

    public static Specification<T> Not<T>(this ISpecification<T> left)
    {
        if (left != null)
        {
            return new Specification<T>(o => !left.IsSatisfiedBy(o));
        }
        return null;
    }
}

Related to command handling:

internal class DeathCommandFailedProcessor : ICommandProcessor
{
    public void Handle(IProcessable command, StreamSecurityClient client)
    {
        client.NotifyException("Death command failed");
    }
}

internal interface ICommandProcessor
{
    void Handle(IProcessable command, StreamSecurityClient client);
}

internal class DeathCommandProcessor : ICommandProcessor
{
    public void Handle(IProcessable command, StreamSecurityClient client)
    {
        client.Die();
    }
}

internal static class ClientExtensions
{
    public static void Die(this StreamSecurityClient client)
    {
        Console.WriteLine("Dead");
    }
}
Improve this question
\$\endgroup\$
6
  • \$\begingroup\$ Does it work as expected? \$\endgroup\$
    – pacmaninbw
    Aug 6, 2016 at 18:23
  • \$\begingroup\$ It does work as expected indeed, I just think I'm doing something wrong with tasks created from async socket methods and re-inventing the wheel \$\endgroup\$
    – HardLuck
    Aug 6, 2016 at 18:26
  • \$\begingroup\$ I can't read all the code now, but one thing I notice is that you "merge" two byte[] one being the message length and the other the real message. Why not send one packet with the length, and then the message? Since it's TCP it shouldn't be a problem. \$\endgroup\$
    – null
    Aug 6, 2016 at 18:40
  • \$\begingroup\$ Just to send it as one message, nothing more \$\endgroup\$
    – HardLuck
    Aug 6, 2016 at 18:49
  • \$\begingroup\$ But TCP doesn't work as "messages" it's a data stream, to me it makes more sense to do it this way. Also the operation of "merging" takes performance, especially if you do it many times. \$\endgroup\$
    – null
    Aug 7, 2016 at 9:52

1 Answer 1

Reset to default
6
+100
\$\begingroup\$

Async/Await

Async/await's main advantage is that you no longer need to use ContinueWith() to chain tasks, making the code much cleaner. Your code seems to more often favor the use of ContinueWith(). A good book or tutorial on this will help you refactor that part of the code.

For example, this code:

public Task<int> DoSomethingAsync()
{
    return Task.Factory.StartNew(() =>
    {
        //...
        return 1;
    });
}

var task = DoSomethingAsync();
task.ContinueWith(value =>
{
     Console.WriteLine("{0}", value);
});

Becomes

public Task<int> DoSomethingAsync()
{
    return Task.Factory.StartNew(() =>
    {
        //...
        return 1;
    });
}

var value = await DoSomethingAsync();
Console.WriteLine("{0}", value);

Event Notification

In the spirit of asynchronicity, I would rather expose some events which would notify interested parties that something has been sent/received than making them wait on ManualResetEvents. Later I noticed that you also use events. In this case, why did you use the ManualResetEvent? As for the delegate type of the events, using EventHandler<> specializations makes the code more idiomatic.

Single Responsibility Principle

A short look upon StreamSecurityClient and one immediately notices that this class is doing a lot. For example, the client shouldn't be concerned in how a command is handled.

The socket notifies you that some bytes have arrived over the wire. A protocol decoder receives all the bytes and splits them into meaningful messages (as mentioned in a comment, a stream of bytes doesn't mean that you receive a full message every time, it's your task to put together ["He", "llo W", "orld"] into ["Hello", "World"]). After you have a specific message, you pass it on to whatever handling mechanism you choose.

Over-engineering

I can't help noticing the large number of patterns you employ. Are you learning design-patterns? Simply throwing in all the patterns one can think off can easily lead to code that is unnecessarily complex, prepared to solve issues that might never arise at the expense of being harder to follow: YAGNI

Improve this answer
\$\endgroup\$
5
  • \$\begingroup\$ Thanks for your input. Could you please put some examples of code improvements if possible based on the recommendations you gave? Especially for async/await sections. Regarding the patterns - no, just trying to be prepared for different kinds of situations, it could be over-engineering indeed. \$\endgroup\$
    – HardLuck
    Aug 8, 2016 at 20:12
  • 1
    \$\begingroup\$ Updated the answer with an example \$\endgroup\$
    – D. Jurcau
    Aug 8, 2016 at 20:20
  • \$\begingroup\$ and for example for the receiving task that should restart itself? Could i please have an example for this as well? \$\endgroup\$
    – HardLuck
    Aug 8, 2016 at 20:52
  • \$\begingroup\$ The main problem with it is Fire and Forget mode that should be used for receiving (I guess) \$\endgroup\$
    – HardLuck
    Aug 8, 2016 at 23:39
  • 1
    \$\begingroup\$ Solved Fire and Forget by using async void methods. Now my code looks better and cleaner \$\endgroup\$
    – HardLuck
    Aug 9, 2016 at 21:57

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

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