update server sync

MediaBrowser.Server.Implementations/Sync/MediaSync.cs

@@ -206,9 +206,12 @@ namespace MediaBrowser.Server.Implementations.Sync
             await dataProvider.Delete(target, localId).ConfigureAwait(false);
         }
 
-        private Task SendFile(IServerSyncProvider provider, string inputPath, LocalItem item, SyncTarget target, CancellationToken cancellationToken)
+        private async Task SendFile(IServerSyncProvider provider, string inputPath, LocalItem item, SyncTarget target, CancellationToken cancellationToken)
         {
-            return provider.SendFile(inputPath, item.LocalPath, target, new Progress<double>(), cancellationToken);
+            using (var stream = _fileSystem.GetFileStream(inputPath, FileMode.Open, FileAccess.Read, FileShare.Read, true))
+            {
+                await provider.SendFile(stream, item.LocalPath, target, new Progress<double>(), cancellationToken).ConfigureAwait(false);
+            }
         }
 
         private string GetLocalId(string serverId, string itemId)

MediaBrowser.Server.Implementations/Sync/MultiProviderSync.cs

@@ -14,12 +14,12 @@ namespace MediaBrowser.Server.Implementations.Sync
 {
     public class MultiProviderSync
     {
-        private readonly ISyncManager _syncManager;
+        private readonly SyncManager _syncManager;
         private readonly IServerApplicationHost _appHost;
         private readonly ILogger _logger;
         private readonly IFileSystem _fileSystem;
 
-        public MultiProviderSync(ISyncManager syncManager, IServerApplicationHost appHost, ILogger logger, IFileSystem fileSystem)
+        public MultiProviderSync(SyncManager syncManager, IServerApplicationHost appHost, ILogger logger, IFileSystem fileSystem)
         {
             _syncManager = syncManager;
             _appHost = appHost;
@@ -54,8 +54,10 @@ namespace MediaBrowser.Server.Implementations.Sync
                     progress.Report(totalProgress);
                 });
 
+                var dataProvider = _syncManager.GetDataProvider(target.Item1, target.Item2);
+
                 await new MediaSync(_logger, _syncManager, _appHost, _fileSystem)
-                    .Sync(target.Item1, target.Item1.GetDataProvider(), target.Item2, innerProgress, cancellationToken)
+                    .Sync(target.Item1, dataProvider, target.Item2, innerProgress, cancellationToken)
                     .ConfigureAwait(false);
 
                 numComplete++;

MediaBrowser.Server.Implementations/Sync/ServerSyncScheduledTask.cs

@@ -46,7 +46,7 @@ namespace MediaBrowser.Server.Implementations.Sync
 
         public Task Execute(CancellationToken cancellationToken, IProgress<double> progress)
         {
-            return new MultiProviderSync(_syncManager, _appHost, _logger, _fileSystem)
+            return new MultiProviderSync((SyncManager)_syncManager, _appHost, _logger, _fileSystem)
                 .Sync(ServerSyncProviders, progress, cancellationToken);
         }
 

MediaBrowser.Server.Implementations/Sync/SyncManager.cs

@@ -21,10 +21,12 @@ using MediaBrowser.Model.Entities;
 using MediaBrowser.Model.Events;
 using MediaBrowser.Model.Logging;
 using MediaBrowser.Model.Querying;
+using MediaBrowser.Model.Serialization;
 using MediaBrowser.Model.Sync;
 using MediaBrowser.Model.Users;
 using MoreLinq;
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
@@ -49,6 +51,7 @@ namespace MediaBrowser.Server.Implementations.Sync
         private readonly IConfigurationManager _config;
         private readonly IUserDataManager _userDataManager;
         private readonly Func<IMediaSourceManager> _mediaSourceManager;
+        private readonly IJsonSerializer _json;
 
         private ISyncProvider[] _providers = { };
 
@@ -58,7 +61,7 @@ namespace MediaBrowser.Server.Implementations.Sync
         public event EventHandler<GenericEventArgs<SyncJobItem>> SyncJobItemUpdated;
         public event EventHandler<GenericEventArgs<SyncJobItem>> SyncJobItemCreated;
 
-        public SyncManager(ILibraryManager libraryManager, ISyncRepository repo, IImageProcessor imageProcessor, ILogger logger, IUserManager userManager, Func<IDtoService> dtoService, IApplicationHost appHost, ITVSeriesManager tvSeriesManager, Func<IMediaEncoder> mediaEncoder, IFileSystem fileSystem, Func<ISubtitleEncoder> subtitleEncoder, IConfigurationManager config, IUserDataManager userDataManager, Func<IMediaSourceManager> mediaSourceManager)
+        public SyncManager(ILibraryManager libraryManager, ISyncRepository repo, IImageProcessor imageProcessor, ILogger logger, IUserManager userManager, Func<IDtoService> dtoService, IApplicationHost appHost, ITVSeriesManager tvSeriesManager, Func<IMediaEncoder> mediaEncoder, IFileSystem fileSystem, Func<ISubtitleEncoder> subtitleEncoder, IConfigurationManager config, IUserDataManager userDataManager, Func<IMediaSourceManager> mediaSourceManager, IJsonSerializer json)
         {
             _libraryManager = libraryManager;
             _repo = repo;
@@ -74,6 +77,7 @@ namespace MediaBrowser.Server.Implementations.Sync
             _config = config;
             _userDataManager = userDataManager;
             _mediaSourceManager = mediaSourceManager;
+            _json = json;
         }
 
         public void AddParts(IEnumerable<ISyncProvider> providers)
@@ -86,6 +90,14 @@ namespace MediaBrowser.Server.Implementations.Sync
             get { return _providers.OfType<IServerSyncProvider>(); }
         }
 
