You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
jellyfin/MediaBrowser.Server.Impleme.../Sync/SyncJobProcessor.cs

494 lines
16 KiB

using MediaBrowser.Controller.Entities;
using MediaBrowser.Controller.Entities.Audio;
using MediaBrowser.Controller.Library;
using MediaBrowser.Controller.Sync;
10 years ago
using MediaBrowser.Controller.TV;
using MediaBrowser.Model.Dlna;
using MediaBrowser.Model.Dto;
10 years ago
using MediaBrowser.Model.Entities;
using MediaBrowser.Model.Logging;
using MediaBrowser.Model.MediaInfo;
10 years ago
using MediaBrowser.Model.Querying;
using MediaBrowser.Model.Session;
using MediaBrowser.Model.Sync;
using MoreLinq;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
namespace MediaBrowser.Server.Implementations.Sync
{
public class SyncJobProcessor
{
private readonly ILibraryManager _libraryManager;
private readonly ISyncRepository _syncRepo;
private readonly ISyncManager _syncManager;
private readonly ILogger _logger;
private readonly IUserManager _userManager;
10 years ago
private readonly ITVSeriesManager _tvSeriesManager;
10 years ago
public SyncJobProcessor(ILibraryManager libraryManager, ISyncRepository syncRepo, ISyncManager syncManager, ILogger logger, IUserManager userManager, ITVSeriesManager tvSeriesManager)
{
_libraryManager = libraryManager;
_syncRepo = syncRepo;
_syncManager = syncManager;
_logger = logger;
_userManager = userManager;
10 years ago
_tvSeriesManager = tvSeriesManager;
}
public async Task EnsureJobItems(SyncJob job)
{
var user = _userManager.GetUserById(job.UserId);
if (user == null)
{
throw new InvalidOperationException("Cannot proceed with sync because user no longer exists.");
}
10 years ago
var items = (await GetItemsForSync(job.Category, job.ParentId, job.RequestedItemIds, user, job.UnwatchedOnly).ConfigureAwait(false))
.ToList();
var jobItems = _syncRepo.GetJobItems(new SyncJobItemQuery
{
JobId = job.Id
}).Items.ToList();
foreach (var item in items)
{
10 years ago
// Respect ItemLimit, if set
if (job.ItemLimit.HasValue)
{
if (jobItems.Count >= job.ItemLimit.Value)
{
break;
}
}
var itemId = item.Id.ToString("N");
var jobItem = jobItems.FirstOrDefault(i => string.Equals(i.ItemId, itemId, StringComparison.OrdinalIgnoreCase));
if (jobItem != null)
{
continue;
}
jobItem = new SyncJobItem
{
Id = Guid.NewGuid().ToString("N"),
ItemId = itemId,
JobId = job.Id,
TargetId = job.TargetId,
DateCreated = DateTime.UtcNow
};
await _syncRepo.Create(jobItem).ConfigureAwait(false);
jobItems.Add(jobItem);
}
jobItems = jobItems
.OrderBy(i => i.DateCreated)
.ToList();
await UpdateJobStatus(job, jobItems).ConfigureAwait(false);
}
10 years ago
public Task UpdateJobStatus(string id)
{
var job = _syncRepo.GetJob(id);
return UpdateJobStatus(job);
}
private Task UpdateJobStatus(SyncJob job)
{
if (job == null)
{
throw new ArgumentNullException("job");
}
var result = _syncRepo.GetJobItems(new SyncJobItemQuery
{
JobId = job.Id
});
return UpdateJobStatus(job, result.Items.ToList());
}
private Task UpdateJobStatus(SyncJob job, List<SyncJobItem> jobItems)
{
job.ItemCount = jobItems.Count;
double pct = 0;
foreach (var item in jobItems)
{
10 years ago
if (item.Status == SyncJobItemStatus.Failed || item.Status == SyncJobItemStatus.Synced || item.Status == SyncJobItemStatus.RemovedFromDevice)
{
pct += 100;
}
else
{
pct += item.Progress ?? 0;
}
}
if (job.ItemCount > 0)
{
pct /= job.ItemCount;
job.Progress = pct;
}
else
{
job.Progress = null;
}
if (pct >= 100)
{
if (jobItems.Any(i => i.Status == SyncJobItemStatus.Failed))
{
job.Status = SyncJobStatus.CompletedWithError;
}
else
{
job.Status = SyncJobStatus.Completed;
}
}
else if (pct.Equals(0))
{
job.Status = SyncJobStatus.Queued;
}
else
{
job.Status = SyncJobStatus.InProgress;
}
return _syncRepo.Update(job);
}
10 years ago
public async Task<IEnumerable<BaseItem>> GetItemsForSync(SyncCategory? category, string parentId, IEnumerable<string> itemIds, User user, bool unwatchedOnly)
{
10 years ago
var items = category.HasValue ?
await GetItemsForSync(category.Value, parentId, user).ConfigureAwait(false) :
itemIds.SelectMany(i => GetItemsForSync(i, user))
10 years ago
.Where(_syncManager.SupportsSync);
if (unwatchedOnly)
{
// Avoid implicitly captured closure
var currentUser = user;
items = items.Where(i =>
{
var video = i as Video;
if (video != null)
{
return !video.IsPlayed(currentUser);
}
return true;
});
}
return items.DistinctBy(i => i.Id);
}
10 years ago
private async Task<IEnumerable<BaseItem>> GetItemsForSync(SyncCategory category, string parentId, User user)
{
var parent = string.IsNullOrWhiteSpace(parentId)
? user.RootFolder
: (Folder)_libraryManager.GetItemById(parentId);
InternalItemsQuery query;
switch (category)
{
case SyncCategory.Latest:
query = new InternalItemsQuery
{
IsFolder = false,
SortBy = new[] { ItemSortBy.DateCreated, ItemSortBy.SortName },
SortOrder = SortOrder.Descending,
Recursive = true
};
break;
case SyncCategory.Resume:
query = new InternalItemsQuery
{
IsFolder = false,
SortBy = new[] { ItemSortBy.DatePlayed, ItemSortBy.SortName },
SortOrder = SortOrder.Descending,
Recursive = true,
IsResumable = true,
MediaTypes = new[] { MediaType.Video }
};
break;
case SyncCategory.NextUp:
return _tvSeriesManager.GetNextUp(new NextUpQuery
{
ParentId = parentId,
UserId = user.Id.ToString("N")
}).Items;
default:
throw new ArgumentException("Unrecognized category: " + category);
}
query.User = user;
var result = await parent.GetItems(query).ConfigureAwait(false);
return result.Items;
}
private IEnumerable<BaseItem> GetItemsForSync(string id, User user)
{
var item = _libraryManager.GetItemById(id);
if (item == null)
{
return new List<BaseItem>();
}
return GetItemsForSync(item, user);
}
private IEnumerable<BaseItem> GetItemsForSync(BaseItem item, User user)
{
var itemByName = item as IItemByName;
if (itemByName != null)
{
var items = user.RootFolder
.GetRecursiveChildren(user);
return itemByName.GetTaggedItems(items);
10 years ago
}
if (item.IsFolder)
{
var folder = (Folder)item;
var items = folder.GetRecursiveChildren(user);
items = items.Where(i => !i.IsFolder);
if (!folder.IsPreSorted)
{
items = items.OrderBy(i => i.SortName);
}
return items;
}
return new[] { item };
}
public async Task EnsureSyncJobs(CancellationToken cancellationToken)
{
var jobResult = _syncRepo.GetJobs(new SyncJobQuery
{
IsCompleted = false
});
foreach (var job in jobResult.Items)
{
cancellationToken.ThrowIfCancellationRequested();
if (job.SyncNewContent)
{
await EnsureJobItems(job).ConfigureAwait(false);
}
}
}
public async Task Sync(IProgress<double> progress, CancellationToken cancellationToken)
{
await EnsureSyncJobs(cancellationToken).ConfigureAwait(false);
var result = _syncRepo.GetJobItems(new SyncJobItemQuery
{
10 years ago
Status = SyncJobItemStatus.Queued
});
var jobItems = result.Items;
var index = 0;
foreach (var item in jobItems)
{
double percent = index;
percent /= result.TotalRecordCount;
progress.Report(100 * percent);
cancellationToken.ThrowIfCancellationRequested();
if (item.Status == SyncJobItemStatus.Queued)
{
await ProcessJobItem(item, cancellationToken).ConfigureAwait(false);
}
var job = _syncRepo.GetJob(item.JobId);
await UpdateJobStatus(job).ConfigureAwait(false);
index++;
}
}
private async Task ProcessJobItem(SyncJobItem jobItem, CancellationToken cancellationToken)
{
var item = _libraryManager.GetItemById(jobItem.ItemId);
if (item == null)
{
jobItem.Status = SyncJobItemStatus.Failed;
_logger.Error("Unable to locate library item for JobItem {0}, ItemId {1}", jobItem.Id, jobItem.ItemId);
await _syncRepo.Update(jobItem).ConfigureAwait(false);
return;
}
var deviceProfile = _syncManager.GetDeviceProfile(jobItem.TargetId);
if (deviceProfile == null)
{
jobItem.Status = SyncJobItemStatus.Failed;
_logger.Error("Unable to locate SyncTarget for JobItem {0}, SyncTargetId {1}", jobItem.Id, jobItem.TargetId);
await _syncRepo.Update(jobItem).ConfigureAwait(false);
return;
}
jobItem.Progress = 0;
jobItem.Status = SyncJobItemStatus.Converting;
var video = item as Video;
if (video != null)
{
10 years ago
await Sync(jobItem, video, deviceProfile, cancellationToken).ConfigureAwait(false);
}
else if (item is Audio)
{
10 years ago
await Sync(jobItem, (Audio)item, deviceProfile, cancellationToken).ConfigureAwait(false);
}
else if (item is Photo)
{
10 years ago
await Sync(jobItem, (Photo)item, deviceProfile, cancellationToken).ConfigureAwait(false);
}
10 years ago
else
{
10 years ago
await SyncGeneric(jobItem, item, deviceProfile, cancellationToken).ConfigureAwait(false);
}
}
10 years ago
private async Task Sync(SyncJobItem jobItem, Video item, DeviceProfile profile, CancellationToken cancellationToken)
{
var options = new VideoOptions
{
Context = EncodingContext.Streaming,
ItemId = item.Id.ToString("N"),
DeviceId = jobItem.TargetId,
Profile = profile,
MediaSources = item.GetMediaSources(false).ToList()
};
var streamInfo = new StreamBuilder().BuildVideoItem(options);
var mediaSource = streamInfo.MediaSource;
10 years ago
jobItem.MediaSourceId = streamInfo.MediaSourceId;
10 years ago
if (streamInfo.PlayMethod == PlayMethod.Transcode)
{
await _syncRepo.Update(jobItem).ConfigureAwait(false);
}
else
{
if (mediaSource.Protocol == MediaProtocol.File)
{
10 years ago
jobItem.OutputPath = mediaSource.Path;
}
if (mediaSource.Protocol == MediaProtocol.Http)
{
10 years ago
jobItem.OutputPath = await DownloadFile(jobItem, mediaSource, cancellationToken).ConfigureAwait(false);
}
throw new InvalidOperationException(string.Format("Cannot direct stream {0} protocol", mediaSource.Protocol));
}
// TODO: Transcode
10 years ago
jobItem.OutputPath = mediaSource.Path;
jobItem.Progress = 50;
jobItem.Status = SyncJobItemStatus.Transferring;
await _syncRepo.Update(jobItem).ConfigureAwait(false);
}
10 years ago
private async Task Sync(SyncJobItem jobItem, Audio item, DeviceProfile profile, CancellationToken cancellationToken)
{
var options = new AudioOptions
{
Context = EncodingContext.Streaming,
ItemId = item.Id.ToString("N"),
DeviceId = jobItem.TargetId,
Profile = profile,
MediaSources = item.GetMediaSources(false).ToList()
};
var streamInfo = new StreamBuilder().BuildAudioItem(options);
var mediaSource = streamInfo.MediaSource;
10 years ago
jobItem.MediaSourceId = streamInfo.MediaSourceId;
10 years ago
if (streamInfo.PlayMethod == PlayMethod.Transcode)
{
await _syncRepo.Update(jobItem).ConfigureAwait(false);
}
else
{
if (mediaSource.Protocol == MediaProtocol.File)
{
10 years ago
jobItem.OutputPath = mediaSource.Path;
}
if (mediaSource.Protocol == MediaProtocol.Http)
{
10 years ago
jobItem.OutputPath = await DownloadFile(jobItem, mediaSource, cancellationToken).ConfigureAwait(false);
}
throw new InvalidOperationException(string.Format("Cannot direct stream {0} protocol", mediaSource.Protocol));
}
// TODO: Transcode
10 years ago
jobItem.OutputPath = mediaSource.Path;
10 years ago
jobItem.Progress = 50;
jobItem.Status = SyncJobItemStatus.Transferring;
await _syncRepo.Update(jobItem).ConfigureAwait(false);
}
10 years ago
private async Task Sync(SyncJobItem jobItem, Photo item, DeviceProfile profile, CancellationToken cancellationToken)
{
10 years ago
jobItem.OutputPath = item.Path;
jobItem.Progress = 50;
jobItem.Status = SyncJobItemStatus.Transferring;
await _syncRepo.Update(jobItem).ConfigureAwait(false);
}
10 years ago
private async Task SyncGeneric(SyncJobItem jobItem, BaseItem item, DeviceProfile profile, CancellationToken cancellationToken)
{
10 years ago
jobItem.OutputPath = item.Path;
jobItem.Progress = 50;
jobItem.Status = SyncJobItemStatus.Transferring;
await _syncRepo.Update(jobItem).ConfigureAwait(false);
}
private async Task<string> DownloadFile(SyncJobItem jobItem, MediaSourceInfo mediaSource, CancellationToken cancellationToken)
{
// TODO: Download
return mediaSource.Path;
}
}
}