From 9911df11e8a96d5d7fffcd8618b63a6adb27701f Mon Sep 17 00:00:00 2001 From: LukePulverenti Date: Fri, 8 Mar 2013 00:08:27 -0500 Subject: extracted provider manager. took more off the kernel --- .../IO/DirectoryWatchers.cs | 543 +++++++++++++++++++++ 1 file changed, 543 insertions(+) create mode 100644 MediaBrowser.Server.Implementations/IO/DirectoryWatchers.cs (limited to 'MediaBrowser.Server.Implementations/IO') diff --git a/MediaBrowser.Server.Implementations/IO/DirectoryWatchers.cs b/MediaBrowser.Server.Implementations/IO/DirectoryWatchers.cs new file mode 100644 index 000000000..c1371a32b --- /dev/null +++ b/MediaBrowser.Server.Implementations/IO/DirectoryWatchers.cs @@ -0,0 +1,543 @@ +using MediaBrowser.Common.ScheduledTasks; +using MediaBrowser.Controller.Configuration; +using MediaBrowser.Controller.Entities; +using MediaBrowser.Controller.IO; +using MediaBrowser.Controller.Library; +using MediaBrowser.Controller.ScheduledTasks; +using MediaBrowser.Model.Logging; +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace MediaBrowser.Server.Implementations.IO +{ + /// + /// Class DirectoryWatchers + /// + public class DirectoryWatchers : IDirectoryWatchers + { + /// + /// The file system watchers + /// + private ConcurrentBag _fileSystemWatchers = new ConcurrentBag(); + /// + /// The update timer + /// + private Timer _updateTimer; + /// + /// The affected paths + /// + private readonly ConcurrentDictionary _affectedPaths = new ConcurrentDictionary(); + + /// + /// A dynamic list of paths that should be ignored. Added to during our own file sytem modifications. + /// + private readonly ConcurrentDictionary _tempIgnoredPaths = new ConcurrentDictionary(StringComparer.OrdinalIgnoreCase); + + /// + /// The timer lock + /// + private readonly object _timerLock = new object(); + + /// + /// Add the path to our temporary ignore list. Use when writing to a path within our listening scope. + /// + /// The path. + public void TemporarilyIgnore(string path) + { + _tempIgnoredPaths[path] = path; + } + + /// + /// Removes the temp ignore. + /// + /// The path. + public void RemoveTempIgnore(string path) + { + string val; + _tempIgnoredPaths.TryRemove(path, out val); + } + + /// + /// Gets or sets the logger. + /// + /// The logger. + private ILogger Logger { get; set; } + + /// + /// Gets or sets the task manager. + /// + /// The task manager. + private ITaskManager TaskManager { get; set; } + + private ILibraryManager LibraryManager { get; set; } + private IServerConfigurationManager ConfigurationManager { get; set; } + + /// + /// Initializes a new instance of the class. + /// + public DirectoryWatchers(ILogManager logManager, ITaskManager taskManager, ILibraryManager libraryManager, IServerConfigurationManager configurationManager) + { + if (taskManager == null) + { + throw new ArgumentNullException("taskManager"); + } + + LibraryManager = libraryManager; + TaskManager = taskManager; + Logger = logManager.GetLogger("DirectoryWatchers"); + ConfigurationManager = configurationManager; + } + + /// + /// Starts this instance. + /// + public void Start() + { + LibraryManager.LibraryChanged += Instance_LibraryChanged; + + var pathsToWatch = new List { LibraryManager.RootFolder.Path }; + + var paths = LibraryManager.RootFolder.Children.OfType() + .SelectMany(f => + { + try + { + // Accessing ResolveArgs could involve file system access + return f.ResolveArgs.PhysicalLocations; + } + catch (IOException) + { + return new string[] {}; + } + + }) + .Where(Path.IsPathRooted); + + foreach (var path in paths) + { + if (!ContainsParentFolder(pathsToWatch, path)) + { + pathsToWatch.Add(path); + } + } + + foreach (var path in pathsToWatch) + { + StartWatchingPath(path); + } + } + + /// + /// Examine a list of strings assumed to be file paths to see if it contains a parent of + /// the provided path. + /// + /// The LST. + /// The path. + /// true if [contains parent folder] [the specified LST]; otherwise, false. + /// path + private static bool ContainsParentFolder(IEnumerable lst, string path) + { + if (string.IsNullOrEmpty(path)) + { + throw new ArgumentNullException("path"); + } + + path = path.TrimEnd(Path.DirectorySeparatorChar); + + return lst.Any(str => + { + //this should be a little quicker than examining each actual parent folder... + var compare = str.TrimEnd(Path.DirectorySeparatorChar); + + return (path.Equals(compare, StringComparison.OrdinalIgnoreCase) || (path.StartsWith(compare, StringComparison.OrdinalIgnoreCase) && path[compare.Length] == Path.DirectorySeparatorChar)); + }); + } + + /// + /// Starts the watching path. + /// + /// The path. + private void StartWatchingPath(string path) + { + // Creating a FileSystemWatcher over the LAN can take hundreds of milliseconds, so wrap it in a Task to do them all in parallel + Task.Run(() => + { + var newWatcher = new FileSystemWatcher(path, "*") { IncludeSubdirectories = true, InternalBufferSize = 32767 }; + + newWatcher.Created += watcher_Changed; + newWatcher.Deleted += watcher_Changed; + newWatcher.Renamed += watcher_Changed; + newWatcher.Changed += watcher_Changed; + + newWatcher.Error += watcher_Error; + + try + { + newWatcher.EnableRaisingEvents = true; + _fileSystemWatchers.Add(newWatcher); + + Logger.Info("Watching directory " + path); + } + catch (IOException ex) + { + Logger.ErrorException("Error watching path: {0}", ex, path); + } + catch (PlatformNotSupportedException ex) + { + Logger.ErrorException("Error watching path: {0}", ex, path); + } + }); + } + + /// + /// Stops the watching path. + /// + /// The path. + private void StopWatchingPath(string path) + { + var watcher = _fileSystemWatchers.FirstOrDefault(f => f.Path.Equals(path, StringComparison.OrdinalIgnoreCase)); + + if (watcher != null) + { + DisposeWatcher(watcher); + } + } + + /// + /// Disposes the watcher. + /// + /// The watcher. + private void DisposeWatcher(FileSystemWatcher watcher) + { + Logger.Info("Stopping directory watching for path {0}", watcher.Path); + + watcher.EnableRaisingEvents = false; + watcher.Dispose(); + + var watchers = _fileSystemWatchers.ToList(); + + watchers.Remove(watcher); + + _fileSystemWatchers = new ConcurrentBag(watchers); + } + + /// + /// Handles the LibraryChanged event of the Kernel + /// + /// The source of the event. + /// The instance containing the event data. + void Instance_LibraryChanged(object sender, ChildrenChangedEventArgs e) + { + if (e.Folder is AggregateFolder && e.HasAddOrRemoveChange) + { + if (e.ItemsRemoved != null) + { + foreach (var item in e.ItemsRemoved.OfType()) + { + StopWatchingPath(item.Path); + } + } + if (e.ItemsAdded != null) + { + foreach (var item in e.ItemsAdded.OfType()) + { + StartWatchingPath(item.Path); + } + } + } + } + + /// + /// Handles the Error event of the watcher control. + /// + /// The source of the event. + /// The instance containing the event data. + async void watcher_Error(object sender, ErrorEventArgs e) + { + var ex = e.GetException(); + var dw = (FileSystemWatcher) sender; + + Logger.ErrorException("Error in Directory watcher for: "+dw.Path, ex); + + if (ex.Message.Contains("network name is no longer available")) + { + //Network either dropped or, we are coming out of sleep and it hasn't reconnected yet - wait and retry + Logger.Warn("Network connection lost - will retry..."); + var retries = 0; + var success = false; + while (!success && retries < 10) + { + await Task.Delay(500).ConfigureAwait(false); + + try + { + dw.EnableRaisingEvents = false; + dw.EnableRaisingEvents = true; + success = true; + } + catch (IOException) + { + Logger.Warn("Network still unavailable..."); + retries++; + } + } + if (!success) + { + Logger.Warn("Unable to access network. Giving up."); + DisposeWatcher(dw); + } + + } + else + { + if (!ex.Message.Contains("BIOS command limit")) + { + Logger.Info("Attempting to re-start watcher."); + + dw.EnableRaisingEvents = false; + dw.EnableRaisingEvents = true; + } + + } + } + + /// + /// Handles the Changed event of the watcher control. + /// + /// The source of the event. + /// The instance containing the event data. + void watcher_Changed(object sender, FileSystemEventArgs e) + { + if (e.ChangeType == WatcherChangeTypes.Created && e.Name == "New folder") + { + return; + } + if (_tempIgnoredPaths.ContainsKey(e.FullPath)) + { + Logger.Info("Watcher requested to ignore change to " + e.FullPath); + return; + } + + Logger.Info("Watcher sees change of type " + e.ChangeType.ToString() + " to " + e.FullPath); + + //Since we're watching created, deleted and renamed we always want the parent of the item to be the affected path + var affectedPath = e.FullPath; + + _affectedPaths.AddOrUpdate(affectedPath, affectedPath, (key, oldValue) => affectedPath); + + lock (_timerLock) + { + if (_updateTimer == null) + { + _updateTimer = new Timer(TimerStopped, null, TimeSpan.FromSeconds(ConfigurationManager.Configuration.FileWatcherDelay), TimeSpan.FromMilliseconds(-1)); + } + else + { + _updateTimer.Change(TimeSpan.FromSeconds(ConfigurationManager.Configuration.FileWatcherDelay), TimeSpan.FromMilliseconds(-1)); + } + } + } + + /// + /// Timers the stopped. + /// + /// The state info. + private async void TimerStopped(object stateInfo) + { + lock (_timerLock) + { + // Extend the timer as long as any of the paths are still being written to. + if (_affectedPaths.Any(p => IsFileLocked(p.Key))) + { + Logger.Info("Timer extended."); + _updateTimer.Change(TimeSpan.FromSeconds(ConfigurationManager.Configuration.FileWatcherDelay), TimeSpan.FromMilliseconds(-1)); + return; + } + + Logger.Info("Timer stopped."); + + _updateTimer.Dispose(); + _updateTimer = null; + } + + var paths = _affectedPaths.Keys.ToList(); + _affectedPaths.Clear(); + + await ProcessPathChanges(paths).ConfigureAwait(false); + } + + /// + /// Try and determine if a file is locked + /// This is not perfect, and is subject to race conditions, so I'd rather not make this a re-usable library method. + /// + /// The path. + /// true if [is file locked] [the specified path]; otherwise, false. + private bool IsFileLocked(string path) + { + try + { + var data = FileSystem.GetFileData(path); + + if (!data.HasValue || data.Value.IsDirectory) + { + return false; + } + } + catch (IOException) + { + return false; + } + + FileStream stream = null; + + try + { + stream = new FileStream(path, FileMode.Open, FileAccess.ReadWrite, FileShare.ReadWrite); + } + catch + { + //the file is unavailable because it is: + //still being written to + //or being processed by another thread + //or does not exist (has already been processed) + return true; + } + finally + { + if (stream != null) + stream.Close(); + } + + //file is not locked + return false; + } + + /// + /// Processes the path changes. + /// + /// The paths. + /// Task. + private async Task ProcessPathChanges(List paths) + { + var itemsToRefresh = paths.Select(Path.GetDirectoryName) + .Select(GetAffectedBaseItem) + .Where(item => item != null) + .Distinct() + .ToList(); + + foreach (var p in paths) Logger.Info(p + " reports change."); + + // If the root folder changed, run the library task so the user can see it + if (itemsToRefresh.Any(i => i is AggregateFolder)) + { + TaskManager.CancelIfRunningAndQueue(); + return; + } + + await Task.WhenAll(itemsToRefresh.Select(i => Task.Run(async () => + { + Logger.Info(i.Name + " (" + i.Path + ") will be refreshed."); + + try + { + await i.ChangedExternally().ConfigureAwait(false); + } + catch (IOException ex) + { + // For now swallow and log. + // Research item: If an IOException occurs, the item may be in a disconnected state (media unavailable) + // Should we remove it from it's parent? + Logger.ErrorException("Error refreshing {0}", ex, i.Name); + } + + }))).ConfigureAwait(false); + } + + /// + /// Gets the affected base item. + /// + /// The path. + /// BaseItem. + private BaseItem GetAffectedBaseItem(string path) + { + BaseItem item = null; + + while (item == null && !string.IsNullOrEmpty(path)) + { + item = LibraryManager.RootFolder.FindByPath(path); + + path = Path.GetDirectoryName(path); + } + + if (item != null) + { + // If the item has been deleted find the first valid parent that still exists + while (!Directory.Exists(item.Path) && !File.Exists(item.Path)) + { + item = item.Parent; + + if (item == null) + { + break; + } + } + } + + return item; + } + + /// + /// Stops this instance. + /// + public void Stop() + { + LibraryManager.LibraryChanged -= Instance_LibraryChanged; + + FileSystemWatcher watcher; + + while (_fileSystemWatchers.TryTake(out watcher)) + { + watcher.Changed -= watcher_Changed; + watcher.EnableRaisingEvents = false; + watcher.Dispose(); + } + + lock (_timerLock) + { + if (_updateTimer != null) + { + _updateTimer.Dispose(); + _updateTimer = null; + } + } + + _affectedPaths.Clear(); + } + + /// + /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. + /// + public void Dispose() + { + Dispose(true); + GC.SuppressFinalize(this); + } + + /// + /// Releases unmanaged and - optionally - managed resources. + /// + /// true to release both managed and unmanaged resources; false to release only unmanaged resources. + protected virtual void Dispose(bool dispose) + { + if (dispose) + { + Stop(); + } + } + } +} -- cgit v1.2.3