Synnefo » Pithos MS Client

using System;

using System.Collections.Generic;

using System.ComponentModel.Composition;

using System.Diagnostics;

using System.Diagnostics.Contracts;

using System.IO;

using System.Linq;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

using Pithos.Interfaces;

using Pithos.Network;

namespace Pithos.Core.Agents

{

    [Export]

    public class NetworkAgent

    {

        private Agent<CloudAction> _agent;

        [Import]

        public IStatusKeeper StatusKeeper { get; set; }

        public IStatusNotification StatusNotification { get; set; }

        [Import]

        public ICloudClient CloudClient { get; set; }

        [Import]

        public FileAgent FileAgent {get;set;}

        /*

        [Import]

        public IPithosWorkflow Workflow { get; set; }

*/

        public string PithosContainer { get; set; }

        public string TrashContainer { get; private set; }

        public int BlockSize { get; set; }

        public string BlockHash { get; set; }

        public void Start(string pithosContainer, string trashContainer, int blockSize, string blockHash)

        {

            if (String.IsNullOrWhiteSpace(pithosContainer))

                throw new ArgumentNullException("pithosContainer");

            if (String.IsNullOrWhiteSpace(trashContainer))

                throw new ArgumentNullException("trashContainer");

            Contract.EndContractBlock();

            PithosContainer = pithosContainer;

            TrashContainer = trashContainer;

            BlockSize = blockSize;

            BlockHash = blockHash;

            _agent = Agent<CloudAction>.Start(inbox =>

            {

                Action loop = null;

                loop = () =>

                {

                    var message = inbox.Receive();

/*

                    var process=Process(message);

                    inbox.LoopAsync(process, loop);

*/

/*

                    process1.ContinueWith(t =>

                    {

                        inbox.DoAsync(loop);

                        if (t.IsFaulted)

                        {

                            var ex = t.Exception.InnerException;

                            if (ex is OperationCanceledException)

                                inbox.Stop();

                        }

                    });

*/

                    //inbox.DoAsync(loop);

                    var process = message.ContinueWith(t =>

                    {

                        var action = t.Result;

                        Process(action);

                        inbox.DoAsync(loop);

                    });

                    process.ContinueWith(t =>

                    {

                        inbox.DoAsync(loop);

                        if (t.IsFaulted)

                        {

                            var ex = t.Exception.InnerException;

                            if (ex is OperationCanceledException)

                                inbox.Stop();

                        }

                    });

                };

                loop();

            });

        }

        public void Post(CloudAction cloudAction)

        {

            if (cloudAction == null)

                throw new ArgumentNullException("cloudAction");

            Contract.EndContractBlock();

            //If the action targets a local file, add a treehash calculation

            if (cloudAction.LocalFile != null)

            {

                cloudAction.TopHash = new Lazy<string>(() => Signature.CalculateTreeHashAsync(cloudAction.LocalFile, 

                                    BlockSize, BlockHash).Result

                                     .TopHash.ToHashString());

            }

            _agent.Post(cloudAction);

        }

        class ObjectInfoByNameComparer:IEqualityComparer<ObjectInfo>

        {

            public bool Equals(ObjectInfo x, ObjectInfo y)

            {

                return x.Name.Equals(y.Name,StringComparison.InvariantCultureIgnoreCase);

            }

            public int GetHashCode(ObjectInfo obj)

            {

                return obj.Name.ToLower().GetHashCode();

            }

        }

        public Task ProcessRemoteFiles(string accountPath,DateTime? since=null)

