Statistics
| Branch: | Revision:

root / trunk / Pithos.Core / Agents / Uploader.cs @ 5e646964

History | View | Annotate | Download (16.1 kB)

1
using System;
2
using System.Collections.Generic;
3
using System.ComponentModel.Composition;
4
using System.Diagnostics;
5
using System.Diagnostics.Contracts;
6
using System.IO;
7
using System.Linq;
8
using System.Net;
9
using System.Reflection;
10
using System.Threading;
11
using System.Threading.Tasks;
12
using Pithos.Interfaces;
13
using Pithos.Network;
14
using log4net;
15

    
16
namespace Pithos.Core.Agents
17
{
18
    [Export(typeof(Uploader))]
19
    public class Uploader
20
    {
21
        private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
22

    
23
        [Import]
24
        private IStatusKeeper StatusKeeper { get; set; }
25

    
26
        
27
        public IStatusNotification StatusNotification { get; set; }
28

    
29
        
30
        //CancellationTokenSource _cts = new CancellationTokenSource();
31
        /*public void SignalStop()
32
        {
33
            _cts.Cancel();
34
        }*/
35

    
36
        public async Task UploadCloudFile(CloudAction action,CancellationToken cancellationToken)
37
        {
38
            if (action == null)
39
                throw new ArgumentNullException("action");
40
            Contract.EndContractBlock();
41

    
42
            using (ThreadContext.Stacks["Operation"].Push("UploadCloudFile"))
43
            {
44
                try
45
                {
46
                    await UnpauseEvent.WaitAsync();
47

    
48
                    var fileInfo = action.LocalFile;
49

    
50
                    if (fileInfo.Extension.Equals("ignore", StringComparison.InvariantCultureIgnoreCase))
51
                        return;
52

    
53
                    if (!Selectives.IsSelected(action.AccountInfo, fileInfo))
54
                        return;
55

    
56
                    //Try to load the action's local state, if it is empty
57
                    if (action.FileState == null)
58
                        action.FileState = StatusKeeper.GetStateByFilePath(fileInfo.FullName);
59
                    if (action.FileState == null)
60
                    {
61
                        Log.WarnFormat("File [{0}] has no local state. It was probably created by a download action", fileInfo.FullName);
62
                        return;
63
                    }
64

    
65
                    var latestState = action.FileState;
66

    
67
                    //Do not upload files in conflict
68
                    if (latestState.FileStatus == FileStatus.Conflict)
69
                    {
70
                        Log.InfoFormat("Skipping file in conflict [{0}]", fileInfo.FullName);
71
                        return;
72
                    }
73
                    //Do not upload files when we have no permission
74
                    if (latestState.FileStatus == FileStatus.Forbidden)
75
                    {
76
                        Log.InfoFormat("Skipping forbidden file [{0}]", fileInfo.FullName);
77
                        return;
78
                    }
79

    
80
                    //Are we targeting our own account or a sharer account?
81
                    var relativePath = fileInfo.AsRelativeTo(action.AccountInfo.AccountPath);
82
                    var accountInfo = relativePath.StartsWith(FolderConstants.OthersFolder) 
83
                                                  ? GetSharerAccount(relativePath, action.AccountInfo) 
84
                                                  : action.AccountInfo;
85

    
86

    
87

    
88
                    var fullFileName = fileInfo.GetProperCapitalization();
89
                    using (var gate = NetworkGate.Acquire(fullFileName, NetworkOperation.Uploading))
90
                    {
91
                        //Abort if the file is already being uploaded or downloaded
92
                        if (gate.Failed)
93
                            return;
94

    
95
                        var cloudFile = action.CloudFile;
96
                        var account = cloudFile.Account ?? accountInfo.UserName;
97
                        try
98
                        {
99

    
100
                            var client = new CloudFilesClient(accountInfo);
101

    
102
                            //Even if GetObjectInfo times out, we can proceed with the upload            
103
                            var cloudInfo = client.GetObjectInfo(account, cloudFile.Container, cloudFile.Name);
104

    
105
                            //If this a shared file
106
                            if (!cloudFile.Account.Equals(action.AccountInfo.UserName,StringComparison.InvariantCultureIgnoreCase))
107
                            {
108
                                //If this is a read-only file, do not upload changes
109
                                if (!cloudInfo.IsWritable(action.AccountInfo.UserName))
110
                                {
111
                                    MakeFileReadOnly(fullFileName);
112
                                    return;
113
                                }
114

    
115
                                //If the file is new, can we upload it?
116
                                if ( !cloudInfo.Exists && !client.CanUpload(account, cloudFile))
117
                                {
118
                                    MakeFileReadOnly(fullFileName);
119
                                    return;
120
                                }
121

    
122
                            }
123

    
124
                            await UnpauseEvent.WaitAsync();
125

    
126
                            if (fileInfo is DirectoryInfo)
127
                            {
128
                                //If the directory doesn't exist the Hash property will be empty
129
                                if (String.IsNullOrWhiteSpace(cloudInfo.Hash))
130
                                    //Go on and create the directory
131
                                    await client.PutObject(account, cloudFile.Container, cloudFile.Name, fullFileName,
132
                                                         String.Empty, "application/directory");
133
                            }
134
                            else
135
                            {
136

    
137
                                var cloudHash = cloudInfo.Hash.ToLower();
138

    
139
                                string topHash;
140
                                TreeHash treeHash;
141
                                using(StatusNotification.GetNotifier("Hashing {0} for Upload", "Finished hashing {0}",fileInfo.Name))
142
                                {
143
                                    treeHash = action.TreeHash.Value;
144
                                    topHash = treeHash.TopHash.ToHashString();
145
                                }
146

    
147
                                //If the file hashes match, abort the upload
148
                                if (cloudInfo != ObjectInfo.Empty && topHash == cloudHash)
149
                                {
150
                                    //but store any metadata changes 
151
                                    StatusKeeper.StoreInfo(fullFileName, cloudInfo);
152
                                    Log.InfoFormat("Skip upload of {0}, hashes match", fullFileName);
153
                                    return;
154
                                }
155

    
156

    
157
                                //Mark the file as modified while we upload it
158
                                StatusKeeper.SetFileOverlayStatus(fullFileName, FileOverlayStatus.Modified);
159
                                //And then upload it
160

    
161
                                //Upload even small files using the Hashmap. The server may already contain
162
                                //the relevant block                                
163

    
164
                                
165

    
166
                                await UploadWithHashMap(accountInfo, cloudFile, fileInfo as FileInfo, cloudFile.Name, treeHash,cancellationToken);
167
                            }
168
                            //If everything succeeds, change the file and overlay status to normal
169
                            StatusKeeper.SetFileState(fullFileName, FileStatus.Unchanged, FileOverlayStatus.Normal, "");
170
                        }
171
                        catch (WebException exc)
172
                        {
173
                            var response = (exc.Response as HttpWebResponse);
174
                            if (response == null)
175
                                throw;
176
                            if (response.StatusCode == HttpStatusCode.Forbidden)
177
                            {
178
                                StatusKeeper.SetFileState(fileInfo.FullName, FileStatus.Forbidden, FileOverlayStatus.Conflict, "Forbidden");
179
                                MakeFileReadOnly(fullFileName);
180
                            }
181
                            else
182
                                //In any other case, propagate the error
183
                                throw;
184
                        }
185
                    }
186
                    //Notify the Shell to update the overlays
187
                    NativeMethods.RaiseChangeNotification(fullFileName);
188
                    StatusNotification.NotifyChangedFile(fullFileName);
189
                }
190
                catch (AggregateException ex)
191
                {
192
                    var exc = ex.InnerException as WebException;
193
                    if (exc == null)
194
                        throw ex.InnerException;
195
                    if (HandleUploadWebException(action, exc))
196
                        return;
197
                    throw;
198
                }
199
                catch (WebException ex)
200
                {
201
                    if (HandleUploadWebException(action, ex))
202
                        return;
203
                    throw;
204
                }
205
                catch (Exception ex)
206
                {
207
                    Log.Error("Unexpected error while uploading file", ex);
208
                    throw;
209
                }
210
            }
211
        }
212

    
213
        private static void MakeFileReadOnly(string fullFileName)
214
        {
215
            var attributes = File.GetAttributes(fullFileName);
216
            //Do not make any modifications if not necessary
217
            if (attributes.HasFlag(FileAttributes.ReadOnly))
218
                return;
219
            File.SetAttributes(fullFileName, attributes | FileAttributes.ReadOnly);
220
        }
221

    
222
        private static AccountInfo GetSharerAccount(string relativePath, AccountInfo accountInfo)
223
        {
224
            var parts = relativePath.Split('\\');
225
            var accountName = parts[1];
226
            var oldName = accountInfo.UserName;
227
            var absoluteUri = accountInfo.StorageUri.AbsoluteUri;
228
            var nameIndex = absoluteUri.IndexOf(oldName, StringComparison.Ordinal);
229
            var root = absoluteUri.Substring(0, nameIndex);
230

    
231
            accountInfo = new AccountInfo
232
                              {
233
                                  UserName = accountName,
234
                                  AccountPath = Path.Combine(accountInfo.AccountPath, parts[0], parts[1]),
235
                                  StorageUri = new Uri(root + accountName),
236
                                  BlockHash = accountInfo.BlockHash,
237
                                  BlockSize = accountInfo.BlockSize,
238
                                  Token = accountInfo.Token
239
                              };
240
            return accountInfo;
241
        }
242

    
243

    
244
        public async Task UploadWithHashMap(AccountInfo accountInfo, ObjectInfo cloudFile, FileInfo fileInfo, string url, TreeHash treeHash, CancellationToken token)
245
        {
246
            if (accountInfo == null)
247
                throw new ArgumentNullException("accountInfo");
248
            if (cloudFile == null)
249
                throw new ArgumentNullException("cloudFile");
250
            if (fileInfo == null)
251
                throw new ArgumentNullException("fileInfo");
252
            if (String.IsNullOrWhiteSpace(url))
253
                throw new ArgumentNullException(url);
254
            if (treeHash == null)
255
                throw new ArgumentNullException("treeHash");
256
            if (String.IsNullOrWhiteSpace(cloudFile.Container))
257
                throw new ArgumentException("Invalid container", "cloudFile");
258
            Contract.EndContractBlock();
259

    
260
           
261
            using (StatusNotification.GetNotifier("Uploading {0}", "Finished Uploading {0}", fileInfo.Name))
262
            {
263
                if (await WaitOrAbort(accountInfo,cloudFile, token)) 
264
                    return;
265

    
266
                var fullFileName = fileInfo.GetProperCapitalization();
267

    
268
                var account = cloudFile.Account ?? accountInfo.UserName;
269
                var container = cloudFile.Container;
270

    
271

    
272
                var client = new CloudFilesClient(accountInfo);
273
                //Send the hashmap to the server            
274
                var missingHashes = await client.PutHashMap(account, container, url, treeHash);
275
                int block = 0;
276
                ReportUploadProgress(fileInfo.Name, block++, missingHashes.Count, fileInfo.Length);
277
                //If the server returns no missing hashes, we are done
278
                while (missingHashes.Count > 0)
279
                {
280

    
281
                    if (await WaitOrAbort(accountInfo,cloudFile, token))
282
                        return;
283

    
284

    
285
                    var buffer = new byte[accountInfo.BlockSize];
286
                    foreach (var missingHash in missingHashes)
287
                    {
288
                        if (await WaitOrAbort(accountInfo,cloudFile, token))
289
                            return;
290

    
291

    
292
                        //Find the proper block
293
                        var blockIndex = treeHash.HashDictionary[missingHash];
294
                        long offset = blockIndex*accountInfo.BlockSize;
295

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

    
298
                        try
299
                        {
300
                            //And upload the block                
301
                            await client.PostBlock(account, container, buffer, 0, read);
302
                            Log.InfoFormat("[BLOCK] Block {0} of {1} uploaded", blockIndex, fullFileName);
303
                        }
304
                        catch (Exception exc)
305
                        {
306
                            Log.Error(String.Format("Uploading block {0} of {1}", blockIndex, fullFileName), exc);
307
                        }
308
                        ReportUploadProgress(fileInfo.Name, block++, missingHashes.Count, fileInfo.Length);
309
                    }
310

    
311
                    token.ThrowIfCancellationRequested();
312
                    //Repeat until there are no more missing hashes                
313
                    missingHashes = await client.PutHashMap(account, container, url, treeHash);
314
                }
315

    
316
                ReportUploadProgress(fileInfo.Name, missingHashes.Count, missingHashes.Count, fileInfo.Length);
317
            }
318
        }
319

    
320
        private async Task<bool> WaitOrAbort(AccountInfo account,ObjectInfo cloudFile, CancellationToken token)
321
        {
322
            token.ThrowIfCancellationRequested();
323
            await UnpauseEvent.WaitAsync();
324
            var shouldAbort = !Selectives.IsSelected(account,cloudFile);
325
            if (shouldAbort)
326
                Log.InfoFormat("Aborting [{0}]",cloudFile.Uri);
327
            return shouldAbort;
328
        }
329

    
330
        private void ReportUploadProgress(string fileName, int block, int totalBlocks, long fileSize)
331
        {
332
            StatusNotification.Notify(totalBlocks == 0
333
                                          ? new ProgressNotification(fileName, "Uploading", 1, 1, fileSize)
334
                                          : new ProgressNotification(fileName, "Uploading", block, totalBlocks, fileSize));
335
        }
336

    
337

    
338
        private bool HandleUploadWebException(CloudAction action, WebException exc)
339
        {
340
            var response = exc.Response as HttpWebResponse;
341
            if (response == null)
342
                throw exc;
343
            if (response.StatusCode == HttpStatusCode.Unauthorized)
344
            {
345
                Log.Error("Not allowed to upload file", exc);
346
                var message = String.Format("Not allowed to uplad file {0}", action.LocalFile.FullName);
347
                StatusKeeper.SetFileState(action.LocalFile.FullName, FileStatus.Unchanged, FileOverlayStatus.Normal, "");
348
                StatusNotification.NotifyChange(message, TraceLevel.Warning);
349
                return true;
350
            }
351
            return false;
352
        }
353

    
354
        [Import]
355
        public Selectives Selectives { get; set; }
356

    
357
        public AsyncManualResetEvent UnpauseEvent { get; set; }
358
    }
359
}