starred/jellyfin-jellyfin-1
1
0
Fork 0
mirror of https://github.com/jellyfin/jellyfin.git synced 2025-12-21 08:15:28 +03:00
Code Issues Packages Projects Releases Wiki Activity

Rewrite WebSocket handling code

Browse Source
This commit is contained in:
Bond_009
2019-12-17 23:15:02 +01:00
committed by Bond_009
parent ce7744806c
commit 976459d3e8
23 changed files with 325 additions and 971 deletions
Download Patch File Download Diff File Expand all files Collapse all files

233
Emby.Server.Implementations/HttpServer/WebSocketConnection.cs
View File

@@ -1,15 +1,16 @@
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 MediaBrowser.Common.Json;
using MediaBrowser.Controller.Net;
using MediaBrowser.Model.Net;
using MediaBrowser.Model.Serialization;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging;
using UtfUnknown;
namespace Emby.Server.Implementations.HttpServer
{
@@ -24,54 +25,46 @@ namespace Emby.Server.Implementations.HttpServer
private readonly ILogger _logger;
/// <summary>
/// The json serializer.
/// The json serializer options.
/// </summary>
private readonly IJsonSerializer _jsonSerializer;
private readonly JsonSerializerOptions _jsonOptions;
/// <summary>
/// The socket.
/// </summary>
private readonly IWebSocket _socket;
private readonly WebSocket _socket;
private bool _disposed = false;
/// <summary>
/// Initializes a new instance of the <see cref="WebSocketConnection" /> class.
/// </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)
public WebSocketConnection(ILogger<WebSocketConnection> logger, WebSocket socket, IPAddress remoteEndPoint)
{
if (socket == null)
{
throw new ArgumentNullException(nameof(socket));
}
if (string.IsNullOrEmpty(remoteEndPoint))
if (remoteEndPoint != null)
{
throw new ArgumentNullException(nameof(remoteEndPoint));
}
if (jsonSerializer == null)
{
throw new ArgumentNullException(nameof(jsonSerializer));
}
if (logger == null)
{
throw new ArgumentNullException(nameof(logger));
}
Id = Guid.NewGuid();
_jsonSerializer = jsonSerializer;
_socket = socket;
_socket.OnReceiveBytes = OnReceiveInternal;
RemoteEndPoint = remoteEndPoint;
_logger = logger;
socket.Closed += OnSocketClosed;
_jsonOptions = JsonDefaults.GetOptions();
}
/// <inheritdoc />
@@ -80,7 +73,7 @@ namespace Emby.Server.Implementations.HttpServer
/// <summary>
/// Gets or sets the remote end point.
/// </summary>
public string RemoteEndPoint { get; private set; }
public IPAddress RemoteEndPoint { get; private set; }
/// <summary>
/// Gets or sets the receive action.
@@ -94,12 +87,6 @@ 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; }
/// <summary>
/// Gets or sets the URL.
/// </summary>
@@ -118,70 +105,6 @@ namespace Emby.Server.Implementations.HttpServer
/// <value>The state.</value>
public WebSocketState State => _socket.State;
void OnSocketClosed(object sender, EventArgs e)
{
Closed?.Invoke(this, EventArgs.Empty);
}
/// <summary>
/// Called when [receive].
/// </summary>
/// <param name="bytes">The bytes.</param>
private void OnReceiveInternal(byte[] bytes)
{
LastActivityDate = DateTime.UtcNow;
if (OnReceive == null)
{
return;
}
var charset = CharsetDetector.DetectFromBytes(bytes).Detected?.EncodingName;
if (string.Equals(charset, "utf-8", StringComparison.OrdinalIgnoreCase))
{
OnReceiveInternal(Encoding.UTF8.GetString(bytes, 0, bytes.Length));
}
else
{
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);
return;
}
if (OnReceive == null)
{
return;
}
try
{
var stub = (WebSocketMessage<object>)_jsonSerializer.DeserializeFromString(message, typeof(WebSocketMessage<object>));
var info = new WebSocketMessageInfo
{
MessageType = stub.MessageType,
Data = stub.Data?.ToString(),
Connection = this
};
OnReceive(info);
}
catch (Exception ex)
{
_logger.LogError(ex, "Error processing web socket message");
}
}
/// <summary>
/// Sends a message asynchronously.
/// </summary>
@@ -197,40 +120,109 @@ namespace Emby.Server.Implementations.HttpServer
throw new ArgumentNullException(nameof(message));
}
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)
{
if (buffer == null)
{
throw new ArgumentNullException(nameof(buffer));
}
cancellationToken.ThrowIfCancellationRequested();
return _socket.SendAsync(buffer, true, cancellationToken);
var json = JsonSerializer.SerializeToUtf8Bytes(message, _jsonOptions);
return _socket.SendAsync(json, WebSocketMessageType.Text, true, cancellationToken);
}
/// <inheritdoc />
public Task SendAsync(string text, CancellationToken cancellationToken)
public async Task ProcessAsync(CancellationToken cancellationToken = default)
{
if (string.IsNullOrEmpty(text))
var pipe = new Pipe();
var writer = pipe.Writer;
ValueWebSocketReceiveResult receiveresult;
do
{
throw new ArgumentNullException(nameof(text));
// Allocate at least 512 bytes from the PipeWriter
Memory<byte> memory = writer.GetMemory(512);
receiveresult = await _socket.ReceiveAsync(memory, cancellationToken);
int bytesRead = receiveresult.Count;
if (bytesRead == 0)
{
continue;
}
// Tell the PipeWriter how much was read from the Socket
writer.Advance(bytesRead);
// Make the data available to the PipeReader
FlushResult flushResult = await writer.FlushAsync();
if (flushResult.IsCompleted)
{
// The PipeReader stopped reading
break;
}
if (receiveresult.EndOfMessage)
{
await ProcessInternal(pipe.Reader).ConfigureAwait(false);
}
} while (_socket.State == WebSocketState.Open && receiveresult.MessageType != WebSocketMessageType.Close);
if (_socket.State == WebSocketState.Open)
{
await _socket.CloseAsync(
WebSocketCloseStatus.NormalClosure,
string.Empty, // REVIEW: human readable explanation as to why the connection is closed.
cancellationToken).ConfigureAwait(false);
}
cancellationToken.ThrowIfCancellationRequested();
Closed?.Invoke(this, EventArgs.Empty);
return _socket.SendAsync(text, true, cancellationToken);
_socket.Dispose();
}
private async Task ProcessInternal(PipeReader reader)
{
LastActivityDate = DateTime.UtcNow;
if (OnReceive == null)
{
return;
}
try
{
var result = await reader.ReadAsync().ConfigureAwait(false);
if (!result.IsCompleted)
{
return;
}
WebSocketMessage<object> stub;
var buffer = result.Buffer;
if (buffer.IsSingleSegment)
{
stub = JsonSerializer.Deserialize<WebSocketMessage<object>>(buffer.FirstSpan, _jsonOptions);
}
else
{
var buf = ArrayPool<byte>.Shared.Rent(Convert.ToInt32(buffer.Length));
try
{
buffer.CopyTo(buf);
stub = JsonSerializer.Deserialize<WebSocketMessage<object>>(buf, _jsonOptions);
}
finally
{
ArrayPool<byte>.Shared.Return(buf);
}
}
var info = new WebSocketMessageInfo
{
MessageType = stub.MessageType,
Data = stub.Data.ToString(),
Connection = this
};
await OnReceive(info).ConfigureAwait(false);
}
catch (JsonException ex)
{
_logger.LogError(ex, "Error processing web socket message");
}
}
/// <inheritdoc />
@@ -246,10 +238,17 @@ namespace Emby.Server.Implementations.HttpServer
/// <param name="dispose"><c>true</c> to release both managed and unmanaged resources; <c>false</c> to release only unmanaged resources.</param>
protected virtual void Dispose(bool dispose)
{
if (_disposed)
{
return;
}
if (dispose)
{
_socket.Dispose();
}
_disposed = true;
}
}
}