code.grnet.gr Git - pithos-ms-client/blob - trunk/Pithos.Core/Agents/Downloader.cs

   1 using System;\r
   2 using System.Collections.Generic;\r
   3 using System.ComponentModel.Composition;\r
   4 using System.Diagnostics.Contracts;\r
   5 using System.IO;\r
   6 using System.Linq;\r
   7 using System.Reflection;\r
   8 using System.Threading;\r
   9 using System.Threading.Tasks;\r
  10 using Pithos.Interfaces;\r
  11 using Pithos.Network;\r
  12 using log4net;\r
  13 \r
  14 namespace Pithos.Core.Agents\r
  15 {\r
  16     [Export(typeof(Downloader))]\r
  17     public class Downloader\r
  18     {\r
  19         private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);\r
  20 \r
  21         [Import]\r
  22         private IStatusKeeper StatusKeeper { get; set; }\r
  23 \r
  24         \r
  25         public IStatusNotification StatusNotification { get; set; }\r
  26 \r
  27 /*\r
  28         private CancellationTokenSource _cts=new CancellationTokenSource();\r
  29 \r
  30         public void SignalStop()\r
  31         {\r
  32             _cts.Cancel();\r
  33         }\r
  34 */\r
  35 \r
  36 \r
  37         //Download a file.\r
  38         public async Task DownloadCloudFile(AccountInfo accountInfo, ObjectInfo cloudFile, string filePath,CancellationToken cancellationToken)\r
  39         {\r
  40             if (accountInfo == null)\r
  41                 throw new ArgumentNullException("accountInfo");\r
  42             if (cloudFile == null)\r
  43                 throw new ArgumentNullException("cloudFile");\r
  44             if (String.IsNullOrWhiteSpace(cloudFile.Account))\r
  45                 throw new ArgumentNullException("cloudFile");\r
  46             if (String.IsNullOrWhiteSpace(cloudFile.Container))\r
  47                 throw new ArgumentNullException("cloudFile");\r
  48             if (String.IsNullOrWhiteSpace(filePath))\r
  49                 throw new ArgumentNullException("filePath");\r
  50             if (!Path.IsPathRooted(filePath))\r
  51                 throw new ArgumentException("The filePath must be rooted", "filePath");\r
  52             Contract.EndContractBlock();\r
  53                 using (ThreadContext.Stacks["Operation"].Push("DownloadCloudFile"))\r
  54                 {\r
  55                    // var cancellationToken=_cts.Token;//  .ThrowIfCancellationRequested();\r
  56 \r
  57                     if (await WaitOrAbort(accountInfo,cloudFile, cancellationToken))\r
  58                         return;\r
  59 \r
  60 \r
  61                     var localPath = FileInfoExtensions.GetProperFilePathCapitalization(filePath);\r
  62                     var relativeUrl = new Uri(cloudFile.Name, UriKind.Relative);\r
  63 \r
  64                     var url = relativeUrl.ToString();\r
  65                     if (cloudFile.Name.EndsWith(".ignore", StringComparison.InvariantCultureIgnoreCase))\r
  66                         return;\r
  67 \r
  68                     if (!Selectives.IsSelected(accountInfo,cloudFile))\r
  69                         return;\r
  70 \r
  71 \r
  72                     //Are we already downloading or uploading the file? \r
  73                     using (var gate = NetworkGate.Acquire(localPath, NetworkOperation.Downloading))\r
  74                     {\r
  75                         if (gate.Failed)\r
  76                             return;\r
  77 \r
  78                         var client = new CloudFilesClient(accountInfo);\r
  79                         var account = cloudFile.Account;\r
  80                         var container = cloudFile.Container;\r
  81 \r
  82                         if (cloudFile.IsDirectory)\r
  83                         {\r
  84                             if (!Directory.Exists(localPath))\r
  85                                 try\r
  86                                 {\r
  87                                     Directory.CreateDirectory(localPath);\r
  88                                     if (Log.IsDebugEnabled)\r
  89                                         Log.DebugFormat("Created Directory [{0}]", localPath);\r
  90                                 }\r
  91                                 catch (IOException)\r
  92                                 {\r
  93                                     var localInfo = new FileInfo(localPath);\r
  94                                     if (localInfo.Exists && localInfo.Length == 0)\r
  95                                     {\r
  96                                         Log.WarnFormat("Malformed directory object detected for [{0}]", localPath);\r
  97                                         localInfo.Delete();\r
  98                                         Directory.CreateDirectory(localPath);\r
  99                                         if (Log.IsDebugEnabled)\r
 100                                             Log.DebugFormat("Created Directory [{0}]", localPath);\r
 101                                     }\r
 102                                 }\r
 103                         }\r
 104                         else\r
 105                         {\r
 106                             var isChanged = IsObjectChanged(cloudFile, localPath);\r
 107                             if (isChanged)\r
 108                             {\r
 109                                 //Retrieve the hashmap from the server\r
 110                                 var serverHash = await client.GetHashMap(account, container, url);\r
 111                                 //If it's a small file\r
 112                                 if (serverHash.Hashes.Count == 1)\r
 113                                     //Download it in one go\r
 114                                     await\r
 115                                         DownloadEntireFileAsync(accountInfo, client, cloudFile, relativeUrl, localPath,cancellationToken);\r
 116                                     //Otherwise download it block by block\r
 117                                 else\r
 118                                     await\r
 119                                         DownloadWithBlocks(accountInfo, client, cloudFile, relativeUrl, localPath,\r
 120                                                            serverHash,cancellationToken);\r
 121 \r
 122                                 if (!cloudFile.IsWritable(accountInfo.UserName))\r
 123                                 {\r
 124                                     var attributes = File.GetAttributes(localPath);\r
 125                                     File.SetAttributes(localPath, attributes | FileAttributes.ReadOnly);\r
 126                                 }\r
 127                             }\r
 128                         }\r
 129 \r
 130                         //Now we can store the object's metadata without worrying about ghost status entries\r
 131                         StatusKeeper.StoreInfo(localPath, cloudFile);\r
 132 \r
 133                     }\r
 134                 }\r
 135            \r
 136         }\r
 137 \r
 138         //Download a file asynchronously using blocks\r
 139         public async Task DownloadWithBlocks(AccountInfo accountInfo, CloudFilesClient client, ObjectInfo cloudFile, Uri relativeUrl, string filePath, TreeHash serverHash, CancellationToken cancellationToken)\r
 140         {\r
 141             if (client == null)\r
 142                 throw new ArgumentNullException("client");\r
 143             if (cloudFile == null)\r
 144                 throw new ArgumentNullException("cloudFile");\r
 145             if (relativeUrl == null)\r
 146                 throw new ArgumentNullException("relativeUrl");\r
 147             if (String.IsNullOrWhiteSpace(filePath))\r
 148                 throw new ArgumentNullException("filePath");\r
 149             if (!Path.IsPathRooted(filePath))\r
 150                 throw new ArgumentException("The filePath must be rooted", "filePath");\r
 151             if (serverHash == null)\r
 152                 throw new ArgumentNullException("serverHash");\r
 153             if (cloudFile.IsDirectory)\r
 154                 throw new ArgumentException("cloudFile is a directory, not a file", "cloudFile");\r
 155             Contract.EndContractBlock();\r
 156 \r
 157             if (await WaitOrAbort(accountInfo,cloudFile, cancellationToken))\r
 158                 return;\r
 159 \r
 160             var fileAgent = GetFileAgent(accountInfo);\r
 161             var localPath = FileInfoExtensions.GetProperFilePathCapitalization(filePath);\r
 162 \r
 163             //Calculate the relative file path for the new file\r
 164             var relativePath = relativeUrl.RelativeUriToFilePath();\r
 165             var blockUpdater = new BlockUpdater(fileAgent.CachePath, localPath, relativePath, serverHash);\r
 166 \r
 167             StatusNotification.SetPithosStatus(PithosStatus.LocalSyncing, String.Format("Calculating hashmap for {0} before download", Path.GetFileName(localPath)));\r
 168             //Calculate the file's treehash\r
 169 \r
 170             //TODO: Should pass cancellation token here\r
 171             var treeHash = await Signature.CalculateTreeHashAsync(localPath, (int)serverHash.BlockSize, serverHash.BlockHash, 2);\r
 172 \r
 173             //And compare it with the server's hash\r
 174             var upHashes = serverHash.GetHashesAsStrings();\r
 175             var localHashes = treeHash.HashDictionary;\r
 176             ReportDownloadProgress(Path.GetFileName(localPath), 0, upHashes.Length, cloudFile.Bytes);\r
 177             for (var i = 0; i < upHashes.Length; i++)\r
 178             {\r
 179                 if (await WaitOrAbort(accountInfo,cloudFile, cancellationToken))\r
 180                     return;\r
 181 \r
 182                 //For every non-matching hash\r
 183                 var upHash = upHashes[i];\r
 184                 if (!localHashes.ContainsKey(upHash))\r
 185                 {\r
 186                     StatusNotification.Notify(new CloudNotification { Data = cloudFile });\r
 187                     ReportDownloadProgress(Path.GetFileName(localPath), i, upHashes.Length, cloudFile.Bytes);\r
 188 \r
 189                     if (blockUpdater.UseOrphan(i, upHash))\r
 190                     {\r
 191                         Log.InfoFormat("[BLOCK GET] ORPHAN FOUND for {0} of {1} for {2}", i, upHashes.Length, localPath);\r
 192                         continue;\r
 193                     }\r
 194                     Log.InfoFormat("[BLOCK GET] START {0} of {1} for {2}", i, upHashes.Length, localPath);\r
 195                     var start = i * serverHash.BlockSize;\r
 196                     //To download the last block just pass a null for the end of the range\r
 197                     long? end = null;\r
 198                     if (i < upHashes.Length - 1)\r
 199                         end = ((i + 1) * serverHash.BlockSize);\r
 200 \r
 201                     //TODO: Pass token here\r
 202                     //Download the missing block\r
 203                     var block = await client.GetBlock(cloudFile.Account, cloudFile.Container, relativeUrl, start, end,cancellationToken);\r
 204 \r
 205                     //and store it\r
 206                     blockUpdater.StoreBlock(i, block);\r
 207 \r
 208 \r
 209                     Log.InfoFormat("[BLOCK GET] FINISH {0} of {1} for {2}", i, upHashes.Length, localPath);\r
 210                 }\r
 211                 ReportDownloadProgress(Path.GetFileName(localPath), i, upHashes.Length, cloudFile.Bytes);\r
 212             }\r
 213 \r
 214             //Want to avoid notifications if no changes were made\r
 215             var hasChanges = blockUpdater.HasBlocks;\r
 216             blockUpdater.Commit();\r
 217 \r
 218             if (hasChanges)\r
 219                 //Notify listeners that a local file has changed\r
 220                 StatusNotification.NotifyChangedFile(localPath);\r
 221 \r
 222             Log.InfoFormat("[BLOCK GET] COMPLETE {0}", localPath);\r
 223         }\r
 224 \r
 225         //Download a small file with a single GET operation\r
 226         private async Task DownloadEntireFileAsync(AccountInfo accountInfo, CloudFilesClient client, ObjectInfo cloudFile, Uri relativeUrl, string filePath, CancellationToken cancellationToken)\r
 227         {\r
 228             if (client == null)\r
 229                 throw new ArgumentNullException("client");\r
 230             if (cloudFile == null)\r
 231                 throw new ArgumentNullException("cloudFile");\r
 232             if (relativeUrl == null)\r
 233                 throw new ArgumentNullException("relativeUrl");\r
 234             if (String.IsNullOrWhiteSpace(filePath))\r
 235                 throw new ArgumentNullException("filePath");\r
 236             if (!Path.IsPathRooted(filePath))\r
 237                 throw new ArgumentException("The localPath must be rooted", "filePath");\r
 238             if (cloudFile.IsDirectory)\r
 239                 throw new ArgumentException("cloudFile is a directory, not a file", "cloudFile");\r
 240             Contract.EndContractBlock();\r
 241 \r
 242             if (await WaitOrAbort(accountInfo,cloudFile, cancellationToken))\r
 243                 return;\r
 244 \r
 245             var localPath = FileInfoExtensions.GetProperFilePathCapitalization(filePath);\r
 246             StatusNotification.SetPithosStatus(PithosStatus.LocalSyncing, String.Format("Downloading {0}", Path.GetFileName(localPath)));\r
 247             StatusNotification.Notify(new CloudNotification { Data = cloudFile });\r
 248             ReportDownloadProgress(Path.GetFileName(localPath), 1, 1, cloudFile.Bytes);\r
 249 \r
 250             var fileAgent = GetFileAgent(accountInfo);\r
 251             //Calculate the relative file path for the new file\r
 252             var relativePath = relativeUrl.RelativeUriToFilePath();\r
 253             //The file will be stored in a temporary location while downloading with an extension .download\r
 254             var tempPath = Path.Combine(fileAgent.CachePath, relativePath + ".download");\r
 255             //Make sure the target folder exists. DownloadFileTask will not create the folder\r
 256             var tempFolder = Path.GetDirectoryName(tempPath);\r
 257             if (!Directory.Exists(tempFolder))\r
 258                 Directory.CreateDirectory(tempFolder);\r
 259 \r
 260             //TODO: Should pass the token here\r
 261 \r
 262             //Download the object to the temporary location\r
 263             await client.GetObject(cloudFile.Account, cloudFile.Container, relativeUrl.ToString(), tempPath,cancellationToken);\r
 264 \r
 265             //Create the local folder if it doesn't exist (necessary for shared objects)\r
 266             var localFolder = Path.GetDirectoryName(localPath);\r
 267             if (!Directory.Exists(localFolder))\r
 268                 try\r
 269                 {\r
 270                     Directory.CreateDirectory(localFolder);\r
 271                 }\r
 272                 catch (IOException)\r
 273                 {\r
 274                     //A file may already exist that has the same name as the new folder.\r
 275                     //This may be an artifact of the way Pithos handles directories\r
 276                     var fileInfo = new FileInfo(localFolder);\r
 277                     if (fileInfo.Exists && fileInfo.Length == 0)\r
 278                     {\r
 279                         Log.WarnFormat("Malformed directory object detected for [{0}]", localFolder);\r
 280                         fileInfo.Delete();\r
 281                         Directory.CreateDirectory(localFolder);\r
 282                     }\r
 283                     else\r
 284                         throw;\r
 285                 }\r
 286             //And move it to its actual location once downloading is finished\r
 287             if (File.Exists(localPath))\r
 288                 File.Replace(tempPath, localPath, null, true);\r
 289             else\r
 290                 File.Move(tempPath, localPath);\r
 291             //Notify listeners that a local file has changed\r
 292             StatusNotification.NotifyChangedFile(localPath);\r
 293 \r
 294 \r
 295         }\r
 296 \r
 297 \r
 298         private void ReportDownloadProgress(string fileName, int block, int totalBlocks, long fileSize)\r
 299         {\r
 300             StatusNotification.Notify(totalBlocks == 0\r
 301                                           ? new ProgressNotification(fileName, "Downloading", 1, 1, fileSize)\r
 302                                           : new ProgressNotification(fileName, "Downloading", block, totalBlocks, fileSize));\r
 303         }\r
 304 \r
 305         private bool IsObjectChanged(ObjectInfo cloudFile, string localPath)\r
 306         {\r
 307             //If the target is a directory, there are no changes to download\r
 308             if (Directory.Exists(localPath))\r
 309                 return false;\r
 310             //If the file doesn't exist, we have a chagne\r
 311             if (!File.Exists(localPath))\r
 312                 return true;\r
 313             //If there is no stored state, we have a change\r
 314             var localState = StatusKeeper.GetStateByFilePath(localPath);\r
 315             if (localState == null)\r
 316                 return true;\r
 317 \r
 318             var info = new FileInfo(localPath);\r
 319             var shortHash = info.ComputeShortHash();\r
 320             //If the file is different from the stored state, we have a change\r
 321             if (localState.ShortHash != shortHash)\r
 322                 return true;\r
 323             //If the top hashes differ, we have a change\r
 324             return (localState.Checksum != cloudFile.Hash);\r
 325         }\r
 326 \r
 327         private static FileAgent GetFileAgent(AccountInfo accountInfo)\r
 328         {\r
 329             return AgentLocator<FileAgent>.Get(accountInfo.AccountPath);\r
 330         }\r
 331 \r
 332         private async Task<bool> WaitOrAbort(AccountInfo account,ObjectInfo cloudFile, CancellationToken token)\r
 333         {\r
 334             token.ThrowIfCancellationRequested();\r
 335             await UnpauseEvent.WaitAsync();\r
 336             var shouldAbort = !Selectives.IsSelected(account,cloudFile);\r
 337             if (shouldAbort)\r
 338                 Log.InfoFormat("Aborting [{0}]", cloudFile.Uri);\r
 339             return shouldAbort;\r
 340         }\r
 341 \r
 342         [Import]\r
 343         public Selectives Selectives { get; set; }\r
 344 \r
 345         public AsyncManualResetEvent UnpauseEvent { get; set; }\r
 346     }\r
 347 }\r