Sockets in C#, how can I asynchronously read and write data through a NetworkStream

Question

[I am limited to Visual Studio 2010, and therefore, C# 4. async and await are not available to me.]

I'm working on network architecture for a project of mine, that sends packets of data over a network between a server and a client, but the client and server must continue running while waiting, so the code has to be non-blocking, so I thought to use the asynchronous methods. However, except for simple synchronous one-time IO, I don't know much about what to do, especially when using a NetworkStream. What I'm trying to do is:

1) Client connects to server

2) Server accepts connection

3) Server waits for data from client

4) Server processes data

5) Server responds to client

6) While connection is open, repeat from 3.

And I would like to use a NetworkStream to wrap the socket. But I am new to asynchronous I/O and i'm not really sure how to do this without blocking other parts of the server/client code when waiting for a response, especially with a NetworkStream. In my research I see examples that use something like this:

while(true){
    socket.BeginAccept(new AsyncCallback(AcceptCallback), socket );
}

But it seems like that loop would still hold up the application. Can anyone give me some pointers (ha) on how exactly to do this? I haven't been able to find many examples that keep the connection open, only Client Connect -> Client Send -> Server Recieve -> Server Send -> Disconnect. I'm not asking for full code, just the general idea with a few snippets.

Solution

Here is a very simple example of using async/await with a NetworkStream:

SocketServer.cs:

class SocketServer
{
    private readonly Socket _listen;

    public SocketServer(int port)
    {
        IPEndPoint listenEndPoint = new IPEndPoint(IPAddress.Loopback, port);
        _listen = new Socket(SocketType.Stream, ProtocolType.Tcp);
        _listen.Bind(listenEndPoint);
        _listen.Listen(1);
        _listen.BeginAccept(_Accept, null);
    }

    public void Stop()
    {
        _listen.Close();
    }

    private async void _Accept(IAsyncResult result)
    {
        try
        {
            using (Socket client = _listen.EndAccept(result))
            using (NetworkStream stream = new NetworkStream(client))
            using (StreamReader reader = new StreamReader(stream))
            using (StreamWriter writer = new StreamWriter(stream))
            {
                Console.WriteLine("SERVER: accepted new client");

                string text;

                while ((text = await reader.ReadLineAsync()) != null)
                {
                    Console.WriteLine("SERVER: received \"" + text + "\"");
                    writer.WriteLine(text);
                    writer.Flush();
                }
            }

            Console.WriteLine("SERVER: end-of-stream");

            // Don't accept a new client until the previous one is done
            _listen.BeginAccept(_Accept, null);
        }
        catch (ObjectDisposedException)
        {
            Console.WriteLine("SERVER: server was closed");
        }
        catch (SocketException e)
        {
            Console.WriteLine("SERVER: Exception: " + e);
        }
    }
}

Program.cs:

class Program
{
    private const int _kport = 54321;

    static void Main(string[] args)
    {
        SocketServer server = new SocketServer(_kport);
        Socket remote = new Socket(SocketType.Stream, ProtocolType.Tcp);
        IPEndPoint remoteEndPoint = new IPEndPoint(IPAddress.Loopback, _kport);

        remote.Connect(remoteEndPoint);

        using (NetworkStream stream = new NetworkStream(remote))
        using (StreamReader reader = new StreamReader(stream))
        using (StreamWriter writer = new StreamWriter(stream))
        {
            Task receiveTask = _Receive(reader);
            string text;

            Console.WriteLine("CLIENT: connected. Enter text to send...");

            while ((text = Console.ReadLine()) != "")
            {
                writer.WriteLine(text);
                writer.Flush();
            }

            remote.Shutdown(SocketShutdown.Send);
            receiveTask.Wait();
        }

        server.Stop();
    }

    private static async Task _Receive(StreamReader reader)
    {
        string receiveText;

        while ((receiveText = await reader.ReadLineAsync()) != null)
        {
            Console.WriteLine("CLIENT: received \"" + receiveText + "\"");
        }

        Console.WriteLine("CLIENT: end-of-stream");
    }
}

It's a very simple example, hosting both the server and client in the same process and accepting just one connection at a time. It's really just for illustration purposes. Real-world scenarios will no doubt include other features to suit their needs.

Here, I'm wrapping the NetworkStreams in StreamReaders and StreamWriters. Note that you have to call Flush() to ensure that the data is actually sent. For better control over the I/O, you can of course use the NetworkStream directly. Just use the Stream.ReadAsync() method instead of StreamReader.ReadLineAsync(). Note also that in my example, writing is synchronous. You can make this asynchronous as well if you like, using the same basic technique as shown for reading.

EDIT:

The OP indicates they are unable to use async/await. Here is a version of the client which uses NetworkStream and the old-style Begin/EndXXX() API (similar changes would be made to the server of course):

using System;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;

namespace TestOldSchoolNetworkStream
{
    class Program
    {
        private const int _kport = 54321;

