// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.

using System;
using System.Binary;
using System.Diagnostics;
using System.Globalization;
using System.IO.Pipelines;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Microsoft.Extensions.WebSockets.Internal
{
    /// <summary>
    /// Provides the default implementation of <see cref="IWebSocketConnection"/>.
    /// </summary>
    /// <remarks>
    /// <para>
    /// This type is thread-safe, as long as only one thread ever calls <see cref="ExecuteAsync"/>. Multiple threads may call <see cref="SendAsync"/> simultaneously
    /// and the sends will block until ongoing send operations complete.
    /// </para>
    /// <para>
    /// The general pattern of having a single thread running <see cref="ExecuteAsync"/> and a separate thread running <see cref="SendAsync"/> will
    /// be thread-safe, as each method interacts with completely separate state.
    /// </para>
    /// </remarks>
    public class WebSocketConnection : IWebSocketConnection
    {
        private WebSocketOptions _options;
        private readonly byte[] _maskingKeyBuffer;
        private readonly IPipeReader _inbound;
        private readonly IPipeWriter _outbound;
        private readonly Timer _pinger;
        private readonly CancellationTokenSource _timerCts = new CancellationTokenSource();
        private Utf8Validator _validator = new Utf8Validator();
        private WebSocketOpcode _currentMessageType = WebSocketOpcode.Continuation;

        // Sends must be serialized between SendAsync, Pinger, and the Close frames sent when invalid messages are received.
        private SemaphoreSlim _sendLock = new SemaphoreSlim(1, 1);

        public string SubProtocol { get; }

        public WebSocketConnectionState State { get; private set; } = WebSocketConnectionState.Created;

        /// <summary>
        /// Constructs a new, unmasked, <see cref="WebSocketConnection"/> from an <see cref="IPipeReader"/> and an <see cref="IPipeWriter"/> that represents an established WebSocket connection (i.e. after handshaking)
        /// </summary>
        /// <param name="inbound">A <see cref="IPipeReader"/> from which frames will be read when receiving.</param>
        /// <param name="outbound">A <see cref="IPipeWriter"/> to which frame will be written when sending.</param>
        public WebSocketConnection(IPipeReader inbound, IPipeWriter outbound) : this(inbound, outbound, options: WebSocketOptions.DefaultUnmasked) { }

        /// <summary>
        /// Constructs a new, unmasked, <see cref="WebSocketConnection"/> from an <see cref="IPipeReader"/> and an <see cref="IPipeWriter"/> that represents an established WebSocket connection (i.e. after handshaking)
        /// </summary>
        /// <param name="inbound">A <see cref="IPipeReader"/> from which frames will be read when receiving.</param>
        /// <param name="outbound">A <see cref="IPipeWriter"/> to which frame will be written when sending.</param>
        /// <param name="subProtocol">The sub-protocol provided during handshaking</param>
        public WebSocketConnection(IPipeReader inbound, IPipeWriter outbound, string subProtocol) : this(inbound, outbound, subProtocol, options: WebSocketOptions.DefaultUnmasked) { }

        /// <summary>
        /// Constructs a new, <see cref="WebSocketConnection"/> from an <see cref="IPipeReader"/> and an <see cref="IPipeWriter"/> that represents an established WebSocket connection (i.e. after handshaking)
        /// </summary>
        /// <param name="inbound">A <see cref="IPipeReader"/> from which frames will be read when receiving.</param>
        /// <param name="outbound">A <see cref="IPipeWriter"/> to which frame will be written when sending.</param>
        /// <param name="options">A <see cref="WebSocketOptions"/> which provides the configuration options for the socket.</param>
        public WebSocketConnection(IPipeReader inbound, IPipeWriter outbound, WebSocketOptions options) : this(inbound, outbound, subProtocol: string.Empty, options: options) { }

        /// <summary>
        /// Constructs a new <see cref="WebSocketConnection"/> from an <see cref="IPipeReader"/> and an <see cref="IPipeWriter"/> that represents an established WebSocket connection (i.e. after handshaking)
        /// </summary>
        /// <param name="inbound">A <see cref="IPipeReader"/> from which frames will be read when receiving.</param>
        /// <param name="outbound">A <see cref="IPipeWriter"/> to which frame will be written when sending.</param>
        /// <param name="subProtocol">The sub-protocol provided during handshaking</param>
        /// <param name="options">A <see cref="WebSocketOptions"/> which provides the configuration options for the socket.</param>
        public WebSocketConnection(IPipeReader inbound, IPipeWriter outbound, string subProtocol, WebSocketOptions options)
        {
            _inbound = inbound;
            _outbound = outbound;
            _options = options;
            SubProtocol = subProtocol;

            if (_options.FixedMaskingKey != null)
            {
                // Use the fixed key directly as the buffer.
                _maskingKeyBuffer = _options.FixedMaskingKey;

                // Clear the MaskingKeyGenerator just to ensure that nobody set it.
                _options.MaskingKeyGenerator = null;
            }
            else if (_options.MaskingKeyGenerator != null)
            {
                // Establish a buffer for the random generator to use
                _maskingKeyBuffer = new byte[4];
            }

            if (_options.PingInterval > TimeSpan.Zero)
            {
                var pingIntervalMillis = (int)_options.PingInterval.TotalMilliseconds;
                // Set up the pinger
                _pinger = new Timer(Pinger, this, pingIntervalMillis, pingIntervalMillis);
            }
        }

        private static void Pinger(object state)
        {
            var connection = (WebSocketConnection)state;

            // If we are cancelled, don't send the ping
            // Also, if we can't immediately acquire the send lock, we're already sending something, so we don't need the ping.
            if (!connection._timerCts.Token.IsCancellationRequested && connection._sendLock.Wait(0))
            {
                // We don't need to wait for this task to complete, we're "tail calling" and
                // we are in a Timer thread-pool thread.
                var ignore = connection.SendCoreLockAcquiredAsync(
                    fin: true,
                    opcode: WebSocketOpcode.Ping,
                    payloadAllocLength: 28,
                    payloadLength: 28,
                    payloadWriter: PingPayloadWriter,
                    payload: DateTime.UtcNow,
                    cancellationToken: connection._timerCts.Token);
            }
        }

        public void Dispose()
        {
            State = WebSocketConnectionState.Closed;
            _pinger?.Dispose();
            _timerCts.Cancel();
            _inbound.Complete();
            _outbound.Complete();
        }

        public Task<WebSocketCloseResult> ExecuteAsync(Func<WebSocketFrame, object, Task> messageHandler, object state)
        {
            if (State == WebSocketConnectionState.Closed)
            {
                throw new ObjectDisposedException(nameof(WebSocketConnection));
            }

            if (State != WebSocketConnectionState.Created)
            {
                throw new InvalidOperationException("Connection is already running.");
            }
            State = WebSocketConnectionState.Connected;
            return ReceiveLoop(messageHandler, state);
        }

        /// <summary>
        /// Sends the specified frame.
        /// </summary>
        /// <param name="frame">The frame to send.</param>
        /// <param name="cancellationToken">A <see cref="CancellationToken"/> that indicates when/if the send is cancelled.</param>
        /// <returns>A <see cref="Task"/> that completes when the message has been written to the outbound stream.</returns>
        // TODO: De-taskify this to allow consumers to create their own awaiter.
        public Task SendAsync(WebSocketFrame frame, CancellationToken cancellationToken)
        {
            if (State == WebSocketConnectionState.Closed)
            {
                throw new ObjectDisposedException(nameof(WebSocketConnection));
            }
            // This clause is a bit of an artificial restriction to ensure people run "Execute". Maybe we don't care?
            else if (State == WebSocketConnectionState.Created)
            {
                throw new InvalidOperationException($"Cannot send until the connection is started using {nameof(ExecuteAsync)}");
            }
            else if (State == WebSocketConnectionState.CloseSent)
            {
                throw new InvalidOperationException("Cannot send after sending a Close frame");
            }

            if (frame.Opcode == WebSocketOpcode.Close)
            {
                throw new InvalidOperationException($"Cannot use {nameof(SendAsync)} to send a Close frame, use {nameof(CloseAsync)} instead.");
            }
            return SendCoreAsync(
                fin: frame.EndOfMessage,
                opcode: frame.Opcode,
                payloadAllocLength: 0, // We don't copy the payload, we append it, so we don't need any alloc for the payload
                payloadLength: frame.Payload.Length,
                payloadWriter: AppendPayloadWriter,
                payload: frame.Payload,
                cancellationToken: cancellationToken);
        }

        /// <summary>
        /// Sends a Close frame to the other party. This does not guarantee that the client will send a responding close frame.
        /// </summary>
        /// <remarks>
        /// If the other party does not respond with a close frame, the connection will remain open and the <see cref="Task{WebSocketCloseResult}"/>
        /// will remain active. Call the <see cref="IDisposable.Dispose"/> method on this instance to forcibly terminate the connection.
        /// </remarks>
        /// <param name="result">A <see cref="WebSocketCloseResult"/> with the payload for the close frame</param>
        /// <param name="cancellationToken">A <see cref="CancellationToken"/> that indicates when/if the send is cancelled.</param>
        /// <returns>A <see cref="Task"/> that completes when the close frame has been sent</returns>
        public async Task CloseAsync(WebSocketCloseResult result, CancellationToken cancellationToken)
        {
            if (State == WebSocketConnectionState.Closed)
            {
                // Already closed
                return;
            }
            else if (State == WebSocketConnectionState.Created)
            {
                throw new InvalidOperationException("Cannot send close frame when the connection hasn't been started");
            }
            else if (State == WebSocketConnectionState.CloseSent)
            {
                throw new InvalidOperationException("Cannot send multiple close frames");
            }

            var payloadSize = result.GetSize();
            await SendCoreAsync(
                fin: true,
                opcode: WebSocketOpcode.Close,
                payloadAllocLength: payloadSize,
                payloadLength: payloadSize,
                payloadWriter: CloseResultPayloadWriter,
                payload: result,
                cancellationToken: cancellationToken);

            _timerCts.Cancel();
            _pinger?.Dispose();

            if (State == WebSocketConnectionState.CloseReceived)
            {
                State = WebSocketConnectionState.Closed;
            }
            else
            {
                State = WebSocketConnectionState.CloseSent;
            }
        }

        private void WriteMaskingKey(Span<byte> buffer)
        {
            if (_options.MaskingKeyGenerator != null)
            {
                // Get a new random mask
                // Until https://github.com/dotnet/corefx/issues/12323 is fixed we need to use this shared buffer and copy model
                // Once we have that fix we should be able to generate the mask directly into the output buffer.
                _options.MaskingKeyGenerator.GetBytes(_maskingKeyBuffer);
            }

            _maskingKeyBuffer.CopyTo(buffer);
        }

        /// <summary>
        /// Terminates the socket abruptly.
        /// </summary>
        public void Abort()
        {
            // We duplicate some work from Dispose here, but that's OK.
            _timerCts.Cancel();
            _inbound.CancelPendingRead();
            _outbound.Complete();
        }

        private async ValueTask<(bool Success, byte OpcodeByte, bool Masked, bool Fin, int Length, uint MaskingKey)> ReadHeaderAsync()
        {
            // Read at least 2 bytes
            var readResult = await _inbound.ReadAtLeastAsync(2);
            if (readResult.IsCancelled || (readResult.IsCompleted && readResult.Buffer.Length < 2))
            {
                _inbound.Advance(readResult.Buffer.End);
                return (Success: false, OpcodeByte: 0, Masked: false, Fin: false, Length: 0, MaskingKey: 0);
            }
            var buffer = readResult.Buffer;

            // Read the opcode and length
            var opcodeByte = buffer.ReadBigEndian<byte>();
            buffer = buffer.Slice(1);

            // Read the first byte of the payload length
            var lengthByte = buffer.ReadBigEndian<byte>();
            buffer = buffer.Slice(1);

            _inbound.Advance(buffer.Start);

            // Determine how much header there still is to read
            var fin = (opcodeByte & 0x80) != 0;
            var masked = (lengthByte & 0x80) != 0;
            var length = lengthByte & 0x7F;

            // Calculate the rest of the header length
            var headerLength = masked ? 4 : 0;
            if (length == 126)
            {
                headerLength += 2;
            }
            else if (length == 127)
            {
                headerLength += 8;
            }

            // Read the next set of header data
            uint maskingKey = 0;
            if (headerLength > 0)
            {
                readResult = await _inbound.ReadAtLeastAsync(headerLength);
                if (readResult.IsCancelled || (readResult.IsCompleted && readResult.Buffer.Length < headerLength))
                {
                    _inbound.Advance(readResult.Buffer.End);
                    return (Success: false, OpcodeByte: 0, Masked: false, Fin: false, Length: 0, MaskingKey: 0);
                }
                buffer = readResult.Buffer;

                // Read extended payload length (if any)
                if (length == 126)
                {
                    length = buffer.ReadBigEndian<ushort>();
                    buffer = buffer.Slice(sizeof(ushort));
                }
                else if (length == 127)
                {
                    var longLen = buffer.ReadBigEndian<ulong>();
                    buffer = buffer.Slice(sizeof(ulong));
                    if (longLen > int.MaxValue)
                    {
                        throw new WebSocketException($"Frame is too large. Maximum frame size is {int.MaxValue} bytes");
                    }
                    length = (int)longLen;
                }

                // Read masking key
                if (masked)
                {
                    var maskingKeyStart = buffer.Start;
                    maskingKey = buffer.Slice(0, sizeof(uint)).ReadBigEndian<uint>();
                    buffer = buffer.Slice(sizeof(uint));
                }

                // Mark the length and masking key consumed
                _inbound.Advance(buffer.Start);
            }

            return (Success: true, opcodeByte, masked, fin, length, maskingKey);
        }

        private async ValueTask<(bool Success, ReadableBuffer Buffer)> ReadPayloadAsync(int length, bool masked, uint maskingKey)
        {
            var payload = default(ReadableBuffer);
            if (length > 0)
            {
                var readResult = await _inbound.ReadAtLeastAsync(length);
                if (readResult.IsCancelled || (readResult.IsCompleted && readResult.Buffer.Length < length))
                {
                    return (Success: false, Buffer: readResult.Buffer);
                }
                var buffer = readResult.Buffer;

                payload = buffer.Slice(0, length);

                if (masked)
                {
                    // Unmask
                    MaskingUtilities.ApplyMask(ref payload, maskingKey);
                }
            }
            return (Success: true, Buffer: payload);
        }

        private async Task<WebSocketCloseResult> ReceiveLoop(Func<WebSocketFrame, object, Task> messageHandler, object state)
        {
            try
            {
                while (true)
                {
                    // WebSocket Frame layout (https://tools.ietf.org/html/rfc6455#section-5.2):
                    //  0                   1                   2                   3
                    //  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
                    // +-+-+-+-+-------+-+-------------+-------------------------------+
                    // |F|R|R|R| opcode|M| Payload len |    Extended payload length    |
                    // |I|S|S|S|  (4)  |A|     (7)     |             (16/64)           |
                    // |N|V|V|V|       |S|             |   (if payload len==126/127)   |
                    // | |1|2|3|       |K|             |                               |
                    // +-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
                    // |     Extended payload length continued, if payload len == 127  |
                    // + - - - - - - - - - - - - - - - +-------------------------------+
                    // |                               |Masking-key, if MASK set to 1  |
                    // +-------------------------------+-------------------------------+
                    // | Masking-key (continued)       |          Payload Data         |
                    // +-------------------------------- - - - - - - - - - - - - - - - +
                    // :                     Payload Data continued ...                :
                    // + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
                    // |                     Payload Data continued ...                |
                    // +---------------------------------------------------------------+

                    var header = await ReadHeaderAsync();
                    if (!header.Success)
                    {
                        break;
                    }

                    // Validate Opcode
                    var opcodeNum = header.OpcodeByte & 0x0F;

                    if ((header.OpcodeByte & 0x70) != 0)
                    {
                        // Reserved bits set, this frame is invalid, close our side and terminate immediately
                        await CloseFromProtocolError("Reserved bits, which are required to be zero, were set.");
                        break;
                    }
                    else if ((opcodeNum >= 0x03 && opcodeNum <= 0x07) || (opcodeNum >= 0x0B && opcodeNum <= 0x0F))
                    {
                        // Reserved opcode
                        await CloseFromProtocolError($"Received frame using reserved opcode: 0x{opcodeNum:X}");
                        break;
                    }
                    var opcode = (WebSocketOpcode)opcodeNum;

                    var payload = await ReadPayloadAsync(header.Length, header.Masked, header.MaskingKey);
                    if (!payload.Success)
                    {
                        _inbound.Advance(payload.Buffer.End);
                        break;
                    }

                    var frame = new WebSocketFrame(header.Fin, opcode, payload.Buffer);

                    // Start a try-finally because we may get an exception while closing, if there's an error
                    // And we need to advance the buffer even if that happens. It wasn't needed above because
                    // we had already parsed the buffer before we verified it, so we had already advanced the
                    // buffer, if we encountered an error while closing we didn't have to advance the buffer.
                    // Side Note: Look at this gloriously aligned comment. You have anurse and brecon to thank for it. Oh wait, I ruined it.
                    try
                    {
                        if (frame.Opcode.IsControl() && !frame.EndOfMessage)
                        {
                            // Control frames cannot be fragmented.
                            await CloseFromProtocolError("Control frames may not be fragmented");
                            break;
                        }
                        else if (_currentMessageType != WebSocketOpcode.Continuation && opcode.IsMessage() && opcode != 0)
                        {
                            await CloseFromProtocolError("Received non-continuation frame during a fragmented message");
                            break;
                        }
                        else if (_currentMessageType == WebSocketOpcode.Continuation && frame.Opcode == WebSocketOpcode.Continuation)
                        {
                            await CloseFromProtocolError("Continuation Frame was received when expecting a new message");
                            break;
                        }

                        if (frame.Opcode == WebSocketOpcode.Close)
                        {
                            return await ProcessCloseFrameAsync(frame);
                        }
                        else
                        {
                            if (frame.Opcode == WebSocketOpcode.Ping)
                            {
                                // Check the ping payload length
                                if (frame.Payload.Length > 125)
                                {
                                    // Payload too long
                                    await CloseFromProtocolError("Ping frame exceeded maximum size of 125 bytes");
                                    break;
                                }

                                await SendCoreAsync(
                                    frame.EndOfMessage,
                                    WebSocketOpcode.Pong,
                                    payloadAllocLength: 0,
                                    payloadLength: frame.Payload.Length,
                                    payloadWriter: AppendPayloadWriter,
                                    payload: frame.Payload,
                                    cancellationToken: CancellationToken.None);
                            }
                            var effectiveOpcode = opcode == WebSocketOpcode.Continuation ? _currentMessageType : opcode;
                            if (effectiveOpcode == WebSocketOpcode.Text && !_validator.ValidateUtf8Frame(frame.Payload, frame.EndOfMessage))
                            {
                                // Drop the frame and immediately close with InvalidPayload
                                await CloseFromProtocolError("An invalid Text frame payload was received", statusCode: WebSocketCloseStatus.InvalidPayloadData);
                                break;
                            }
                            else if (_options.PassAllFramesThrough || (frame.Opcode != WebSocketOpcode.Ping && frame.Opcode != WebSocketOpcode.Pong))
                            {
                                await messageHandler(frame, state);
                            }
                        }
                    }
                    finally
                    {
                        if (frame.Payload.Length > 0)
                        {
                            _inbound.Advance(frame.Payload.End);
                        }
                    }

                    if (header.Fin)
                    {
                        // Reset the UTF8 validator
                        _validator.Reset();

                        // If it's a non-control frame, reset the message type tracker
                        if (opcode.IsMessage())
                        {
                            _currentMessageType = WebSocketOpcode.Continuation;
                        }
                    }
                    // If there isn't a current message type, and this was a fragmented message frame, set the current message type
                    else if (!header.Fin && _currentMessageType == WebSocketOpcode.Continuation && opcode.IsMessage())
                    {
                        _currentMessageType = opcode;
                    }
                }
            }
            catch
            {
                // Abort the socket and rethrow
                Abort();
                throw;
            }
            return WebSocketCloseResult.AbnormalClosure;
        }

        private async ValueTask<WebSocketCloseResult> ProcessCloseFrameAsync(WebSocketFrame frame)
        {
            // Allowed frame lengths:
            //  0 - No body
            //  2 - Code with no reason phrase
            //  >2 - Code and reason phrase (must be valid UTF-8)
            if (frame.Payload.Length > 125)
            {
                await CloseFromProtocolError("Close frame payload too long. Maximum size is 125 bytes");
                return WebSocketCloseResult.AbnormalClosure;
            }
            else if ((frame.Payload.Length == 1) || (frame.Payload.Length > 2 && !Utf8Validator.ValidateUtf8(frame.Payload.Slice(2))))
            {
                await CloseFromProtocolError("Close frame payload invalid");
                return WebSocketCloseResult.AbnormalClosure;
            }

            ushort? actualStatusCode;
            var closeResult = ParseCloseFrame(frame.Payload, frame, out actualStatusCode);

            // Verify the close result
            if (actualStatusCode != null)
            {
                var statusCode = actualStatusCode.Value;
                if (statusCode < 1000 || statusCode == 1004 || statusCode == 1005 || statusCode == 1006 || (statusCode > 1011 && statusCode < 3000))
                {
                    await CloseFromProtocolError($"Invalid close status: {statusCode}.");
                    return WebSocketCloseResult.AbnormalClosure;
                }
            }

            return closeResult;
        }

        private async Task CloseFromProtocolError(string reason, WebSocketCloseStatus statusCode = WebSocketCloseStatus.ProtocolError)
        {
            var closeResult = new WebSocketCloseResult(
                statusCode,
                reason);
            await CloseAsync(closeResult, CancellationToken.None);

            // We can now terminate our connection, according to the spec.
            Abort();
        }

        private WebSocketCloseResult ParseCloseFrame(ReadableBuffer payload, WebSocketFrame frame, out ushort? actualStatusCode)
        {
            // Update state
            if (State == WebSocketConnectionState.CloseSent)
            {
                State = WebSocketConnectionState.Closed;
            }
            else
            {
                State = WebSocketConnectionState.CloseReceived;
            }

            // Process the close frame
            WebSocketCloseResult closeResult;
            if (!WebSocketCloseResult.TryParse(frame.Payload, out closeResult, out actualStatusCode))
            {
                closeResult = WebSocketCloseResult.Empty;
            }

            return closeResult;
        }

        private static unsafe void PingPayloadWriter(WritableBuffer output, uint maskingKeyValue, int payloadLength, DateTime timestamp)
        {
            var maskingKey = new Span<byte>(&maskingKeyValue, sizeof(uint));

            var payload = output.Buffer.Slice(0, payloadLength);

            // TODO: Don't put this string on the heap? Is there a way to do that without re-implementing ToString?
            // Ideally we'd like to render the string directly to the output buffer.
            var str = timestamp.ToString("O", CultureInfo.InvariantCulture);

            ArraySegment<byte> buffer;
            if (payload.TryGetArray(out buffer))
            {
                // Fast path - Write the encoded bytes directly out.
                Encoding.UTF8.GetBytes(str, 0, str.Length, buffer.Array, buffer.Offset);
            }
            else
            {
                // TODO: Could use TryGetPointer, GetBytes does take a byte*, but it seems like just waiting until we have a version that uses Span is best.
                // Slow path - Allocate a heap buffer for the encoded bytes before writing them out.
                Encoding.UTF8.GetBytes(str).CopyTo(payload.Span);
            }

            if (maskingKey.Length > 0)
            {
                MaskingUtilities.ApplyMask(payload.Span, maskingKey);
            }

            output.Advance(payloadLength);
        }

        private static unsafe void CloseResultPayloadWriter(WritableBuffer output, uint maskingKeyValue, int payloadLength, WebSocketCloseResult result)
        {
            var maskingKey = new Span<byte>(&maskingKeyValue, sizeof(uint));

            // Write the close payload out
            var payload = output.Buffer.Slice(0, payloadLength).Span;
            result.WriteTo(ref output);

            if (maskingKey.Length > 0)
            {
                MaskingUtilities.ApplyMask(payload, maskingKey);
            }
        }

        private static unsafe void AppendPayloadWriter(WritableBuffer output, uint maskingKeyValue, int payloadLength, ReadableBuffer payload)
        {
            var maskingKey = new Span<byte>(&maskingKeyValue, sizeof(uint));

            if (maskingKey.Length > 0)
            {
                // Mask the payload in it's own buffer
                MaskingUtilities.ApplyMask(ref payload, maskingKey);
            }

            output.Append(payload);
        }

        private Task SendCoreAsync<T>(bool fin, WebSocketOpcode opcode, int payloadAllocLength, int payloadLength, Action<WritableBuffer, uint, int, T> payloadWriter, T payload, CancellationToken cancellationToken)
        {
            if (_sendLock.Wait(0))
            {
                return SendCoreLockAcquiredAsync(fin, opcode, payloadAllocLength, payloadLength, payloadWriter, payload, cancellationToken);
            }
            else
            {
                return SendCoreWaitForLockAsync(fin, opcode, payloadAllocLength, payloadLength, payloadWriter, payload, cancellationToken);
            }
        }

        private async Task SendCoreWaitForLockAsync<T>(bool fin, WebSocketOpcode opcode, int payloadAllocLength, int payloadLength, Action<WritableBuffer, uint, int, T> payloadWriter, T payload, CancellationToken cancellationToken)
        {
            await _sendLock.WaitAsync(cancellationToken);
            await SendCoreLockAcquiredAsync(fin, opcode, payloadAllocLength, payloadLength, payloadWriter, payload, cancellationToken);
        }

        private async Task SendCoreLockAcquiredAsync<T>(bool fin, WebSocketOpcode opcode, int payloadAllocLength, int payloadLength, Action<WritableBuffer, uint, int, T> payloadWriter, T payload, CancellationToken cancellationToken)
        {
            try
            {
                // Ensure the lock is held
                Debug.Assert(_sendLock.CurrentCount == 0);

                // Base header size is 2 bytes.
                WritableBuffer buffer;
                var allocSize = CalculateAllocSize(payloadAllocLength, payloadLength);

                // Allocate a buffer
                buffer = _outbound.Alloc(minimumSize: allocSize);
                Debug.Assert(buffer.Buffer.Length >= allocSize);

                // Write the opcode and FIN flag
                var opcodeByte = (byte)opcode;
                if (fin)
                {
                    opcodeByte |= 0x80;
                }
                buffer.WriteBigEndian(opcodeByte);

                // Write the length and mask flag
                WritePayloadLength(payloadLength, buffer);

                WritePayload(ref buffer, payloadLength, payloadWriter, payload);

                // Flush.
                await buffer.FlushAsync();
            }
            finally
            {
                // Unlock.
                _sendLock.Release();
            }
        }

        private void WritePayload<T>(ref WritableBuffer buffer, int payloadLength, Action<WritableBuffer, uint, int, T> payloadWriter, T payload)
        {
            var maskingKey = Span<byte>.Empty;
            var keySize = sizeof(uint);

            if (_maskingKeyBuffer != null)
            {
                // Get a span of the output buffer for the masking key, write it there, then advance the write head.
                maskingKey = buffer.Buffer.Slice(0, keySize).Span;
                WriteMaskingKey(maskingKey);
                buffer.Advance(keySize);

                // Write the payload
                payloadWriter(buffer, maskingKey.Read<uint>(), payloadLength, payload);
            }
            else
            {
                // Write the payload un-masked
                payloadWriter(buffer, 0, payloadLength, payload);
            }
        }

        private int CalculateAllocSize(int payloadAllocLength, int payloadLength)
        {
            var allocSize = 2;
            if (payloadLength > ushort.MaxValue)
            {
                // We're going to need an 8-byte length
                allocSize += 8;
            }
            else if (payloadLength > 125)
            {
                // We're going to need a 2-byte length
                allocSize += 2;
            }
            if (_maskingKeyBuffer != null)
            {
                // We need space for the masking key
                allocSize += 4;
            }

            // We may need space for the payload too
            return allocSize + payloadAllocLength;
        }

        private void WritePayloadLength(int payloadLength, WritableBuffer buffer)
        {
            var maskingByte = _maskingKeyBuffer != null ? 0x80 : 0x00;

            if (payloadLength > ushort.MaxValue)
            {
                buffer.WriteBigEndian((byte)(0x7F | maskingByte));

                // 8-byte length
                buffer.WriteBigEndian((ulong)payloadLength);
            }
            else if (payloadLength > 125)
            {
                buffer.WriteBigEndian((byte)(0x7E | maskingByte));

                // 2-byte length
                buffer.WriteBigEndian((ushort)payloadLength);
            }
            else
            {
                // 1-byte length
                buffer.WriteBigEndian((byte)(payloadLength | maskingByte));
            }
        }
    }
}