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.Net;

using System.Reflection;

using System.Security.Cryptography;

using System.Threading;

using System.Threading.Tasks;

using Pithos.Interfaces;

using Pithos.Network;

using log4net;

namespace Pithos.Core.Agents

{

    [Export(typeof(Uploader))]

    public class Uploader

    {

        private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);

        [Import]

        private IStatusKeeper StatusKeeper { get; set; }

        [Import]

        private IPithosSettings Settings { get; set; }

        public IStatusNotification StatusNotification { get; set; }

        //CancellationTokenSource _cts = new CancellationTokenSource();

        /*public void SignalStop()

        {

            _cts.Cancel();

        }*/

        public async Task UploadCloudFile(CloudUploadAction action,CancellationToken cancellationToken)

        {

            if (action == null)

                throw new ArgumentNullException("action");

            Contract.EndContractBlock();

            using (ThreadContext.Stacks["Operation"].Push("UploadCloudFile"))

            {

                try

                {

                    await UnpauseEvent.WaitAsync().ConfigureAwait(false);

                    var fileInfo = action.LocalFile;

                    var progress=new Progress<double>(d=>

                        StatusNotification.Notify(new StatusNotification(String.Format("Merkle Hashing for Upload {0:p} of {1}",d,fileInfo.Name))));

                    TreeHash localTreeHash;

                    using (StatusNotification.GetNotifier("Merkle Hashing for Upload {0}", "Merkle Hashed for Upload {0}", fileInfo.Name))

                    {

                        localTreeHash = Signature.CalculateTreeHashAsync(fileInfo,

                                                                         action.AccountInfo.BlockSize,

                                                                         action.AccountInfo.BlockHash, Settings.HashingParallelism,cancellationToken,progress);

                    }

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

                        return;

                    if (!Selectives.IsSelected(action.AccountInfo, fileInfo) && !action.IsCreation)

                        return;

                    //Try to load the action's local state, if it is empty

                    if (action.FileState == null)

                        action.FileState = StatusKeeper.GetStateByFilePath(fileInfo.FullName);

                    if (action.FileState != null)

                    {

                        /*

                                                Log.WarnFormat("File [{0}] has no local state. It was probably created by a download action", fileInfo.FullName);

                                                return;

                        */

                        var latestState = action.FileState;

                        //Do not upload files in conflict

                        if (latestState.FileStatus == FileStatus.Conflict)

                        {

                            Log.InfoFormat("Skipping file in conflict [{0}]", fileInfo.FullName);

                            return;

                        }

                        //Do not upload files when we have no permission

                        if (latestState.FileStatus == FileStatus.Forbidden)

                        {

                            Log.InfoFormat("Skipping forbidden file [{0}]", fileInfo.FullName);

                            return;

                        }

                    }

                    //Are we targeting our own account or a sharer account?

                    var relativePath = fileInfo.AsRelativeTo(action.AccountInfo.AccountPath);

                    var accountInfo = relativePath.StartsWith(FolderConstants.OthersFolder) 

                                                  ? GetSharerAccount(relativePath, action.AccountInfo) 

                                                  : action.AccountInfo;

                    var fullFileName = fileInfo.GetProperCapitalization();

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

                    {

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

                        if (gate.Failed)

                            return;

                        var cloudFile = action.CloudFile;

                        var account = cloudFile.Account ?? accountInfo.UserName;

                        try

                        {

                            var client = new CloudFilesClient(accountInfo);

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

                            var cloudInfo = client.GetObjectInfo(account, cloudFile.Container, cloudFile.Name);

                            //If this a shared file

                            if (!cloudFile.Account.Equals(action.AccountInfo.UserName,StringComparison.InvariantCultureIgnoreCase))

                            {

/*

                                if (!cloudInfo.IsWritable(action.AccountInfo.UserName))

                                {

                                    MakeFileReadOnly(fullFileName);

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

                                    return;

                                }

*/

                                //If this is a read-only file, do not upload changes

                                if ( !cloudInfo.IsWritable(action.AccountInfo.UserName) ||

                                    //If the file is new, but we can't upload it

                                    (!cloudInfo.Exists && !client.CanUpload(account, cloudFile)) )

                                {

                                    MakeFileReadOnly(fullFileName);

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

                                    return;

                                }

                            }

                            await UnpauseEvent.WaitAsync().ConfigureAwait(false);

                            fileInfo.Refresh();

                            //Does the file still exist or was it deleted/renamed?

                            if (fileInfo.Exists)

                            {

                                if (fileInfo is DirectoryInfo)

                                {

                                    //If the directory doesn't exist the Hash property will be empty

                                    if (String.IsNullOrWhiteSpace(cloudInfo.X_Object_Hash))

                                        //Go on and create the directory

                                        await

                                            client.PutObject(account, cloudFile.Container, cloudFile.Name, fullFileName,

                                                             String.Empty, "application/directory");

                                    //If the upload is in response to a Folder create with Selective Sync enabled

                                    if (action.IsCreation)

                                    {

                                        //Add the folder to the Selected URls

                                        var selectiveUri = new Uri(client.RootAddressUri, cloudFile.Uri);

                                        Selectives.AddUri(accountInfo, selectiveUri);

                                        Selectives.Save(accountInfo);

                                    }

                                }

                                else

                                {

                                    var cloudHash = cloudInfo.X_Object_Hash.ToLower();

                                    string topHash;

                                    TreeHash treeHash;

                                    using (

                                        StatusNotification.GetNotifier("Hashing {0} for Upload", "Finished hashing {0}",

                                                                       fileInfo.Name))

                                    {

                                        treeHash = localTreeHash ?? action.TreeHash.Value;

                                        topHash = treeHash.TopHash.ToHashString();

                                    }

                                    //If the file hashes match, abort the upload

                                    if (cloudInfo != ObjectInfo.Empty && (topHash == cloudHash))

                                    {

                                        //but store any metadata changes 

                                        StatusKeeper.StoreInfo(fullFileName, cloudInfo);

                                        Log.InfoFormat("Skip upload of {0}, hashes match", fullFileName);

                                        return;

                                    }

                                    //Mark the file as modified while we upload it

                                    await

                                        StatusKeeper.SetFileOverlayStatus(fullFileName, FileOverlayStatus.Modified).

                                            ConfigureAwait(false);

                                    //And then upload it

                                    //Upload even small files using the Hashmap. The server may already contain

                                    //the relevant block                                

                                    await

                                        UploadWithHashMap(accountInfo, cloudFile, fileInfo as FileInfo, cloudFile.Name,

                                                          treeHash, cancellationToken).ConfigureAwait(false);

                                }

                                var currentInfo = client.GetObjectInfo(cloudFile.Account, cloudFile.Container,

                                                                       cloudFile.Name);

                                //If there is no stored ObjectID in the file state, add it

                                if (action.FileState == null || action.FileState.ObjectID == null)

                                {

                                    StatusKeeper.StoreInfo(fullFileName, currentInfo);

                                }

                                else

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

                                    StatusKeeper.SetFileState(fullFileName, FileStatus.Unchanged,

                                                              FileOverlayStatus.Normal, "");

                            }

                            else

                            {

                                StatusKeeper.ClearFileStatus(fullFileName);

                            }

                        }

                        catch (WebException exc)

                        {

                            var response = (exc.Response as HttpWebResponse);

                            if (response == null)

                                throw;

                            if (response.StatusCode == HttpStatusCode.Forbidden)

                            {

                                StatusKeeper.SetFileState(fileInfo.FullName, FileStatus.Forbidden, FileOverlayStatus.Conflict, "Forbidden");

                                MakeFileReadOnly(fullFileName);

                            }

                            else

                                //In any other case, propagate the error

                                throw;

                        }

                    }

                    //Notify the Shell to update the overlays

                    NativeMethods.RaiseChangeNotification(fullFileName);

                    StatusNotification.NotifyChangedFile(fullFileName);

                }

                catch (AggregateException ex)

                {

                    var exc = ex.InnerException as WebException;

                    if (exc == null)

                        throw ex.InnerException;

                    if (HandleUploadWebException(action, exc))

                        return;

                    throw;

                }

                catch (WebException ex)

                {

                    if (HandleUploadWebException(action, ex))

                        return;

                    throw;

                }

                catch (Exception ex)

                {

                    Log.Error("Unexpected error while uploading file", ex);

                    throw;

                }

            }

        }

