Revert "Removed Lazy implementation."

This commit is contained in:
cvium 2020-11-28 09:50:16 +01:00
parent 5cd5a7d4ce
commit 1a0d8aef80
12 changed files with 64 additions and 95 deletions

View File

@ -4,9 +4,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Net.WebSockets; using System.Net.WebSockets;
using System.Threading.Tasks; using System.Threading.Tasks;
using Emby.Server.Implementations.Session; using Jellyfin.Data.Events;
using Jellyfin.Api.WebSocketListeners;
using MediaBrowser.Controller;
using MediaBrowser.Controller.Net; using MediaBrowser.Controller.Net;
using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
@ -15,21 +13,24 @@ namespace Emby.Server.Implementations.HttpServer
{ {
public class WebSocketManager : IWebSocketManager public class WebSocketManager : IWebSocketManager
{ {
private readonly IServerApplicationHost _appHost; private readonly Lazy<IEnumerable<IWebSocketListener>> _webSocketListeners;
private readonly ILogger<WebSocketManager> _logger; private readonly ILogger<WebSocketManager> _logger;
private readonly ILoggerFactory _loggerFactory; private readonly ILoggerFactory _loggerFactory;
private bool _disposed = false; private bool _disposed = false;
public WebSocketManager( public WebSocketManager(
IServerApplicationHost appHost, Lazy<IEnumerable<IWebSocketListener>> webSocketListeners,
ILogger<WebSocketManager> logger, ILogger<WebSocketManager> logger,
ILoggerFactory loggerFactory) ILoggerFactory loggerFactory)
{ {
_appHost = appHost; _webSocketListeners = webSocketListeners;
_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)
{ {
@ -38,8 +39,6 @@ namespace Emby.Server.Implementations.HttpServer
return; return;
} }
var listener = _appHost.Resolve<ISessionWebSocketListener>();
try try
{ {
_logger.LogInformation("WS {IP} request", context.Connection.RemoteIpAddress); _logger.LogInformation("WS {IP} request", context.Connection.RemoteIpAddress);
@ -55,7 +54,7 @@ namespace Emby.Server.Implementations.HttpServer
OnReceive = ProcessWebSocketMessageReceived OnReceive = ProcessWebSocketMessageReceived
}; };
listener?.ProcessWebSocketConnected(connection); WebSocketConnected?.Invoke(this, new GenericEventArgs<IWebSocketConnection>(connection));
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);
@ -81,12 +80,16 @@ namespace Emby.Server.Implementations.HttpServer
return Task.CompletedTask; return Task.CompletedTask;
} }
Parallel.Invoke( IEnumerable<Task> GetTasks()
() => _appHost.Resolve<IActivityLogWebSocketListener>(), {
() => _appHost.Resolve<IScheduledTasksWebSocketListener>(), var listeners = _webSocketListeners.Value;
() => _appHost.Resolve<ISessionInfoWebSocketListener>()); foreach (var x in listeners)
{
yield return x.ProcessMessageAsync(result);
}
}
return Task.CompletedTask; return Task.WhenAll(GetTasks());
} }
} }
} }

View File

@ -1,30 +0,0 @@
namespace Emby.Server.Implementations.Session
{
using System.Threading.Tasks;
using Jellyfin.Data.Events;
using MediaBrowser.Controller.Net;
/// <summary>
/// Defines the <see cref="ISessionWebSocketListener" />.
/// </summary>
public interface ISessionWebSocketListener
{
/// <summary>
/// Runs processes due to a WebSocket connection event.
/// </summary>
/// <param name="websocketConnection">The <see cref="IWebSocketConnection"/> instance.</param>
void ProcessWebSocketConnected(IWebSocketConnection websocketConnection);
/// <summary>
/// Disposes the object.
/// </summary>
void Dispose();
/// <summary>
/// Processes a message.
/// </summary>
/// <param name="message">The <see cref="WebSocketMessageInfo"/>.</param>
/// <returns>A <see cref="Task"/>.</returns>
Task ProcessMessageAsync(WebSocketMessageInfo message);
}
}

View File

