Got at least one data fetching method working; turns out, we can't use a patched LogicStack to get the data

This commit is contained in:
2026-01-14 22:11:11 +01:00
parent 40a8431464
commit 3f7122d30a
350 changed files with 41444 additions and 119 deletions

View File

@@ -0,0 +1,22 @@
namespace EmbedIO.WebSockets.Internal
{
/// <summary>
/// Indicates whether a WebSocket frame is the final frame of a message.
/// </summary>
/// <remarks>
/// The values of this enumeration are defined in
/// <see href="http://tools.ietf.org/html/rfc6455#section-5.2">Section 5.2</see> of RFC 6455.
/// </remarks>
internal enum Fin : byte
{
/// <summary>
/// Equivalent to numeric value 0. Indicates more frames of a message follow.
/// </summary>
More = 0x0,
/// <summary>
/// Equivalent to numeric value 1. Indicates the final frame of a message.
/// </summary>
Final = 0x1,
}
}

View File

@@ -0,0 +1,29 @@
using System.IO;
using System.Threading;
using System.Threading.Tasks;
namespace EmbedIO.WebSockets.Internal
{
internal class FragmentBuffer : MemoryStream
{
private readonly bool _fragmentsCompressed;
private readonly Opcode _fragmentsOpcode;
public FragmentBuffer(Opcode frameOpcode, bool frameIsCompressed)
{
_fragmentsOpcode = frameOpcode;
_fragmentsCompressed = frameIsCompressed;
}
public void AddPayload(MemoryStream data) => data.CopyTo(this, 1024);
public async Task<MessageEventArgs> GetMessage(CompressionMethod compression)
{
var data = _fragmentsCompressed
? await this.CompressAsync(compression, false, CancellationToken.None).ConfigureAwait(false)
: this;
return new MessageEventArgs(_fragmentsOpcode, data.ToArray());
}
}
}

View File

@@ -0,0 +1,22 @@
namespace EmbedIO.WebSockets.Internal
{
/// <summary>
/// Indicates whether the payload data of a WebSocket frame is masked.
/// </summary>
/// <remarks>
/// The values of this enumeration are defined in
/// <see href="http://tools.ietf.org/html/rfc6455#section-5.2">Section 5.2</see> of RFC 6455.
/// </remarks>
internal enum Mask : byte
{
/// <summary>
/// Equivalent to numeric value 0. Indicates not masked.
/// </summary>
Off = 0x0,
/// <summary>
/// Equivalent to numeric value 1. Indicates masked.
/// </summary>
On = 0x1,
}
}

View File

@@ -0,0 +1,114 @@
using System;
using Swan;
namespace EmbedIO.WebSockets.Internal
{
/// <summary>
/// Represents the event data for the <see cref="Internal.WebSocket.OnMessage"/> event.
/// </summary>
/// <remarks>
/// <para>
/// That event occurs when the <see cref="Internal.WebSocket"/> receives
/// a message or a ping if the <see cref="Internal.WebSocket.EmitOnPing"/>
/// property is set to <c>true</c>.
/// </para>
/// <para>
/// If you would like to get the message data, you should access
/// the <see cref="Data"/> or <see cref="RawData"/> property.
/// </para>
/// </remarks>
internal class MessageEventArgs : EventArgs
{
private readonly byte[] _rawData;
private string? _data;
private bool _dataSet;
internal MessageEventArgs(WebSocketFrame frame)
{
Opcode = frame.Opcode;
_rawData = frame.PayloadData.ApplicationData.ToArray();
}
internal MessageEventArgs(Opcode opcode, byte[] rawData)
{
if ((ulong)rawData.Length > PayloadData.MaxLength)
throw new WebSocketException(CloseStatusCode.TooBig);
Opcode = opcode;
_rawData = rawData;
}
/// <summary>
/// Gets the message data as a <see cref="string"/>.
/// </summary>
/// <value>
/// A <see cref="string"/> that represents the message data if its type is
/// text or ping and if decoding it to a string has successfully done;
/// otherwise, <see langword="null"/>.
/// </value>
public string? Data
{
get
{
SetData();
return _data;
}
}
/// <summary>
/// Gets a value indicating whether the message type is binary.
/// </summary>
/// <value>
/// <c>true</c> if the message type is binary; otherwise, <c>false</c>.
/// </value>
public bool IsBinary => Opcode == Opcode.Binary;
/// <summary>
/// Gets a value indicating whether the message type is ping.
/// </summary>
/// <value>
/// <c>true</c> if the message type is ping; otherwise, <c>false</c>.
/// </value>
public bool IsPing => Opcode == Opcode.Ping;
/// <summary>
/// Gets a value indicating whether the message type is text.
/// </summary>
/// <value>
/// <c>true</c> if the message type is text; otherwise, <c>false</c>.
/// </value>
public bool IsText => Opcode == Opcode.Text;
/// <summary>
/// Gets the message data as an array of <see cref="byte"/>.
/// </summary>
/// <value>
/// An array of <see cref="byte"/> that represents the message data.
/// </value>
public byte[] RawData
{
get
{
SetData();
return _rawData;
}
}
internal Opcode Opcode { get; }
private void SetData()
{
if (_dataSet)
return;
if (Opcode == Opcode.Binary)
{
_dataSet = true;
return;
}
_data = _rawData.ToText();
_dataSet = true;
}
}
}

View File

