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

   1 using System;\r
   2 using System.Collections.Generic;\r
   3 using System.ComponentModel.Composition;\r
   4 using System.Diagnostics;\r
   5 using System.Diagnostics.Contracts;\r
   6 using System.IO;\r
   7 using System.Linq;\r
   8 using System.Net;\r
   9 using System.Reflection;\r
  10 using System.Threading;\r
  11 using System.Threading.Tasks;\r
  12 using Pithos.Interfaces;\r
  13 using Pithos.Network;\r
  14 using log4net;\r
  15 \r
  16 namespace Pithos.Core.Agents\r
  17 {\r
  18     [Export(typeof(Uploader))]\r
  19     public class Uploader\r
  20     {\r
  21         private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);\r
  22 \r
  23         [Import]\r
  24         private IStatusKeeper StatusKeeper { get; set; }\r
  25 \r
  26         \r
  27         public IStatusNotification StatusNotification { get; set; }\r
  28 \r
  29         \r
  30         //CancellationTokenSource _cts = new CancellationTokenSource();\r
  31         /*public void SignalStop()\r
  32         {\r
  33             _cts.Cancel();\r
  34         }*/\r
  35 \r
  36         public async Task UploadCloudFile(CloudAction action,CancellationToken cancellationToken)\r
  37         {\r
  38             if (action == null)\r
  39                 throw new ArgumentNullException("action");\r
  40             Contract.EndContractBlock();\r
  41 \r
  42             using (ThreadContext.Stacks["Operation"].Push("UploadCloudFile"))\r
  43             {\r
  44                 try\r
  45                 {\r
  46                     await UnpauseEvent.WaitAsync();\r
  47 \r
  48                     var fileInfo = action.LocalFile;\r
  49 \r
  50                     if (fileInfo.Extension.Equals("ignore", StringComparison.InvariantCultureIgnoreCase))\r
  51                         return;\r
  52 \r
  53                     if (!Selectives.IsSelected(action.AccountInfo, fileInfo))\r
  54                         return;\r
  55 \r
  56                     //Try to load the action's local state, if it is empty\r
  57                     if (action.FileState == null)\r
  58                         action.FileState = StatusKeeper.GetStateByFilePath(fileInfo.FullName);\r
  59                     if (action.FileState == null)\r
  60                     {\r
  61                         Log.WarnFormat("File [{0}] has no local state. It was probably created by a download action", fileInfo.FullName);\r
  62                         return;\r
  63                     }\r
  64 \r
  65                     var latestState = action.FileState;\r
  66 \r
  67                     //Do not upload files in conflict\r
  68                     if (latestState.FileStatus == FileStatus.Conflict)\r
  69                     {\r
  70                         Log.InfoFormat("Skipping file in conflict [{0}]", fileInfo.FullName);\r
  71                         return;\r
  72                     }\r
  73                     //Do not upload files when we have no permission\r
  74                     if (latestState.FileStatus == FileStatus.Forbidden)\r
  75                     {\r
  76                         Log.InfoFormat("Skipping forbidden file [{0}]", fileInfo.FullName);\r
  77                         return;\r
  78                     }\r
  79 \r
  80                     //Are we targeting our own account or a sharer account?\r
  81                     var relativePath = fileInfo.AsRelativeTo(action.AccountInfo.AccountPath);\r
  82                     var accountInfo = relativePath.StartsWith(FolderConstants.OthersFolder) \r
  83                                                   ? GetSharerAccount(relativePath, action.AccountInfo) \r
  84                                                   : action.AccountInfo;\r
  85 \r
  86 \r
  87 \r
  88                     var fullFileName = fileInfo.GetProperCapitalization();\r
  89                     using (var gate = NetworkGate.Acquire(fullFileName, NetworkOperation.Uploading))\r
  90                     {\r
  91                         //Abort if the file is already being uploaded or downloaded\r
  92                         if (gate.Failed)\r
  93                             return;\r
  94 \r
  95                         var cloudFile = action.CloudFile;\r
  96                         var account = cloudFile.Account ?? accountInfo.UserName;\r
  97                         try\r
  98                         {\r
  99 \r
 100                             var client = new CloudFilesClient(accountInfo);\r
 101 \r
 102                             //Even if GetObjectInfo times out, we can proceed with the upload            \r
 103                             var cloudInfo = client.GetObjectInfo(account, cloudFile.Container, cloudFile.Name);\r
 104 \r
 105                             //If this a shared file\r
 106                             if (!cloudFile.Account.Equals(action.AccountInfo.UserName,StringComparison.InvariantCultureIgnoreCase))\r
 107                             {\r
 108                                 //If this is a read-only file, do not upload changes\r
 109                                 if (!cloudInfo.IsWritable(action.AccountInfo.UserName))\r
 110                                 {\r
 111                                     MakeFileReadOnly(fullFileName);\r
 112                                     return;\r
 113                                 }\r
 114 \r
 115                                 //If the file is new, can we upload it?\r
 116                                 if ( !cloudInfo.Exists && !client.CanUpload(account, cloudFile))\r
 117                                 {\r
 118                                     MakeFileReadOnly(fullFileName);\r
 119                                     return;\r
 120                                 }\r
 121 \r
 122                             }\r
 123 \r
 124                             await UnpauseEvent.WaitAsync();\r
 125 \r
 126                             if (fileInfo is DirectoryInfo)\r
 127                             {\r
 128                                 //If the directory doesn't exist the Hash property will be empty\r
 129                                 if (String.IsNullOrWhiteSpace(cloudInfo.Hash))\r
 130                                     //Go on and create the directory\r
 131                                     await client.PutObject(account, cloudFile.Container, cloudFile.Name, fullFileName,\r
 132                                                          String.Empty, "application/directory");\r
 133                             }\r
 134                             else\r
 135                             {\r
 136 \r
 137                                 var cloudHash = cloudInfo.Hash.ToLower();\r
 138 \r
 139                                 string topHash;\r
 140                                 TreeHash treeHash;\r
 141                                 using(StatusNotification.GetNotifier("Hashing {0} for Upload", "Finished hashing {0}",fileInfo.Name))\r
 142                                 {\r
 143                                     treeHash = action.TreeHash.Value;\r
 144                                     topHash = treeHash.TopHash.ToHashString();\r
 145                                 }\r
 146 \r
 147                                 //If the file hashes match, abort the upload\r
 148                                 if (cloudInfo != ObjectInfo.Empty && topHash == cloudHash)\r
 149                                 {\r
 150                                     //but store any metadata changes \r
 151                                     StatusKeeper.StoreInfo(fullFileName, cloudInfo);\r
 152                                     Log.InfoFormat("Skip upload of {0}, hashes match", fullFileName);\r
 153                                     return;\r
 154                                 }\r
 155 \r
 156 \r
 157                                 //Mark the file as modified while we upload it\r
 158                                 StatusKeeper.SetFileOverlayStatus(fullFileName, FileOverlayStatus.Modified);\r
 159                                 //And then upload it\r
 160 \r
 161                                 //Upload even small files using the Hashmap. The server may already contain\r
 162                                 //the relevant block                                \r
 163 \r
 164                                 \r
 165 \r
 166                                 await UploadWithHashMap(accountInfo, cloudFile, fileInfo as FileInfo, cloudFile.Name, treeHash,cancellationToken);\r
 167                             }\r
 168                             //If everything succeeds, change the file and overlay status to normal\r
 169                             StatusKeeper.SetFileState(fullFileName, FileStatus.Unchanged, FileOverlayStatus.Normal, "");\r
 170                         }\r
 171                         catch (WebException exc)\r
 172                         {\r
 173                             var response = (exc.Response as HttpWebResponse);\r
 174                             if (response == null)\r
 175                                 throw;\r
 176                             if (response.StatusCode == HttpStatusCode.Forbidden)\r
 177                             {\r
 178                                 StatusKeeper.SetFileState(fileInfo.FullName, FileStatus.Forbidden, FileOverlayStatus.Conflict, "Forbidden");\r
 179                                 MakeFileReadOnly(fullFileName);\r
 180                             }\r
 181                             else\r
 182                                 //In any other case, propagate the error\r
 183                                 throw;\r
 184                         }\r
 185                     }\r
 186                     //Notify the Shell to update the overlays\r
 187                     NativeMethods.RaiseChangeNotification(fullFileName);\r
 188                     StatusNotification.NotifyChangedFile(fullFileName);\r
 189                 }\r
 190                 catch (AggregateException ex)\r
 191                 {\r
 192                     var exc = ex.InnerException as WebException;\r
 193                     if (exc == null)\r
 194                         throw ex.InnerException;\r
 195                     if (HandleUploadWebException(action, exc))\r
 196                         return;\r
 197                     throw;\r
 198                 }\r
 199                 catch (WebException ex)\r
 200                 {\r
 201                     if (HandleUploadWebException(action, ex))\r
 202                         return;\r
 203                     throw;\r
 204                 }\r
 205                 catch (Exception ex)\r
 206                 {\r
 207                     Log.Error("Unexpected error while uploading file", ex);\r
 208                     throw;\r
 209                 }\r
 210             }\r
 211         }\r
 212 \r
 213         private static void MakeFileReadOnly(string fullFileName)\r
 214         {\r
 215             var attributes = File.GetAttributes(fullFileName);\r
 216             //Do not make any modifications if not necessary\r
 217             if (attributes.HasFlag(FileAttributes.ReadOnly))\r
 218                 return;\r
 219             File.SetAttributes(fullFileName, attributes | FileAttributes.ReadOnly);\r
 220         }\r
 221 \r
 222         private static AccountInfo GetSharerAccount(string relativePath, AccountInfo accountInfo)\r
 223         {\r
 224             var parts = relativePath.Split('\\');\r
 225             var accountName = parts[1];\r
 226             var oldName = accountInfo.UserName;\r
 227             var absoluteUri = accountInfo.StorageUri.AbsoluteUri;\r
 228             var nameIndex = absoluteUri.IndexOf(oldName, StringComparison.Ordinal);\r
 229             var root = absoluteUri.Substring(0, nameIndex);\r
 230 \r
 231             accountInfo = new AccountInfo\r
 232                               {\r
 233                                   UserName = accountName,\r
 234                                   AccountPath = Path.Combine(accountInfo.AccountPath, parts[0], parts[1]),\r
 235                                   StorageUri = new Uri(root + accountName),\r
 236                                   BlockHash = accountInfo.BlockHash,\r
 237                                   BlockSize = accountInfo.BlockSize,\r
 238                                   Token = accountInfo.Token\r
 239                               };\r
 240             return accountInfo;\r
 241         }\r
 242 \r
 243 \r
 244         public async Task UploadWithHashMap(AccountInfo accountInfo, ObjectInfo cloudFile, FileInfo fileInfo, string url, TreeHash treeHash, CancellationToken token)\r
 245         {\r
 246             if (accountInfo == null)\r
 247                 throw new ArgumentNullException("accountInfo");\r
 248             if (cloudFile == null)\r
 249                 throw new ArgumentNullException("cloudFile");\r
 250             if (fileInfo == null)\r
 251                 throw new ArgumentNullException("fileInfo");\r
 252             if (String.IsNullOrWhiteSpace(url))\r
 253                 throw new ArgumentNullException(url);\r
 254             if (treeHash == null)\r
 255                 throw new ArgumentNullException("treeHash");\r
 256             if (String.IsNullOrWhiteSpace(cloudFile.Container))\r
 257                 throw new ArgumentException("Invalid container", "cloudFile");\r
 258             Contract.EndContractBlock();\r
 259 \r
 260            \r
 261             using (StatusNotification.GetNotifier("Uploading {0}", "Finished Uploading {0}", fileInfo.Name))\r
 262             {\r
 263                 if (await WaitOrAbort(cloudFile, token)) \r
 264                     return;\r
 265 \r
 266                 var fullFileName = fileInfo.GetProperCapitalization();\r
 267 \r
 268                 var account = cloudFile.Account ?? accountInfo.UserName;\r
 269                 var container = cloudFile.Container;\r
 270 \r
 271 \r
 272                 var client = new CloudFilesClient(accountInfo);\r
 273                 //Send the hashmap to the server            \r
 274                 var missingHashes = await client.PutHashMap(account, container, url, treeHash);\r
 275                 int block = 0;\r
 276                 ReportUploadProgress(fileInfo.Name, block++, missingHashes.Count, fileInfo.Length);\r
 277                 //If the server returns no missing hashes, we are done\r
 278                 while (missingHashes.Count > 0)\r
 279                 {\r
 280 \r
 281                     if (await WaitOrAbort(cloudFile, token))\r
 282                         return;\r
 283 \r
 284 \r
 285                     var buffer = new byte[accountInfo.BlockSize];\r
 286                     foreach (var missingHash in missingHashes)\r
 287                     {\r
 288                         if (await WaitOrAbort(cloudFile, token))\r
 289                             return;\r
 290 \r
 291 \r
 292                         //Find the proper block\r
 293                         var blockIndex = treeHash.HashDictionary[missingHash];\r
 294                         long offset = blockIndex*accountInfo.BlockSize;\r
 295 \r
 296                         var read = fileInfo.Read(buffer, offset, accountInfo.BlockSize);\r
 297 \r
 298                         try\r
 299                         {\r
 300                             //And upload the block                \r
 301                             await client.PostBlock(account, container, buffer, 0, read);\r
 302                             Log.InfoFormat("[BLOCK] Block {0} of {1} uploaded", blockIndex, fullFileName);\r
 303                         }\r
 304                         catch (Exception exc)\r
 305                         {\r
 306                             Log.Error(String.Format("Uploading block {0} of {1}", blockIndex, fullFileName), exc);\r
 307                         }\r
 308                         ReportUploadProgress(fileInfo.Name, block++, missingHashes.Count, fileInfo.Length);\r
 309                     }\r
 310 \r
 311                     token.ThrowIfCancellationRequested();\r
 312                     //Repeat until there are no more missing hashes                \r
 313                     missingHashes = await client.PutHashMap(account, container, url, treeHash);\r
 314                 }\r
 315 \r
 316                 ReportUploadProgress(fileInfo.Name, missingHashes.Count, missingHashes.Count, fileInfo.Length);\r
 317             }\r
 318         }\r
 319 \r
 320         private async Task<bool> WaitOrAbort(ObjectInfo cloudFile, CancellationToken token)\r
 321         {\r
 322             token.ThrowIfCancellationRequested();\r
 323             await UnpauseEvent.WaitAsync();\r
 324             var shouldAbort = !Selectives.IsSelected(cloudFile);\r
 325             if (shouldAbort)\r
 326                 Log.InfoFormat("Aborting [{0}]",cloudFile.Uri);\r
 327             return shouldAbort;\r
 328         }\r
 329 \r
 330         private void ReportUploadProgress(string fileName, int block, int totalBlocks, long fileSize)\r
 331         {\r
 332             StatusNotification.Notify(totalBlocks == 0\r
 333                                           ? new ProgressNotification(fileName, "Uploading", 1, 1, fileSize)\r
 334                                           : new ProgressNotification(fileName, "Uploading", block, totalBlocks, fileSize));\r
 335         }\r
 336 \r
 337 \r
 338         private bool HandleUploadWebException(CloudAction action, WebException exc)\r
 339         {\r
 340             var response = exc.Response as HttpWebResponse;\r
 341             if (response == null)\r
 342                 throw exc;\r
 343             if (response.StatusCode == HttpStatusCode.Unauthorized)\r
 344             {\r
 345                 Log.Error("Not allowed to upload file", exc);\r
 346                 var message = String.Format("Not allowed to uplad file {0}", action.LocalFile.FullName);\r
 347                 StatusKeeper.SetFileState(action.LocalFile.FullName, FileStatus.Unchanged, FileOverlayStatus.Normal, "");\r
 348                 StatusNotification.NotifyChange(message, TraceLevel.Warning);\r
 349                 return true;\r
 350             }\r
 351             return false;\r
 352         }\r
 353 \r
 354         [Import]\r
 355         public Selectives Selectives { get; set; }\r
 356 \r
 357         public AsyncManualResetEvent UnpauseEvent { get; set; }\r
 358     }\r
 359 }\r