        {            

            Trace.CorrelationManager.StartLogicalOperation();

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

            var listObjects = Task.Factory.StartNewDelayed(10000).ContinueWith(t =>

                CloudClient.ListObjects(PithosContainer,since));

            DateTime nextSince = DateTime.Now.AddSeconds(-1);

            var enqueueFiles = listObjects.ContinueWith(task =>

            {

                if (task.IsFaulted)

                {

                    //ListObjects failed at this point, need to reschedule

                    Trace.TraceError("[FAIL] ListObjects in ProcessRemoteFiles with {0}", task.Exception);

                    ProcessRemoteFiles(accountPath, since);

                    return;

                }

                Trace.CorrelationManager.StartLogicalOperation("Listener");

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

                var remoteObjects = task.Result;

                var remote=from info in remoteObjects

                                  let name=info.Name

                                  where !name.EndsWith(".ignore",StringComparison.InvariantCultureIgnoreCase) &&

                                  !name.StartsWith("fragments/",StringComparison.InvariantCultureIgnoreCase)

                                select info;

                var commonObjects = new List<Tuple<ObjectInfo, FileInfo,FileState>>();

                var remoteOnly = new List<ObjectInfo>();

                //In order to avoid multiple iterations over the files, we iterate only once

                //over the remote files

                foreach (var objectInfo in remote)

                {

                    var relativePath= objectInfo.Name.RelativeUrlToFilePath();// fileInfo.AsRelativeUrlTo(FileAgent.RootPath);

                    //and remove any matching objects from the list, adding them to the commonObjects list

                    if (FileAgent.Exists(relativePath))

                    {

                        var localFile = FileAgent.GetFileInfo(relativePath);

                        var state=FileState.FindByFilePath(localFile.FullName);                        

                        commonObjects.Add(Tuple.Create(objectInfo, localFile,state));

                    }

                    else

                        //If there is no match we add them to the localFiles list

                        remoteOnly.Add(objectInfo);

                }

                //At the end of the iteration, the *remote* list will contain the files that exist 

                //only on the server

                //Remote files should be downloaded

                var actionsForRemote = from upFile in remoteOnly

                                       select new CloudAction(CloudActionType.DownloadUnconditional,upFile);

                //Common files should be checked on a per-case basis to detect differences, which is newer

                var actionsForCommon = from pair in commonObjects

                                       let objectInfo = pair.Item1

                                       let localFile = pair.Item2

                                       let state=pair.Item3

                                       select new CloudAction(CloudActionType.MustSynch, 

                                           localFile, objectInfo,state);

                //Collect all the actions

                var allActions = actionsForRemote.Union(actionsForCommon);

                //And remove those that are already being processed by the agent

                var distinctActions =allActions

                        .Except(_agent.GetEnumerable(), new PithosMonitor.LocalFileComparer())

                        .ToList();

                //Queue all the actions

                foreach (var message in distinctActions)

                {

                    Post(message);

                }

                if(remoteOnly.Count>0)

                    StatusNotification.NotifyChange(String.Format("Processing {0} new files", remoteOnly.Count));

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

                Trace.CorrelationManager.StopLogicalOperation();

            });

            var loop = enqueueFiles.ContinueWith(t =>

            {

                if (t.IsFaulted)

                {

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

                }

                else

                {

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

                }

                ProcessRemoteFiles(accountPath, nextSince);

            });

            return loop;

        }

        private Task Process(Task<CloudAction> action)

        {

            return action.ContinueWith(t=> Process(t.Result));

        }

        private void Process(CloudAction action)