@@ -0,0 +1,74 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using Swan;
using EmbedIO.Net.Internal;
namespace EmbedIO.WebSockets.Internal
{
internal class PayloadData
{
public const ulong MaxLength = long.MaxValue;
private readonly byte[] _data;
private ushort? _code;
internal PayloadData(byte[] data)
{
_data = data;
}
internal PayloadData(ushort code = 1005, string? reason = null)
{
_code = code;
_data = code == 1005 ? Array.Empty<byte>() : Append(code, reason);
}
internal MemoryStream ApplicationData => new MemoryStream(_data);
internal ulong Length => (ulong)_data.Length;
internal ushort Code
{
get
{
if (!_code.HasValue)
{
_code = _data.Length > 1
? BitConverter.ToUInt16(_data.Take(2).ToArray().ToHostOrder(Endianness.Big), 0)
: (ushort)1005;
}
return _code.Value;
}
}
internal bool HasReservedCode => _data.Length > 1 && (Code == (ushort)CloseStatusCode.Undefined ||
Code == (ushort)CloseStatusCode.NoStatus ||
Code == (ushort)CloseStatusCode.Abnormal ||
Code == (ushort)CloseStatusCode.TlsHandshakeFailure);
public override string ToString() => BitConverter.ToString(_data);
internal static byte[] Append(ushort code, string? reason)
{
var ret = code.ToByteArray(Endianness.Big);
if (string.IsNullOrEmpty(reason)) return ret;
var buff = new List<byte>(ret);
buff.AddRange(Encoding.UTF8.GetBytes(reason));
return buff.ToArray();
}
internal void Mask(byte[] key)
{
for (long i = 0; i < _data.Length; i++)
_data[i] = (byte)(_data[i] ^ key[i % 4]);
}
internal byte[] ToArray() => _data;
}
}

View File

@@ -0,0 +1,22 @@
namespace EmbedIO.WebSockets.Internal
{
/// <summary>
/// Indicates whether each RSV (RSV1, RSV2, and RSV3) of a WebSocket frame is non-zero.
/// </summary>
/// <remarks>
/// The values of this enumeration are defined in
/// <see href="http://tools.ietf.org/html/rfc6455#section-5.2">Section 5.2</see> of RFC 6455.
/// </remarks>
internal enum Rsv : byte
{
/// <summary>
/// Equivalent to numeric value 0. Indicates zero.
/// </summary>
Off = 0x0,
/// <summary>
/// Equivalent to numeric value 1. Indicates non-zero.
/// </summary>
On = 0x1,
}
}

View File

@@ -0,0 +1,66 @@
using System;
using System.IO;
using System.IO.Compression;
using System.Threading;
using System.Threading.Tasks;
namespace EmbedIO.WebSockets.Internal
{
internal static class StreamExtensions
{
private static readonly byte[] LastByte = { 0x00 };
// Compresses or decompresses a stream using the specified compression method.
public static async Task<MemoryStream> CompressAsync(
this Stream @this,
CompressionMethod method,
bool compress,
CancellationToken cancellationToken)
{
@this.Position = 0;
var targetStream = new MemoryStream();
switch (method)
{
case CompressionMethod.Deflate:
if (compress)
{
using var compressor = new DeflateStream(targetStream, CompressionMode.Compress, true);
await @this.CopyToAsync(compressor, 1024, cancellationToken).ConfigureAwait(false);
await @this.CopyToAsync(compressor).ConfigureAwait(false);
// WebSocket use this
targetStream.Write(LastByte, 0, 1);
targetStream.Position = 0;
}
else
{
using var compressor = new DeflateStream(@this, CompressionMode.Decompress);
await compressor.CopyToAsync(targetStream).ConfigureAwait(false);
}
break;
case CompressionMethod.Gzip:
if (compress)
{
using var compressor = new GZipStream(targetStream, CompressionMode.Compress, true);
await @this.CopyToAsync(compressor).ConfigureAwait(false);
}
else
{
using var compressor = new GZipStream(@this, CompressionMode.Decompress);
await compressor.CopyToAsync(targetStream).ConfigureAwait(false);
}
break;
case CompressionMethod.None:
await @this.CopyToAsync(targetStream).ConfigureAwait(false);
break;
default:
throw new ArgumentOutOfRangeException(nameof(method), method, null);
}
return targetStream;
}
}
}

View File

@@ -0,0 +1,66 @@
using System;
using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;
namespace EmbedIO.WebSockets.Internal
{
internal sealed class SystemWebSocket : IWebSocket
{
public SystemWebSocket(System.Net.WebSockets.WebSocket webSocket)
{
UnderlyingWebSocket = webSocket;
}
~SystemWebSocket()
{
Dispose(false);
}
public System.Net.WebSockets.WebSocket UnderlyingWebSocket { get; }
public WebSocketState State => UnderlyingWebSocket.State;
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
/// <inheritdoc />
public Task SendAsync(byte[] buffer, bool isText, CancellationToken cancellationToken = default)
=> UnderlyingWebSocket.SendAsync(
new ArraySegment<byte>(buffer),
isText ? WebSocketMessageType.Text : WebSocketMessageType.Binary,
true,
cancellationToken);
/// <inheritdoc />
public Task CloseAsync(CancellationToken cancellationToken = default) =>
UnderlyingWebSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken);
/// <inheritdoc />
public Task CloseAsync(CloseStatusCode code, string? comment = null, CancellationToken cancellationToken = default)=>
UnderlyingWebSocket.CloseAsync(MapCloseStatus(code), comment ?? string.Empty, cancellationToken);
private void Dispose(bool disposing)
{
if (!disposing)
return;
UnderlyingWebSocket.Dispose();
}
private WebSocketCloseStatus MapCloseStatus(CloseStatusCode code) => code switch {
CloseStatusCode.Normal => WebSocketCloseStatus.NormalClosure,
CloseStatusCode.ProtocolError => WebSocketCloseStatus.ProtocolError,
CloseStatusCode.InvalidData => WebSocketCloseStatus.InvalidPayloadData,
CloseStatusCode.UnsupportedData => WebSocketCloseStatus.InvalidPayloadData,
CloseStatusCode.PolicyViolation => WebSocketCloseStatus.PolicyViolation,
CloseStatusCode.TooBig => WebSocketCloseStatus.MessageTooBig,
CloseStatusCode.MandatoryExtension => WebSocketCloseStatus.MandatoryExtension,
CloseStatusCode.ServerError => WebSocketCloseStatus.InternalServerError,
_ => throw new ArgumentOutOfRangeException(nameof(code), code, null)
};
}
}

