11
\$\begingroup\$

After some investigation, I implemented an asynchronous TCP server as per the following example. During my investigation I was unable to find an example that cleanly shuts down the server; after some experimenting I was able to furnish my code with this functionality. I would appreciate a review of the service code for whether I might be doing something stupid/dangerous. (Please note that I have stripped out parameter validation, etc. in order to improve readability).

public class AsyncTcpServer : IDisposable
{
    public class DataReceivedEventArgs : EventArgs
    {
        public NetworkStream Stream { get; private set; }

        public DataReceivedEventArgs(NetworkStream stream)
        {
            Stream = stream;
        }
    }

    public event EventHandler<DataReceivedEventArgs> OnDataReceived;

    public AsyncTcpServer(IPAddress address, int port)
    {
        _listener = new TcpListener(address, port);
    }

    public void Start()
    {
        _listener.Start();
        _isListening = true;
        WaitForClientConnection();
    }

    public void Stop()
    {
        _isListening = false;
        _listener.Stop();
    }

    public void Dispose()
    {
        Stop();
    }

    private void WaitForClientConnection()
    {
        _listener.BeginAcceptTcpClient(HandleClientConnection, _listener);
    }

    private void HandleClientConnection(IAsyncResult result)
    {
        if (!_isListening)
        {
            return;
        }

        var server = result.AsyncState as TcpListener;
        var client = _listener.EndAcceptTcpClient(result);

        WaitForClientConnection();

        OnDataReceived?.Invoke(this, new DataReceivedEventArgs(client.GetStream()));
    }

    private readonly TcpListener _listener;
    private volatile bool _isListening = false;
}

The following test verifies the asynchronous nature of the service (test completes under 10 seconds for 5x client connections blocking for 5 seconds each).

[TestMethod]
public void TestSendReceiveDataFromMultipleClientsConcurrently()
{
    using (var server = new AsyncTcpServer(IPAddress.Any, 54001))
    {
        server.OnDataReceived += (sender, e) =>
        {
            var bytesRead = 0;
            do
            {
                // Read buffer, discarding data
                bytesRead = e.Stream.Read(new byte[1024], 0, 1024);
            }
            while (bytesRead > 0 && e.Stream.DataAvailable);

            // Simulate long running task
            Thread.Sleep(5000);

            var response = Encoding.ASCII.GetBytes("Who's there?");
            e.Stream.Write(response, 0, response.Length);
        };

        server.Start();
        var tasks = new List<Task>();

        for (var i = 0; i < 5; ++i)
        {
            tasks.Add(Task.Run(() =>
            {
                var response = new byte[1024];

                using (var client = new TcpClient())
                {
                    client.Connect("127.0.0.1", 54001);

                    using (var stream = client.GetStream())
                    {
                        var request = Encoding.ASCII.GetBytes("Knock, knock...");
                        stream.Write(request, 0, request.Length);
                        stream.Read(response, 0, response.Length);

                        Assert.AreEqual("Who's there?", Encoding.ASCII.GetString(response).TrimEnd('\0'));
                    }
                }
            }));
        }

        Assert.IsTrue(Task.WaitAll(tasks.ToArray(), 10000));
    }
}
\$\endgroup\$
13
\$\begingroup\$

This is an ugly old pattern. Why don't just try the newer async/await? You already use Task in your tests anyway.

In order to implement it the awaitable way you just need to use a different API, in this case AcceptTcpClientAsync and build everything on top of it. With the CancellationToken you can now better control the server.

public class TcpServer : IDisposable
{
    private readonly TcpListener _listener;
    private CancellationTokenSource _tokenSource;
    private bool _listening;    
    private CancellationToken _token;

    public event EventHandler<DataReceivedEventArgs> OnDataReceived;

    public TcpServer(IPAddress address, int port)
    {
        _listener = new TcpListener(address, port);
    }

    public bool Listening => _listening;