        private static void MakeFileReadOnly(string fullFileName)

        {

            var attributes = File.GetAttributes(fullFileName);

            //Do not make any modifications if not necessary

            if (attributes.HasFlag(FileAttributes.ReadOnly))

                return;

            File.SetAttributes(fullFileName, attributes | FileAttributes.ReadOnly);            

        }

        private static AccountInfo GetSharerAccount(string relativePath, AccountInfo accountInfo)

        {

            var parts = relativePath.Split('\\');

            var accountName = parts[1];

            var oldName = accountInfo.UserName;

            var absoluteUri = accountInfo.StorageUri.AbsoluteUri;

            var nameIndex = absoluteUri.IndexOf(oldName, StringComparison.Ordinal);

            var root = absoluteUri.Substring(0, nameIndex);

            accountInfo = new AccountInfo

                              {

                                  UserName = accountName,

                                  AccountPath = Path.Combine(accountInfo.AccountPath, parts[0], parts[1]),

                                  StorageUri = new Uri(root + accountName),

                                  BlockHash = accountInfo.BlockHash,

                                  BlockSize = accountInfo.BlockSize,

                                  Token = accountInfo.Token

                              };

            return accountInfo;

        }

        public async Task UploadWithHashMap(AccountInfo accountInfo, ObjectInfo cloudFile, FileInfo fileInfo, string url, TreeHash treeHash, CancellationToken token)