View File

@@ -0,0 +1,29 @@
namespace EmbedIO.WebSockets.Internal
{
/// <summary>
/// Represents a wrapper around a regular WebSocketContext.
/// </summary>
/// <inheritdoc />
internal sealed class SystemWebSocketReceiveResult : IWebSocketReceiveResult
{
private readonly System.Net.WebSockets.WebSocketReceiveResult _results;
/// <summary>
/// Initializes a new instance of the <see cref="SystemWebSocketReceiveResult"/> class.
/// </summary>
/// <param name="results">The results.</param>
public SystemWebSocketReceiveResult(System.Net.WebSockets.WebSocketReceiveResult results)
{
_results = results;
}
/// <inheritdoc/>
public int Count => _results.Count;
/// <inheritdoc/>
public bool EndOfMessage=> _results.EndOfMessage;
/// <inheritdoc/>
public int MessageType => (int) _results.MessageType;
}
}

View File

@@ -0,0 +1,561 @@
using System;
using System.Collections.Concurrent;
using System.IO;
using System.Net.WebSockets;
using System.Security.Cryptography;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using EmbedIO.Net.Internal;
using Swan;
using Swan.Logging;
namespace EmbedIO.WebSockets.Internal
{
/// <summary>
/// Implements the WebSocket interface.
/// </summary>
/// <remarks>
/// The WebSocket class provides a set of methods and properties for two-way communication using
/// the WebSocket protocol (<see href="http://tools.ietf.org/html/rfc6455">RFC 6455</see>).
/// </remarks>
internal sealed class WebSocket : IWebSocket
{
public const string SupportedVersion = "13";
private readonly object _stateSyncRoot = new ();
private readonly ConcurrentQueue<MessageEventArgs> _messageEventQueue = new ();
private readonly Action _closeConnection;
private readonly TimeSpan _waitTime = TimeSpan.FromSeconds(1);
private volatile WebSocketState _readyState;
private AutoResetEvent? _exitReceiving;
private FragmentBuffer? _fragmentsBuffer;
private volatile bool _inMessage;
private AutoResetEvent? _receivePong;
private Stream? _stream;
private WebSocket(HttpConnection connection)
{
_closeConnection = connection.ForceClose;
_stream = connection.Stream;
_readyState = WebSocketState.Open;
}
~WebSocket()
{
Dispose(false);
}
/// <summary>
/// Occurs when the <see cref="WebSocket"/> receives a message.
/// </summary>
public event EventHandler<MessageEventArgs>? OnMessage;
/// <inheritdoc />
public WebSocketState State => _readyState;
internal CompressionMethod Compression { get; } = CompressionMethod.None;
internal bool EmitOnPing { get; set; }
internal bool InContinuation { get; private set; }
/// <inheritdoc />
public Task SendAsync(byte[] buffer, bool isText, CancellationToken cancellationToken) => SendAsync(buffer, isText ? Opcode.Text : Opcode.Binary, cancellationToken);
/// <inheritdoc />
public Task CloseAsync(CancellationToken cancellationToken = default) => CloseAsync(CloseStatusCode.Normal, cancellationToken: cancellationToken);
/// <inheritdoc />
public Task CloseAsync(
CloseStatusCode code = CloseStatusCode.Undefined,
string? reason = null,
CancellationToken cancellationToken = default)
{
bool CheckParametersForClose()
{
if (code == CloseStatusCode.NoStatus && !string.IsNullOrEmpty(reason))
{
"'code' cannot have a reason.".Trace(nameof(WebSocket));
return false;
}
if (code == CloseStatusCode.MandatoryExtension)
{
"'code' cannot be used by a server.".Trace(nameof(WebSocket));
return false;
}
if (!string.IsNullOrEmpty(reason) && Encoding.UTF8.GetBytes(reason).Length > 123)
{
"The size of 'reason' is greater than the allowable max size.".Trace(nameof(WebSocket));
return false;
}
return true;
}
if (_readyState != WebSocketState.Open)
{
return Task.CompletedTask;
}
if (code != CloseStatusCode.Undefined && !CheckParametersForClose())
{
return Task.CompletedTask;
}
if (code == CloseStatusCode.NoStatus)
{
return InternalCloseAsync(cancellationToken: cancellationToken);
}
var send = !IsOpcodeReserved(code);
return InternalCloseAsync(new PayloadData((ushort)code, reason), send, send, cancellationToken);
}
/// <summary>
/// Sends a ping using the WebSocket connection.
/// </summary>
/// <returns>
/// <c>true</c> if the <see cref="WebSocket"/> receives a pong to this ping in a time;
/// otherwise, <c>false</c>.
/// </returns>
public Task<bool> PingAsync() => PingAsync(WebSocketFrame.EmptyPingBytes, _waitTime);
/// <summary>
/// Sends a ping with the specified <paramref name="message"/> using the WebSocket connection.
/// </summary>
/// <returns>
/// <c>true</c> if the <see cref="WebSocket"/> receives a pong to this ping in a time;
/// otherwise, <c>false</c>.
/// </returns>
/// <param name="message">
/// A <see cref="string"/> that represents the message to send.
/// </param>
public Task<bool> PingAsync(string message)
{
if (string.IsNullOrEmpty(message))
{
return PingAsync();
}
var data = Encoding.UTF8.GetBytes(message);
if (data.Length <= 125)
{
return PingAsync(WebSocketFrame.CreatePingFrame(data).ToArray(), _waitTime);
}
"A message has greater than the allowable max size.".Error(nameof(PingAsync));
return Task.FromResult(false);
}
/// <summary>
/// Sends binary <paramref name="data" /> using the WebSocket connection.
/// </summary>
/// <param name="data">An array of <see cref="byte" /> that represents the binary data to send.</param>
/// <param name="opcode">The opcode.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <returns>
/// A task that represents the asynchronous of send
/// binary data using websocket.
/// </returns>
#pragma warning disable CA1801 // Unused parameter
public async Task SendAsync(byte[] data, Opcode opcode, CancellationToken cancellationToken = default)
#pragma warning restore CA1801
{
if (_readyState != WebSocketState.Open)
{
throw new WebSocketException(CloseStatusCode.Normal, $"This operation isn\'t available in: {_readyState}");
}
using var stream = new WebSocketStream(data, opcode, Compression);
foreach (var frame in stream.GetFrames())
{
await Send(frame).ConfigureAwait(false);
}
}
/// <inheritdoc />
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
internal static async Task<WebSocket> AcceptAsync(HttpListenerContext httpContext, string acceptedProtocol)
{
static string CreateResponseKey(string clientKey)
{
const string Guid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
var buff = new StringBuilder(clientKey, 64).Append(Guid);
#pragma warning disable CA5350 // Do Not Use Weak Cryptographic Algorithms
using var sha1 = SHA1.Create();
return Convert.ToBase64String(sha1.ComputeHash(Encoding.UTF8.GetBytes(buff.ToString())));
#pragma warning restore CA5350 // Do Not Use Weak Cryptographic Algorithms
}
var requestHeaders = httpContext.Request.Headers;
var webSocketKey = requestHeaders[HttpHeaderNames.SecWebSocketKey];
if (string.IsNullOrEmpty(webSocketKey))
{
throw new WebSocketException(CloseStatusCode.ProtocolError, $"Includes no {HttpHeaderNames.SecWebSocketKey} header, or it has an invalid value.");
}
var webSocketVersion = requestHeaders[HttpHeaderNames.SecWebSocketVersion];
if (webSocketVersion == null || webSocketVersion != SupportedVersion)
{
throw new WebSocketException(CloseStatusCode.ProtocolError, $"Includes no {HttpHeaderNames.SecWebSocketVersion} header, or it has an invalid value.");
}
var handshakeResponse = new WebSocketHandshakeResponse(httpContext);
handshakeResponse.Headers[HttpHeaderNames.SecWebSocketAccept] = CreateResponseKey(webSocketKey);
if (acceptedProtocol.Length > 0)
{
handshakeResponse.Headers[HttpHeaderNames.SecWebSocketProtocol] = acceptedProtocol;
}
var bytes = Encoding.UTF8.GetBytes(handshakeResponse.ToString());
await httpContext.Connection.Stream.WriteAsync(bytes, 0, bytes.Length).ConfigureAwait(false);
// Signal the original response that headers have been sent.
httpContext.HttpListenerResponse.HeadersSent = true;
var socket = new WebSocket(httpContext.Connection);
socket.Open();
return socket;
}
internal async Task<bool> PingAsync(byte[] frameAsBytes, TimeSpan timeout)
{
if (_readyState != WebSocketState.Open)
{
return false;
}
await _stream.WriteAsync(frameAsBytes, 0, frameAsBytes.Length).ConfigureAwait(false);
return _receivePong != null && _receivePong.WaitOne(timeout);
}
private static bool IsOpcodeReserved(CloseStatusCode code)
=> code == CloseStatusCode.Undefined
|| code == CloseStatusCode.NoStatus
|| code == CloseStatusCode.Abnormal
|| code == CloseStatusCode.TlsHandshakeFailure;
#pragma warning disable CA1801 // Unused parameter
private void Dispose(bool disposing)
#pragma warning restore CA1801
{
try
{
InternalCloseAsync(new PayloadData((ushort)CloseStatusCode.Away)).Await();
}
catch
{
// Ignored
}
}
private async Task InternalCloseAsync(
PayloadData? payloadData = null,
bool send = true,
bool receive = true,
CancellationToken cancellationToken = default)
{
lock (_stateSyncRoot)
{
if (_readyState == WebSocketState.CloseReceived || _readyState == WebSocketState.CloseSent)
{
"The closing is already in progress.".Trace(nameof(InternalCloseAsync));
return;
}
if (_readyState == WebSocketState.Closed)
{
"The connection has been closed.".Trace(nameof(InternalCloseAsync));
return;
}
send = send && _readyState == WebSocketState.Open;
receive = receive && send;
_readyState = WebSocketState.CloseSent;
}
"Begin closing the connection.".Trace(nameof(InternalCloseAsync));
var bytes = send ? WebSocketFrame.CreateCloseFrame(payloadData).ToArray() : null;
await CloseHandshakeAsync(bytes, receive, cancellationToken).ConfigureAwait(false);
ReleaseResources();
"End closing the connection.".Trace(nameof(InternalCloseAsync));
lock (_stateSyncRoot)
{
_readyState = WebSocketState.Closed;
}
}
private async Task CloseHandshakeAsync(
byte[]? frameAsBytes,
bool receive,
CancellationToken cancellationToken)
{
var sent = frameAsBytes != null;
if (sent)
{
await _stream.WriteAsync(frameAsBytes, 0, frameAsBytes.Length, cancellationToken).ConfigureAwait(false);
}
if (receive && sent)
{
_ = _exitReceiving?.WaitOne(_waitTime);
}
}
private void Fatal(string message, Exception? exception = null)
=> Fatal(message, (exception as WebSocketException)?.Code ?? CloseStatusCode.Abnormal);
private void Fatal(string message, CloseStatusCode code)
=> InternalCloseAsync(new PayloadData((ushort)code, message), !IsOpcodeReserved(code), false).Await();
private void Message()
{
if (_inMessage || _messageEventQueue.IsEmpty || _readyState != WebSocketState.Open)
{
return;
}
_inMessage = true;
if (_messageEventQueue.TryDequeue(out var e))
{
Messages(e);
}
}
private void Messages(MessageEventArgs e)
{
try
{
OnMessage?.Invoke(this, e);
}
catch (Exception ex)
{
ex.Log(nameof(WebSocket));
}
if (!_messageEventQueue.TryDequeue(out e) || _readyState != WebSocketState.Open)
{
_inMessage = false;
return;
}
_ = Task.Run(() => Messages(e));
}
private void Open()
{
_inMessage = true;
StartReceiving();
if (!_messageEventQueue.TryDequeue(out var e) || _readyState != WebSocketState.Open)
{
_inMessage = false;
return;
}
Messages(e);
}
private Task ProcessCloseFrame(WebSocketFrame frame) => InternalCloseAsync(frame.PayloadData, !frame.PayloadData.HasReservedCode, false);
private async Task ProcessDataFrame(WebSocketFrame frame)
{
if (frame.IsCompressed)
{
using var ms = await frame.PayloadData.ApplicationData.CompressAsync(Compression, false, CancellationToken.None).ConfigureAwait(false);
_messageEventQueue.Enqueue(new MessageEventArgs(frame.Opcode, ms.ToArray()));
}
else
{
_messageEventQueue.Enqueue(new MessageEventArgs(frame));
}
}
private async Task ProcessFragmentFrame(WebSocketFrame frame)
{
if (!InContinuation)
{
// Must process first fragment.
if (frame.Opcode == Opcode.Cont)
{
return;
}
_fragmentsBuffer = new FragmentBuffer(frame.Opcode, frame.IsCompressed);
InContinuation = true;
}
_fragmentsBuffer.AddPayload(frame.PayloadData.ApplicationData);
if (frame.Fin == Fin.Final)
{
using (_fragmentsBuffer)
{
_messageEventQueue.Enqueue(await _fragmentsBuffer.GetMessage(Compression).ConfigureAwait(false));
}
_fragmentsBuffer = null;
InContinuation = false;
}
}
private Task ProcessPingFrame(WebSocketFrame frame)
{
if (EmitOnPing)
{
_messageEventQueue.Enqueue(new MessageEventArgs(frame));
}
return Send(new WebSocketFrame(Opcode.Pong, frame.PayloadData));
}
private void ProcessPongFrame()
{
_ = _receivePong?.Set();
"Received a pong.".Trace(nameof(ProcessPongFrame));
}
private async Task<bool> ProcessReceivedFrame(WebSocketFrame frame)
{
if (frame.IsFragment)
{
await ProcessFragmentFrame(frame).ConfigureAwait(false);
}
else
{
switch (frame.Opcode)
{
case Opcode.Text:
case Opcode.Binary:
await ProcessDataFrame(frame).ConfigureAwait(false);
break;
case Opcode.Ping:
await ProcessPingFrame(frame).ConfigureAwait(false);
break;
case Opcode.Pong:
ProcessPongFrame();
break;
case Opcode.Close:
await ProcessCloseFrame(frame).ConfigureAwait(false);
break;
default:
Fatal($"Unsupported frame received: {frame.PrintToString()}", CloseStatusCode.PolicyViolation);
return false;
}
}
return true;
}
private void ReleaseResources()
{
_closeConnection();
_stream = null;
if (_fragmentsBuffer != null)
{
_fragmentsBuffer.Dispose();
_fragmentsBuffer = null;
InContinuation = false;
}
if (_receivePong != null)
{
_receivePong.Dispose();
_receivePong = null;
}
if (_exitReceiving == null)
{
return;
}
_exitReceiving.Dispose();
_exitReceiving = null;
}
private Task Send(WebSocketFrame frame)
{
lock (_stateSyncRoot)
{
if (_readyState != WebSocketState.Open)
{
"The sending has been interrupted.".Error(nameof(Send));
return Task.Delay(0);
}
}
var frameAsBytes = frame.ToArray();
return _stream.WriteAsync(frameAsBytes, 0, frameAsBytes.Length);
}
private void StartReceiving()
{
while (_messageEventQueue.TryDequeue(out _))
{
// do nothing
}
_exitReceiving = new AutoResetEvent(false);
_receivePong = new AutoResetEvent(false);
var frameStream = new WebSocketFrameStream(_stream);
_ = Task.Run(async () =>
{
while (_readyState == WebSocketState.Open)
{
try
{
var frame = await frameStream.ReadFrameAsync(this).ConfigureAwait(false);
if (frame == null)
{
return;
}
var result = await ProcessReceivedFrame(frame).ConfigureAwait(false);
if (!result || _readyState == WebSocketState.Closed)
{
_ = _exitReceiving?.Set();
return;
}
_ = Task.Run(Message);
}
catch (Exception ex)
{
Fatal("An exception has occurred while receiving.", ex);
}
}
});
}
}
}