        {

            if (action==null)

                throw new ArgumentNullException("action");

            Contract.EndContractBlock();

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

            var localFile = action.LocalFile;

            var cloudFile = action.CloudFile;

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

                                    : Path.Combine(FileAgent.RootPath, cloudFile.Name.RelativeUrlToFilePath());

            try

            {

                switch (action.Action)

                {

                    case CloudActionType.UploadUnconditional:

                        UploadCloudFile(localFile, action.LocalHash.Value,action.TopHash.Value);

                        break;

                    case CloudActionType.DownloadUnconditional:

                        DownloadCloudFile(PithosContainer, new Uri(cloudFile.Name,UriKind.Relative), downloadPath);

                        break;

                    case CloudActionType.DeleteCloud:

                        DeleteCloudFile(cloudFile.Name);

                        break;

                    case CloudActionType.RenameCloud:

                        RenameCloudFile(action.OldFileName, action.NewPath, action.NewFileName);

                        break;

                    case CloudActionType.MustSynch:

                        if (File.Exists(downloadPath))

                        {                            

                            var cloudHash = cloudFile.Hash;

                            var localHash = action.LocalHash.Value;

                            var topHash = action.TopHash.Value;

                            //Not enough to compare only the local hashes, also have to compare the tophashes

                            if (!cloudHash.Equals(localHash, StringComparison.InvariantCultureIgnoreCase) &&

                                !cloudHash.Equals(topHash, StringComparison.InvariantCultureIgnoreCase))

                            {

                                var lastLocalTime = localFile.LastWriteTime;

                                var lastUpTime = cloudFile.Last_Modified;

                                if (lastUpTime <= lastLocalTime)

                                {

                                    //Local change while the app was down or Files in conflict

                                    //Maybe need to store version as well, to check who has the latest version

                                    //StatusKeeper.SetFileOverlayStatus(downloadPath, FileOverlayStatus.Conflict);

                                    UploadCloudFile(localFile, action.LocalHash.Value,action.TopHash.Value);

                                }

                                else

                                {

                                    var status = StatusKeeper.GetFileStatus(downloadPath);

                                    switch (status)

                                    {

                                        case FileStatus.Unchanged:

                                            //It he cloud file has a later date, it was modified by another user or computer.

                                            //If the local file's status is Unchanged, we should go on and download the cloud file

                                            DownloadCloudFile(PithosContainer, new Uri(action.CloudFile.Name,UriKind.Relative), downloadPath);

                                            break;

                                        case FileStatus.Modified:

                                            //If the local file is Modified, we may have a conflict. In this case we should mark the file as Conflict

                                            //We can't ensure that a file modified online since the last time will appear as Modified, unless we 

                                            //index all files before we start listening.

                                            StatusKeeper.SetFileOverlayStatus(downloadPath, FileOverlayStatus.Conflict);

                                            break;

                                        case FileStatus.Created:

                                            //If the local file is Created, it means that the local and cloud files aren't related yet have the same name

                                            //In this case we must mark the file as in conflict

                                            //Other cases should never occur. Mark them as Conflict as well but log a warning

                                            StatusKeeper.SetFileOverlayStatus(downloadPath,FileOverlayStatus.Conflict);

                                            break;

                                        default:

                                            //If the local file is Created, it means that the local and cloud files aren't related yet have the same name

                                            //In this case we must mark the file as in conflict

                                            //Other cases should never occur. Mark them as Conflict as well but log a warning

                                            StatusKeeper.SetFileOverlayStatus(downloadPath,FileOverlayStatus.Conflict);

                                            Trace.TraceWarning("Unexcepted status {0} for file {1}->{2}",status,downloadPath,action.CloudFile.Name);

                                            break;

                                    }

                                }

                            }

                        }

                        else

                            DownloadCloudFile(PithosContainer, new Uri(action.CloudFile.Name,UriKind.Relative), downloadPath);

                        break;

                }

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

            }

            catch (OperationCanceledException)

            {

                throw;

            }

            catch (Exception exc)

            {

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

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

                _agent.Post(action);

            }

        }

        private void RenameCloudFile(string oldFileName, string newPath, string newFileName)

        {

            if (String.IsNullOrWhiteSpace(oldFileName))

                throw new ArgumentNullException("oldFileName");

            if (String.IsNullOrWhiteSpace(oldFileName))

                throw new ArgumentNullException("newPath");

            if (String.IsNullOrWhiteSpace(oldFileName))

                throw new ArgumentNullException("newFileName");

            Contract.EndContractBlock();

            //The local file is already renamed

            this.StatusKeeper.SetFileOverlayStatus(newPath, FileOverlayStatus.Modified);

            CloudClient.MoveObject(PithosContainer, oldFileName, PithosContainer, newFileName);

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

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

            NativeMethods.RaiseChangeNotification(newPath);

        }

        private void DeleteCloudFile(string fileName)

