diff options
Diffstat (limited to 'Emby.Server.Implementations/HttpServer/WebSocketConnection.cs')
| -rw-r--r-- | Emby.Server.Implementations/HttpServer/WebSocketConnection.cs | 327 |
1 files changed, 152 insertions, 175 deletions
diff --git a/Emby.Server.Implementations/HttpServer/WebSocketConnection.cs b/Emby.Server.Implementations/HttpServer/WebSocketConnection.cs index e9d0bac74..5f25f6980 100644 --- a/Emby.Server.Implementations/HttpServer/WebSocketConnection.cs +++ b/Emby.Server.Implementations/HttpServer/WebSocketConnection.cs @@ -1,50 +1,76 @@ -using System; +using System; +using System.Buffers; +using System.IO.Pipelines; +using System.Net; using System.Net.WebSockets; using System.Text; +using System.Text.Json; using System.Threading; using System.Threading.Tasks; -using Emby.Server.Implementations.Net; +using Jellyfin.Extensions.Json; using MediaBrowser.Controller.Net; using MediaBrowser.Model.Net; -using MediaBrowser.Model.Serialization; -using MediaBrowser.Model.Services; +using MediaBrowser.Model.Session; +using Microsoft.AspNetCore.Http; using Microsoft.Extensions.Logging; -using UtfUnknown; namespace Emby.Server.Implementations.HttpServer { /// <summary> - /// Class WebSocketConnection + /// Class WebSocketConnection. /// </summary> - public class WebSocketConnection : IWebSocketConnection + public class WebSocketConnection : IWebSocketConnection, IDisposable { - public event EventHandler<EventArgs> Closed; + /// <summary> + /// The logger. + /// </summary> + private readonly ILogger<WebSocketConnection> _logger; /// <summary> - /// The _socket + /// The json serializer options. /// </summary> - private readonly IWebSocket _socket; + private readonly JsonSerializerOptions _jsonOptions; /// <summary> - /// The _remote end point + /// The socket. /// </summary> - public string RemoteEndPoint { get; private set; } + private readonly WebSocket _socket; /// <summary> - /// The logger + /// Initializes a new instance of the <see cref="WebSocketConnection" /> class. /// </summary> - private readonly ILogger _logger; + /// <param name="logger">The logger.</param> + /// <param name="socket">The socket.</param> + /// <param name="remoteEndPoint">The remote end point.</param> + /// <param name="query">The query.</param> + public WebSocketConnection( + ILogger<WebSocketConnection> logger, + WebSocket socket, + IPAddress? remoteEndPoint, + IQueryCollection query) + { + _logger = logger; + _socket = socket; + RemoteEndPoint = remoteEndPoint; + QueryString = query; + + _jsonOptions = JsonDefaults.Options; + LastActivityDate = DateTime.Now; + } + + /// <inheritdoc /> + public event EventHandler<EventArgs>? Closed; /// <summary> - /// The _json serializer + /// Gets the remote end point. /// </summary> - private readonly IJsonSerializer _jsonSerializer; + public IPAddress? RemoteEndPoint { get; } /// <summary> /// Gets or sets the receive action. /// </summary> /// <value>The receive action.</value> - public Func<WebSocketMessageInfo, Task> OnReceive { get; set; } + public Func<WebSocketMessageInfo, Task>? OnReceive { get; set; } /// <summary> /// Gets the last activity date. @@ -52,221 +78,172 @@ namespace Emby.Server.Implementations.HttpServer /// <value>The last activity date.</value> public DateTime LastActivityDate { get; private set; } - /// <summary> - /// Gets the id. - /// </summary> - /// <value>The id.</value> - public Guid Id { get; private set; } + /// <inheritdoc /> + public DateTime LastKeepAliveDate { get; set; } /// <summary> - /// Gets or sets the URL. + /// Gets the query string. /// </summary> - /// <value>The URL.</value> - public string Url { get; set; } + /// <value>The query string.</value> + public IQueryCollection QueryString { get; } + /// <summary> - /// Gets or sets the query string. + /// Gets the state. /// </summary> - /// <value>The query string.</value> - public QueryParamCollection QueryString { get; set; } + /// <value>The state.</value> + public WebSocketState State => _socket.State; /// <summary> - /// Initializes a new instance of the <see cref="WebSocketConnection" /> class. + /// Sends a message asynchronously. /// </summary> - /// <param name="socket">The socket.</param> - /// <param name="remoteEndPoint">The remote end point.</param> - /// <param name="jsonSerializer">The json serializer.</param> - /// <param name="logger">The logger.</param> - /// <exception cref="ArgumentNullException">socket</exception> - public WebSocketConnection(IWebSocket socket, string remoteEndPoint, IJsonSerializer jsonSerializer, ILogger logger) + /// <typeparam name="T">The type of the message.</typeparam> + /// <param name="message">The message.</param> + /// <param name="cancellationToken">The cancellation token.</param> + /// <returns>Task.</returns> + public Task SendAsync<T>(WebSocketMessage<T> message, CancellationToken cancellationToken) { - if (socket == null) - { - throw new ArgumentNullException(nameof(socket)); - } - if (string.IsNullOrEmpty(remoteEndPoint)) - { - throw new ArgumentNullException(nameof(remoteEndPoint)); - } - if (jsonSerializer == null) - { - throw new ArgumentNullException(nameof(jsonSerializer)); - } - if (logger == null) - { - throw new ArgumentNullException(nameof(logger)); - } + var json = JsonSerializer.SerializeToUtf8Bytes(message, _jsonOptions); + return _socket.SendAsync(json, WebSocketMessageType.Text, true, cancellationToken); + } - Id = Guid.NewGuid(); - _jsonSerializer = jsonSerializer; - _socket = socket; - _socket.OnReceiveBytes = OnReceiveInternal; + /// <inheritdoc /> + public async Task ProcessAsync(CancellationToken cancellationToken = default) + { + var pipe = new Pipe(); + var writer = pipe.Writer; - var memorySocket = socket as IMemoryWebSocket; - if (memorySocket != null) + ValueWebSocketReceiveResult receiveresult; + do { - memorySocket.OnReceiveMemoryBytes = OnReceiveInternal; - } + // Allocate at least 512 bytes from the PipeWriter + Memory<byte> memory = writer.GetMemory(512); + try + { + receiveresult = await _socket.ReceiveAsync(memory, cancellationToken).ConfigureAwait(false); + } + catch (WebSocketException ex) + { + _logger.LogWarning("WS {IP} error receiving data: {Message}", RemoteEndPoint, ex.Message); + break; + } - RemoteEndPoint = remoteEndPoint; - _logger = logger; + int bytesRead = receiveresult.Count; + if (bytesRead == 0) + { + break; + } - socket.Closed += socket_Closed; - } + // Tell the PipeWriter how much was read from the Socket + writer.Advance(bytesRead); - void socket_Closed(object sender, EventArgs e) - { - Closed?.Invoke(this, EventArgs.Empty); - } + // Make the data available to the PipeReader + FlushResult flushResult = await writer.FlushAsync(cancellationToken).ConfigureAwait(false); + if (flushResult.IsCompleted) + { + // The PipeReader stopped reading + break; + } - /// <summary> - /// Called when [receive]. - /// </summary> - /// <param name="bytes">The bytes.</param> - private void OnReceiveInternal(byte[] bytes) - { - LastActivityDate = DateTime.UtcNow; + LastActivityDate = DateTime.UtcNow; - if (OnReceive == null) - { - return; + if (receiveresult.EndOfMessage) + { + await ProcessInternal(pipe.Reader).ConfigureAwait(false); + } } - var charset = CharsetDetector.DetectFromBytes(bytes).Detected?.EncodingName; + while ((_socket.State == WebSocketState.Open || _socket.State == WebSocketState.Connecting) + && receiveresult.MessageType != WebSocketMessageType.Close); - if (string.Equals(charset, "utf-8", StringComparison.OrdinalIgnoreCase)) - { - OnReceiveInternal(Encoding.UTF8.GetString(bytes, 0, bytes.Length)); - } - else + Closed?.Invoke(this, EventArgs.Empty); + + if (_socket.State == WebSocketState.Open + || _socket.State == WebSocketState.CloseReceived + || _socket.State == WebSocketState.CloseSent) { - OnReceiveInternal(Encoding.ASCII.GetString(bytes, 0, bytes.Length)); + await _socket.CloseAsync( + WebSocketCloseStatus.NormalClosure, + string.Empty, + cancellationToken).ConfigureAwait(false); } } - /// <summary> - /// Called when [receive]. - /// </summary> - /// <param name="memory">The memory block.</param> - /// <param name="length">The length of the memory block.</param> - private void OnReceiveInternal(Memory<byte> memory, int length) + private async Task ProcessInternal(PipeReader reader) { - LastActivityDate = DateTime.UtcNow; + ReadResult result = await reader.ReadAsync().ConfigureAwait(false); + ReadOnlySequence<byte> buffer = result.Buffer; if (OnReceive == null) { + // Tell the PipeReader how much of the buffer we have consumed + reader.AdvanceTo(buffer.End); return; } - var bytes = memory.Slice(0, length).ToArray(); - - var charset = CharsetDetector.DetectFromBytes(bytes).Detected?.EncodingName; - - if (string.Equals(charset, "utf-8", StringComparison.OrdinalIgnoreCase)) + WebSocketMessage<object>? stub; + long bytesConsumed = 0; + try { - OnReceiveInternal(Encoding.UTF8.GetString(bytes, 0, bytes.Length)); + stub = DeserializeWebSocketMessage(buffer, out bytesConsumed); } - else + catch (JsonException ex) { - OnReceiveInternal(Encoding.ASCII.GetString(bytes, 0, bytes.Length)); - } - } - - private void OnReceiveInternal(string message) - { - LastActivityDate = DateTime.UtcNow; - - if (!message.StartsWith("{", StringComparison.OrdinalIgnoreCase)) - { - // This info is useful sometimes but also clogs up the log - _logger.LogDebug("Received web socket message that is not a json structure: {message}", message); + // Tell the PipeReader how much of the buffer we have consumed + reader.AdvanceTo(buffer.End); + _logger.LogError(ex, "Error processing web socket message: {Data}", Encoding.UTF8.GetString(buffer)); return; } - if (OnReceive == null) + if (stub == null) { + _logger.LogError("Error processing web socket message"); return; } - try - { - var stub = (WebSocketMessage<object>)_jsonSerializer.DeserializeFromString(message, typeof(WebSocketMessage<object>)); + // Tell the PipeReader how much of the buffer we have consumed + reader.AdvanceTo(buffer.GetPosition(bytesConsumed)); - var info = new WebSocketMessageInfo - { - MessageType = stub.MessageType, - Data = stub.Data == null ? null : stub.Data.ToString(), - Connection = this - }; + _logger.LogDebug("WS {IP} received message: {@Message}", RemoteEndPoint, stub); - OnReceive(info); - } - catch (Exception ex) + if (stub.MessageType == SessionMessageType.KeepAlive) { - _logger.LogError(ex, "Error processing web socket message"); + await SendKeepAliveResponse().ConfigureAwait(false); } - } - - /// <summary> - /// Sends a message asynchronously. - /// </summary> - /// <typeparam name="T"></typeparam> - /// <param name="message">The message.</param> - /// <param name="cancellationToken">The cancellation token.</param> - /// <returns>Task.</returns> - /// <exception cref="ArgumentNullException">message</exception> - public Task SendAsync<T>(WebSocketMessage<T> message, CancellationToken cancellationToken) - { - if (message == null) + else { - throw new ArgumentNullException(nameof(message)); + await OnReceive( + new WebSocketMessageInfo + { + MessageType = stub.MessageType, + Data = stub.Data?.ToString(), // Data can be null + Connection = this + }).ConfigureAwait(false); } - - var json = _jsonSerializer.SerializeToString(message); - - return SendAsync(json, cancellationToken); } - /// <summary> - /// Sends a message asynchronously. - /// </summary> - /// <param name="buffer">The buffer.</param> - /// <param name="cancellationToken">The cancellation token.</param> - /// <returns>Task.</returns> - public Task SendAsync(byte[] buffer, CancellationToken cancellationToken) + internal WebSocketMessage<object>? DeserializeWebSocketMessage(ReadOnlySequence<byte> bytes, out long bytesConsumed) { - if (buffer == null) - { - throw new ArgumentNullException(nameof(buffer)); - } - - cancellationToken.ThrowIfCancellationRequested(); - - return _socket.SendAsync(buffer, true, cancellationToken); + var jsonReader = new Utf8JsonReader(bytes); + var ret = JsonSerializer.Deserialize<WebSocketMessage<object>>(ref jsonReader, _jsonOptions); + bytesConsumed = jsonReader.BytesConsumed; + return ret; } - public Task SendAsync(string text, CancellationToken cancellationToken) + private Task SendKeepAliveResponse() { - if (string.IsNullOrEmpty(text)) - { - throw new ArgumentNullException(nameof(text)); - } - - cancellationToken.ThrowIfCancellationRequested(); - - return _socket.SendAsync(text, true, cancellationToken); + LastKeepAliveDate = DateTime.UtcNow; + return SendAsync( + new WebSocketMessage<string> + { + MessageId = Guid.NewGuid(), + MessageType = SessionMessageType.KeepAlive + }, CancellationToken.None); } - /// <summary> - /// Gets the state. - /// </summary> - /// <value>The state.</value> - public WebSocketState State => _socket.State; - - /// <summary> - /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. - /// </summary> + /// <inheritdoc /> public void Dispose() { Dispose(true); + GC.SuppressFinalize(this); } /// <summary> |