View File

@@ -0,0 +1,100 @@
using System;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Net;
using System.Security.Principal;
using System.Threading;
using EmbedIO.Sessions;
using EmbedIO.Utilities;
namespace EmbedIO.WebSockets.Internal
{
internal sealed class WebSocketContext : IWebSocketContext
{
internal WebSocketContext(
IHttpContextImpl httpContext,
string webSocketVersion,
IEnumerable<string> requestedProtocols,
string acceptedProtocol,
IWebSocket webSocket,
CancellationToken cancellationToken)
{
Id = UniqueIdGenerator.GetNext();
CancellationToken = cancellationToken;
HttpContextId = httpContext.Id;
Session = httpContext.Session;
Items = httpContext.Items;
LocalEndPoint = httpContext.LocalEndPoint;
RemoteEndPoint = httpContext.RemoteEndPoint;
RequestUri = httpContext.Request.Url;
Headers = httpContext.Request.Headers;
Origin = Headers[HttpHeaderNames.Origin];
RequestedProtocols = requestedProtocols;
AcceptedProtocol = acceptedProtocol;
WebSocketVersion = webSocketVersion;
Cookies = httpContext.Request.Cookies;
User = httpContext.User;
IsAuthenticated = httpContext.User.Identity.IsAuthenticated;
IsLocal = httpContext.Request.IsLocal;
IsSecureConnection = httpContext.Request.IsSecureConnection;
WebSocket = webSocket;
}
/// <inheritdoc />
public string Id { get; }
/// <inheritdoc />
public CancellationToken CancellationToken { get; }
/// <inheritdoc />
public string HttpContextId { get; }
/// <inheritdoc />
public ISessionProxy Session { get; }
/// <inheritdoc />
public IDictionary<object, object> Items { get; }
/// <inheritdoc />
public IPEndPoint LocalEndPoint { get; }
/// <inheritdoc />
public IPEndPoint RemoteEndPoint { get; }
/// <inheritdoc />
public Uri RequestUri { get; }
/// <inheritdoc />
public NameValueCollection Headers { get; }
/// <inheritdoc />
public string Origin { get; }
/// <inheritdoc />
public IEnumerable<string> RequestedProtocols { get; }
/// <inheritdoc />
public string AcceptedProtocol { get; }
/// <inheritdoc />
public string WebSocketVersion { get; }
/// <inheritdoc />
public ICookieCollection Cookies { get; }
/// <inheritdoc />
public IPrincipal User { get; }
/// <inheritdoc />
public bool IsAuthenticated { get; }
/// <inheritdoc />
public bool IsLocal { get; }
/// <inheritdoc />
public bool IsSecureConnection { get; }
/// <inheritdoc />
public IWebSocket WebSocket { get; }
}
}

