Synnefo » Pithos MS Client

using System;

using System.Collections.Concurrent;

using System.Collections.Generic;

using System.ComponentModel.Composition;

using System.Diagnostics;

using System.Diagnostics.Contracts;

using System.IO;

using System.Linq;

using System.Net.NetworkInformation;

using System.Security.Cryptography;

using System.ServiceModel.Description;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

using Microsoft.WindowsAPICodePack.Net;

using Pithos.Interfaces;

using System.ServiceModel;

namespace Pithos.Core

{

    [Export(typeof(PithosMonitor))]

    public class PithosMonitor:IDisposable

    {

        [Import]

        public IPithosSettings Settings{get;set;}

        [Import]

        public IStatusKeeper StatusKeeper { get; set; }

        [Import]

        public IPithosWorkflow Workflow { get; set; }

        [Import]

        public ICloudClient CloudClient { get; set; }

        [Import]

        public ICloudClient CloudListeningClient { get; set; }

        private ServiceHost _statusService { get; set; }

        private FileSystemWatcher _watcher;

        public bool Pause

        {

            get { return _watcher == null || !_watcher.EnableRaisingEvents; }

            set

            {

                if (_watcher!=null)

                    _watcher.EnableRaisingEvents = !value;

                if (value)

                {

                    StatusKeeper.SetPithosStatus(PithosStatus.SyncPaused);

                }

                else

                {

                    StatusKeeper.SetPithosStatus(PithosStatus.InSynch);

                }

            }

        }

        CancellationTokenSource _cancellationSource;

        readonly BlockingCollection<WorkflowState> _fileEvents = new BlockingCollection<WorkflowState>();

        readonly BlockingCollection<WorkflowState> _uploadEvents = new BlockingCollection<WorkflowState>();

        public void Start()

        {

            if (_cancellationSource != null)

            {

                if (!_cancellationSource.IsCancellationRequested)

                    return;

            }

            _cancellationSource = new CancellationTokenSource();

            string path = Settings.PithosPath;

            var proxyUri = ProxyFromSettings();            

            CloudClient.Proxy = proxyUri;

            IndexLocalFiles(path);

            StartMonitoringFiles(path);

            StartStatusService();

            StartNetwork();

        }

        private Uri ProxyFromSettings()

        {            

            if (Settings.UseManualProxy)

            {

                var proxyUri = new UriBuilder

                                   {

                                       Host = Settings.ProxyServer, 

                                       Port = Settings.ProxyPort

                                   };

                if (Settings.ProxyAuthentication)

                {

                    proxyUri.UserName = Settings.ProxyUsername;

                    proxyUri.Password = Settings.ProxyPassword;

                }

                return proxyUri.Uri;

            }

            return null;

        }

        private void IndexLocalFiles(string path)

        {

            Trace.TraceInformation("[START] Inxed Local");

            try

            {

                var files =

                    from filePath in Directory.EnumerateFiles(path, "*", SearchOption.AllDirectories).AsParallel()

                    select filePath;

                StatusKeeper.StoreUnversionedFiles(files);

                var newFiles= FileState.FindAllByProperty("OverlayStatus", FileOverlayStatus.Unversioned)

                            .Select(state=>state.FilePath);

                foreach (var newFile in newFiles)

                {

                    _uploadEvents.Add(new WorkflowState

                                          {

                                              Path = newFile,

                                              FileName = Path.GetFileName(newFile),

                                              TriggeringChange = WatcherChangeTypes.Created

                                          });

                }

            }

            catch (Exception exc)

            {

                Trace.TraceError("[ERROR] Index Local - {0}", exc);

            }

            finally

            {

                Trace.TraceInformation("[END] Inxed Local");

            }

        }

        private void StartStatusService()