        {            

            if (String.IsNullOrWhiteSpace(fileName))

                throw new ArgumentNullException("fileName");

            if (Path.IsPathRooted(fileName))

                throw new ArgumentException("The fileName should not be rooted","fileName");

            Contract.EndContractBlock();

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

            CloudClient.MoveObject(PithosContainer, fileName, TrashContainer, fileName);

            this.StatusKeeper.ClearFileStatus(fileName);

            this.StatusKeeper.RemoveFileOverlayStatus(fileName);

        }

        //Download a file.

        private void DownloadCloudFile(string container, Uri relativeUrl, string localPath)

        {

            if (String.IsNullOrWhiteSpace(container))

                throw new ArgumentNullException("container");

            if (relativeUrl==null)

                throw new ArgumentNullException("relativeUrl");

            if (String.IsNullOrWhiteSpace(localPath))

                throw new ArgumentNullException("localPath");

            if (!Path.IsPathRooted(localPath))

                throw new ArgumentException("The localPath must be rooted", "localPath");

            Contract.EndContractBlock();

            var url = relativeUrl.ToString();

            if (url.EndsWith(".ignore",StringComparison.InvariantCultureIgnoreCase))

                return;

            //Are we already downloading or uploading the file? 

            using (var gate=NetworkGate.Acquire(localPath, NetworkOperation.Downloading))

            {

                if (gate.Failed)

                    return;

                //The file's hashmap will be stored in the same location with the extension .hashmap

                //var hashPath = Path.Combine(FileAgent.FragmentsPath, relativePath + ".hashmap");

                //Retrieve the hashmap from the server

                var getHashMap = CloudClient.GetHashMap(container,url);                

                var downloadTask= getHashMap.ContinueWith(t =>

                {

                    var serverHash=t.Result;

                    //If it's a small file

                    return serverHash.Hashes.Count == 1 

                        //Download it in one go

                        ? DownloadEntireFile(container, relativeUrl, localPath) 

                        //Otherwise download it block by block

                        : DownloadWithBlocks(container, relativeUrl, localPath, serverHash);

                });

                //Retrieve the object's metadata

                var getInfo = downloadTask.ContinueWith(t =>

                            CloudClient.GetObjectInfo(container, url));

                //And store it

                var storeInfo = getInfo.ContinueWith(t =>

                            StatusKeeper.StoreInfo(localPath, t.Result));

                storeInfo.Wait();

                StatusNotification.NotifyChangedFile(localPath);

            }

        }

        //Download a small file with a single GET operation

        private Task DownloadEntireFile(string container, Uri relativeUrl, string localPath)

        {

            if (String.IsNullOrWhiteSpace(container))

                throw new ArgumentNullException("container");

            if (relativeUrl == null)

                throw new ArgumentNullException("relativeUrl");

            if (String.IsNullOrWhiteSpace(localPath))

                throw new ArgumentNullException("localPath");

            if (!Path.IsPathRooted(localPath))

                throw new ArgumentException("The localPath must be rooted", "localPath");

            Contract.EndContractBlock();

            //Calculate the relative file path for the new file

            var relativePath = relativeUrl.RelativeUriToFilePath();

            //The file will be stored in a temporary location while downloading with an extension .download

            var tempPath = Path.Combine(FileAgent.FragmentsPath, relativePath + ".download");

            //Make sure the target folder exists. DownloadFileTask will not create the folder

            var directoryPath = Path.GetDirectoryName(tempPath);

            if (!Directory.Exists(directoryPath))

                Directory.CreateDirectory(directoryPath);

            //Download the object to the temporary location

            var getObject = CloudClient.GetObject(container, relativeUrl.ToString(), tempPath).ContinueWith(t =>

            {

                //And move it to its actual location once downloading is finished

                if (File.Exists(localPath))

                    File.Replace(tempPath,localPath,null,true);

                else

                    File.Move(tempPath,localPath);

            });

            return getObject;

        }

        public Task DownloadWithBlocks(string container,Uri relativeUrl, string localPath,TreeHash serverHash)

