Buffered asynchronous writer

Question

Imagine I have connection in which I have to write numbers, represented by this contract:

interface ISocket
{
    Task WriteAsync(Byte[], Int32 offset, Int32 count, CancellationToken cancel);
}

There is one single thread calling this WriteAsync method.

Till now each caller to ISocket was buffering independently and knows when it has to flush, however I would like to centralize that logic. Also I want to make sure that when such buffer is being transmitted, I can take advantage of async/await and release the thread pool thread.

My idea would be to create something like this:

class SocketWriter
{
    private ISocket _socket;
    private Byte[] _buffer; // assume 8K
    private Int32 _cursor;

    [ ... ]

    private async Task FlushCacheIfNotEnoughSpace(Int32 length, CancellationToken cancel)
    {
        if(!FitsInBuffer(length))
        {
            await _socket.WriteAsync(_buffer, 0, _cursor, cancel).ConfigureAwait(false);
            _cursor = 0;
        }
    }

    [ ... ]

    public async Task WriteAsync(UInt64 value, CancellationToken cancel)
    {
        await FlushCacheIfNotEnoughSpace(8, cancel).ConfigureAwait(false);
        CopyToByteArray(value);
    }

    public Task Flush(CancellationToken cancel)
    {
        return FlushCacheIfAny(cancel);
    }
}

There are some memory constraints, so I cannot create more than one of those Byte[] buffers. CopyToByteArray requires enough space in the _buffer and increments _cursor accordingly.

Note that FlushCacheIfNotEnoughSpace only awaits if the given length does not fit the current buffer.

My intention is that when the buffer is not complete and there is space for another number this method call executes synchronously, or properly awaits using IO completion ports otherwise:

await _writer.WriteAsync(1UL, cancel).ConfigureAwait(false);

I wonder if there is a better or more elegant way of doing this, because it seems like a lot of "noise":

await _writer.WriteAsync(obj.Member1, cancel).ConfigureAwait(false);
await _writer.WriteAsync(obj.Member2, cancel).ConfigureAwait(false);
await _writer.WriteAsync(obj.Member3, cancel).ConfigureAwait(false);
await _writer.WriteAsync(obj.Member4, cancel).ConfigureAwait(false);

Answer 1

OK, now it is SOLID :)

Let's do not mix infrastructure and application code. Unfortunatly a lot of stuff is missing in .NET, so infrastructure first.

AsyncStream:

public interface IAsyncStream
{
    Task WriteAsync(
        byte[] buffer, 
        int offset, 
        int count, 
        CancellationToken cancellationToken);        

    Task FlushAsync(CancellationToken cancellationToken);
}

BufferedAsyncStream:

public class BufferedAsyncStream : IAsyncStream
{
    public BufferedAsyncStream(IAsyncStream inner, long bufferSize = 8192)
    {
        Inner = inner;
        Buffer = new byte[bufferSize];
        BufferStream = new MemoryStream(Buffer);
    }

    public async Task WriteAsync(
         byte[] buffer, 
         int offset, 
         int count, 
         CancellationToken cancellationToken)
    {
        if (FreeSpace < count)
            await FlushAsync(cancellationToken)
                .ConfigureAwait(false);

        BufferStream.Write(buffer, offset, count);
    }

    public async Task FlushAsync(CancellationToken cancellationToken)
    {
        await Inner.WriteAsync(Buffer, 0, Allocated, cancellationToken)
            .ConfigureAwait(false);

        await Inner.FlushAsync(cancellationToken)
            .ConfigureAwait(false);

        BufferStream.Seek(0, SeekOrigin.Begin);
    }

    int FreeSpace => (int)BufferStream.Length - (int)BufferStream.Position;
    int Allocated => (int)BufferStream.Position;
    IAsyncStream Inner { get; }
    byte[] Buffer { get; }
    MemoryStream BufferStream { get; }
}

AsyncWriter:

public class AsyncWriter
{
    public AsyncWriter(IAsyncStream stream)
    {
        Stream = stream;
    }

    public Task WriteAsync(ulong value, CancellationToken cancellationToken)
    {
        var buffer = BitConverter.GetBytes(value);
        return Stream.WriteAsync(buffer, 0, buffer.Length, cancellationToken);
    }

    public Task FlushAsync(CancellationToken cancellationToken)
    {
        return Stream.FlushAsync(cancellationToken);
    }

    IAsyncStream Stream { get; }
}

Now application code. ISocket being adapted to IAsyncStream:

class SocketStream : IAsyncStream
{
    public SocketStream(ISocket socket)
    {
        Socket = socket;
    }

    public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
    {
        return Socket.WriteAsync(buffer, offset, count, cancellationToken);
    }

    public Task FlushAsync(CancellationToken cancellationToken)
    {
        return Task.CompletedTask;
    }

    ISocket Socket { get; }        
}

So, we can construct writer in the following way:

var writer = new AsyncWriter(
    new BufferedAsyncStream(
        new SocketStream(socket)));

UPDATE You could use this overload to remove "noise":

public class AsyncWriter
{
     ...
    public async Task WriteAsync(
        CancellationToken cancellationToken, params ulong[] values)
    {
        foreach (var value in values)
            await WriteAsync(value, cancellationToken)
                .ConfigureAwait(false);
    }

Now it is this way:

  await writer
     .WriteAsync(
         cancellationToken, 
         obj.Member1, obj.Member2, obj.Member3, obj.Member4)
     .ConfigureAwait(false);