        {

            // Create a ServiceHost for the CalculatorService type and provide the base address.

            var baseAddress = new Uri("net.pipe://localhost/pithos");

            _statusService = new ServiceHost(typeof(StatusService), baseAddress);

            var binding = new NetNamedPipeBinding(NetNamedPipeSecurityMode.None);

            _statusService.AddServiceEndpoint(typeof(IStatusService), binding, "net.pipe://localhost/pithos/statuscache");

            _statusService.AddServiceEndpoint(typeof (ISettingsService), binding, "net.pipe://localhost/pithos/settings");

            //// Add a mex endpoint

            var smb = new ServiceMetadataBehavior

                          {

                              HttpGetEnabled = true, 

                              HttpGetUrl = new Uri("http://localhost:30000/pithos/mex")

                          };

            _statusService.Description.Behaviors.Add(smb);

            _statusService.Open();

        }

        private void StopStatusService()

        {

            if (_statusService == null)

                return;

            if (_statusService.State == CommunicationState.Faulted)

                _statusService.Abort();

            else if (_statusService.State != CommunicationState.Closed)

                _statusService.Close();

            _statusService = null;

        }

        private void StartNetwork()

        {

            bool connected = NetworkListManager.IsConnectedToInternet;

            //If we are not connected retry later

            if (!connected)

            {

                Task.Factory.StartNewDelayed(10000, StartNetwork);

                return;

            }

            try

            {

                CloudClient.Authenticate(Settings.UserName, Settings.ApiKey);

                StartListening(Settings.PithosPath);

                StartSending();

            }

            catch (Exception)

            {

                //Faild to authenticate due to network or account error

                //Retry after a while

                Task.Factory.StartNewDelayed(10000, StartNetwork);

            }

        }

        internal enum CloudActionType

        {

            Upload=0,

            Download,

            UploadUnconditional,

            DownloadUnconditional,

            DeleteLocal,

            DeleteCloud

        }

        internal class ListenerAction

        {

            public CloudActionType Action { get; set; }

            public FileInfo LocalFile { get; set; }

            public ObjectInfo CloudFile { get; set; }

            public Lazy<string> LocalHash { get; set; }

            public ListenerAction(CloudActionType action, FileInfo localFile, ObjectInfo cloudFile)

            {

                Action = action;

                LocalFile = localFile;

                CloudFile = cloudFile;

                LocalHash=new Lazy<string>(()=>Signature.CalculateHash(LocalFile.FullName),LazyThreadSafetyMode.ExecutionAndPublication);

            }

        }

        internal class LocalFileComparer:EqualityComparer<ListenerAction>

        {

            public override bool Equals(ListenerAction x, ListenerAction y)

            {

                if (x.Action != y.Action)

                    return false;

                if (x.LocalFile != null && y.LocalFile != null && !x.LocalFile.FullName.Equals(y.LocalFile.FullName))

                    return false;

                if (x.CloudFile != null && y.CloudFile != null && !x.CloudFile.Hash.Equals(y.CloudFile.Hash))

                    return false;

                if (x.CloudFile == null ^ y.CloudFile == null ||

                    x.LocalFile == null ^ y.LocalFile == null)

                    return false;

                return true;

            }

            public override int GetHashCode(ListenerAction obj)

            {

                var hash1 = (obj.LocalFile == null) ? int.MaxValue : obj.LocalFile.FullName.GetHashCode();

                var hash2 = (obj.CloudFile == null) ? int.MaxValue : obj.CloudFile.Hash.GetHashCode();

                var hash3 = obj.Action.GetHashCode();

                return hash1 ^ hash2 & hash3;

            }

        }

        private BlockingCollection<ListenerAction> _listenerActions=new BlockingCollection<ListenerAction>();

        private Timer timer;

        private void StartListening(string accountPath)

        {

            ProcessRemoteFiles(accountPath);

            Task.Factory.StartNew(ProcessListenerActions);

        }

        private Task ProcessRemoteFiles(string accountPath)

