NTDLS.StreamFraming 1.2.15

NTDLS.StreamFraming

NTDLS.StreamFraming is a set of extension methods for Stream (typically TCP/IP/NetworkStream) that enables reliable framing, compression, optional encryption, two-way communication and support for asynchronous query/reply. Messages are guaranteed to be received in their entirety and in the order in which they were sent.

📦 Be sure to check out the NuGet package: https://www.nuget.org/packages/NTDLS.StreamFraming

When would you want to use this?

TCP/IP streams do not preserve message boundaries: bytes written by one side can arrive fragmented, or concatenated with other messages on the receiving side. If you are building a client/server application over sockets and need to send structured messages reliably, NTDLS.StreamFraming handles all of the plumbing for you:

• Message reassembly - fragments are buffered and combined back into complete frames, even when multiple messages arrive in a single read.
• Integrity checking - every frame carries a CRC16 checksum; corrupt frames are detected and skipped.
• Compression - frame payloads are compressed before being written to the stream.
• Optional encryption - supply an IStreamFramingEncryptionProvider to encrypt payloads before framing and decrypt them after receipt.
• Two communication patterns - fire-and-forget notifications and request/response queries with replies routed back to the waiting caller.
• Typed payloads - your payload classes are serialized automatically and reconstructed on the other end, so your handlers receive the original objects.

Usage examples

The examples below use an established TcpClient connection. The payload classes used throughout are shown at the end of this section.

Sending a notification frame

using var tcpStream = tcpClient.GetStream();

//Assemble a message frame and write it to the stream:
tcpStream.WriteNotificationFrame(new MyMessage("Hello!"));

WriteNotificationFrame() serializes the payload, compresses it, wraps it in a frame header (delimiter, size, CRC16) and writes it to the stream. It is a fire-and-forget operation - no reply is expected.

Receiving a notification frame

var frameBuffer = new FrameBuffer();

using var tcpStream = tcpClient.GetStream();

while (tcpClient.Connected)
{
    //Read from the stream, assemble the bytes into the original messages and call
    //  the handler for each message that was received.
    if (!tcpStream.ReadAndProcessFrames(frameBuffer, ProcessFrameNotification))
    {
        break; //ReadAndProcessFrames() returns false when we have been disconnected.
    }
}

//Handler for a frame written with WriteNotificationFrame().
//Note that the original class is received and we use pattern matching to determine what we are receiving.
private static void ProcessFrameNotification(IFramePayloadNotification payload)
{
    //We received a message, see if it is of the type "MyMessage".
    if (payload is MyMessage myMessage)
    {
        Console.WriteLine($"Received [Notification]: '{myMessage.Text}'");
    }
}

ReadAndProcessFrames() reads whatever bytes are available, determines whether they make up a full frame, only a fragment, or a concatenation of multiple frames and fragments. Any full frames are validated, decompressed, deserialized and passed to the callback. A FrameBuffer instance must be reused across calls so that partial frames can be completed as more data arrives.

Sending a query frame and awaiting a reply

using var tcpStream = tcpClient.GetStream();

//Send a query and await the typed reply asynchronously:
MyQueryReply reply = await tcpStream.WriteQueryFrameAsync<MyQueryReply>(new MyQuery("Hello client!"));
Console.WriteLine($"Received [QueryReply] from client: '{reply.Text}'");

//In this example, we still have to call ReadAndProcessFrames() even though we are not
//  supplying it with any callbacks because it receives the replies from the query
//  that is sent above and routes them to the correct waiting query task.
if (!tcpStream.ReadAndProcessFrames(frameBuffer))
{
    break; //The peer disconnected.
}

WriteQueryFrameAsync<T>() writes the query and returns a Task<T> that completes when the peer's reply arrives (or throws when the timeout expires). There is also a non-async overload, WriteQueryFrame<T>(), which returns a Task<T> without blocking the calling thread.

Receiving a query frame and replying

var frameBuffer = new FrameBuffer();

using var tcpStream = tcpClient.GetStream();

while (tcpClient.Connected)
{
    //Read from the stream, assemble the bytes into the original messages and call
    //  the handler for each message that was received.
    if (!tcpStream.ReadAndProcessFrames(frameBuffer, ProcessFrameNotification, ProcessFrameQuery))
    {
        break; //Disconnected.
    }
}

//Handler for a frame written with WriteNotificationFrame().
private static void ProcessFrameNotification(IFramePayloadNotification payload)
{
    if (payload is MyMessage myMessage)
    {
        Console.WriteLine($"Received [Notification]: '{myMessage.Text}'");
    }
}

//Handler for a frame written with WriteQueryFrame()/WriteQueryFrameAsync().
//The returned reply is framed and sent back to the originator, where it is routed
//to the appropriate waiting asynchronous task.
private static IFramePayloadQueryReply ProcessFrameQuery(IFramePayloadQuery payload)
{
    //We received a message, see if it is of the type "MyQuery", if so reply with MyQueryReply.
    if (payload is MyQuery myQuery)
    {
        Console.WriteLine($"Received [Query]: '{myQuery.Text}'");

        return new MyQueryReply("Hello Server!");
    }

    throw new Exception("The query type was unhandled.");
}

When a query frame is received, ReadAndProcessFrames() invokes the query callback and automatically writes the returned reply back to the stream. On the originating side, the reply is matched to the waiting query by frame ID.

Payload classes

These are the payload classes used in the examples. Payloads must implement one of the payload interfaces and be JSON serializable.

Used in the examples to send a one-way notification:

internal class MyMessage : IFramePayloadNotification
{
    public string Text { get; set; }

    public MyMessage(string text)
    {
        Text = text;
    }
}

Used in the examples to send a query that expects a reply:

internal class MyQuery : IFramePayloadQuery
{
    public string Text { get; set; }

    public MyQuery(string text)
    {
        Text = text;
    }
}

Used in the examples to reply to a received query:

internal class MyQueryReply : IFramePayloadQueryReply
{
    public string Text { get; set; }

    public MyQueryReply(string text)
    {
        Text = text;
    }
}

License

MIT