+        private readonly ConcurrentDictionary<string, ISyncDataProvider> _dataProviders =
+            new ConcurrentDictionary<string, ISyncDataProvider>(StringComparer.OrdinalIgnoreCase);
+ 
+        public ISyncDataProvider GetDataProvider(IServerSyncProvider provider, SyncTarget target)
+        {
+            return _dataProviders.GetOrAdd(target.Id, key => new TargetDataProvider(provider, target, _appHost.SystemId, _logger, _json, _fileSystem, _config.CommonApplicationPaths));
+        }
+
         public async Task<SyncJobCreationResult> CreateJob(SyncJobRequest request)
         {
             var processor = GetSyncJobProcessor();

MediaBrowser.Server.Implementations/Sync/TargetDataProvider.cs

@@ -0,0 +1,242 @@
+using MediaBrowser.Common.Configuration;
+using MediaBrowser.Common.Extensions;
+using MediaBrowser.Common.IO;
+using MediaBrowser.Controller.Sync;
+using MediaBrowser.Model.Logging;
+using MediaBrowser.Model.Serialization;
+using MediaBrowser.Model.Sync;
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MediaBrowser.Server.Implementations.Sync
+{
+    public class TargetDataProvider : ISyncDataProvider
+    {
+        private readonly SyncTarget _target;
+        private readonly IServerSyncProvider _provider;
+
+        private readonly SemaphoreSlim _dataLock = new SemaphoreSlim(1, 1);
+        private List<LocalItem> _items;
+
+        private readonly ILogger _logger;
+        private readonly IJsonSerializer _json;
+        private readonly IFileSystem _fileSystem;
+        private readonly IApplicationPaths _appPaths;
+        private readonly string _serverId;
+
+        private readonly SemaphoreSlim _cacheFileLock = new SemaphoreSlim(1, 1);
+
+        public TargetDataProvider(IServerSyncProvider provider, SyncTarget target, string serverId, ILogger logger, IJsonSerializer json, IFileSystem fileSystem, IApplicationPaths appPaths)
+        {
+            _logger = logger;
+            _json = json;
+            _provider = provider;
+            _target = target;
+            _fileSystem = fileSystem;
+            _appPaths = appPaths;
+            _serverId = serverId;
+        }
+
+        private string GetCachePath()
+        {
+            return Path.Combine(_appPaths.DataPath, "sync", _target.Id.GetMD5().ToString("N") + ".json");
+        }
+
+        private string GetRemotePath()
+        {
+            var parts = new List<string>
+            {
+                _serverId,
+                "data.json"
+            };
+
+            return _provider.GetFullPath(parts, _target);
+        }
+
+        private async Task CacheData(Stream stream)
+        {
+            var cachePath = GetCachePath();
+
+            await _cacheFileLock.WaitAsync().ConfigureAwait(false);
+
+            try
+            {
+                Directory.CreateDirectory(Path.GetDirectoryName(cachePath));
+                using (var fileStream = _fileSystem.GetFileStream(cachePath, FileMode.Create, FileAccess.Write, FileShare.Read, true))
+                {
+                    await stream.CopyToAsync(fileStream).ConfigureAwait(false);
+                }
+            }
+            catch (Exception ex)
+            {
+                _logger.ErrorException("Error saving sync data to {0}", ex, cachePath);
+            }
+            finally
+            {
+                _cacheFileLock.Release();
+            }
+        }
+
+        private async Task EnsureData(CancellationToken cancellationToken)
+        {
+            if (_items == null)
+            {
+                try
+                {
+                    using (var stream = await _provider.GetFile(GetRemotePath(), _target, new Progress<double>(), cancellationToken))
+                    {
+                        _items = _json.DeserializeFromStream<List<LocalItem>>(stream);
+                    }
+                }
+                catch (FileNotFoundException)
+                {
+                    _items = new List<LocalItem>();
+                }
+                catch (DirectoryNotFoundException)
+                {
+                    _items = new List<LocalItem>();
+                }
+
+                using (var memoryStream = new MemoryStream())
+                {
+                    _json.SerializeToStream(_items, memoryStream);
+                    
+                    // Now cache it
+                    memoryStream.Position = 0;
+                    await CacheData(memoryStream).ConfigureAwait(false);
+                }
+            }
+        }
+
+        private async Task SaveData(CancellationToken cancellationToken)
+        {
+            using (var stream = new MemoryStream())
+            {
+                _json.SerializeToStream(_items, stream);
+
+                // Save to sync provider
+                stream.Position = 0;
+                await _provider.SendFile(stream, GetRemotePath(), _target, new Progress<double>(), cancellationToken).ConfigureAwait(false);
+
+                // Now cache it
+                stream.Position = 0;
+                await CacheData(stream).ConfigureAwait(false);
+            }
+        }
+
+        private async Task<T> GetData<T>(Func<List<LocalItem>, T> dataFactory)
+        {
+            await _dataLock.WaitAsync().ConfigureAwait(false);
+
+            try
+            {
+                await EnsureData(CancellationToken.None).ConfigureAwait(false);
+
+                return dataFactory(_items);
+            }
+            finally
+            {
+                _dataLock.Release();
+            }
+        }
+
+        private async Task UpdateData(Func<List<LocalItem>, List<LocalItem>> action)
+        {
+            await _dataLock.WaitAsync().ConfigureAwait(false);
+
+            try
+            {
+                await EnsureData(CancellationToken.None).ConfigureAwait(false);
+
+                _items = action(_items);
+
+                await SaveData(CancellationToken.None).ConfigureAwait(false);
+            }
+            finally
+            {
+                _dataLock.Release();
+            }
+        }
+
+        public Task<List<string>> GetServerItemIds(SyncTarget target, string serverId)
+        {
+            return GetData(items => items.Where(i => string.Equals(i.ServerId, serverId, StringComparison.OrdinalIgnoreCase)).Select(i => i.ItemId).ToList());
+        }
+
+        public Task AddOrUpdate(SyncTarget target, LocalItem item)
+        {
+            return UpdateData(items =>
+            {
+                var list = items.Where(i => !string.Equals(i.Id, item.Id, StringComparison.OrdinalIgnoreCase))
+                    .ToList();
+
+                list.Add(item);
+
+                return list;
+            });
+        }
+
+        public Task Delete(SyncTarget target, string id)
+        {
+            return UpdateData(items => items.Where(i => !string.Equals(i.Id, id, StringComparison.OrdinalIgnoreCase)).ToList());
+        }
+
+        public Task<LocalItem> Get(SyncTarget target, string id)
+        {
+            return GetData(items => items.FirstOrDefault(i => string.Equals(i.Id, id, StringComparison.OrdinalIgnoreCase)));
+        }
+
+        private async Task<List<LocalItem>> GetCachedData()
+        {
+            if (_items == null)
+            {
+                await _cacheFileLock.WaitAsync().ConfigureAwait(false);
+
+                try
+                {
+                    if (_items == null)
+                    {
+                        try
+                        {
+                            _items = _json.DeserializeFromFile<List<LocalItem>>(GetCachePath());
+                        }
+                        catch (FileNotFoundException)
+                        {
+                            _items = new List<LocalItem>();
+                        }
+                        catch (DirectoryNotFoundException)
+                        {
+                            _items = new List<LocalItem>();
+                        }
+                    }
+                }
+                finally
+                {
+                    _cacheFileLock.Release();
+                }
+            }
+
+            return _items.ToList();
+        }
+
+        public async Task<List<string>> GetCachedServerItemIds(SyncTarget target, string serverId)
+        {
+            var items = await GetCachedData().ConfigureAwait(false);
+
+            return items.Where(i => string.Equals(i.ServerId, serverId, StringComparison.OrdinalIgnoreCase))
+                    .Select(i => i.ItemId)
+                    .ToList();
+        }
+
+        public async Task<LocalItem> GetCachedItem(SyncTarget target, string id)
+        {
+            var items = await GetCachedData().ConfigureAwait(false);
+
+            return items.FirstOrDefault(i => string.Equals(i.Id, id, StringComparison.OrdinalIgnoreCase));
+        }
+    }
+}