View File

@@ -0,0 +1,239 @@
using System;
using System.Collections.Generic;
using System.Globalization;
using System.IO;
using EmbedIO.Net.Internal;
using Swan;
namespace EmbedIO.WebSockets.Internal
{
internal class WebSocketFrame
{
internal static readonly byte[] EmptyPingBytes = CreatePingFrame().ToArray();
internal WebSocketFrame(Opcode opcode, PayloadData payloadData)
: this(Fin.Final, opcode, payloadData)
{
}
internal WebSocketFrame(Fin fin, Opcode opcode, byte[] data, bool compressed)
: this(fin, opcode, new PayloadData(data), compressed)
{
}
internal WebSocketFrame(
Fin fin,
Opcode opcode,
PayloadData payloadData,
bool compressed = false)
{
Fin = fin;
Rsv1 = IsOpcodeData(opcode) && compressed ? Rsv.On : Rsv.Off;
Rsv2 = Rsv.Off;
Rsv3 = Rsv.Off;
Opcode = opcode;
var len = payloadData.Length;
if (len < 126)
{
PayloadLength = (byte)len;
ExtendedPayloadLength = Array.Empty<byte>();
}
else if (len < 0x010000)
{
PayloadLength = 126;
ExtendedPayloadLength = ((ushort)len).ToByteArray(Endianness.Big);
}
else
{
PayloadLength = 127;
ExtendedPayloadLength = len.ToByteArray(Endianness.Big);
}
Mask = Mask.Off;
MaskingKey = Array.Empty<byte>();
PayloadData = payloadData;
}
internal WebSocketFrame(
Fin fin,
Rsv rsv1,
Rsv rsv2,
Rsv rsv3,
Opcode opcode,
Mask mask,
byte payloadLength)
{
Fin = fin;
Rsv1 = rsv1;
Rsv2 = rsv2;
Rsv3 = rsv3;
Opcode = opcode;
Mask = mask;
PayloadLength = payloadLength;
}
public byte[]? ExtendedPayloadLength { get; internal set; }
public Fin Fin { get; internal set; }
public bool IsCompressed => Rsv1 == Rsv.On;
public bool IsFragment => Fin == Fin.More || Opcode == Opcode.Cont;
public bool IsMasked => Mask == Mask.On;
public Mask Mask { get; internal set; }
public byte[] MaskingKey { get; internal set; }
public Opcode Opcode { get; internal set; }
public PayloadData PayloadData { get; internal set; }
public byte PayloadLength { get; internal set; }
public Rsv Rsv1 { get; internal set; }
public Rsv Rsv2 { get; internal set; }
public Rsv Rsv3 { get; internal set; }
internal int ExtendedPayloadLengthCount => PayloadLength < 126 ? 0 : (PayloadLength == 126 ? 2 : 8);
internal ulong FullPayloadLength => PayloadLength < 126
? PayloadLength
: PayloadLength == 126
? BitConverter.ToUInt16(ExtendedPayloadLength.ToHostOrder(Endianness.Big), 0)
: BitConverter.ToUInt64(ExtendedPayloadLength.ToHostOrder(Endianness.Big), 0);
public IEnumerator<byte> GetEnumerator() => ((IEnumerable<byte>)ToArray()).GetEnumerator();
public string PrintToString()
{
// Payload Length
var payloadLen = PayloadLength;
// Extended Payload Length
var extPayloadLen = payloadLen > 125 ? FullPayloadLength.ToString(CultureInfo.InvariantCulture) : string.Empty;
// Masking Key
var maskingKey = BitConverter.ToString(MaskingKey);
// Payload Data
var payload = payloadLen == 0
? string.Empty
: payloadLen > 125
? "---"
: Opcode == Opcode.Text && !(IsFragment || IsMasked || IsCompressed)
? PayloadData.ApplicationData.ToArray().ToText()
: PayloadData.ToString();
return $@"
FIN: {Fin}
RSV1: {Rsv1}
RSV2: {Rsv2}
RSV3: {Rsv3}
Opcode: {Opcode}
MASK: {Mask}
Payload Length: {payloadLen}
Extended Payload Length: {extPayloadLen}
Masking Key: {maskingKey}
Payload Data: {payload}";
}
public byte[] ToArray()
{
using var buff = new MemoryStream();
var header = (int)Fin;
header = (header << 1) + (int)Rsv1;
header = (header << 1) + (int)Rsv2;
header = (header << 1) + (int)Rsv3;
header = (header << 4) + (int)Opcode;
header = (header << 1) + (int)Mask;
header = (header << 7) + PayloadLength;
buff.Write(((ushort)header).ToByteArray(Endianness.Big), 0, 2);
if (PayloadLength > 125)
{
buff.Write(ExtendedPayloadLength, 0, PayloadLength == 126 ? 2 : 8);
}
if (Mask == Mask.On)
{
buff.Write(MaskingKey, 0, 4);
}
if (PayloadLength > 0)
{
var bytes = PayloadData.ToArray();
if (PayloadLength < 127)
{
buff.Write(bytes, 0, bytes.Length);
}
else
{
using var input = new MemoryStream(bytes);
input.CopyTo(buff, 1024);
}
}
return buff.ToArray();
}
public override string ToString() => BitConverter.ToString(ToArray());
internal static WebSocketFrame CreateCloseFrame(PayloadData? payloadData) => new (Fin.Final, Opcode.Close, payloadData ?? new PayloadData());
internal static WebSocketFrame CreatePingFrame() => new (Fin.Final, Opcode.Ping, new PayloadData());
internal static WebSocketFrame CreatePingFrame(byte[] data) => new (Fin.Final, Opcode.Ping, new PayloadData(data));
internal void Validate(WebSocket webSocket)
{
if (!IsMasked)
{
throw new WebSocketException(CloseStatusCode.ProtocolError, "A frame from a client isn't masked.");
}
if (webSocket.InContinuation && (Opcode == Opcode.Text || Opcode == Opcode.Binary))
{
throw new WebSocketException(CloseStatusCode.ProtocolError,
"A data frame has been received while receiving continuation frames.");
}
if (IsCompressed && webSocket.Compression == CompressionMethod.None)
{
throw new WebSocketException(CloseStatusCode.ProtocolError,
"A compressed frame has been received without any agreement for it.");
}
if (Rsv2 == Rsv.On)
{
throw new WebSocketException(CloseStatusCode.ProtocolError,
"The RSV2 of a frame is non-zero without any negotiation for it.");
}
if (Rsv3 == Rsv.On)
{
throw new WebSocketException(CloseStatusCode.ProtocolError,
"The RSV3 of a frame is non-zero without any negotiation for it.");
}
}
internal void Unmask()
{
if (Mask == Mask.Off)
{
return;
}
Mask = Mask.Off;
PayloadData.Mask(MaskingKey);
MaskingKey = Array.Empty<byte>();
}
private static bool IsOpcodeData(Opcode opcode) => opcode == Opcode.Text || opcode == Opcode.Binary;
}
}