        {

            if (String.IsNullOrWhiteSpace(container))

                throw new ArgumentNullException("container");

            if (relativeUrl == null)

                throw new ArgumentNullException("relativeUrl");

            if (String.IsNullOrWhiteSpace(localPath))

                throw new ArgumentNullException("localPath");

            if (!Path.IsPathRooted(localPath))

                throw new ArgumentException("The localPath must be rooted", "localPath");

            if(serverHash==null)

                throw new ArgumentNullException("serverHash");

            Contract.EndContractBlock();

            //Calculate the relative file path for the new file

            var relativePath = relativeUrl.RelativeUriToFilePath();

            //The file will be stored in a temporary location while downloading with an extension .download

            var tempPath = Path.Combine(FileAgent.FragmentsPath, relativePath + ".download");

            var directoryPath = Path.GetDirectoryName(tempPath);

            if (!Directory.Exists(directoryPath))

                Directory.CreateDirectory(directoryPath);

            //If the local file exists we should make a copy of it to the

            //fragments folder, unless a newer temp copy already exists, which

            //means there is an interrupted download

            if (ShouldCopy(localPath, tempPath))

                File.Copy(localPath, tempPath, true);    

            //Set the size of the file to the size specified in the treehash

            //This will also create an empty file if the file doesn't exist            

            SetFileSize(tempPath, serverHash.Bytes);

            return Task.Factory.StartNew(() =>

            {

                //Calculate the temp file's treehash

                var treeHash = Signature.CalculateTreeHashAsync(tempPath, this.BlockSize,BlockHash).Result;

                //And compare it with the server's hash

                var upHashes = serverHash.GetHashesAsStrings();

                var localHashes = treeHash.HashDictionary;

                for (int i = 0; i < upHashes.Length; i++)

                {

                    //For every non-matching hash

                    if (!localHashes.ContainsKey(upHashes[i]))

                    {

                        Trace.TraceInformation("[BLOCK GET] START {0} of {1} for {2}",i,upHashes.Length,localPath);

                        var start = i*BlockSize;

                        long? end = null;

                        if (i < upHashes.Length - 1 )

                            end= ((i + 1)*BlockSize) ;

                        //Get its block

                        var blockTask = CloudClient.GetBlock(container, relativeUrl,

                                                            start, end);

                        blockTask.ContinueWith(b =>

                        {

                            //And apply it to the temp file

                            var buffer = b.Result;

                            var stream =FileAsync.OpenWrite(tempPath);

                            stream.Seek(start,SeekOrigin.Begin);

                            return stream.WriteAsync(buffer, 0, buffer.Length)

                                .ContinueWith(s => stream.Close());

                        }).Unwrap()

                        .Wait();

                        Trace.TraceInformation("[BLOCK GET] FINISH {0} of {1} for {2}", i, upHashes.Length, localPath);

                    }

                }

                //Replace the existing file with the temp

                if (File.Exists(localPath))

                    File.Replace(tempPath, localPath, null, true);

                else

                    File.Move(tempPath, localPath);

                Trace.TraceInformation("[BLOCK GET] COMPLETE {0}", localPath);

            });

        }

        //Change the file's size, possibly truncating or adding to it

        private static void SetFileSize(string filePath, long fileSize)

        {

            if (String.IsNullOrWhiteSpace(filePath))

                throw new ArgumentNullException("filePath");

            if (!Path.IsPathRooted(filePath))

                throw new ArgumentException("The filePath must be rooted", "filePath");

            if (fileSize<0)

                throw new ArgumentOutOfRangeException("fileSize");

            Contract.EndContractBlock();

            using (var stream = File.Open(filePath, FileMode.OpenOrCreate, FileAccess.Write))

            {

                stream.SetLength(fileSize);

            }

        }

        //Check whether we should copy the local file to a temp path        

        private static bool ShouldCopy(string localPath, string tempPath)

        {

            //No need to copy if there is no file

            if (!File.Exists(localPath))            

                return false;

            //If there is no temp file, go ahead and copy

            if (!File.Exists(tempPath))                

                return true;

            //If there is a temp file and is newer than the actual file, don't copy

            var localLastWrite = File.GetLastWriteTime(localPath);

            var tempLastWrite = File.GetLastWriteTime(tempPath);

            //This could mean there is an interrupted download in progress

            return (tempLastWrite < localLastWrite);

        }