    public async Task StartAsync(CancellationToken? token = null)
    {
        _tokenSource = CancellationTokenSource.CreateLinkedTokenSource(token ?? new CancellationToken());
        _token = _tokenSource.Token;
        _listener.Start();
        _listening = true;

        try
        {
            while (!_token.IsCancellationRequested)
            {   
                await Task.Run(async () =>
                {
                    var tcpClientTask = _listener.AcceptTcpClientAsync();
                    var result = await tcpClientTask;
                    OnDataReceived?.Invoke(this, new DataReceivedEventArgs(result.GetStream()));
                }, _token);
            }
        }
        finally
        {
            _listener.Stop();
            _listening = false;
        }
    }

    public void Stop()
    {
        _tokenSource?.Cancel();
    }

    public void Dispose()
    {
        Stop();
    }
}

EventArgs as a nested class, no no no ;-)


Test

The old test won't work anymore as now you need to make everything async/await.

  • the event handler now becomes async (sender, e)
  • Thread.Sleep is now await Task.Delay(3000);
  • the server needs to run async so you need a Task.Run(async () => {..}
  • at the end you wait for the server with await serverTask;

I used the console for output as I run this in LINQPad and added the thread-id so to see where it runs.

using (var server = new TcpServer(IPAddress.Any, 54001))
{
    server.OnDataReceived += async (sender, e) =>
    {
        var bytesRead = 0;
        do
        {
            // Read buffer, discarding data
            bytesRead = e.Stream.Read(new byte[1024], 0, 1024);
        }
        while (bytesRead > 0 && e.Stream.DataAvailable);

        // Simulate long running task
        Console.WriteLine($"Doing some heavy response processing now. [{Thread.CurrentThread.ManagedThreadId}]");
        await Task.Delay(3000);
        Console.WriteLine($"Finished processing. [{Thread.CurrentThread.ManagedThreadId}]");

        var response = Encoding.ASCII.GetBytes("Who's there?");
        e.Stream.Write(response, 0, response.Length);
    };

    Task.Run(async () =>
    {
        var serverTask = server.StartAsync();

        var tasks = new List<Task>();

        for (var i = 0; i < 5; ++i)
        {
            tasks.Add(Task.Run(() =>
            {
                var response = new byte[1024];

                using (var client = new TcpClient())
                {
                    client.Connect("127.0.0.1", 54001);

                    using (var stream = client.GetStream())
                    {
                        var request = Encoding.ASCII.GetBytes("Knock, knock...");
                        stream.Write(request, 0, request.Length);
                        stream.Read(response, 0, response.Length);

                        //Assert.AreEqual("Who's there?", Encoding.ASCII.GetString(response).TrimEnd('\0'));
                        Console.WriteLine($"Who's there? Echo: " + Encoding.ASCII.GetString(response).TrimEnd('\0') + $" [{Thread.CurrentThread.ManagedThreadId}]");
                    }
                }
            }));
        }

        //Assert.IsTrue(Task.WaitAll(tasks.ToArray(), 10000));
        Console.WriteLine($"IsTrue: " + Task.WaitAll(tasks.ToArray(), 10000));

        await serverTask;
    });

}

\$\endgroup\$
  • \$\begingroup\$ Thanks for the input. I have put the provided code through the same unit test, but the test now runs for 25s, which indicates that the client requests are not processed concurrently. Apart from the class name used in the test, I simply updated the line server.Start() to var serverTask = server.StartAsync(). I am not well versed in using async/await, so the possibility is there that I am doing something wrong. Also, I am curious; why are nested classes considered a bad thing? \$\endgroup\$ – RiaanDP Jan 3 '17 at 7:00
  • \$\begingroup\$ @RiaanDP You were on the right track. I've added an updated example test. \$\endgroup\$ – t3chb0t Jan 3 '17 at 8:02
  • 1
    \$\begingroup\$ @RiaanDP nested classes are hard to find because they're hidden. Most people rely on intellisense and would like to have a quick access to classes etc. If you hide it inside another class you always need to use the parent class to use the nested one. This becomes unnecessarily verbose. Nested classes are ok if they are private. Public nested classes, structs, enums etc. are mostly inconvenient to use. Use namespaces if you want to group something. \$\endgroup\$ – t3chb0t Jan 3 '17 at 8:05
  • \$\begingroup\$ @t3chb0t Why not using ReadAsync, WriteAsync? I am trying to figure out how to do that \$\endgroup\$ – cd491415 Feb 27 at 0:19

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.