mirror of https://github.com/jellyfin/jellyfin.git
Remove circular dependency between websocket listeners and manager
This commit is contained in:
parent
3ae39d44da
commit
65e6211c03
|
@ -2,9 +2,9 @@
|
||||||
|
|
||||||
using System;
|
using System;
|
||||||
using System.Collections.Generic;
|
using System.Collections.Generic;
|
||||||
|
using System.Linq;
|
||||||
using System.Net.WebSockets;
|
using System.Net.WebSockets;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using Jellyfin.Data.Events;
|
|
||||||
using MediaBrowser.Controller.Net;
|
using MediaBrowser.Controller.Net;
|
||||||
using Microsoft.AspNetCore.Http;
|
using Microsoft.AspNetCore.Http;
|
||||||
using Microsoft.Extensions.Logging;
|
using Microsoft.Extensions.Logging;
|
||||||
|
@ -13,32 +13,23 @@ namespace Emby.Server.Implementations.HttpServer
|
||||||
{
|
{
|
||||||
public class WebSocketManager : IWebSocketManager
|
public class WebSocketManager : IWebSocketManager
|
||||||
{
|
{
|
||||||
private readonly Lazy<IEnumerable<IWebSocketListener>> _webSocketListeners;
|
private readonly IWebSocketListener[] _webSocketListeners;
|
||||||
private readonly ILogger<WebSocketManager> _logger;
|
private readonly ILogger<WebSocketManager> _logger;
|
||||||
private readonly ILoggerFactory _loggerFactory;
|
private readonly ILoggerFactory _loggerFactory;
|
||||||
|
|
||||||
private bool _disposed = false;
|
|
||||||
|
|
||||||
public WebSocketManager(
|
public WebSocketManager(
|
||||||
Lazy<IEnumerable<IWebSocketListener>> webSocketListeners,
|
IEnumerable<IWebSocketListener> webSocketListeners,
|
||||||
ILogger<WebSocketManager> logger,
|
ILogger<WebSocketManager> logger,
|
||||||
ILoggerFactory loggerFactory)
|
ILoggerFactory loggerFactory)
|
||||||
{
|
{
|
||||||
_webSocketListeners = webSocketListeners;
|
_webSocketListeners = webSocketListeners.ToArray();
|
||||||
_logger = logger;
|
_logger = logger;
|
||||||
_loggerFactory = loggerFactory;
|
_loggerFactory = loggerFactory;
|
||||||
}
|
}
|
||||||
|
|
||||||
public event EventHandler<GenericEventArgs<IWebSocketConnection>> WebSocketConnected;
|
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public async Task WebSocketRequestHandler(HttpContext context)
|
public async Task WebSocketRequestHandler(HttpContext context)
|
||||||
{
|
{
|
||||||
if (_disposed)
|
|
||||||
{
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
_logger.LogInformation("WS {IP} request", context.Connection.RemoteIpAddress);
|
_logger.LogInformation("WS {IP} request", context.Connection.RemoteIpAddress);
|
||||||
|
@ -54,7 +45,13 @@ namespace Emby.Server.Implementations.HttpServer
|
||||||
OnReceive = ProcessWebSocketMessageReceived
|
OnReceive = ProcessWebSocketMessageReceived
|
||||||
};
|
};
|
||||||
|
|
||||||
WebSocketConnected?.Invoke(this, new GenericEventArgs<IWebSocketConnection>(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);
|
await connection.ProcessAsync().ConfigureAwait(false);
|
||||||
_logger.LogInformation("WS {IP} closed", context.Connection.RemoteIpAddress);
|
_logger.LogInformation("WS {IP} closed", context.Connection.RemoteIpAddress);
|
||||||
|
@ -75,21 +72,13 @@ namespace Emby.Server.Implementations.HttpServer
|
||||||
/// <param name="result">The result.</param>
|
/// <param name="result">The result.</param>
|
||||||
private Task ProcessWebSocketMessageReceived(WebSocketMessageInfo result)
|
private Task ProcessWebSocketMessageReceived(WebSocketMessageInfo result)
|
||||||
{
|
{
|
||||||
if (_disposed)
|
var tasks = new Task[_webSocketListeners.Length];
|
||||||
|
for (var i = 0; i < _webSocketListeners.Length; ++i)
|
||||||
{
|
{
|
||||||
return Task.CompletedTask;
|
tasks[i] = _webSocketListeners[i].ProcessMessageAsync(result);
|
||||||
}
|
}
|
||||||
|
|
||||||
IEnumerable<Task> GetTasks()
|
return Task.WhenAll(tasks);
|
||||||
{
|
|
||||||
var listeners = _webSocketListeners.Value;
|
|
||||||
foreach (var x in listeners)
|
|
||||||
{
|
|
||||||
yield return x.ProcessMessageAsync(result);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return Task.WhenAll(GetTasks());
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -4,7 +4,6 @@ using System.Linq;
|
||||||
using System.Net.WebSockets;
|
using System.Net.WebSockets;
|
||||||
using System.Threading;
|
using System.Threading;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using Jellyfin.Data.Events;
|
|
||||||
using MediaBrowser.Controller.Net;
|
using MediaBrowser.Controller.Net;
|
||||||
using MediaBrowser.Controller.Session;
|
using MediaBrowser.Controller.Session;
|
||||||
using MediaBrowser.Model.Net;
|
using MediaBrowser.Model.Net;
|
||||||
|
@ -22,35 +21,17 @@ namespace Emby.Server.Implementations.Session
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// The timeout in seconds after which a WebSocket is considered to be lost.
|
/// The timeout in seconds after which a WebSocket is considered to be lost.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public const int WebSocketLostTimeout = 60;
|
private const int WebSocketLostTimeout = 60;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// The keep-alive interval factor; controls how often the watcher will check on the status of the WebSockets.
|
/// The keep-alive interval factor; controls how often the watcher will check on the status of the WebSockets.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public const float IntervalFactor = 0.2f;
|
private const float IntervalFactor = 0.2f;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// The ForceKeepAlive factor; controls when a ForceKeepAlive is sent.
|
/// The ForceKeepAlive factor; controls when a ForceKeepAlive is sent.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public const float ForceKeepAliveFactor = 0.75f;
|
private const float ForceKeepAliveFactor = 0.75f;
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// The _session manager.
|
|
||||||
/// </summary>
|
|
||||||
private readonly ISessionManager _sessionManager;
|
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// The _logger.
|
|
||||||
/// </summary>
|
|
||||||
private readonly ILogger<SessionWebSocketListener> _logger;
|
|
||||||
private readonly ILoggerFactory _loggerFactory;
|
|
||||||
|
|
||||||
private readonly IWebSocketManager _webSocketManager;
|
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// The KeepAlive cancellation token.
|
|
||||||
/// </summary>
|
|
||||||
private CancellationTokenSource _keepAliveCancellationToken;
|
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Lock used for accesing the KeepAlive cancellation token.
|
/// Lock used for accesing the KeepAlive cancellation token.
|
||||||
|
@ -63,42 +44,68 @@ namespace Emby.Server.Implementations.Session
|
||||||
private readonly HashSet<IWebSocketConnection> _webSockets = new HashSet<IWebSocketConnection>();
|
private readonly HashSet<IWebSocketConnection> _webSockets = new HashSet<IWebSocketConnection>();
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Lock used for accesing the WebSockets watchlist.
|
/// Lock used for accessing the WebSockets watchlist.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private readonly object _webSocketsLock = new object();
|
private readonly object _webSocketsLock = new object();
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The _session manager.
|
||||||
|
/// </summary>
|
||||||
|
private readonly ISessionManager _sessionManager;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The _logger.
|
||||||
|
/// </summary>
|
||||||
|
private readonly ILogger<SessionWebSocketListener> _logger;
|
||||||
|
private readonly ILoggerFactory _loggerFactory;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The KeepAlive cancellation token.
|
||||||
|
/// </summary>
|
||||||
|
private CancellationTokenSource _keepAliveCancellationToken;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Initializes a new instance of the <see cref="SessionWebSocketListener" /> class.
|
/// Initializes a new instance of the <see cref="SessionWebSocketListener" /> class.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="logger">The logger.</param>
|
/// <param name="logger">The logger.</param>
|
||||||
/// <param name="sessionManager">The session manager.</param>
|
/// <param name="sessionManager">The session manager.</param>
|
||||||
/// <param name="loggerFactory">The logger factory.</param>
|
/// <param name="loggerFactory">The logger factory.</param>
|
||||||
/// <param name="webSocketManager">The HTTP server.</param>
|
|
||||||
public SessionWebSocketListener(
|
public SessionWebSocketListener(
|
||||||
ILogger<SessionWebSocketListener> logger,
|
ILogger<SessionWebSocketListener> logger,
|
||||||
ISessionManager sessionManager,
|
ISessionManager sessionManager,
|
||||||
ILoggerFactory loggerFactory,
|
ILoggerFactory loggerFactory)
|
||||||
IWebSocketManager webSocketManager)
|
|
||||||
{
|
{
|
||||||
_logger = logger;
|
_logger = logger;
|
||||||
_sessionManager = sessionManager;
|
_sessionManager = sessionManager;
|
||||||
_loggerFactory = loggerFactory;
|
_loggerFactory = loggerFactory;
|
||||||
_webSocketManager = webSocketManager;
|
|
||||||
|
|
||||||
webSocketManager.WebSocketConnected += OnServerManagerWebSocketConnected;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private async void OnServerManagerWebSocketConnected(object sender, GenericEventArgs<IWebSocketConnection> e)
|
/// <inheritdoc />
|
||||||
|
public void Dispose()
|
||||||
{
|
{
|
||||||
var session = GetSession(e.Argument.QueryString, e.Argument.RemoteEndPoint.ToString());
|
StopKeepAlive();
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Processes the message.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="message">The message.</param>
|
||||||
|
/// <returns>Task.</returns>
|
||||||
|
public Task ProcessMessageAsync(WebSocketMessageInfo message)
|
||||||
|
=> Task.CompletedTask;
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task ProcessWebSocketConnectedAsync(IWebSocketConnection connection)
|
||||||
|
{
|
||||||
|
var session = GetSession(connection.QueryString, connection.RemoteEndPoint.ToString());
|
||||||
if (session != null)
|
if (session != null)
|
||||||
{
|
{
|
||||||
EnsureController(session, e.Argument);
|
EnsureController(session, connection);
|
||||||
await KeepAliveWebSocket(e.Argument).ConfigureAwait(false);
|
await KeepAliveWebSocket(connection).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
else
|
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);
|
return _sessionManager.GetSessionByAuthenticationToken(token, deviceId, remoteEndpoint);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
|
||||||
public void Dispose()
|
|
||||||
{
|
|
||||||
_webSocketManager.WebSocketConnected -= OnServerManagerWebSocketConnected;
|
|
||||||
StopKeepAlive();
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Processes the message.
|
|
||||||
/// </summary>
|
|
||||||
/// <param name="message">The message.</param>
|
|
||||||
/// <returns>Task.</returns>
|
|
||||||
public Task ProcessMessageAsync(WebSocketMessageInfo message)
|
|
||||||
=> Task.CompletedTask;
|
|
||||||
|
|
||||||
private void EnsureController(SessionInfo session, IWebSocketConnection connection)
|
private void EnsureController(SessionInfo session, IWebSocketConnection connection)
|
||||||
{
|
{
|
||||||
var controllerInfo = session.EnsureController<WebSocketController>(
|
var controllerInfo = session.EnsureController<WebSocketController>(
|
||||||
|
|
|
@ -58,7 +58,7 @@ namespace Jellyfin.Api.WebSocketListeners
|
||||||
|
|
||||||
private void OnEntryCreated(object? sender, GenericEventArgs<ActivityLogEntry> e)
|
private void OnEntryCreated(object? sender, GenericEventArgs<ActivityLogEntry> e)
|
||||||
{
|
{
|
||||||
SendData(true);
|
SendData(true).GetAwaiter().GetResult();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -82,13 +82,11 @@ namespace Jellyfin.Server
|
||||||
ServiceCollection.AddSingleton<IUserManager, UserManager>();
|
ServiceCollection.AddSingleton<IUserManager, UserManager>();
|
||||||
ServiceCollection.AddSingleton<IDisplayPreferencesManager, DisplayPreferencesManager>();
|
ServiceCollection.AddSingleton<IDisplayPreferencesManager, DisplayPreferencesManager>();
|
||||||
|
|
||||||
ServiceCollection.AddScoped<IWebSocketListener, SessionWebSocketListener>();
|
// TODO search the assemblies instead of adding them manually?
|
||||||
ServiceCollection.AddScoped<IWebSocketListener, ActivityLogWebSocketListener>();
|
ServiceCollection.AddSingleton<IWebSocketListener, SessionWebSocketListener>();
|
||||||
ServiceCollection.AddScoped<IWebSocketListener, ScheduledTasksWebSocketListener>();
|
ServiceCollection.AddSingleton<IWebSocketListener, ActivityLogWebSocketListener>();
|
||||||
ServiceCollection.AddScoped<IWebSocketListener, SessionInfoWebSocketListener>();
|
ServiceCollection.AddSingleton<IWebSocketListener, ScheduledTasksWebSocketListener>();
|
||||||
|
ServiceCollection.AddSingleton<IWebSocketListener, SessionInfoWebSocketListener>();
|
||||||
// TODO fix circular dependency on IWebSocketManager
|
|
||||||
ServiceCollection.AddScoped(serviceProvider => new Lazy<IEnumerable<IWebSocketListener>>(serviceProvider.GetRequiredService<IEnumerable<IWebSocketListener>>));
|
|
||||||
|
|
||||||
base.RegisterServices();
|
base.RegisterServices();
|
||||||
}
|
}
|
||||||
|
|
|
@ -92,6 +92,8 @@ namespace MediaBrowser.Controller.Net
|
||||||
return Task.CompletedTask;
|
return Task.CompletedTask;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public Task ProcessWebSocketConnectedAsync(IWebSocketConnection connection) => Task.CompletedTask;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Starts sending messages over a web socket.
|
/// Starts sending messages over a web socket.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
|
|
@ -3,7 +3,7 @@ using System.Threading.Tasks;
|
||||||
namespace MediaBrowser.Controller.Net
|
namespace MediaBrowser.Controller.Net
|
||||||
{
|
{
|
||||||
/// <summary>
|
/// <summary>
|
||||||
///This is an interface for listening to messages coming through a web socket connection.
|
/// Interface for listening to messages coming through a web socket connection.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public interface IWebSocketListener
|
public interface IWebSocketListener
|
||||||
{
|
{
|
||||||
|
@ -13,5 +13,12 @@ namespace MediaBrowser.Controller.Net
|
||||||
/// <param name="message">The message.</param>
|
/// <param name="message">The message.</param>
|
||||||
/// <returns>Task.</returns>
|
/// <returns>Task.</returns>
|
||||||
Task ProcessMessageAsync(WebSocketMessageInfo message);
|
Task ProcessMessageAsync(WebSocketMessageInfo message);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Processes a new web socket connection.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="connection">An instance of the <see cref="IWebSocketConnection"/> interface.</param>
|
||||||
|
/// <returns>Task.</returns>
|
||||||
|
Task ProcessWebSocketConnectedAsync(IWebSocketConnection connection);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,7 +1,4 @@
|
||||||
using System;
|
|
||||||
using System.Collections.Generic;
|
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using Jellyfin.Data.Events;
|
|
||||||
using Microsoft.AspNetCore.Http;
|
using Microsoft.AspNetCore.Http;
|
||||||
|
|
||||||
namespace MediaBrowser.Controller.Net
|
namespace MediaBrowser.Controller.Net
|
||||||
|
@ -11,11 +8,6 @@ namespace MediaBrowser.Controller.Net
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public interface IWebSocketManager
|
public interface IWebSocketManager
|
||||||
{
|
{
|
||||||
/// <summary>
|
|
||||||
/// Occurs when [web socket connected].
|
|
||||||
/// </summary>
|
|
||||||
event EventHandler<GenericEventArgs<IWebSocketConnection>> WebSocketConnected;
|
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// The HTTP request handler.
|
/// The HTTP request handler.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
|
Loading…
Reference in New Issue