        {

            Trace.TraceInformation("[LISTENER] Scheduled");    

            return Task.Factory.StartNewDelayed(10000)

                .ContinueWith(t=>CloudClient.ListObjects("PITHOS"))

                .ContinueWith(task =>

                                  {

                                      Trace.TraceInformation("[LISTENER] Start Processing");

                                      var remoteObjects = task.Result;

/*

                                      if (remoteObjects.Count == 0)

                                          return;

*/

                                      var pithosDir = new DirectoryInfo(accountPath);

                                      var remoteFiles = from info in remoteObjects

                                                    select info.Name;

                                      var onlyLocal = from localFile in pithosDir.EnumerateFiles()

                                                      where !remoteFiles.Contains(localFile.Name) 

                                                      select new ListenerAction(CloudActionType.UploadUnconditional, localFile,null);

                                      var localNames =pithosDir.EnumerateFiles().Select(info => info.Name);

                                      var onlyRemote = from upFile in remoteObjects

                                                       where !localNames.Contains(upFile.Name)

                                                       select new ListenerAction(CloudActionType.DownloadUnconditional,null,upFile);

                                      var commonObjects = from  upFile in remoteObjects

                                                            join  localFile in pithosDir.EnumerateFiles()

                                                                on upFile.Name equals localFile.Name 

                                                            select new ListenerAction(CloudActionType.Download, localFile, upFile);

                                      var uniques =

                                          onlyLocal.Union(onlyRemote).Union(commonObjects)

                                              .Except(_listenerActions,new LocalFileComparer());

                                      _listenerActions.AddFromEnumerable(uniques, false);

                                      Trace.TraceInformation("[LISTENER] End Processing");

                                  }

                ).ContinueWith(t=>

                {

                    if (t.IsFaulted)

                    {

                        Trace.TraceError("[LISTENER] Exception: {0}",t.Exception);                                           

                    }

                    else

                    {

                        Trace.TraceInformation("[LISTENER] Finished");                                           

                    }                    

                    ProcessRemoteFiles(accountPath);

                });

        }

        private void ProcessListenerActions()

        {

            foreach(var action in _listenerActions.GetConsumingEnumerable())

            {

                Trace.TraceInformation("[ACTION] Start Processing {0}:{1}->{2}",action.Action,action.LocalFile,action.CloudFile);

                var localFile = action.LocalFile;

                var cloudFile = action.CloudFile;

                var downloadPath = (cloudFile == null)? String.Empty

                                        : Path.Combine(Settings.PithosPath,cloudFile.Name);

                try

                {

                    switch (action.Action)

                    {

                        case CloudActionType.UploadUnconditional:

                            UploadCloudFile(localFile.Name,localFile.Length,localFile.FullName,action.LocalHash.Value);

                            break;

                        case CloudActionType.DownloadUnconditional:

                            DownloadCloudFile("PITHOS",cloudFile.Name,downloadPath);

                            break;

                        case CloudActionType.Download:

                            if (File.Exists(downloadPath))

                            {

                                if (cloudFile.Hash !=action.LocalHash.Value)

                                {

                                    var lastLocalTime =localFile.LastWriteTime;

                                    var lastUpTime =cloudFile.Last_Modified;

                                    if (lastUpTime <=lastLocalTime)

                                    {

                                        //Files in conflict

                                        StatusKeeper.SetFileOverlayStatus(downloadPath,FileOverlayStatus.Conflict);

                                    }

                                    else

                                        DownloadCloudFile("PITHOS",action.CloudFile.Name,downloadPath);

                                }

                            }

                            else

                                DownloadCloudFile("PITHOS",action.CloudFile.Name,downloadPath);

                            break;

                    }

                    Trace.TraceInformation("[ACTION] End Processing {0}:{1}->{2}", action.Action, action.LocalFile, action.CloudFile);

                }

                catch (Exception exc)

                {

                    Trace.TraceError("[REQUEUE] {0} : {1} -> {2} due to exception\r\n{3}",

                                    action.Action, action.LocalFile,action.CloudFile,exc);

                    Trace.TraceError(exc.ToString());

                    _listenerActions.Add(action);

                }

            }

        }

        private void DownloadCloudFile(string container, string fileName, string localPath)

