timothy/ersatztv
1
0
Fork 0
Code Issues 51 Pull Requests 5 Actions Packages Projects Releases 9 Wiki Activity

rework concurrency (#1199)

Browse Source
This commit is contained in:
Jason Dove
2023-03-08 21:23:18 -06:00
committed by GitHub
parent e7e20de502
commit 234e93349b
28 changed files with 323 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ErsatzTV/Services/EmbyService.cs
-37
View File
@@ -4,8 +4,6 @@ using ErsatzTV.Application;
using ErsatzTV.Application.Emby;
using ErsatzTV.Core;
using ErsatzTV.Core.Domain;
using ErsatzTV.Core.Errors;
using ErsatzTV.Core.Interfaces.Locking;
using MediatR;
namespace ErsatzTV.Services;
@@ -55,9 +53,6 @@ public class EmbyService : BackgroundService
case SynchronizeEmbyLibraries synchronizeEmbyLibraries:
requestTask = SynchronizeLibraries(synchronizeEmbyLibraries, cancellationToken);
break;
case ISynchronizeEmbyLibraryById synchronizeEmbyLibraryById:
requestTask = SynchronizeEmbyLibrary(synchronizeEmbyLibraryById, cancellationToken);
break;
default:
throw new NotSupportedException($"Unsupported request type: {request.GetType().Name}");
}
@@ -126,36 +121,4 @@ public class EmbyService : BackgroundService
request.EmbyMediaSourceId,
error.Value));
}
private async Task SynchronizeEmbyLibrary(ISynchronizeEmbyLibraryById request, CancellationToken cancellationToken)
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
Either<BaseError, string> result = await mediator.Send(request, cancellationToken);
result.BiIter(
name => _logger.LogDebug("Done synchronizing emby library {Name}", name),
error =>
{
if (error is ScanIsNotRequired)
{
_logger.LogDebug(
"Scan is not required for emby library {LibraryId} at this time",
request.EmbyLibraryId);
}
else
{
_logger.LogWarning(
"Unable to synchronize emby library {LibraryId}: {Error}",
request.EmbyLibraryId,
error.Value);
}
});
if (entityLocker.IsLibraryLocked(request.EmbyLibraryId))
{
entityLocker.UnlockLibrary(request.EmbyLibraryId);
}
}
}
ErsatzTV/Services/JellyfinService.cs
-79
View File
@@ -4,8 +4,6 @@ using ErsatzTV.Application;
using ErsatzTV.Application.Jellyfin;
using ErsatzTV.Core;
using ErsatzTV.Core.Domain;
using ErsatzTV.Core.Errors;
using ErsatzTV.Core.Interfaces.Locking;
using MediatR;
namespace ErsatzTV.Services;
@@ -52,15 +50,6 @@ public class JellyfinService : BackgroundService
case SynchronizeJellyfinMediaSources synchronizeJellyfinMediaSources:
requestTask = SynchronizeSources(synchronizeJellyfinMediaSources, cancellationToken);
break;
case SynchronizeJellyfinAdminUserId synchronizeJellyfinAdminUserId:
requestTask = SynchronizeAdminUserId(synchronizeJellyfinAdminUserId, cancellationToken);
break;
case SynchronizeJellyfinLibraries synchronizeJellyfinLibraries:
requestTask = SynchronizeLibraries(synchronizeJellyfinLibraries, cancellationToken);
break;
case ISynchronizeJellyfinLibraryById synchronizeJellyfinLibraryById:
requestTask = SynchronizeJellyfinLibrary(synchronizeJellyfinLibraryById, cancellationToken);
break;
default:
throw new NotSupportedException($"Unsupported request type: {request.GetType().Name}");
}
@@ -115,72 +104,4 @@ public class JellyfinService : BackgroundService
error.Value);
});
}
private async Task SynchronizeLibraries(SynchronizeJellyfinLibraries request, CancellationToken cancellationToken)
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
Either<BaseError, Unit> result = await mediator.Send(request, cancellationToken);
result.BiIter(
_ => _logger.LogInformation(
"Successfully synchronized Jellyfin libraries for source {MediaSourceId}",
request.JellyfinMediaSourceId),
error => _logger.LogWarning(
"Unable to synchronize Jellyfin libraries for source {MediaSourceId}: {Error}",
request.JellyfinMediaSourceId,
error.Value));
}
private async Task SynchronizeAdminUserId(
SynchronizeJellyfinAdminUserId request,
CancellationToken cancellationToken)
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
Either<BaseError, Unit> result = await mediator.Send(request, cancellationToken);
result.BiIter(
_ => _logger.LogInformation(
"Successfully synchronized Jellyfin admin user id for source {MediaSourceId}",
request.JellyfinMediaSourceId),
error => _logger.LogWarning(
"Unable to synchronize Jellyfin admin user id for source {MediaSourceId}: {Error}",
request.JellyfinMediaSourceId,
error.Value));
}
private async Task SynchronizeJellyfinLibrary(
ISynchronizeJellyfinLibraryById request,
CancellationToken cancellationToken)
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
Either<BaseError, string> result = await mediator.Send(request, cancellationToken);
result.BiIter(
name => _logger.LogDebug("Done synchronizing jellyfin library {Name}", name),
error =>
{
if (error is ScanIsNotRequired)
{
_logger.LogDebug(
"Scan is not required for jellyfin library {LibraryId} at this time",
request.JellyfinLibraryId);
}
else
{
_logger.LogWarning(
"Unable to synchronize jellyfin library {LibraryId}: {Error}",
request.JellyfinLibraryId,
error.Value);
}
});
if (entityLocker.IsLibraryLocked(request.JellyfinLibraryId))
{
entityLocker.UnlockLibrary(request.JellyfinLibraryId);
}
}
}
ErsatzTV/Services/PlexService.cs
-39
View File
@@ -4,8 +4,6 @@ using ErsatzTV.Application;
using ErsatzTV.Application.Plex;
using ErsatzTV.Core;
using ErsatzTV.Core.Domain;
using ErsatzTV.Core.Errors;
using ErsatzTV.Core.Interfaces.Locking;
using MediatR;
namespace ErsatzTV.Services;
@@ -58,9 +56,6 @@ public class PlexService : BackgroundService
case SynchronizePlexLibraries synchronizePlexLibrariesRequest:
requestTask = SynchronizeLibraries(synchronizePlexLibrariesRequest, cancellationToken);
break;
case ISynchronizePlexLibraryById synchronizePlexLibraryById:
requestTask = SynchronizePlexLibrary(synchronizePlexLibraryById, cancellationToken);
break;
default:
throw new NotSupportedException($"Unsupported request type: {request.GetType().Name}");
}
@@ -148,38 +143,4 @@ public class PlexService : BackgroundService
request.PlexMediaSourceId,
error.Value));
}
private async Task SynchronizePlexLibrary(
ISynchronizePlexLibraryById request,
CancellationToken cancellationToken)
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
Either<BaseError, string> result = await mediator.Send(request, cancellationToken);
result.BiIter(
name => _logger.LogDebug("Done synchronizing plex library {Name}", name),
error =>
{
if (error is ScanIsNotRequired)
{
_logger.LogDebug(
"Scan is not required for plex library {LibraryId} at this time",
request.PlexLibraryId);
}
else
{
_logger.LogWarning(
"Unable to synchronize plex library {LibraryId}: {Error}",
request.PlexLibraryId,
error.Value);
}
});
if (entityLocker.IsLibraryLocked(request.PlexLibraryId))
{
entityLocker.UnlockLibrary(request.PlexLibraryId);
}
}
}
ErsatzTV/Services/ScannerService.cs
+259
View File
@@ -0,0 +1,259 @@
using System.Threading.Channels;
using Bugsnag;
using ErsatzTV.Application;
using ErsatzTV.Application.Emby;
using ErsatzTV.Application.Jellyfin;
using ErsatzTV.Application.MediaSources;
using ErsatzTV.Application.Plex;
using ErsatzTV.Core;
using ErsatzTV.Core.Errors;
using ErsatzTV.Core.Interfaces.Locking;
using MediatR;
namespace ErsatzTV.Services;
public class ScannerService : BackgroundService
{
private readonly ChannelReader<IScannerBackgroundServiceRequest> _channel;
private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly ILogger<ScannerService> _logger;
public ScannerService(
ChannelReader<IScannerBackgroundServiceRequest> channel,
IServiceScopeFactory serviceScopeFactory,
ILogger<ScannerService> logger)
{
_channel = channel;
_serviceScopeFactory = serviceScopeFactory;
_logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken cancellationToken)
{
try
{
_logger.LogInformation("Scanner service started");
await foreach (IScannerBackgroundServiceRequest request in _channel.ReadAllAsync(cancellationToken))
{
try
{
Task requestTask;
switch (request)
{
case ISynchronizePlexLibraryById synchronizePlexLibraryById:
requestTask = SynchronizePlexLibrary(synchronizePlexLibraryById, cancellationToken);
break;
case SynchronizeJellyfinAdminUserId synchronizeJellyfinAdminUserId:
requestTask = SynchronizeAdminUserId(synchronizeJellyfinAdminUserId, cancellationToken);
break;
case SynchronizeJellyfinLibraries synchronizeJellyfinLibraries:
requestTask = SynchronizeLibraries(synchronizeJellyfinLibraries, cancellationToken);
break;
case ISynchronizeJellyfinLibraryById synchronizeJellyfinLibraryById:
requestTask = SynchronizeJellyfinLibrary(synchronizeJellyfinLibraryById, cancellationToken);
break;
case ISynchronizeEmbyLibraryById synchronizeEmbyLibraryById:
requestTask = SynchronizeEmbyLibrary(synchronizeEmbyLibraryById, cancellationToken);
break;
case IScanLocalLibrary scanLocalLibrary:
requestTask = SynchronizeLocalLibrary(scanLocalLibrary, cancellationToken);
break;
default:
throw new NotSupportedException($"Unsupported request type: {request.GetType().Name}");
}
await requestTask;
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to process scanner background service request");
try
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IClient client = scope.ServiceProvider.GetRequiredService<IClient>();
client.Notify(ex);
}
catch (Exception)
{
// do nothing
}
}
}
}
catch (Exception ex) when (ex is TaskCanceledException or OperationCanceledException)
{
_logger.LogInformation("Plex service shutting down");
}
}
private async Task SynchronizeLocalLibrary(IScanLocalLibrary request, CancellationToken cancellationToken)
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
Either<BaseError, string> scanResult = await mediator.Send(request, cancellationToken);
scanResult.BiIter(
name => _logger.LogDebug(
"Done scanning local library {Library}",
name),
error =>
{
if (error is ScanIsNotRequired)
{
_logger.LogDebug(
"Scan is not required for local library {LibraryId} at this time",
request.LibraryId);
}
else
{
_logger.LogWarning(
"Unable to scan local library {LibraryId}: {Error}",
request.LibraryId,
error.Value);
}
});
if (entityLocker.IsLibraryLocked(request.LibraryId))
{
entityLocker.UnlockLibrary(request.LibraryId);
}
}
private async Task SynchronizePlexLibrary(
ISynchronizePlexLibraryById request,
CancellationToken cancellationToken)
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
Either<BaseError, string> result = await mediator.Send(request, cancellationToken);
result.BiIter(
name => _logger.LogDebug("Done synchronizing plex library {Name}", name),
error =>
{
if (error is ScanIsNotRequired)
{
_logger.LogDebug(
"Scan is not required for plex library {LibraryId} at this time",
request.PlexLibraryId);
}
else
{
_logger.LogWarning(
"Unable to synchronize plex library {LibraryId}: {Error}",
request.PlexLibraryId,
error.Value);
}
});
if (entityLocker.IsLibraryLocked(request.PlexLibraryId))
{
entityLocker.UnlockLibrary(request.PlexLibraryId);
}
}
private async Task SynchronizeAdminUserId(
SynchronizeJellyfinAdminUserId request,
CancellationToken cancellationToken)
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
Either<BaseError, Unit> result = await mediator.Send(request, cancellationToken);
result.BiIter(
_ => _logger.LogInformation(
"Successfully synchronized Jellyfin admin user id for source {MediaSourceId}",
request.JellyfinMediaSourceId),
error => _logger.LogWarning(
"Unable to synchronize Jellyfin admin user id for source {MediaSourceId}: {Error}",
request.JellyfinMediaSourceId,
error.Value));
}
private async Task SynchronizeLibraries(SynchronizeJellyfinLibraries request, CancellationToken cancellationToken)
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
Either<BaseError, Unit> result = await mediator.Send(request, cancellationToken);
result.BiIter(
_ => _logger.LogInformation(
"Successfully synchronized Jellyfin libraries for source {MediaSourceId}",
request.JellyfinMediaSourceId),
error => _logger.LogWarning(
"Unable to synchronize Jellyfin libraries for source {MediaSourceId}: {Error}",
request.JellyfinMediaSourceId,
error.Value));
}
private async Task SynchronizeJellyfinLibrary(
ISynchronizeJellyfinLibraryById request,
CancellationToken cancellationToken)
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
Either<BaseError, string> result = await mediator.Send(request, cancellationToken);
result.BiIter(
name => _logger.LogDebug("Done synchronizing jellyfin library {Name}", name),
error =>
{
if (error is ScanIsNotRequired)
{
_logger.LogDebug(
"Scan is not required for jellyfin library {LibraryId} at this time",
request.JellyfinLibraryId);
}
else
{
_logger.LogWarning(
"Unable to synchronize jellyfin library {LibraryId}: {Error}",
request.JellyfinLibraryId,
error.Value);
}
});
if (entityLocker.IsLibraryLocked(request.JellyfinLibraryId))
{
entityLocker.UnlockLibrary(request.JellyfinLibraryId);
}
}
private async Task SynchronizeEmbyLibrary(ISynchronizeEmbyLibraryById request, CancellationToken cancellationToken)
{
using IServiceScope scope = _serviceScopeFactory.CreateScope();
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
Either<BaseError, string> result = await mediator.Send(request, cancellationToken);
result.BiIter(
name => _logger.LogDebug("Done synchronizing emby library {Name}", name),
error =>
{
if (error is ScanIsNotRequired)
{
_logger.LogDebug(
"Scan is not required for emby library {LibraryId} at this time",
request.EmbyLibraryId);
}
else
{
_logger.LogWarning(
"Unable to synchronize emby library {LibraryId}: {Error}",
request.EmbyLibraryId,
error.Value);
}
});
if (entityLocker.IsLibraryLocked(request.EmbyLibraryId))
{
entityLocker.UnlockLibrary(request.EmbyLibraryId);
}
}
}
ErsatzTV/Services/SchedulerService.cs
+7 -15
View File
@@ -18,28 +18,22 @@ namespace ErsatzTV.Services;
public class SchedulerService : BackgroundService
{
private readonly ChannelWriter<IEmbyBackgroundServiceRequest> _embyWorkerChannel;
private readonly ChannelWriter<IScannerBackgroundServiceRequest> _scannerWorkerChannel;
private readonly IEntityLocker _entityLocker;
private readonly ChannelWriter<IJellyfinBackgroundServiceRequest> _jellyfinWorkerChannel;
private readonly ILogger<SchedulerService> _logger;
private readonly ChannelWriter<IPlexBackgroundServiceRequest> _plexWorkerChannel;
private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly ChannelWriter<IBackgroundServiceRequest> _workerChannel;
public SchedulerService(
IServiceScopeFactory serviceScopeFactory,
ChannelWriter<IBackgroundServiceRequest> workerChannel,
ChannelWriter<IPlexBackgroundServiceRequest> plexWorkerChannel,
ChannelWriter<IJellyfinBackgroundServiceRequest> jellyfinWorkerChannel,
ChannelWriter<IEmbyBackgroundServiceRequest> embyWorkerChannel,
ChannelWriter<IScannerBackgroundServiceRequest> scannerWorkerChannel,
IEntityLocker entityLocker,
ILogger<SchedulerService> logger)
{
_serviceScopeFactory = serviceScopeFactory;
_workerChannel = workerChannel;
_plexWorkerChannel = plexWorkerChannel;
_jellyfinWorkerChannel = jellyfinWorkerChannel;
_embyWorkerChannel = embyWorkerChannel;
_scannerWorkerChannel = scannerWorkerChannel;
_entityLocker = entityLocker;
_logger = logger;
}
@@ -200,9 +194,7 @@ public class SchedulerService : BackgroundService
{
if (_entityLocker.LockLibrary(libraryId))
{
await _workerChannel.WriteAsync(
new ScanLocalLibraryIfNeeded(libraryId),
cancellationToken);
await _scannerWorkerChannel.WriteAsync(new ScanLocalLibraryIfNeeded(libraryId), cancellationToken);
}
}
}
@@ -216,7 +208,7 @@ public class SchedulerService : BackgroundService
{
if (_entityLocker.LockLibrary(library.Id))
{
await _plexWorkerChannel.WriteAsync(
await _scannerWorkerChannel.WriteAsync(
new SynchronizePlexLibraryByIdIfNeeded(library.Id),
cancellationToken);
}
@@ -232,7 +224,7 @@ public class SchedulerService : BackgroundService
{
if (_entityLocker.LockLibrary(library.Id))
{
await _jellyfinWorkerChannel.WriteAsync(
await _scannerWorkerChannel.WriteAsync(
new SynchronizeJellyfinLibraryByIdIfNeeded(library.Id),
cancellationToken);
}
@@ -248,7 +240,7 @@ public class SchedulerService : BackgroundService
{
if (_entityLocker.LockLibrary(library.Id))
{
await _embyWorkerChannel.WriteAsync(
await _scannerWorkerChannel.WriteAsync(
new SynchronizeEmbyLibraryByIdIfNeeded(library.Id),
cancellationToken);
}
ErsatzTV/Services/WorkerService.cs
-39
View File
@@ -3,13 +3,9 @@ using Bugsnag;
using ErsatzTV.Application;
using ErsatzTV.Application.Maintenance;
using ErsatzTV.Application.MediaCollections;
using ErsatzTV.Application.MediaSources;
using ErsatzTV.Application.Playouts;
using ErsatzTV.Application.Search;
using ErsatzTV.Application.Subtitles;
using ErsatzTV.Core;
using ErsatzTV.Core.Errors;
using ErsatzTV.Core.Interfaces.Locking;
using MediatR;
namespace ErsatzTV.Services;
@@ -48,7 +44,6 @@ public class WorkerService : BackgroundService
try
{
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
switch (request)
{
@@ -63,40 +58,6 @@ public class WorkerService : BackgroundService
buildPlayout.PlayoutId,
error.Value));
break;
case IScanLocalLibrary scanLocalLibrary:
Either<BaseError, string> scanResult = await mediator.Send(
scanLocalLibrary,
cancellationToken);
scanResult.BiIter(
name => _logger.LogDebug(
"Done scanning local library {Library}",
name),
error =>
{
if (error is ScanIsNotRequired)
{
_logger.LogDebug(
"Scan is not required for local library {LibraryId} at this time",
scanLocalLibrary.LibraryId);
}
else
{
_logger.LogWarning(
"Unable to scan local library {LibraryId}: {Error}",
scanLocalLibrary.LibraryId,
error.Value);
}
});
if (entityLocker.IsLibraryLocked(scanLocalLibrary.LibraryId))
{
entityLocker.UnlockLibrary(scanLocalLibrary.LibraryId);
}
break;
case RebuildSearchIndex rebuildSearchIndex:
await mediator.Send(rebuildSearchIndex, cancellationToken);
break;
case DeleteOrphanedArtwork deleteOrphanedArtwork:
_logger.LogInformation("Deleting orphaned artwork from the database");
await mediator.Send(deleteOrphanedArtwork, cancellationToken);