//******************************************************************************************************//  FileTracker.cs - Gbtc////  Copyright © 2014, Grid Protection Alliance.  All Rights Reserved.////  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See//  the NOTICE file distributed with this work for additional information regarding copyright ownership.//  The GPA licenses this file to you under the MIT License (MIT), the "License"; you may//  not use this file except in compliance with the License. You may obtain a copy of the License at:////      http://www.opensource.org/licenses/MIT////  Unless agreed to in writing, the subject software distributed under the License is distributed on an//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the//  License for the specific language governing permissions and limitations.////  Code Modification History://  ----------------------------------------------------------------------------------------------------//  04/29/2014 - Stephen C. Wills//       Generated original version of source code.////****************************************************************************************************** using System;using System.Collections.Generic;using System.IO;using System.Linq;using System.Threading;using System.Threading.Tasks;using System.Timers;using GSF.Threading;using CancellationToken = GSF.Threading.CancellationToken;using Timer = System.Timers.Timer; namespace GSF.IO{    /// <summary>    /// Defines strategies for enumerating    /// files in the watch directories.    /// </summary>    public enum FileEnumerationStrategy    {        /// <summary>        /// Enumerate all files sequentially.        /// </summary>        Sequential,         /// <summary>        /// Enumerates the watch directories in parallel,        /// but subdirectories are processed sequentially.        /// </summary>        ParallelWatchDirectories,         /// <summary>        /// Enumerates every directory, including subdirectories, in parallel.        /// </summary>        ParallelSubdirectories,         /// <summary>        /// Does not enumerate directories, relying only on        /// the file watcher to handle file processing events.        /// </summary>        None    }     /// <summary>    /// Arguments to events triggered by the <see cref="FileProcessor"/>.    /// </summary>    public class FileProcessorEventArgs : EventArgs    {        #region [ Members ]         // Fields        private readonly string m_fullPath;        private readonly bool m_raisedByFileWatcher;        private readonly Func<int> m_retryCounter;        private bool m_requeue;         #endregion         #region [ Constructors ]         /// <summary>        /// Creates a new instance of the <see cref="FileProcessorEventArgs"/> class.        /// </summary>        /// <param name="fullPath">The full path to the file to be processed.</param>        /// <param name="raisedByFileWatcher">Flag indicating whether this event was raised by the file watcher.</param>        /// <param name="retryCounter">The function that provides the value for <see cref="RetryCount"/>.</param>        public FileProcessorEventArgs(string fullPath, bool raisedByFileWatcher, Func<int> retryCounter)        {            m_fullPath = fullPath;            m_raisedByFileWatcher = raisedByFileWatcher;            m_retryCounter = retryCounter;        }         #endregion         #region [ Properties ]         /// <summary>        /// Gets the full path to the file to be processed.        /// </summary>        public string FullPath        {            get            {                return m_fullPath;            }        }         /// <summary>        /// Gets the flag that indicates whether this event was raised by the file watcher.        /// </summary>        public bool RaisedByFileWatcher        {            get            {                return m_raisedByFileWatcher;            }        }         /// <summary>        /// Gets the number of file processing attempts.        /// </summary>        public int RetryCount        {            get            {                return m_retryCounter();            }        }         /// <summary>        /// Gets or sets a flag that indicates whether the        /// file should be requeued by the file processor.        /// </summary>        public bool Requeue        {            get            {                return m_requeue;            }            set            {                m_requeue = value;            }        }         #endregion    }     /// <summary>    /// Tracks files processed in a list of directories, and    /// notifies when new files become available to be processed.    /// </summary>    public sealed class FileProcessor : IDisposable    {        #region [ Members ]         // Nested Types        private sealed class TrackedDirectory : IDisposable        {            #region [ Members ]             // Events            private event EventHandler<EventArgs<List<string>>> ActivelyVisitedPathsRequested;             // Fields            private readonly FileProcessor m_fileProcessor;            private SafeFileWatcher m_fileWatcher;            private CancellationToken m_cancellationToken;            private bool m_disposed;             #endregion             #region [ Constructors ]             public TrackedDirectory(FileProcessor fileProcessor, string path)            {                m_fileProcessor = fileProcessor;                Path = path;                CreateFileWatcher();            }             #endregion             #region [ Properties ]             public string Path { get; }             public List<string> ActivelyEnumeratedPaths            {                get                {                    List<string> activelyVisitedPaths = new List<string>();                    EventArgs<List<string>> args = new EventArgs<List<string>>(activelyVisitedPaths);                    ActivelyVisitedPathsRequested?.Invoke(this, args);                    return activelyVisitedPaths;                }            }             #endregion             #region [ Methods ]             public async Task EnumerateAsync(FileEnumerationStrategy fileEnumerationStrategy)            {                if (m_disposed)                    return;                 CancellationToken cancellationToken = new CancellationToken();                Interlocked.Exchange(ref m_cancellationToken, cancellationToken);                DirectoryInfo directory = new DirectoryInfo(Path);                await EnumerateDirectoryAsync(directory, fileEnumerationStrategy, cancellationToken);            }             public void CancelEnumeration()            {                CancellationToken cancellationToken = Interlocked.CompareExchange(ref m_cancellationToken, null, null);                cancellationToken?.Cancel();            }             public void CheckFileWatcher()            {                if (m_disposed)                    return;                 if (!m_fileWatcher.EnableRaisingEvents)                {                    DisposeFileWatcher();                    CreateFileWatcher();                }            }             public void Dispose()            {                if (m_disposed)                    return;                 try                {                    CancelEnumeration();                    DisposeFileWatcher();                }                finally                {                    m_disposed = true;                }            }             private async Task EnumerateDirectoryAsync(DirectoryInfo directory, FileEnumerationStrategy fileEnumerationStrategy, CancellationToken cancellationToken)            {                bool parallelSubdirectories =                    fileEnumerationStrategy == FileEnumerationStrategy.ParallelSubdirectories;                 LogicalThread thread = LogicalThread.CurrentThread;                 if (parallelSubdirectories)                {                    LogicalThreadScheduler scheduler = m_fileProcessor.m_threadScheduler;                    thread = scheduler.CreateThread();                    await thread.Join();                }                 async Task ForEach(IEnumerable<Task> tasks)                {                    foreach (Task task in tasks)                    {                        if (cancellationToken.IsCancelled)                            return;                         await task;                        await thread.Yield();                    }                }                 LogicalThread processingThread = m_fileProcessor.m_processingThread;                string activePath = null;                 async Task VisitSubdirectoryAsync(DirectoryInfo subdirectory)                {                    activePath = subdirectory.FullName;                     if (cancellationToken.IsCancelled)                        return;                     if (m_fileProcessor.MatchesFolderExclusion(subdirectory.FullName))                        return;                     await EnumerateDirectoryAsync(subdirectory, fileEnumerationStrategy, cancellationToken);                }                 async Task<Task> VisitFileAsync(FileInfo file)                {                    activePath = file.FullName;                     if (cancellationToken.IsCancelled)                        return Task.CompletedTask;                     async Task InvokeOnProcessingThread(Action action)                    {                        await processingThread.Join();                        action();                    }                     DateTime lastWriteTime = file.LastWriteTimeUtc;                    void Process() => m_fileProcessor.TouchAndProcess(file.FullName, lastWriteTime, false);                    void Skip() => m_fileProcessor.TouchAndSkip(file.FullName, lastWriteTime);                     if (!m_fileProcessor.MatchesFilter(file.FullName))                        return InvokeOnProcessingThread(Skip);                     Task processTask = InvokeOnProcessingThread(Process);                    await thread.Yield();                    return processTask;                }                 async Task EnumerateSubdirectoriesAsync()                {                    Func<IEnumerable<Task>, Task> whenAll =                        parallelSubdirectories ? Task.WhenAll : ForEach;                     IEnumerable<Task> subdirectoryTasks = FilePath                        .EnumerateDirectories(directory, "*", SearchOption.TopDirectoryOnly, m_fileProcessor.OnError)                        .Select(VisitSubdirectoryAsync);                     await whenAll(subdirectoryTasks);                }                 async Task EnumerateFilesAsync()                {                    IEnumerable<Task<Task>> fileTasks = FilePath                        .EnumerateFiles(directory, "*", SearchOption.TopDirectoryOnly)                        .Select(VisitFileAsync);                     List<Task> processTasks = new List<Task>();                     foreach (Task<Task> fileTask in fileTasks)                        processTasks.Add(await fileTask);                     await Task.WhenAll(processTasks);                }                 EventHandler<EventArgs<List<string>>> handler = (sender, args) =>                {                    if (!(activePath is null))                        args.Argument.Add(activePath);                };                 try                {                    ActivelyVisitedPathsRequested += handler;                     if (parallelSubdirectories)                    {                        Task subdirectoriesTask = EnumerateSubdirectoriesAsync();                        await EnumerateFilesAsync();                        ActivelyVisitedPathsRequested -= handler;                        await subdirectoriesTask;                    }                    else                    {                        await EnumerateFilesAsync();                        await EnumerateSubdirectoriesAsync();                    }                }                finally                {                    ActivelyVisitedPathsRequested -= handler;                }            }             private void CreateFileWatcher()            {                m_fileWatcher = new SafeFileWatcher(Path);                m_fileWatcher.InternalBufferSize = m_fileProcessor.InternalBufferSize;                m_fileWatcher.IncludeSubdirectories = true;                 m_fileWatcher.Created += m_fileProcessor.Watcher_Created;                m_fileWatcher.Changed += m_fileProcessor.Watcher_Changed;                m_fileWatcher.Renamed += m_fileProcessor.Watcher_Renamed;                m_fileWatcher.Deleted += m_fileProcessor.Watcher_Deleted;                m_fileWatcher.Error += m_fileProcessor.Watcher_Error;                 m_fileWatcher.EnableRaisingEvents = true;            }             private void DisposeFileWatcher()            {                m_fileWatcher.Created -= m_fileProcessor.Watcher_Created;                m_fileWatcher.Changed -= m_fileProcessor.Watcher_Changed;                m_fileWatcher.Renamed -= m_fileProcessor.Watcher_Renamed;                m_fileWatcher.Deleted -= m_fileProcessor.Watcher_Deleted;                m_fileWatcher.Error -= m_fileProcessor.Watcher_Error;                m_fileWatcher.Dispose();            }             #endregion        }         // Constants         /// <summary>        /// Default value for the <see cref="Filter"/> property.        /// </summary>        public const string DefaultFilter = @"**\*";         /// <summary>        /// Default value for the <see cref="FolderExclusion"/> property.        /// </summary>        public const string DefaultFolderExclusion = "";         /// <summary>        /// Default value for the <see cref="TrackChanges"/> property.        /// </summary>        public const bool DefaultTrackChanges = false;         /// <summary>        /// Default value for the <see cref="InternalBufferSize"/> property.        /// </summary>        public const int DefaultInternalBufferSize = 8192;         /// <summary>        /// Default value for the <see cref="EnumerationStrategy"/> property.        /// </summary>        public const FileEnumerationStrategy DefaultEnumerationStrategy = FileEnumerationStrategy.ParallelSubdirectories;         // Events         /// <summary>        /// Event triggered when a file is to be processed.        /// </summary>        public event EventHandler<FileProcessorEventArgs> Processing;         /// <summary>        /// Event triggered when an unexpected error occurs during        /// normal operation of the <see cref="FileProcessor"/>.        /// </summary>        public event EventHandler<ErrorEventArgs> Error;         // Fields        private string m_filter;        private string m_folderExclusion;        private bool m_trackChanges;        private int m_internalBufferSize;        private FileEnumerationStrategy m_enumerationStrategy;        private bool m_orderedEnumeration;         private readonly object m_trackedDirectoriesLock;        private readonly List<TrackedDirectory> m_trackedDirectories;        private readonly TaskSynchronizedOperation m_enumerationOperation;         private readonly LogicalThreadScheduler m_threadScheduler;        private readonly LogicalThread m_processingThread;        private readonly LogicalThread m_watcherThread;        private readonly LogicalThread m_sequentialEnumerationThread;        private readonly Timer m_fileWatchTimer;        private readonly ManagedCancellationTokenSource m_requeueTokenSource;         private readonly Dictionary<string, DateTime> m_touchedFiles;         private int m_processedFileCount;        private int m_skippedFileCount;        private int m_requeuedFileCount;         private bool m_disposed;         #endregion         #region [ Constructors ]         /// <summary>        /// Creates a new instance of the <see cref="FileProcessor"/> class.        /// </summary>        public FileProcessor()        {            m_filter = DefaultFilter;            m_folderExclusion = DefaultFolderExclusion;            m_trackChanges = DefaultTrackChanges;            m_internalBufferSize = DefaultInternalBufferSize;            m_enumerationStrategy = DefaultEnumerationStrategy;             m_trackedDirectoriesLock = new object();            m_trackedDirectories = new List<TrackedDirectory>();            m_enumerationOperation = new TaskSynchronizedOperation(EnumerateWatchDirectoriesAsync, OnError);             m_threadScheduler = new LogicalThreadScheduler(2);            m_threadScheduler.UnhandledException += (sender, args) => OnError(args.Argument);            m_processingThread = m_threadScheduler.CreateThread();            m_watcherThread = m_threadScheduler.CreateThread();            m_sequentialEnumerationThread = m_threadScheduler.CreateThread();            m_fileWatchTimer = new Timer(15000);            m_fileWatchTimer.Elapsed += FileWatchTimer_Elapsed;            m_requeueTokenSource = new ManagedCancellationTokenSource();             m_touchedFiles = new Dictionary<string, DateTime>(StringComparer.OrdinalIgnoreCase);        }         #endregion 

Lines 1–500 of 1142. Load more