        {

            using (var upstream = CloudClient.GetObject(container, fileName))

            using (var fileStream = File.OpenWrite(localPath))

            {

                upstream.CopyTo(fileStream);

            }

        }

        private void StartMonitoringFiles(string path)

        {

            _watcher = new FileSystemWatcher(path);

            _watcher.Changed += OnFileEvent;

            _watcher.Created += OnFileEvent;

            _watcher.Deleted += OnFileEvent;

            _watcher.Renamed += OnRenameEvent;

            _watcher.EnableRaisingEvents = true;

            Task.Factory.StartNew(() =>

                                      {

                                          foreach (var state in _fileEvents.GetConsumingEnumerable())

                                          {

                                              try

                                              {

                                                  UpdateFileStatus(state);

                                                  UpdateOverlayStatus(state);

                                                  UpdateFileChecksum(state);

                                                  _uploadEvents.Add(state);                                                  

                                              }

                                              catch (OperationCanceledException)

                                              {

                                                  throw;

                                              }

                                              catch(Exception ex)

                                              {}

                                          }

                                      },_cancellationSource.Token);

        }

        private void StartSending()

        {

            Task.Factory.StartNew(() =>

                                      {

                                          foreach (var state in _uploadEvents.GetConsumingEnumerable())

                                          {

                                              try

                                              {

                                                  SynchToCloud(state);

                                              }

                                              catch (OperationCanceledException)

                                              {

                                                  throw;

                                              }

                                              catch(Exception ex)

                                              {}

                                          }

                                      },_cancellationSource.Token);

        }

        private WorkflowState SynchToCloud(WorkflowState state)

        {

            if (state.Skip)

                return state;

            string path = state.Path;

            string fileName = Path.GetFileName(path);

            switch(state.Status)

            {

                case FileStatus.Created:

                case FileStatus.Modified:

                    var info = new FileInfo(path);

                    long fileSize = info.Length;

                    UploadCloudFile(fileName, fileSize, path,state.Hash);

                    break;

                case FileStatus.Deleted:

                    DeleteCloudFile(fileName);

                    break;

                case FileStatus.Renamed:

                    RenameCloudFile(state);

                    break;

            }

            return state;

        }

        private void RenameCloudFile(WorkflowState state)

        {

            this.StatusKeeper.SetFileOverlayStatus(state.Path, FileOverlayStatus.Synch);

            CloudClient.MoveObject("PITHOS", state.OldFileName, state.FileName);

            this.StatusKeeper.SetFileStatus(state.Path, FileStatus.Unchanged);

            this.StatusKeeper.SetFileOverlayStatus(state.Path, FileOverlayStatus.Normal);

            Workflow.RaiseChangeNotification(state.Path);

        }

        private void DeleteCloudFile(string fileName)

        {

            Contract.Requires(!Path.IsPathRooted(fileName));

            this.StatusKeeper.SetFileOverlayStatus(fileName, FileOverlayStatus.Synch);

            CloudClient.DeleteObject("PITHOS", fileName);

            this.StatusKeeper.ClearFileStatus(fileName);

            this.StatusKeeper.RemoveFileOverlayStatus(fileName);            

        }

        private void UploadCloudFile(string fileName, long fileSize, string path,string hash)

        {

            Contract.Requires(!Path.IsPathRooted(fileName));

            //Even if GetObjectInfo times out, we can proceed with the upload

            var info=CloudClient.GetObjectInfo("PITHOS", fileName);

            if ( hash != info.Hash)

            {

                this.StatusKeeper.SetFileOverlayStatus(path, FileOverlayStatus.Synch);

                CloudClient.PutObject("PITHOS", fileName, path, hash).Wait();

            }

            this.StatusKeeper.SetFileStatus(path,FileStatus.Unchanged);

            this.StatusKeeper.SetFileOverlayStatus(path,FileOverlayStatus.Normal);

            Workflow.RaiseChangeNotification(path);

        }