View File

@@ -0,0 +1,148 @@
using System;
using System.IO;
using System.Threading.Tasks;
using Swan;
namespace EmbedIO.WebSockets.Internal
{
internal class WebSocketFrameStream
{
private readonly bool _unmask;
private readonly Stream? _stream;
public WebSocketFrameStream(Stream? stream, bool unmask = false)
{
_stream = stream;
_unmask = unmask;
}
internal async Task<WebSocketFrame?> ReadFrameAsync(WebSocket webSocket)
{
if (_stream == null) return null;
var frame = ProcessHeader(await _stream.ReadBytesAsync(2).ConfigureAwait(false));
await ReadExtendedPayloadLengthAsync(frame).ConfigureAwait(false);
await ReadMaskingKeyAsync(frame).ConfigureAwait(false);
await ReadPayloadDataAsync(frame).ConfigureAwait(false);
if (_unmask)
frame.Unmask();
frame.Validate(webSocket);
frame.Unmask();
return frame;
}
private static bool IsOpcodeData(byte opcode) => opcode == 0x1 || opcode == 0x2;
private static bool IsOpcodeControl(byte opcode) => opcode > 0x7 && opcode < 0x10;
private static WebSocketFrame ProcessHeader(byte[] header)
{
if (header.Length != 2)
throw new WebSocketException("The header of a frame cannot be read from the stream.");
// FIN
var fin = (header[0] & 0x80) == 0x80 ? Fin.Final : Fin.More;
// RSV1
var rsv1 = (header[0] & 0x40) == 0x40 ? Rsv.On : Rsv.Off;
// RSV2
var rsv2 = (header[0] & 0x20) == 0x20 ? Rsv.On : Rsv.Off;
// RSV3
var rsv3 = (header[0] & 0x10) == 0x10 ? Rsv.On : Rsv.Off;
// Opcode
var opcode = (byte)(header[0] & 0x0f);
// MASK
var mask = (header[1] & 0x80) == 0x80 ? Mask.On : Mask.Off;
// Payload Length
var payloadLen = (byte)(header[1] & 0x7f);
var err = !Enum.IsDefined(typeof(Opcode), opcode) ? "An unsupported opcode."
: !IsOpcodeData(opcode) && rsv1 == Rsv.On ? "A non data frame is compressed."
: IsOpcodeControl(opcode) && fin == Fin.More ? "A control frame is fragmented."
: IsOpcodeControl(opcode) && payloadLen > 125 ? "A control frame has a long payload length."
: null;
if (err != null)
throw new WebSocketException(CloseStatusCode.ProtocolError, err);
return new WebSocketFrame(fin, rsv1, rsv2, rsv3, (Opcode)opcode, mask, payloadLen);
}
private async Task ReadExtendedPayloadLengthAsync(WebSocketFrame frame)
{
var len = frame.ExtendedPayloadLengthCount;
if (len == 0)
{
frame.ExtendedPayloadLength = Array.Empty<byte>();
return;
}
var bytes = await _stream.ReadBytesAsync(len).ConfigureAwait(false);
if (bytes.Length != len)
{
throw new WebSocketException(
"The extended payload length of a frame cannot be read from the stream.");
}
frame.ExtendedPayloadLength = bytes;
}
private async Task ReadMaskingKeyAsync(WebSocketFrame frame)
{
var len = frame.IsMasked ? 4 : 0;
if (len == 0)
{
frame.MaskingKey = Array.Empty<byte>();
return;
}
var bytes = await _stream.ReadBytesAsync(len).ConfigureAwait(false);
if (bytes.Length != len)
{
throw new WebSocketException(
"The masking key of a frame cannot be read from the stream.");
}
frame.MaskingKey = bytes;
}
private async Task ReadPayloadDataAsync(WebSocketFrame frame)
{
var len = frame.FullPayloadLength;
if (len == 0)
{
frame.PayloadData = new PayloadData();
return;
}
if (len > PayloadData.MaxLength)
throw new WebSocketException(CloseStatusCode.TooBig, "A frame has a long payload length.");
var bytes = frame.PayloadLength < 127
? await _stream.ReadBytesAsync((int)len).ConfigureAwait(false)
: await _stream.ReadBytesAsync((int)len, 1024).ConfigureAwait(false);
if (bytes.Length != (int)len)
{
throw new WebSocketException(
"The payload data of a frame cannot be read from the stream.");
}
frame.PayloadData = new PayloadData(bytes);
}
}
}

