#nullable disable using System; using System.Collections.Generic; using System.Linq; using System.Net.WebSockets; using System.Threading; using System.Threading.Tasks; using Jellyfin.Api.Extensions; using MediaBrowser.Controller.Net; using MediaBrowser.Controller.Session; using MediaBrowser.Model.Net; using MediaBrowser.Model.Session; using Microsoft.AspNetCore.Http; using Microsoft.Extensions.Logging; namespace Emby.Server.Implementations.Session { /// /// Class SessionWebSocketListener. /// public sealed class SessionWebSocketListener : IWebSocketListener, IDisposable { /// /// The timeout in seconds after which a WebSocket is considered to be lost. /// private const int WebSocketLostTimeout = 60; /// /// The keep-alive interval factor; controls how often the watcher will check on the status of the WebSockets. /// private const float IntervalFactor = 0.2f; /// /// The ForceKeepAlive factor; controls when a ForceKeepAlive is sent. /// private const float ForceKeepAliveFactor = 0.75f; /// /// Lock used for accessing the KeepAlive cancellation token. /// private readonly object _keepAliveLock = new object(); /// /// The WebSocket watchlist. /// private readonly HashSet _webSockets = new HashSet(); /// /// Lock used for accessing the WebSockets watchlist. /// private readonly object _webSocketsLock = new object(); private readonly ISessionManager _sessionManager; private readonly ILogger _logger; private readonly ILoggerFactory _loggerFactory; /// /// The KeepAlive cancellation token. /// private CancellationTokenSource _keepAliveCancellationToken; /// /// Initializes a new instance of the class. /// /// The logger. /// The session manager. /// The logger factory. public SessionWebSocketListener( ILogger logger, ISessionManager sessionManager, ILoggerFactory loggerFactory) { _logger = logger; _sessionManager = sessionManager; _loggerFactory = loggerFactory; } /// public void Dispose() { StopKeepAlive(); } /// /// Processes the message. /// /// The message. /// Task. public Task ProcessMessageAsync(WebSocketMessageInfo message) => Task.CompletedTask; /// public async Task ProcessWebSocketConnectedAsync(IWebSocketConnection connection, HttpContext httpContext) { var session = await GetSession(httpContext, connection.RemoteEndPoint?.ToString()).ConfigureAwait(false); if (session != null) { EnsureController(session, connection); await KeepAliveWebSocket(connection).ConfigureAwait(false); } else { _logger.LogWarning("Unable to determine session based on query string: {0}", httpContext.Request.QueryString); } } private async Task GetSession(HttpContext httpContext, string remoteEndpoint) { if (!httpContext.User.Identity?.IsAuthenticated ?? false) { return null; } var deviceId = httpContext.User.GetDeviceId(); if (httpContext.Request.Query.TryGetValue("deviceId", out var queryDeviceId)) { deviceId = queryDeviceId; } return await _sessionManager.GetSessionByAuthenticationToken(httpContext.User.GetToken(), deviceId, remoteEndpoint) .ConfigureAwait(false); } private void EnsureController(SessionInfo session, IWebSocketConnection connection) { var controllerInfo = session.EnsureController( s => new WebSocketController(_loggerFactory.CreateLogger(), s, _sessionManager)); var controller = (WebSocketController)controllerInfo.Item1; controller.AddWebSocket(connection); _sessionManager.OnSessionControllerConnected(session); } /// /// Called when a WebSocket is closed. /// /// The WebSocket. /// The event arguments. private void OnWebSocketClosed(object sender, EventArgs e) { var webSocket = (IWebSocketConnection)sender; _logger.LogDebug("WebSocket {0} is closed.", webSocket); RemoveWebSocket(webSocket); } /// /// Adds a WebSocket to the KeepAlive watchlist. /// /// The WebSocket to monitor. private async Task KeepAliveWebSocket(IWebSocketConnection webSocket) { lock (_webSocketsLock) { if (!_webSockets.Add(webSocket)) { _logger.LogWarning("Multiple attempts to keep alive single WebSocket {0}", webSocket); return; } webSocket.Closed += OnWebSocketClosed; webSocket.LastKeepAliveDate = DateTime.UtcNow; StartKeepAlive(); } // Notify WebSocket about timeout try { await SendForceKeepAlive(webSocket).ConfigureAwait(false); } catch (WebSocketException exception) { _logger.LogWarning(exception, "Cannot send ForceKeepAlive message to WebSocket {0}.", webSocket); } } /// /// Removes a WebSocket from the KeepAlive watchlist. /// /// The WebSocket to remove. private void RemoveWebSocket(IWebSocketConnection webSocket) { lock (_webSocketsLock) { if (!_webSockets.Remove(webSocket)) { _logger.LogWarning("WebSocket {0} not on watchlist.", webSocket); } else { webSocket.Closed -= OnWebSocketClosed; } } } /// /// Starts the KeepAlive watcher. /// private void StartKeepAlive() { lock (_keepAliveLock) { if (_keepAliveCancellationToken is null) { _keepAliveCancellationToken = new CancellationTokenSource(); // Start KeepAlive watcher _ = RepeatAsyncCallbackEvery( KeepAliveSockets, TimeSpan.FromSeconds(WebSocketLostTimeout * IntervalFactor), _keepAliveCancellationToken.Token); } } } /// /// Stops the KeepAlive watcher. /// private void StopKeepAlive() { lock (_keepAliveLock) { if (_keepAliveCancellationToken != null) { _keepAliveCancellationToken.Cancel(); _keepAliveCancellationToken.Dispose(); _keepAliveCancellationToken = null; } } lock (_webSocketsLock) { foreach (var webSocket in _webSockets) { webSocket.Closed -= OnWebSocketClosed; } _webSockets.Clear(); } } /// /// Checks status of KeepAlive of WebSockets. /// private async Task KeepAliveSockets() { List inactive; List lost; lock (_webSocketsLock) { _logger.LogDebug("Watching {0} WebSockets.", _webSockets.Count); inactive = _webSockets.Where(i => { var elapsed = (DateTime.UtcNow - i.LastKeepAliveDate).TotalSeconds; return (elapsed > WebSocketLostTimeout * ForceKeepAliveFactor) && (elapsed < WebSocketLostTimeout); }).ToList(); lost = _webSockets.Where(i => (DateTime.UtcNow - i.LastKeepAliveDate).TotalSeconds >= WebSocketLostTimeout).ToList(); } if (inactive.Count > 0) { _logger.LogInformation("Sending ForceKeepAlive message to {0} inactive WebSockets.", inactive.Count); } foreach (var webSocket in inactive) { try { await SendForceKeepAlive(webSocket).ConfigureAwait(false); } catch (WebSocketException exception) { _logger.LogInformation(exception, "Error sending ForceKeepAlive message to WebSocket."); lost.Add(webSocket); } } lock (_webSocketsLock) { if (lost.Count > 0) { _logger.LogInformation("Lost {0} WebSockets.", lost.Count); foreach (var webSocket in lost) { // TODO: handle session relative to the lost webSocket RemoveWebSocket(webSocket); } } if (_webSockets.Count == 0) { StopKeepAlive(); } } } /// /// Sends a ForceKeepAlive message to a WebSocket. /// /// The WebSocket. /// Task. private Task SendForceKeepAlive(IWebSocketConnection webSocket) { return webSocket.SendAsync( new WebSocketMessage { MessageType = SessionMessageType.ForceKeepAlive, Data = WebSocketLostTimeout }, CancellationToken.None); } /// /// Runs a given async callback once every specified interval time, until cancelled. /// /// The async callback. /// The interval time. /// The cancellation token. /// Task. private async Task RepeatAsyncCallbackEvery(Func callback, TimeSpan interval, CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested) { await callback().ConfigureAwait(false); try { await Task.Delay(interval, cancellationToken).ConfigureAwait(false); } catch (TaskCanceledException) { return; } } } } }