        static void Main(string[] args)
        {
            SocketServer server = new SocketServer(_kport);
            Socket remote = new Socket(SocketType.Stream, ProtocolType.Tcp);
            IPEndPoint remoteEndPoint = new IPEndPoint(IPAddress.Loopback, _kport);

            remote.Connect(remoteEndPoint);

            using (NetworkStream stream = new NetworkStream(remote))
            {
                // For convenience, These variables are local and captured by the
                // anonymous method callback. A less-primitive implementation would
                // encapsulate the client state in a separate class, where these objects
                // would be kept. The instance of this object would be then passed to the
                // completion callback, or the receive method itself would contain the
                // completion callback itself.
                ManualResetEvent receiveMonitor = new ManualResetEvent(false);
                byte[] rgbReceive = new byte[8192];
                char[] rgch = new char[Encoding.UTF8.GetMaxCharCount(rgbReceive.Length)];
                Decoder decoder = Encoding.UTF8.GetDecoder();
                StringBuilder receiveBuffer = new StringBuilder();

                stream.BeginRead(rgbReceive, 0, rgbReceive.Length, result =>
                {
                    _Receive(stream, rgbReceive, rgch, decoder, receiveBuffer, receiveMonitor, result);
                }, null);

                string text;

                Console.WriteLine("CLIENT: connected. Enter text to send...");

                while ((text = Console.ReadLine()) != "")
                {
                    byte[] rgbSend = Encoding.UTF8.GetBytes(text + Environment.NewLine);

                    remote.BeginSend(rgbSend, 0, rgbSend.Length, SocketFlags.None, _Send, Tuple.Create(remote, rgbSend.Length));
                }

                remote.Shutdown(SocketShutdown.Send);
                receiveMonitor.WaitOne();
            }

            server.Stop();
        }

        private static void _Receive(NetworkStream stream, byte[] rgb, char[] rgch, Decoder decoder, StringBuilder receiveBuffer, EventWaitHandle monitor, IAsyncResult result)
        {
            try
            {
                int byteCount = stream.EndRead(result);
                string fullLine = null;

                if (byteCount > 0)
                {
                    int charCount = decoder.GetChars(rgb, 0, byteCount, rgch, 0);

                    receiveBuffer.Append(rgch, 0, charCount);

                    int newLineIndex = IndexOf(receiveBuffer, Environment.NewLine);

                    if (newLineIndex >= 0)
                    {
                        fullLine = receiveBuffer.ToString(0, newLineIndex);
                        receiveBuffer.Remove(0, newLineIndex + Environment.NewLine.Length);
                    }

                    stream.BeginRead(rgb, 0, rgb.Length, result1 =>
                    {
                        _Receive(stream, rgb, rgch, decoder, receiveBuffer, monitor, result1);
                    }, null);
                }
                else
                {
                    Console.WriteLine("CLIENT: end-of-stream");
                    fullLine = receiveBuffer.ToString();
                    monitor.Set();
                }

                if (!string.IsNullOrEmpty(fullLine))
                {
                    Console.WriteLine("CLIENT: received \"" + fullLine + "\"");
                }
            }
            catch (IOException e)
            {
                Console.WriteLine("CLIENT: Exception: " + e);
            }
        }

        private static int IndexOf(StringBuilder sb, string text)
        {
            for (int i = 0; i < sb.Length - text.Length + 1; i++)
            {
                bool match = true;

                for (int j = 0; j < text.Length; j++)
                {
                    if (sb[i + j] != text[j])
                    {
                        match = false;
                        break;
                    }
                }

                if (match)
                {
                    return i;
                }
            }

            return -1;
        }

        private static void _Send(IAsyncResult result)
        {
            try
            {
                Tuple<Socket, int> state = (Tuple<Socket, int>)result.AsyncState;
                int actualLength = state.Item1.EndSend(result);

                if (state.Item2 != actualLength)
                {
                    // Should never happen...the async operation should not complete until
                    // the full buffer has been successfully sent, 
                    Console.WriteLine("CLIENT: send completed with only partial success");
                }
            }
            catch (IOException e)
            {
                Console.WriteLine("CLIENT: Exception: " + e);
            }
        }
    }
}

Note that this code, even in spite of leaving out a bunch of exception-handling logic, is considerably longer, at least in part due to the fact that TextReader has no built-in asynchronous API, and so the processing of the input data is much more verbose here. Of course, this is for a simple line-based text exchange protocol. Other protocols may be more or less complex in terms of the data-unpacking aspects, but the underlying read and write elements of the NetworkStream would be the same.

Comments

[AlphaModder]

As good as this answer is, I am limited to VS2010, and consequentely, C# 4. So I can't use async or await. I'll add this to my question.

[gabba]

Is it serves only one connection?

[Peter Duniho]

Yes...this example only allows one connection at a time. However, that's just to keep the example simpler. It's easy enough to change this example to support more than one connection: simply move the BeginAccept() call in SocketServer._Accept() to before the loop reading from the current socket. Then it will allow multiple connections concurrently.

[Peter Duniho]

@AlphaMCubed: note the earlier comment directing you to a link showing how to enable async/await in VS2010/.NET 4. That said, in case that still doesn't work for you, I've added a version of the client to the answer. I don't think this is considerably different from other examples you might find lying around, but I hope it's still useful. Feel free to post a new question if you have trouble with specific implementation issues.

[AlphaModder]

@PeterDuniho Thanks, I didn't see that. I will try it now.