        private Dictionary<WatcherChangeTypes, FileStatus> _statusDict = new Dictionary<WatcherChangeTypes, FileStatus>

        {

            {WatcherChangeTypes.Created,FileStatus.Created},

            {WatcherChangeTypes.Changed,FileStatus.Modified},

            {WatcherChangeTypes.Deleted,FileStatus.Deleted},

            {WatcherChangeTypes.Renamed,FileStatus.Renamed}

        };

        private WorkflowState UpdateFileStatus(WorkflowState  state)

        {

            string path = state.Path;

            FileStatus status = _statusDict[state.TriggeringChange];

            var oldStatus = Workflow.StatusKeeper.GetFileStatus(path);

            if (status == oldStatus)

            {

                state.Status = status;

                state.Skip = true;

                return state;

            }

            if (state.Status == FileStatus.Renamed)

                Workflow.ClearFileStatus(path);                

            state.Status = Workflow.SetFileStatus(path, status);                

            return state;

        }

        private WorkflowState UpdateOverlayStatus(WorkflowState state)

        {            

            if (state.Skip)

                return state;

            switch (state.Status)

            {

                case FileStatus.Created:

                case FileStatus.Modified:

                    this.StatusKeeper.SetFileOverlayStatus(state.Path, FileOverlayStatus.Modified);

                    break;

                case FileStatus.Deleted:

                    this.StatusKeeper.RemoveFileOverlayStatus(state.Path);

                    break;

                case FileStatus.Renamed:

                    this.StatusKeeper.RemoveFileOverlayStatus(state.OldPath);

                    this.StatusKeeper.SetFileOverlayStatus(state.Path, FileOverlayStatus.Modified);

                    break;

                case FileStatus.Unchanged:

                    this.StatusKeeper.SetFileOverlayStatus(state.Path, FileOverlayStatus.Normal);

                    break;

            }

            if (state.Status==FileStatus.Deleted)

                Workflow.RaiseChangeNotification(Path.GetDirectoryName(state.Path));

            else

                Workflow.RaiseChangeNotification(state.Path);

            return state;

        }

        private WorkflowState UpdateFileChecksum(WorkflowState state)

        {

            if (state.Skip)

                return state;

            if (state.Status == FileStatus.Deleted)

                return state;

            string path = state.Path;

            string hash = Signature.CalculateHash(path);

            StatusKeeper.UpdateFileChecksum(path, hash);

            state.Hash = hash;

            return state;

        }

        private FileSystemEventArgs CalculateSignature(FileSystemEventArgs arg)

        {

            Debug.WriteLine(String.Format("{0} {1} {2}", arg.ChangeType, arg.Name, arg.FullPath), "INFO");

            return arg;

        }

        void OnFileEvent(object sender, FileSystemEventArgs e)

        {

            _fileEvents.Add(new WorkflowState{Path=e.FullPath,FileName = e.Name,TriggeringChange=e.ChangeType});            

        }

        void OnRenameEvent(object sender, RenamedEventArgs e)

        {

            _fileEvents.Add(new WorkflowState { OldPath=e.OldFullPath,OldFileName=e.OldName,

                Path = e.FullPath, FileName = e.Name, TriggeringChange = e.ChangeType });

        }

        public void Stop()

        {

            if (_watcher != null)

            {

                _watcher.Changed -= OnFileEvent;

                _watcher.Created -= OnFileEvent;

                _watcher.Deleted -= OnFileEvent;

                _watcher.Renamed -= OnRenameEvent;

                _watcher.Dispose();

            }

            _watcher = null;

            _fileEvents.CompleteAdding();

            if (timer != null)

                timer.Dispose();

            timer = null;

            StopStatusService();

        }

        ~PithosMonitor()

        {

            Dispose(false);

        }

        public void Dispose()

        {

            Dispose(true);

            GC.SuppressFinalize(this);

        }

        protected virtual void Dispose(bool disposing)

        {

            if (disposing)

            {

                Stop();

            }

        }

    }

}