reading line-oriented data using C# Async sockets

Question

I am new to C#, but have done TCP and socket programming many years. I would like to create a program that does async I/O (receiving only) from multiple servers. The data is text and is delimited by newlines (very easy).

My question is about whether I can assume that if I use a StreamReader in my async read callback function, whether I can assume that

1. It will be able to get at least a complete line of data, and
2. That it will not somehow block.

The way I understand it, I wantto do a StreamReader ReadLine in my async read callback. But what in the BeginRead would know that I really want to do line-oriented data? I obviously don't know the length to read at that point.

I hope this makes sense, and thanks in advance!

Mitch

Answer 1

Based on my understanding of your question: you're intending to use the StreamReader like a blocking stream, yet you're using asynchronous sockets... even if my understanding is wrong, please still see the suggestions below since they also cover the case where you're really intending to use the StreamReader for asynchronous reading.

1. It will be able to get at least a complete line of data, and
   Yes and no: it will read a complete line of data as long as the underlying buffer has a new line, however, this is an async event so once the StreamReader runs out of data (i.e. it reaches the last new line) it will no longer have anything to read, even if you get another async callback. Basically, you bind your receive buffer to the SocketAsyncEventArgs and when you get a callback, your buffer is filled. What you do with the buffer after that is up to you (you may feed it into a StreamReader to read the line input).

2. That it will not somehow block.
   It won't block, but it won't get all the data you're looking for depending on how you implement your async callback. Which brings me to my next point...

My next point: I don't know if this is a continuous stream of data, or just some line delimited "batch" of data, however, you have several options:

1. Use 1 memory stream and a read/write stream associated with that memory stream. Whenever you get a callback you write the buffer using the StreamWriter and you read the lines using the StreamReader until you can no longer read any new lines.
2. Keep writing the buffer to a stream until you receive no more bytes from the socket (i.e. you get a 0 for the bytes received) and then use that stream to initialize your StreamReader, then read all the lines. This assumes that you have a fixed size of incoming data, rather than a continuous flow of "input".
3. Create a blocking reader and let it live on a separate thread, it only reads when the async read raises a flag of some sort (think ManualResetEvent or something along the lines). It kinda defeats the purpose of asynchronous sockets, but it still lets you take advantage of the benefits. It's a little better than pure blocking sockets, because the blocking is entirely controlled by you.

I would assume that your best option is #1, but the other 2 may not be out of the question depending on what you're trying to do.

On a side note: I've created an async HTTP client, so feel free to take a look at it and scavenge whatever you find useful. I would assume that these two functions are going to be the most important for you:

Begin Receiving:

private void BeginReceive()
{
    if ( _clientState == EClientState.Receiving)
    {
        if (_asyncTask.BytesReceived != 0 && _asyncTask.TotalBytesReceived <= _maxPageSize)
        {
            SocketAsyncEventArgs e = new SocketAsyncEventArgs();
            e.SetBuffer(_asyncTask.ReceiveBuffer, 0, _asyncTask.ReceiveBuffer.Length);
            e.Completed += new EventHandler<SocketAsyncEventArgs>(ReceiveCallback);
            e.UserToken = _asyncTask.Host;

            bool comletedAsync = false;
            try
            {
                comletedAsync = _socket.ReceiveAsync(e);
            }
            catch (SocketException se)
            {
                Console.WriteLine("Error receiving data from: " + _asyncTask.Host);
                Console.WriteLine("SocketException: {0} Error Code: {1}", se.Message, se.NativeErrorCode);

                ChangeState(EClientState.Failed);
            }

            if (!comletedAsync)
            {
                // The call completed synchronously so invoke the callback ourselves
                ReceiveCallback(this, e);
            }
        }
        else
        {
            //Console.WriteLine("Num bytes received: " + _asyncTask.TotalBytesReceived);
            ChangeState(EClientState.ReceiveDone);
        }
    }
}

And the ReceiveCallback:

private void ReceiveCallback(object sender, SocketAsyncEventArgs args)
{
    lock (_sync) // re-entrant lock
    {
        // Fast fail: should not be receiving data if the client
        // is not in a receiving state.
        if (_clientState == EClientState.Receiving)
        {
            String host = (String)args.UserToken;

            if (_asyncTask.Host == host && args.SocketError == SocketError.Success)
            {
                try
                {
                    Encoding encoding = Encoding.ASCII;
                    _asyncTask.BytesReceived = args.BytesTransferred;
                    _asyncTask.TotalBytesReceived += _asyncTask.BytesReceived;

                    // ********************************************
                    // You will need to replace the following line:
                    _asyncTask.DocSource += encoding.GetString(_asyncTask.ReceiveBuffer, 0, _asyncTask.BytesReceived);

                    // Write the contents of the ReceiveBuffer to a Stream using a StreamWriter

                    // Use the StreamReader associated with the same Stream to read the next line(s)
                    // ********************************************
                    BeginReceive();
                }
                catch (SocketException e)
                {
                    Console.WriteLine("Error receiving data from: " + host);
                    Console.WriteLine("SocketException: {0} Error Code: {1}", e.Message, e.NativeErrorCode);

                    ChangeState(EClientState.Failed);
                }
            }
            else if (_asyncTask.Host != host)
            {
                Console.WriteLine("Warning: received a callback for {0}, but the client is currently working on {1}.",
                    host, _asyncTask.Host);
            }
            else
            {
                Console.WriteLine("Socket Error: {0} when receiving from {1}",
                   args.SocketError,
                   _asyncTask.Host);
                ChangeState(EClientState.Failed);
            }
        }
    }
} 

Answer 2

From your post I think that you need to create your own class, and wrap SYNCHRONOUS socket into it. You will Read() the socket in a blocking fashion, buffer the data, split lines and fire up event to the class consumer when whole line is available.

Answer 3

To the best of my knowledge, you won't be quite able to do this with built-in classes.

You can use StreamReader on a NetworkStream to read whole lines, but you will need to code the async part yourself as StreamReader does not expose an async interface. Alternatively, you could use NetworkStream.BeginRead which is async, but then you would have to write the code to make it recognize line-based input.