        {

            if (accountInfo == null)

                throw new ArgumentNullException("accountInfo");

            if (cloudFile == null)

                throw new ArgumentNullException("cloudFile");

            if (fileInfo == null)

                throw new ArgumentNullException("fileInfo");

            if (String.IsNullOrWhiteSpace(url))

                throw new ArgumentNullException(url);

            if (treeHash == null)

                throw new ArgumentNullException("treeHash");

            if (String.IsNullOrWhiteSpace(cloudFile.Container))

                throw new ArgumentException("Invalid container", "cloudFile");

            Contract.EndContractBlock();

            if (await WaitOrAbort(accountInfo, cloudFile, token).ConfigureAwait(false))

                return;

            var fullFileName = fileInfo.GetProperCapitalization();

            var account = cloudFile.Account ?? accountInfo.UserName;

            var container = cloudFile.Container;

            int block = 0;

            var client = new CloudFilesClient(accountInfo);

            //Send the hashmap to the server            

            var missingHashes = await client.PutHashMap(account, container, url, treeHash).ConfigureAwait(false);

            ReportUploadProgress(fileInfo.Name, block, 0, missingHashes.Count, fileInfo.Length);

            //If the server returns no missing hashes, we are done

            client.UploadProgressChanged += (sender, args) =>

                                            ReportUploadProgress(fileInfo.Name, block, args.ProgressPercentage,

                                                                 missingHashes.Count, fileInfo.Length);

            while (missingHashes.Count > 0)

            {

                block = 0;

                if (await WaitOrAbort(accountInfo, cloudFile, token).ConfigureAwait(false))

                    return;

                var buffer = new byte[accountInfo.BlockSize];

                foreach (var missingHash in missingHashes)

                {

                    if (await WaitOrAbort(accountInfo, cloudFile, token).ConfigureAwait(false))

                        return;

                    //Find the proper block

                    long blockIndex = treeHash.HashDictionary[missingHash];

                    long offset = blockIndex*accountInfo.BlockSize;

                    Debug.Assert(offset >= 0,

                                 String.Format("Negative Offset! BlockIndex {0} BlockSize {1}", blockIndex,

                                               accountInfo.BlockSize));

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

                    try

                    {

                        //And upload the block                

                        await client.PostBlock(account, container, buffer, 0, read, token).ConfigureAwait(false);

                        token.ThrowIfCancellationRequested();

                        Log.InfoFormat("[BLOCK] Block {0} of {1} uploaded", blockIndex, fullFileName);

                    }

                    catch (TaskCanceledException exc)

                    {

                        throw new OperationCanceledException(token);

                    }

                    catch (Exception exc)

                    {

                        Log.Error(String.Format("Uploading block {0} of {1}", blockIndex, fullFileName), exc);

                    }

                    ReportUploadProgress(fileInfo.Name, block++, 100, missingHashes.Count, fileInfo.Length);

                }

                token.ThrowIfCancellationRequested();

                //Repeat until there are no more missing hashes                

                missingHashes = await client.PutHashMap(account, container, url, treeHash).ConfigureAwait(false);

            }

            ReportUploadProgress(fileInfo.Name, missingHashes.Count, 0, missingHashes.Count, fileInfo.Length);

        }

        private async Task<bool> WaitOrAbort(AccountInfo account,ObjectInfo cloudFile, CancellationToken token)

        {

            token.ThrowIfCancellationRequested();

            await UnpauseEvent.WaitAsync().ConfigureAwait(false);

            var shouldAbort = !Selectives.IsSelected(account,cloudFile);

            if (shouldAbort)

                Log.InfoFormat("Aborting [{0}]",cloudFile.Uri);

            return shouldAbort;

        }

        private void ReportUploadProgress(string fileName, int block, int blockPercentage, int totalBlocks, long fileSize)

        {

            StatusNotification.Notify(totalBlocks == 0

                                          ? new ProgressNotification(fileName, "Uploading", 1,blockPercentage, 1, fileSize)

                                          : new ProgressNotification(fileName, "Uploading", block, blockPercentage, totalBlocks, fileSize));

        }

        private bool HandleUploadWebException(CloudAction action, WebException exc)

        {

            var response = exc.Response as HttpWebResponse;

            if (response == null)

                throw exc;

            if (response.StatusCode == HttpStatusCode.Unauthorized)

            {

                Log.Error("Not allowed to upload file", exc);

                var message = String.Format("Not allowed to uplad file {0}", action.LocalFile.FullName);

                StatusKeeper.SetFileState(action.LocalFile.FullName, FileStatus.Unchanged, FileOverlayStatus.Normal, "");

                StatusNotification.NotifyChange(message, TraceLevel.Warning);

                return true;

            }

            return false;

        }

        [Import]

        public Selectives Selectives { get; set; }

        public AsyncManualResetEvent UnpauseEvent { get; set; }

    }

}