        private void UploadCloudFile(FileInfo fileInfo, string hash,string topHash)

        {

            if (fileInfo==null)

                throw new ArgumentNullException("fileInfo");

            if (String.IsNullOrWhiteSpace(hash))

                throw new ArgumentNullException("hash");

            Contract.EndContractBlock();

            if (fileInfo.Extension.Equals("ignore", StringComparison.InvariantCultureIgnoreCase))

                return;

            var url = fileInfo.AsRelativeUrlTo(FileAgent.RootPath);

            var fullFileName = fileInfo.FullName;

            using(var gate=NetworkGate.Acquire(fullFileName,NetworkOperation.Uploading))

            {

                //Abort if the file is already being uploaded or downloaded

                if (gate.Failed)

                    return;

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

                var info = CloudClient.GetObjectInfo(PithosContainer, url);

                //If the file hashes match, abort the upload

                if (hash.Equals(info.Hash, StringComparison.InvariantCultureIgnoreCase) ||

                    topHash.Equals(info.Hash, StringComparison.InvariantCultureIgnoreCase))

                {

                    //but store any metadata changes 

                    this.StatusKeeper.StoreInfo(fullFileName, info);

                    Trace.TraceInformation("Skip upload of {0}, hashes match", fullFileName);

                    return;

                }

                //Mark the file as modified while we upload it

                var setStatus = Task.Factory.StartNew(() =>

                        StatusKeeper.SetFileOverlayStatus(fullFileName, FileOverlayStatus.Modified));

                //And then upload it

                //If the file is larger than the block size, try a hashmap PUT

                if (fileInfo.Length > BlockSize )

                {

                    //To upload using a hashmap

                    //First, calculate the tree hash

                    var treeHash = Signature.CalculateTreeHashAsync(fileInfo.FullName, BlockSize, BlockHash);                    

                    var putHashMap = setStatus.ContinueWith(t=>

                        UploadWithHashMap(fileInfo,url,treeHash));

                    putHashMap.Wait();

                }

                else

                {

                    //Otherwise do a regular PUT

                    var put = setStatus.ContinueWith(t =>

                        CloudClient.PutObject(PithosContainer,url,fullFileName,hash));

                    put.Wait();

                }

                //If everything succeeds, change the file and overlay status to normal

                this.StatusKeeper.SetFileState(fullFileName, FileStatus.Unchanged, FileOverlayStatus.Normal);

            }

            //Notify the Shell to update the overlays

            NativeMethods.RaiseChangeNotification(fullFileName);

            StatusNotification.NotifyChangedFile(fullFileName);

        }

        public void UploadWithHashMap(FileInfo fileInfo,string url,Task<TreeHash> treeHash)

        {

            var fullFileName = fileInfo.FullName;

            //Send the hashmap to the server            

            var hashPut = CloudClient.PutHashMap(PithosContainer, url, treeHash.Result);

            var missingHashes = hashPut.Result;

            if (missingHashes.Count == 0)

                return;

            var buffer = new byte[BlockSize];                      

            foreach (var missingHash in missingHashes)

            {

                int blockIndex = -1;

                try

                {

                    //Find the proper block

                    blockIndex = treeHash.Result.HashDictionary[missingHash];

                    var offset = blockIndex*BlockSize;

                    var read = fileInfo.Read(buffer, offset, BlockSize);

                    if (read > 0)

                    {

                        //Copy the actual block data out of the buffer

                        var data = new byte[read];

                        Buffer.BlockCopy(buffer, 0, data, 0, read);

                        //And POST them

                        CloudClient.PostBlock(PithosContainer, data).Wait();

                        Trace.TraceInformation("[BLOCK] Block {0} of {1} uploaded", blockIndex,

                            fullFileName);

                    }

                }

                catch (Exception exc)

                {

                    Trace.TraceError("[ERROR] uploading block {0} of {1}\n{2}", blockIndex, fullFileName, exc);

                }

            }

            UploadWithHashMap(fileInfo, url, treeHash);

        }

    }

}