@ -17,7 +17,7 @@ namespace Emby.Server.Implementations.Session
/// <summary> /// <summary>
/// Class SessionWebSocketListener. /// Class SessionWebSocketListener.
/// </summary> /// </summary>
public sealed class SessionWebSocketListener : ISessionWebSocketListener, IDisposable public sealed class SessionWebSocketListener : IWebSocketListener, IDisposable
{ {
/// <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.
@ -45,13 +45,15 @@ namespace Emby.Server.Implementations.Session
private readonly ILogger<SessionWebSocketListener> _logger; private readonly ILogger<SessionWebSocketListener> _logger;
private readonly ILoggerFactory _loggerFactory; private readonly ILoggerFactory _loggerFactory;
private readonly IWebSocketManager _webSocketManager;
/// <summary> /// <summary>
/// The KeepAlive cancellation token. /// The KeepAlive cancellation token.
/// </summary> /// </summary>
private CancellationTokenSource _keepAliveCancellationToken; private CancellationTokenSource _keepAliveCancellationToken;
/// <summary> /// <summary>
/// Lock used for accessing the KeepAlive cancellation token. /// Lock used for accesing the KeepAlive cancellation token.
/// </summary> /// </summary>
private readonly object _keepAliveLock = new object(); private readonly object _keepAliveLock = new object();
@ -61,7 +63,7 @@ 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 accessing the WebSockets watchlist. /// Lock used for accesing the WebSockets watchlist.
/// </summary> /// </summary>
private readonly object _webSocketsLock = new object(); private readonly object _webSocketsLock = new object();
@ -71,28 +73,32 @@ namespace Emby.Server.Implementations.Session
/// <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;
} }
/// <inheritdoc/> private async void OnServerManagerWebSocketConnected(object sender, GenericEventArgs<IWebSocketConnection> e)
public async void ProcessWebSocketConnected(IWebSocketConnection websocketConnection)
{ {
var session = GetSession(websocketConnection.QueryString, websocketConnection.RemoteEndPoint.ToString()); var session = GetSession(e.Argument.QueryString, e.Argument.RemoteEndPoint.ToString());
if (session != null) if (session != null)
{ {
EnsureController(session, websocketConnection); EnsureController(session, e.Argument);
await KeepAliveWebSocket(websocketConnection).ConfigureAwait(false); await KeepAliveWebSocket(e.Argument).ConfigureAwait(false);
} }
else else
{ {
_logger.LogWarning("Unable to determine session based on query string: {Querystring}", websocketConnection.QueryString); _logger.LogWarning("Unable to determine session based on query string: {0}", e.Argument.QueryString);
} }
} }
@ -116,6 +122,7 @@ namespace Emby.Server.Implementations.Session
/// <inheritdoc /> /// <inheritdoc />
public void Dispose() public void Dispose()
{ {
_webSocketManager.WebSocketConnected -= OnServerManagerWebSocketConnected;
StopKeepAlive(); StopKeepAlive();
} }

View File

@ -11,7 +11,7 @@ namespace Jellyfin.Api.WebSocketListeners
/// <summary> /// <summary>
/// Class SessionInfoWebSocketListener. /// Class SessionInfoWebSocketListener.
/// </summary> /// </summary>
public class ActivityLogWebSocketListener : BasePeriodicWebSocketListener<ActivityLogEntry[], WebSocketListenerState>, IActivityLogWebSocketListener public class ActivityLogWebSocketListener : BasePeriodicWebSocketListener<ActivityLogEntry[], WebSocketListenerState>
{ {
/// <summary> /// <summary>
/// The _kernel. /// The _kernel.

View File

@ -1,10 +0,0 @@
#pragma warning disable CA1040 // Avoid empty interfaces
namespace Jellyfin.Api.WebSocketListeners
{
/// <summary>
/// Defines the <see cref="IActivityLogWebSocketListener" />.
/// </summary>
public interface IActivityLogWebSocketListener
{
}
}

View File

@ -1,10 +0,0 @@
#pragma warning disable CA1040 // Avoid empty interfaces
namespace Jellyfin.Api.WebSocketListeners
{
/// <summary>
/// Defines the <see cref="IScheduledTasksWebSocketListener" />.
/// </summary>
public interface IScheduledTasksWebSocketListener
{
}
}

View File

@ -1,10 +0,0 @@
#pragma warning disable CA1040 // Avoid empty interfaces
namespace Jellyfin.Api.WebSocketListeners
{
/// <summary>
/// Defines the <see cref="ISessionInfoWebSocketListener" />.
/// </summary>
public interface ISessionInfoWebSocketListener
{
}
}

View File

@ -12,7 +12,7 @@ namespace Jellyfin.Api.WebSocketListeners
/// <summary> /// <summary>
/// Class ScheduledTasksWebSocketListener. /// Class ScheduledTasksWebSocketListener.
/// </summary> /// </summary>
public class ScheduledTasksWebSocketListener : BasePeriodicWebSocketListener<IEnumerable<TaskInfo>, WebSocketListenerState>, IScheduledTasksWebSocketListener public class ScheduledTasksWebSocketListener : BasePeriodicWebSocketListener<IEnumerable<TaskInfo>, WebSocketListenerState>
{ {
/// <summary> /// <summary>
/// Gets or sets the task manager. /// Gets or sets the task manager.
@ -66,19 +66,19 @@ namespace Jellyfin.Api.WebSocketListeners
private void OnTaskCompleted(object? sender, TaskCompletionEventArgs e) private void OnTaskCompleted(object? sender, TaskCompletionEventArgs e)
{ {
SendData(true).GetAwaiter().GetResult(); SendData(true);
e.Task.TaskProgress -= OnTaskProgress; e.Task.TaskProgress -= OnTaskProgress;
} }
private void OnTaskExecuting(object? sender, GenericEventArgs<IScheduledTaskWorker> e) private void OnTaskExecuting(object? sender, GenericEventArgs<IScheduledTaskWorker> e)
{ {
SendData(true).GetAwaiter().GetResult(); SendData(true);
e.Argument.TaskProgress += OnTaskProgress; e.Argument.TaskProgress += OnTaskProgress;
} }
private void OnTaskProgress(object? sender, GenericEventArgs<double> e) private void OnTaskProgress(object? sender, GenericEventArgs<double> e)
{ {
SendData(false).GetAwaiter().GetResult(); SendData(false);
} }
} }
} }

