baseline: OpenMU upstream b5a0961 (fresh source)

src/Network/PacketPipeReaderBase.cs

@@ -0,0 +1,156 @@
+// <copyright file="PacketPipeReaderBase.cs" company="MUnique">
+// Licensed under the MIT License. See LICENSE file in the project root for full license information.
+// </copyright>
+
+namespace MUnique.OpenMU.Network;
+
+using System.Buffers;
+using System.IO.Pipelines;
+using Pipelines.Sockets.Unofficial;
+
+/// <summary>
+/// Base class for all classes which read mu online data packets from a <see cref="PipeReader"/>.
+/// </summary>
+/// <remarks>
+/// Things to consider here:
+///  * Do not call FlushAsync if the reader can't start until FlushAsync finishes, as that may cause a deadlock.
+///  * Ensure that only one context "owns" a PipeReader or PipeWriter or accesses them. These types are not thread-safe.
+/// </remarks>
+public abstract class PacketPipeReaderBase
+{
+    private readonly byte[] _headerBuffer = new byte[3];
+
+    /// <summary>
+    /// Gets or sets the <see cref="PipeReader"/> from which the packets can be read from at <see cref="ReadSourceAsync"/>.
+    /// </summary>
+    protected PipeReader Source { get; set; } = null!; // will be set in derived classes
+
+    /// <summary>
+    /// Reads the mu online packet.
+    /// </summary>
+    /// <param name="packet">The mu online packet.</param>
+    /// <returns><see langword="true" />, if the flush was successful or not required.<see langword="false" />, if the pipe reader is completed and no longer reading data.</returns>
+    protected abstract ValueTask<bool> ReadPacketAsync(ReadOnlySequence<byte> packet);
+
+    /// <summary>
+    /// Called when the <see cref="Source"/> completed.
+    /// </summary>
+    /// <param name="exception">The exception, if any occurred; Otherwise, <c>null</c>.</param>
+    protected abstract ValueTask OnCompleteAsync(Exception? exception);
+
+    /// <summary>
+    /// Reads from the <see cref="Source"/> until it's completed or cancelled.
+    /// </summary>
+    /// <returns>The task.</returns>
+    protected async Task ReadSourceAsync()
+    {
+        if (this.Source is null)
+        {
+            throw new InvalidOperationException("Source must be set before.");
+        }
+
+        try
+        {
+            while (true)
+            {
+                var completed = await this.ReadBufferAsync().ConfigureAwait(false);
+
+                // Stop reading if there's no more data coming
+                if (completed)
+                {
+                    break;
+                }
+            }
+        }
+        catch (ConnectionAbortedException)
+        {
+            // we can ignore that.
+        }
+        catch (Exception e)
+        {
+            await this.OnCompleteAsync(e).ConfigureAwait(false);
+            return;
+        }
+
+        await this.OnCompleteAsync(null).ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// Tries to flush the writer.
+    /// </summary>
+    /// <param name="pipeWriter">The pipe writer.</param>
+    /// <returns>
+    ///   <see langword="true" />, if the flush was successful or not required.<see langword="false" />, if the pipe reader is completed and no longer reading data.
+    /// </returns>
+    protected async ValueTask<bool> TryFlushWriterAsync(PipeWriter pipeWriter)
+    {
+        if (pipeWriter is { CanGetUnflushedBytes: true, UnflushedBytes: 0 })
+        {
+            // It was flushed already in the background.
+            return true;
+        }
+
+        // todo: what happens if it was flushed in the background in the meantime? race-condition?
+        var flushResult = await pipeWriter.FlushAsync().ConfigureAwait(false);
+        return !flushResult.IsCompleted;
+    }
+
+    private async Task<bool> ReadBufferAsync()
+    {
+        ReadResult result = await this.Source.ReadAsync().ConfigureAwait(false);
+        ReadOnlySequence<byte> buffer = result.Buffer;
+        int? length = null;
+        var readingCancelledOrCompleted = false;
+        do
+        {
+            if (buffer.Length > 2)
+            {
+                // peek the length of the next packet
+                buffer.Slice(0, 3).CopyTo(this._headerBuffer);
+                length = this._headerBuffer.AsSpan().GetPacketSize();
+                if (length == 0)
+                {
+                    var exception = new InvalidPacketHeaderException(this._headerBuffer, result.Buffer, buffer.Start);
+
+                    // Notify our source, that we don't intend to read anymore.
+                    await this.Source.CompleteAsync(exception).ConfigureAwait(false);
+
+                    await this.OnCompleteAsync(exception).ConfigureAwait(false);
+                    throw exception;
+                }
+            }
+
+            if (length is > 0 && buffer.Length >= length)
+            {
+                var packet = buffer.Slice(0, length.Value);
+                if (!await this.ReadPacketAsync(packet).ConfigureAwait(false))
+                {
+                    readingCancelledOrCompleted = true;
+                    break;
+                }
+
+                buffer = buffer.Slice(buffer.GetPosition(length.Value), buffer.End);
+                length = null;
+            }
+            else
+            {
+                // read more
+                break;
+            }
+        }
+        while (buffer.Length > 2);
+
+        if (result.IsCanceled || result.IsCompleted)
+        {
+            // Not possible to advance any further, e.g. because of a disconnected network connection.
+            await this.OnCompleteAsync(null).ConfigureAwait(false);
+        }
+        else
+        {
+            // Tell the PipeReader how much of the buffer we have consumed
+            this.Source.AdvanceTo(buffer.Start);
+        }
+
+        return result.IsCompleted || readingCancelledOrCompleted;
+    }
+}