From 65e6211c035c2269584220f1a3dcc0bb37374e01 Mon Sep 17 00:00:00 2001 From: cvium Date: Sat, 28 Nov 2020 11:21:53 +0100 Subject: Remove circular dependency between websocket listeners and manager --- .../HttpServer/WebSocketManager.cs | 41 ++++------- .../Session/SessionWebSocketListener.cs | 80 ++++++++++------------ 2 files changed, 51 insertions(+), 70 deletions(-) (limited to 'Emby.Server.Implementations') diff --git a/Emby.Server.Implementations/HttpServer/WebSocketManager.cs b/Emby.Server.Implementations/HttpServer/WebSocketManager.cs index 71ece80a75..d6cf6233e4 100644 --- a/Emby.Server.Implementations/HttpServer/WebSocketManager.cs +++ b/Emby.Server.Implementations/HttpServer/WebSocketManager.cs @@ -2,9 +2,9 @@ using System; using System.Collections.Generic; +using System.Linq; using System.Net.WebSockets; using System.Threading.Tasks; -using Jellyfin.Data.Events; using MediaBrowser.Controller.Net; using Microsoft.AspNetCore.Http; using Microsoft.Extensions.Logging; @@ -13,32 +13,23 @@ namespace Emby.Server.Implementations.HttpServer { public class WebSocketManager : IWebSocketManager { - private readonly Lazy> _webSocketListeners; + private readonly IWebSocketListener[] _webSocketListeners; private readonly ILogger _logger; private readonly ILoggerFactory _loggerFactory; - private bool _disposed = false; - public WebSocketManager( - Lazy> webSocketListeners, + IEnumerable webSocketListeners, ILogger logger, ILoggerFactory loggerFactory) { - _webSocketListeners = webSocketListeners; + _webSocketListeners = webSocketListeners.ToArray(); _logger = logger; _loggerFactory = loggerFactory; } - public event EventHandler> WebSocketConnected; - /// public async Task WebSocketRequestHandler(HttpContext context) { - if (_disposed) - { - return; - } - try { _logger.LogInformation("WS {IP} request", context.Connection.RemoteIpAddress); @@ -54,7 +45,13 @@ namespace Emby.Server.Implementations.HttpServer OnReceive = ProcessWebSocketMessageReceived }; - WebSocketConnected?.Invoke(this, new GenericEventArgs(connection)); + var tasks = new Task[_webSocketListeners.Length]; + for (var i = 0; i < _webSocketListeners.Length; ++i) + { + tasks[i] = _webSocketListeners[i].ProcessWebSocketConnectedAsync(connection); + } + + await Task.WhenAll(tasks).ConfigureAwait(false); await connection.ProcessAsync().ConfigureAwait(false); _logger.LogInformation("WS {IP} closed", context.Connection.RemoteIpAddress); @@ -75,21 +72,13 @@ namespace Emby.Server.Implementations.HttpServer /// The result. private Task ProcessWebSocketMessageReceived(WebSocketMessageInfo result) { - if (_disposed) - { - return Task.CompletedTask; - } - - IEnumerable GetTasks() + var tasks = new Task[_webSocketListeners.Length]; + for (var i = 0; i < _webSocketListeners.Length; ++i) { - var listeners = _webSocketListeners.Value; - foreach (var x in listeners) - { - yield return x.ProcessMessageAsync(result); - } + tasks[i] = _webSocketListeners[i].ProcessMessageAsync(result); } - return Task.WhenAll(GetTasks()); + return Task.WhenAll(tasks); } } } diff --git a/Emby.Server.Implementations/Session/SessionWebSocketListener.cs b/Emby.Server.Implementations/Session/SessionWebSocketListener.cs index a5f8479537..169eaefd8b 100644 --- a/Emby.Server.Implementations/Session/SessionWebSocketListener.cs +++ b/Emby.Server.Implementations/Session/SessionWebSocketListener.cs @@ -4,7 +4,6 @@ using System.Linq; using System.Net.WebSockets; using System.Threading; using System.Threading.Tasks; -using Jellyfin.Data.Events; using MediaBrowser.Controller.Net; using MediaBrowser.Controller.Session; using MediaBrowser.Model.Net; @@ -22,50 +21,48 @@ namespace Emby.Server.Implementations.Session /// /// The timeout in seconds after which a WebSocket is considered to be lost. /// - public const int WebSocketLostTimeout = 60; + private const int WebSocketLostTimeout = 60; /// /// The keep-alive interval factor; controls how often the watcher will check on the status of the WebSockets. /// - public const float IntervalFactor = 0.2f; + private const float IntervalFactor = 0.2f; /// /// The ForceKeepAlive factor; controls when a ForceKeepAlive is sent. /// - public const float ForceKeepAliveFactor = 0.75f; + private const float ForceKeepAliveFactor = 0.75f; /// - /// The _session manager. + /// Lock used for accesing the KeepAlive cancellation token. /// - private readonly ISessionManager _sessionManager; + private readonly object _keepAliveLock = new object(); /// - /// The _logger. + /// The WebSocket watchlist. /// - private readonly ILogger _logger; - private readonly ILoggerFactory _loggerFactory; - - private readonly IWebSocketManager _webSocketManager; + private readonly HashSet _webSockets = new HashSet(); /// - /// The KeepAlive cancellation token. + /// Lock used for accessing the WebSockets watchlist. /// - private CancellationTokenSource _keepAliveCancellationToken; + private readonly object _webSocketsLock = new object(); /// - /// Lock used for accesing the KeepAlive cancellation token. + /// The _session manager. /// - private readonly object _keepAliveLock = new object(); + private readonly ISessionManager _sessionManager; /// - /// The WebSocket watchlist. + /// The _logger. /// - private readonly HashSet _webSockets = new HashSet(); + private readonly ILogger _logger; + private readonly ILoggerFactory _loggerFactory; /// - /// Lock used for accesing the WebSockets watchlist. + /// The KeepAlive cancellation token. /// - private readonly object _webSocketsLock = new object(); + private CancellationTokenSource _keepAliveCancellationToken; /// /// Initializes a new instance of the class. @@ -73,32 +70,42 @@ namespace Emby.Server.Implementations.Session /// The logger. /// The session manager. /// The logger factory. - /// The HTTP server. public SessionWebSocketListener( ILogger logger, ISessionManager sessionManager, - ILoggerFactory loggerFactory, - IWebSocketManager webSocketManager) + ILoggerFactory loggerFactory) { _logger = logger; _sessionManager = sessionManager; _loggerFactory = loggerFactory; - _webSocketManager = webSocketManager; + } - webSocketManager.WebSocketConnected += OnServerManagerWebSocketConnected; + /// + public void Dispose() + { + StopKeepAlive(); } - private async void OnServerManagerWebSocketConnected(object sender, GenericEventArgs e) + /// + /// Processes the message. + /// + /// The message. + /// Task. + public Task ProcessMessageAsync(WebSocketMessageInfo message) + => Task.CompletedTask; + + /// + public async Task ProcessWebSocketConnectedAsync(IWebSocketConnection connection) { - var session = GetSession(e.Argument.QueryString, e.Argument.RemoteEndPoint.ToString()); + var session = GetSession(connection.QueryString, connection.RemoteEndPoint.ToString()); if (session != null) { - EnsureController(session, e.Argument); - await KeepAliveWebSocket(e.Argument).ConfigureAwait(false); + EnsureController(session, connection); + await KeepAliveWebSocket(connection).ConfigureAwait(false); } else { - _logger.LogWarning("Unable to determine session based on query string: {0}", e.Argument.QueryString); + _logger.LogWarning("Unable to determine session based on query string: {0}", connection.QueryString); } } @@ -119,21 +126,6 @@ namespace Emby.Server.Implementations.Session return _sessionManager.GetSessionByAuthenticationToken(token, deviceId, remoteEndpoint); } - /// - public void Dispose() - { - _webSocketManager.WebSocketConnected -= OnServerManagerWebSocketConnected; - StopKeepAlive(); - } - - /// - /// Processes the message. - /// - /// The message. - /// Task. - public Task ProcessMessageAsync(WebSocketMessageInfo message) - => Task.CompletedTask; - private void EnsureController(SessionInfo session, IWebSocketConnection connection) { var controllerInfo = session.EnsureController( -- cgit v1.2.3