View File

@ -11,7 +11,7 @@ namespace Jellyfin.Api.WebSocketListeners
/// <summary> /// <summary>
/// Class SessionInfoWebSocketListener. /// Class SessionInfoWebSocketListener.
/// </summary> /// </summary>
public class SessionInfoWebSocketListener : BasePeriodicWebSocketListener<IEnumerable<SessionInfo>, WebSocketListenerState>, ISessionInfoWebSocketListener public class SessionInfoWebSocketListener : BasePeriodicWebSocketListener<IEnumerable<SessionInfo>, WebSocketListenerState>
{ {
private readonly ISessionManager _sessionManager; private readonly ISessionManager _sessionManager;

View File

@ -11,6 +11,7 @@ using Jellyfin.Server.Implementations;
using Jellyfin.Server.Implementations.Activity; using Jellyfin.Server.Implementations.Activity;
using Jellyfin.Server.Implementations.Events; using Jellyfin.Server.Implementations.Events;
using Jellyfin.Server.Implementations.Users; using Jellyfin.Server.Implementations.Users;
using MediaBrowser.Common.Net;
using MediaBrowser.Controller; using MediaBrowser.Controller;
using MediaBrowser.Controller.BaseItemManager; using MediaBrowser.Controller.BaseItemManager;
using MediaBrowser.Controller.Drawing; using MediaBrowser.Controller.Drawing;
@ -81,10 +82,14 @@ namespace Jellyfin.Server
ServiceCollection.AddSingleton<IUserManager, UserManager>(); ServiceCollection.AddSingleton<IUserManager, UserManager>();
ServiceCollection.AddSingleton<IDisplayPreferencesManager, DisplayPreferencesManager>(); ServiceCollection.AddSingleton<IDisplayPreferencesManager, DisplayPreferencesManager>();
ServiceCollection.AddScoped<IActivityLogWebSocketListener, ActivityLogWebSocketListener>(); ServiceCollection.AddScoped<IWebSocketListener, ActivityLogWebSocketListener>();
ServiceCollection.AddScoped<IScheduledTasksWebSocketListener, ScheduledTasksWebSocketListener>(); ServiceCollection.AddScoped<IWebSocketListener, ScheduledTasksWebSocketListener>();
ServiceCollection.AddScoped<ISessionInfoWebSocketListener, SessionInfoWebSocketListener>(); ServiceCollection.AddScoped<IWebSocketListener, SessionInfoWebSocketListener>();
ServiceCollection.AddScoped<ISessionWebSocketListener, SessionWebSocketListener>(); // This one has to be last as DI will select it for parameterization.
ServiceCollection.AddScoped<IWebSocketListener, SessionWebSocketListener>();
// TODO fix circular dependency on IWebSocketManager
ServiceCollection.AddScoped(serviceProvider => new Lazy<IEnumerable<IWebSocketListener>>(serviceProvider.GetRequiredService<IEnumerable<IWebSocketListener>>));
base.RegisterServices(); base.RegisterServices();
} }

View File

@ -25,10 +25,19 @@ namespace Jellyfin.Server.Middleware
/// </summary> /// </summary>
/// <param name="httpContext">The current HTTP context.</param> /// <param name="httpContext">The current HTTP context.</param>
/// <param name="webSocketManager">The WebSocket connection manager.</param> /// <param name="webSocketManager">The WebSocket connection manager.</param>
/// <param name="websocketListener">Session manager instance.</param>
/// <returns>The async task.</returns> /// <returns>The async task.</returns>
public async Task Invoke( public async Task Invoke(
HttpContext httpContext, HttpContext httpContext,
IWebSocketManager webSocketManager) IWebSocketManager webSocketManager,
#pragma warning disable CA1801
#pragma warning disable IDE0060
// TODO: Workaround. see https://github.com/jellyfin/jellyfin/pull/3194
// Do not remove this parameter. It uses DI to create a SessionWebSocketListener which is
// required for webSocketManager events.
IWebSocketListener websocketListener)
#pragma warning restore IDE0060 // Remove unused parameter
#pragma warning restore CA1801
{ {
if (!httpContext.WebSockets.IsWebSocketRequest) if (!httpContext.WebSockets.IsWebSocketRequest)
{ {

View File

@ -11,6 +11,11 @@ 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>