Initial commit
This commit is contained in:
Jason Dove
@@ -0,0 +1,46 @@
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using ErsatzTV.Core.Domain;
|
||||
using ErsatzTV.Core.Interfaces.FFmpeg;
|
||||
using LanguageExt;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace ErsatzTV.Services
|
||||
{
|
||||
public class FFmpegLocatorService : IHostedService
|
||||
{
|
||||
private readonly ILogger<FFmpegLocatorService> _logger;
|
||||
private readonly IServiceScopeFactory _serviceScopeFactory;
|
||||
|
||||
public FFmpegLocatorService(
|
||||
IServiceScopeFactory serviceScopeFactory,
|
||||
ILogger<FFmpegLocatorService> logger)
|
||||
{
|
||||
_serviceScopeFactory = serviceScopeFactory;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
using IServiceScope scope = _serviceScopeFactory.CreateScope();
|
||||
IFFmpegLocator ffmpegLocator = scope.ServiceProvider.GetRequiredService<IFFmpegLocator>();
|
||||
|
||||
// check for ffmpeg and ffprobe in the last known/configured location
|
||||
// otherwise search using which/where and save any located executables
|
||||
Option<string> maybeFFmpegPath = await ffmpegLocator.ValidatePath("ffmpeg", ConfigElementKey.FFmpegPath);
|
||||
maybeFFmpegPath.Match(
|
||||
path => _logger.LogInformation("Located ffmpeg at {Path}", path),
|
||||
() => _logger.LogWarning("Failed to locate ffmpeg executable"));
|
||||
|
||||
Option<string> maybeFFprobePath =
|
||||
await ffmpegLocator.ValidatePath("ffprobe", ConfigElementKey.FFprobePath);
|
||||
maybeFFprobePath.Match(
|
||||
path => _logger.LogInformation("Located ffprobe at {Path}", path),
|
||||
() => _logger.LogWarning("Failed to locate ffprobe executable"));
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,144 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Channels;
|
||||
using System.Threading.Tasks;
|
||||
using ErsatzTV.Application;
|
||||
using ErsatzTV.Application.MediaSources.Commands;
|
||||
using ErsatzTV.Core;
|
||||
using ErsatzTV.Core.Domain;
|
||||
using LanguageExt;
|
||||
using MediatR;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Unit = LanguageExt.Unit;
|
||||
|
||||
namespace ErsatzTV.Services
|
||||
{
|
||||
public class PlexService : BackgroundService
|
||||
{
|
||||
private readonly ChannelReader<IPlexBackgroundServiceRequest> _channel;
|
||||
private readonly ILogger<PlexService> _logger;
|
||||
private readonly IServiceScopeFactory _serviceScopeFactory;
|
||||
|
||||
public PlexService(
|
||||
ChannelReader<IPlexBackgroundServiceRequest> channel,
|
||||
IServiceScopeFactory serviceScopeFactory,
|
||||
ILogger<PlexService> logger)
|
||||
{
|
||||
_channel = channel;
|
||||
_serviceScopeFactory = serviceScopeFactory;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (!File.Exists(FileSystemLayout.PlexSecretsPath))
|
||||
{
|
||||
await File.WriteAllTextAsync(FileSystemLayout.PlexSecretsPath, "{}", cancellationToken);
|
||||
}
|
||||
|
||||
_logger.LogInformation(
|
||||
"Plex service started; secrets are at {PlexSecretsPath}",
|
||||
FileSystemLayout.PlexSecretsPath);
|
||||
|
||||
// synchronize sources on startup
|
||||
List<PlexMediaSource> sources = await SynchronizeSources(
|
||||
new SynchronizePlexMediaSources(),
|
||||
cancellationToken);
|
||||
foreach (PlexMediaSource source in sources)
|
||||
{
|
||||
await SynchronizeLibraries(new SynchronizePlexLibraries(source.Id), cancellationToken);
|
||||
}
|
||||
|
||||
await foreach (IPlexBackgroundServiceRequest request in _channel.ReadAllAsync(cancellationToken))
|
||||
{
|
||||
try
|
||||
{
|
||||
Task requestTask = request switch
|
||||
{
|
||||
TryCompletePlexPinFlow pinRequest => CompletePinFlow(pinRequest, cancellationToken),
|
||||
SynchronizePlexMediaSources sourcesRequest => SynchronizeSources(
|
||||
sourcesRequest,
|
||||
cancellationToken)
|
||||
};
|
||||
|
||||
await requestTask;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to process poll for Plex auth token request");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<List<PlexMediaSource>> SynchronizeSources(
|
||||
SynchronizePlexMediaSources request,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
using IServiceScope scope = _serviceScopeFactory.CreateScope();
|
||||
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
|
||||
|
||||
Either<BaseError, List<PlexMediaSource>> result = await mediator.Send(request, cancellationToken);
|
||||
return result.Match(
|
||||
sources =>
|
||||
{
|
||||
if (sources.Any())
|
||||
{
|
||||
_logger.LogInformation("Successfully synchronized plex media sources");
|
||||
}
|
||||
|
||||
return sources;
|
||||
},
|
||||
error =>
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"Unable to synchronize plex media sources: {Error}",
|
||||
error.Value);
|
||||
return new List<PlexMediaSource>();
|
||||
});
|
||||
}
|
||||
|
||||
private async Task CompletePinFlow(
|
||||
TryCompletePlexPinFlow request,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
using IServiceScope scope = _serviceScopeFactory.CreateScope();
|
||||
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
|
||||
|
||||
Either<BaseError, bool> result = await mediator.Send(request, cancellationToken);
|
||||
result.BiIter(
|
||||
success =>
|
||||
{
|
||||
if (success)
|
||||
{
|
||||
_logger.LogInformation("Successfully authenticated with plex");
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger.LogInformation("Plex authentication timeout");
|
||||
}
|
||||
},
|
||||
error => _logger.LogWarning("Unable to poll plex token: {Error}", error.Value));
|
||||
}
|
||||
|
||||
private async Task SynchronizeLibraries(SynchronizePlexLibraries 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 plex libraries for source {MediaSourceId}",
|
||||
request.PlexMediaSourceId),
|
||||
error => _logger.LogWarning(
|
||||
"Unable to synchronize plex libraries for source {MediaSourceId}: {Error}",
|
||||
request.PlexMediaSourceId,
|
||||
error.Value));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Channels;
|
||||
using System.Threading.Tasks;
|
||||
using ErsatzTV.Application;
|
||||
using ErsatzTV.Application.MediaSources.Commands;
|
||||
using ErsatzTV.Application.Playouts.Commands;
|
||||
using ErsatzTV.Infrastructure.Data;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
|
||||
namespace ErsatzTV.Services
|
||||
{
|
||||
public class SchedulerService : IHostedService
|
||||
{
|
||||
private readonly ChannelWriter<IBackgroundServiceRequest> _channel;
|
||||
private readonly IServiceScopeFactory _serviceScopeFactory;
|
||||
private Timer _timer;
|
||||
|
||||
public SchedulerService(
|
||||
IServiceScopeFactory serviceScopeFactory,
|
||||
ChannelWriter<IBackgroundServiceRequest> channel)
|
||||
{
|
||||
_serviceScopeFactory = serviceScopeFactory;
|
||||
_channel = channel;
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_timer = new Timer(
|
||||
async _ => await DoWork(cancellationToken),
|
||||
null,
|
||||
TimeSpan.FromSeconds(0), // fire immediately
|
||||
TimeSpan.FromHours(1)); // repeat every hour
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_timer?.Change(Timeout.Infinite, 0);
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private async Task DoWork(CancellationToken cancellationToken)
|
||||
{
|
||||
await BuildPlayouts(cancellationToken);
|
||||
await ScanLocalMediaSources(cancellationToken);
|
||||
}
|
||||
|
||||
|
||||
private async Task BuildPlayouts(CancellationToken cancellationToken)
|
||||
{
|
||||
using IServiceScope scope = _serviceScopeFactory.CreateScope();
|
||||
TvContext dbContext = scope.ServiceProvider.GetRequiredService<TvContext>();
|
||||
|
||||
List<int> playoutIds = await dbContext.Playouts.Map(p => p.Id).ToListAsync(cancellationToken);
|
||||
foreach (int playoutId in playoutIds)
|
||||
{
|
||||
await _channel.WriteAsync(new BuildPlayout(playoutId), cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ScanLocalMediaSources(CancellationToken cancellationToken)
|
||||
{
|
||||
using IServiceScope scope = _serviceScopeFactory.CreateScope();
|
||||
TvContext dbContext = scope.ServiceProvider.GetRequiredService<TvContext>();
|
||||
|
||||
List<int> localMediaSourceIds = await dbContext.LocalMediaSources
|
||||
.Map(ms => ms.Id)
|
||||
.ToListAsync(cancellationToken);
|
||||
|
||||
foreach (int mediaSourceId in localMediaSourceIds)
|
||||
{
|
||||
await _channel.WriteAsync(new ScanLocalMediaSource(mediaSourceId), cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Channels;
|
||||
using System.Threading.Tasks;
|
||||
using ErsatzTV.Application;
|
||||
using ErsatzTV.Application.MediaItems.Commands;
|
||||
using ErsatzTV.Application.MediaSources.Commands;
|
||||
using ErsatzTV.Application.Playouts.Commands;
|
||||
using ErsatzTV.Core;
|
||||
using LanguageExt;
|
||||
using MediatR;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Unit = LanguageExt.Unit;
|
||||
|
||||
namespace ErsatzTV.Services
|
||||
{
|
||||
public class WorkerService : BackgroundService
|
||||
{
|
||||
private readonly ChannelReader<IBackgroundServiceRequest> _channel;
|
||||
private readonly ILogger<WorkerService> _logger;
|
||||
private readonly IServiceScopeFactory _serviceScopeFactory;
|
||||
|
||||
public WorkerService(
|
||||
ChannelReader<IBackgroundServiceRequest> channel,
|
||||
IServiceScopeFactory serviceScopeFactory,
|
||||
ILogger<WorkerService> logger)
|
||||
{
|
||||
_channel = channel;
|
||||
_serviceScopeFactory = serviceScopeFactory;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("Worker service started");
|
||||
|
||||
await foreach (IBackgroundServiceRequest request in _channel.ReadAllAsync(cancellationToken))
|
||||
{
|
||||
try
|
||||
{
|
||||
using IServiceScope scope = _serviceScopeFactory.CreateScope();
|
||||
IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
|
||||
|
||||
switch (request)
|
||||
{
|
||||
case BuildPlayout buildPlayout:
|
||||
Either<BaseError, Unit> buildPlayoutResult = await mediator.Send(
|
||||
buildPlayout,
|
||||
cancellationToken);
|
||||
buildPlayoutResult.BiIter(
|
||||
_ => _logger.LogDebug("Built playout {PlayoutId}", buildPlayout.PlayoutId),
|
||||
error => _logger.LogWarning(
|
||||
"Unable to build playout {PlayoutId}: {Error}",
|
||||
buildPlayout.PlayoutId,
|
||||
error.Value));
|
||||
break;
|
||||
case RefreshMediaItem refreshMediaItem:
|
||||
string type = refreshMediaItem switch
|
||||
{
|
||||
RefreshMediaItemMetadata => "metadata",
|
||||
RefreshMediaItemStatistics => "statistics",
|
||||
RefreshMediaItemCollections => "collections",
|
||||
_ => ""
|
||||
};
|
||||
|
||||
// TODO: different request types for different media source types?
|
||||
Either<BaseError, Unit> refreshMediaItemResult =
|
||||
await mediator.Send(refreshMediaItem, cancellationToken);
|
||||
refreshMediaItemResult.Match(
|
||||
_ => _logger.LogDebug(
|
||||
$"Refreshed {type} for media item {{MediaItemId}}",
|
||||
refreshMediaItem.MediaItemId),
|
||||
error => _logger.LogWarning(
|
||||
$"Unable to refresh {type} for media item {{MediaItemId}}: {{Error}}",
|
||||
refreshMediaItem.MediaItemId,
|
||||
error.Value));
|
||||
break;
|
||||
case ScanLocalMediaSource scanLocalMediaSource:
|
||||
Either<BaseError, string> scanResult = await mediator.Send(
|
||||
scanLocalMediaSource,
|
||||
cancellationToken);
|
||||
scanResult.BiIter(
|
||||
name => _logger.LogDebug(
|
||||
"Done scanning local media source {MediaSource}",
|
||||
name),
|
||||
error => _logger.LogWarning(
|
||||
"Unable to scan local media source {MediaSourceId}: {Error}",
|
||||
scanLocalMediaSource.MediaSourceId,
|
||||
error.Value));
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to process background service request");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user