How to properly read line using sockets through asynchronous callback

Question

I have a library that is connected to some network service using TCP sockets and randomly receives a data from it. I need to process these data line by line and for that I have 2 options:

Create a new thread (I don't want to do that) in which I have never ending loop which calls StreamReader.ReadLine() (I don't want to spawn new threads as this is a library and threads should be fully under control of main program)

Using async callback which gets called every time some data arrives to stream buffer. I currently use this option, but I am having troubles getting lines only. I hacked out this simple solution:

    // This function reset the callback after it's processed
    private void resetCallback()
    {
        if (this.networkStream == null)
            return;
        if (!string.IsNullOrEmpty(this.lineBuffer) && this.lineBuffer.EndsWith("\n"))
        {
            this.processOutput(this.lineBuffer);
            this.lineBuffer = "";
        }
        AsyncCallback callback = new AsyncCallback(OnReceive);
        this.networkStream.BeginRead(buffer, 0, buffer.Length, callback, this.networkStream);
    }

    // This function gets called every time some data arrives to buffer
    private void OnReceive(IAsyncResult data)
    {
        if (this.networkStream == null)
            return;
        int bytes = this.networkStream.EndRead(data);
        string text = System.Text.Encoding.UTF8.GetString(buffer, 0, bytes);
        if (!text.Contains("\n"))
        {
            this.lineBuffer += text;
        }
        else
        {
            List<string> parts = new List<string>(text.Split('\n'));
            while (parts.Count > 0)
            {
                this.lineBuffer += parts[0];
                if (parts.Count > 1)
                {
                    this.processOutput(this.lineBuffer + "\n");
                    this.lineBuffer = "";
                }
                parts.RemoveAt(0);
            }
        }
        this.resetCallback();
    }

As you can see I am using very nasty solution where I am basically checking in every "packet" of data that are received on buffer:

• Whether data in buffer are whole line (ends with new line)
• Whether data in buffer contains more than 1 line (new line is somewhere in middle of data, or there are more than 1 new line)
• Data in buffer contains only a part of a line (no new line in text)

The problem here is that callback function can be called any time when some data are received, and these data can be a line, part of a line, or even multiple lines.

Based on the new line I am storing data in another buffers and when I finally get a new line, I process it somehow.

This is actually working just fine, but I am still wondering if there isn't a better solution that is more clean and doesn't require such a hacking in order to read the stream line by line without using threads?

Answer 1

Please note commenter Damien_The_Unbeliever's point about the issue with partial UTF8 characters. As he says, there's nothing in TCP that would guarantee that you only receive whole characters; a sequence of bytes in the stream can be interrupted at any point, including mid-character.

The usual way to address this would be to using an instance of a Decoder (which you can retrieve from the appropriate Encoding subclass, e.g. Encoding.UTF8.GetDecoder()). A decoder instance will buffer characters for you, returning only whole characters as they are available.

But in your case, there is a much easier way: use the TextReader.ReadLineAsync() method.

For example, here's an asynchronous method which will process each line of text read from the stream, with the returned task for the method completing only when the stream itself has reached the end (i.e. graceful closure of the socket):

async Task ProcessLines()
{
    using (StreamReader reader = new StreamReader(
        this.networkStream, Encoding.UTF8, false, 1024, true))
    {
        string line;

        while ((line = await reader.ReadLineAsync()) != null)
        {
            this.processOutput(line);
        }
    }

    // Clean up here. I.e. send any remaining response to remote endpoint,
    // call Socket.Shutdown(SocketShutdown.Both), and then close the
    // socket.
}

You would call that (preferably awaiting the result in another async method, though that would depend on the exact context of the caller) from wherever you call resetCallback() now. Given the lack of a good, minimal, complete code example a more specific explanation than that can't be provided.

The key is that, being an async method, the method will return as soon as you call ReadLineAsync() (assuming the call doesn't complete immediately), and will resume execution later once that operation completes, i.e. a complete line of text is available and can be returned.

This is the standard idiom now in C# for dealing with this kind of asynchronous operation. It allows you to write the code practically as if you are doing everything synchronously, while the compiler rewrites the code for you to actually implement it asynchronously.

(As an aside: you may want to consider using the usual .NET conventions, i.e. Pascal casing, for method names, instead of the Java-style camel-casing. That will help readers more readily understand your code examples).