View File

@@ -0,0 +1,29 @@
using System;
namespace EmbedIO.WebSockets.Internal
{
/// <summary>
/// Represents a WS Receive result.
/// </summary>
internal sealed class WebSocketReceiveResult : IWebSocketReceiveResult
{
internal WebSocketReceiveResult(int count, Opcode code)
{
if (count < 0)
throw new ArgumentOutOfRangeException(nameof(count));
Count = count;
EndOfMessage = code == Opcode.Close;
MessageType = code == Opcode.Text ? 0 : 1;
}
/// <inheritdoc />
public int Count { get; }
/// <inheritdoc />
public bool EndOfMessage { get; }
/// <inheritdoc />
public int MessageType { get; }
}
}

View File

@@ -0,0 +1,91 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using Swan;
namespace EmbedIO.WebSockets.Internal
{
internal class WebSocketStream : MemoryStream
{
internal const int FragmentLength = 1016;
private readonly CompressionMethod _compression;
private readonly Opcode _opcode;
public WebSocketStream(byte[] data, Opcode opcode, CompressionMethod compression)
: base(data)
{
_compression = compression;
_opcode = opcode;
}
public IEnumerable<WebSocketFrame> GetFrames()
{
var compressed = _compression != CompressionMethod.None;
var stream = compressed
? this.CompressAsync(_compression, true, CancellationToken.None).Await()
: this;
var len = stream.Length;
/* Not fragmented */
if (len == 0)
{
yield return new WebSocketFrame(Fin.Final, _opcode, Array.Empty<byte>(), compressed);
yield break;
}
var quo = len / FragmentLength;
var rem = (int)(len % FragmentLength);
byte[] buff;
if (quo == 0)
{
buff = new byte[rem];
if (stream.Read(buff, 0, rem) == rem)
yield return new WebSocketFrame(Fin.Final, _opcode, buff, compressed);
yield break;
}
buff = new byte[FragmentLength];
if (quo == 1 && rem == 0)
{
if (stream.Read(buff, 0, FragmentLength) == FragmentLength)
yield return new WebSocketFrame(Fin.Final, _opcode, buff, compressed);
yield break;
}
/* Send fragmented */
// Begin
if (stream.Read(buff, 0, FragmentLength) != FragmentLength)
yield break;
yield return new WebSocketFrame(Fin.More, _opcode, buff, compressed);
var n = rem == 0 ? quo - 2 : quo - 1;
for (var i = 0; i < n; i++)
{
if (stream.Read(buff, 0, FragmentLength) != FragmentLength)
yield break;
yield return new WebSocketFrame(Fin.More, Opcode.Cont, buff, compressed);
}
// End
if (rem == 0)
rem = FragmentLength;
else
buff = new byte[rem];
if (stream.Read(buff, 0, rem) == rem)
yield return new WebSocketFrame(Fin.Final, Opcode.Cont, buff, compressed);
}
}
}