Statistics
| Branch: | Revision:

root / trunk / Pithos.Core / Agents / FileAgent.cs @ 5d7b20e8

History | View | Annotate | Download (21.9 kB)

1
#region
2
/* -----------------------------------------------------------------------
3
 * <copyright file="FileAgent.cs" company="GRNet">
4
 * 
5
 * Copyright 2011-2012 GRNET S.A. All rights reserved.
6
 *
7
 * Redistribution and use in source and binary forms, with or
8
 * without modification, are permitted provided that the following
9
 * conditions are met:
10
 *
11
 *   1. Redistributions of source code must retain the above
12
 *      copyright notice, this list of conditions and the following
13
 *      disclaimer.
14
 *
15
 *   2. Redistributions in binary form must reproduce the above
16
 *      copyright notice, this list of conditions and the following
17
 *      disclaimer in the documentation and/or other materials
18
 *      provided with the distribution.
19
 *
20
 *
21
 * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
22
 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
23
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
24
 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
25
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
26
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
27
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
28
 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
29
 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
30
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
31
 * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
32
 * POSSIBILITY OF SUCH DAMAGE.
33
 *
34
 * The views and conclusions contained in the software and
35
 * documentation are those of the authors and should not be
36
 * interpreted as representing official policies, either expressed
37
 * or implied, of GRNET S.A.
38
 * </copyright>
39
 * -----------------------------------------------------------------------
40
 */
41
#endregion
42
using System;
43
using System.Collections.Generic;
44
using System.Diagnostics.Contracts;
45
using System.IO;
46
using System.Linq;
47
using System.Reflection;
48
using System.Threading.Tasks;
49
using Pithos.Interfaces;
50
using Pithos.Network;
51
using log4net;
52

    
53
namespace Pithos.Core.Agents
54
{
55
//    [Export]
56
    public class FileAgent
57
    {
58
        private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
59

    
60
        Agent<WorkflowState> _agent;
61
        private FileSystemWatcher _watcher;
62
        private FileSystemWatcherAdapter _adapter;
63

    
64
        //[Import]
65
        public IStatusKeeper StatusKeeper { get; set; }
66

    
67
        public IStatusNotification StatusNotification { get; set; }
68
        //[Import]
69
        public IPithosWorkflow Workflow { get; set; }
70
        //[Import]
71
        public WorkflowAgent WorkflowAgent { get; set; }
72

    
73
        private AccountInfo AccountInfo { get; set; }
74

    
75
        internal string RootPath { get;  set; }
76
        
77
        private FileEventIdleBatch _eventIdleBatch;
78

    
79
        public TimeSpan IdleTimeout { get; set; }
80

    
81

    
82
        private void ProcessBatchedEvents(Dictionary<string, FileSystemEventArgs[]> fileEvents)
83
        {
84
            StatusNotification.SetPithosStatus(PithosStatus.LocalSyncing,String.Format("Uploading {0} files",fileEvents.Count));
85
            foreach (var fileEvent in fileEvents)
86
            {
87
                var filePath = fileEvent.Key;
88
                var changes = fileEvent.Value;
89
                
90
                if (Ignore(filePath)) continue;
91
                                
92
                foreach (var change in changes)
93
                {
94
                    if (change.ChangeType == WatcherChangeTypes.Renamed)
95
                    {
96
                        var rename = (RenamedEventArgs) change;
97
                        _agent.Post(new WorkflowState
98
                                        {
99
                                            AccountInfo = AccountInfo,
100
                                            OldPath = rename.OldFullPath,
101
                                            OldFileName = rename.OldName,
102
                                            Path = rename.FullPath,
103
                                            FileName = rename.Name,
104
                                            TriggeringChange = rename.ChangeType
105
                                        });
106
                    }
107
                    else
108
                        _agent.Post(new WorkflowState
109
                        {
110
                            AccountInfo = AccountInfo,
111
                            Path = change.FullPath,
112
                            FileName = change.Name,
113
                            TriggeringChange = change.ChangeType
114
                        });                        
115
                }
116
            }
117
            StatusNotification.SetPithosStatus(PithosStatus.LocalComplete);
118
        }
119

    
120
        public void Start(AccountInfo accountInfo,string rootPath)
121
        {
122
            if (accountInfo==null)
123
                throw new ArgumentNullException("accountInfo");
124
            if (String.IsNullOrWhiteSpace(rootPath))
125
                throw new ArgumentNullException("rootPath");
126
            if (!Path.IsPathRooted(rootPath))
127
                throw new ArgumentException("rootPath must be an absolute path","rootPath");
128
            if (IdleTimeout == null)
129
                throw new InvalidOperationException("IdleTimeout must have a valid value");
130
            Contract.EndContractBlock();
131

    
132
            AccountInfo = accountInfo;
133
            RootPath = rootPath;
134

    
135
            _eventIdleBatch = new FileEventIdleBatch((int)IdleTimeout.TotalMilliseconds, ProcessBatchedEvents);
136

    
137
            _watcher = new FileSystemWatcher(rootPath) {IncludeSubdirectories = true,InternalBufferSize=8*4096};
138
            _adapter = new FileSystemWatcherAdapter(_watcher);
139

    
140
            _adapter.Changed += OnFileEvent;
141
            _adapter.Created += OnFileEvent;
142
            _adapter.Deleted += OnFileEvent;
143
            //_adapter.Renamed += OnRenameEvent;
144
            _adapter.Moved += OnMoveEvent;
145
            _watcher.EnableRaisingEvents = true;
146

    
147

    
148
            _agent = Agent<WorkflowState>.Start(inbox =>
149
            {
150
                Action loop = null;
151
                loop = () =>
152
                {
153
                    var message = inbox.Receive();
154
                    var process=message.Then(Process,inbox.CancellationToken);                    
155
                    inbox.LoopAsync(process,loop,ex=>
156
                        Log.ErrorFormat("[ERROR] File Event Processing:\r{0}", ex));
157
                };
158
                loop();
159
            });
160
        }
161

    
162
        private Task<object> Process(WorkflowState state)
163
        {
164
            if (state==null)
165
                throw new ArgumentNullException("state");
166
            Contract.EndContractBlock();
167

    
168
            if (Ignore(state.Path))
169
                return CompletedTask<object>.Default;
170

    
171
            var networkState = NetworkGate.GetNetworkState(state.Path);
172
            //Skip if the file is already being downloaded or uploaded and 
173
            //the change is create or modify
174
            if (networkState != NetworkOperation.None &&
175
                (
176
                    state.TriggeringChange == WatcherChangeTypes.Created ||
177
                    state.TriggeringChange == WatcherChangeTypes.Changed
178
                ))
179
                return CompletedTask<object>.Default;
180

    
181
            try
182
            {
183
                //StatusKeeper.EnsureFileState(state.Path);
184
                
185
                UpdateFileStatus(state);
186
                UpdateOverlayStatus(state);
187
                UpdateFileChecksum(state);
188
                WorkflowAgent.Post(state);
189
            }
190
            catch (IOException exc)
191
            {
192
                if (File.Exists(state.Path))
193
                {
194
                    Log.WarnFormat("File access error occured, retrying {0}\n{1}", state.Path, exc);
195
                    _agent.Post(state);
196
                }
197
                else
198
                {
199
                    Log.WarnFormat("File {0} does not exist. Will be ignored\n{1}", state.Path, exc);
200
                }
201
            }
202
            catch (Exception exc)
203
            {
204
                Log.WarnFormat("Error occured while indexing{0}. The file will be skipped\n{1}",
205
                               state.Path, exc);
206
            }
207
            return CompletedTask<object>.Default;
208
        }
209

    
210
        public bool Pause
211
        {
212
            get { return _watcher == null || !_watcher.EnableRaisingEvents; }
213
            set
214
            {
215
                if (_watcher != null)
216
                    _watcher.EnableRaisingEvents = !value;                
217
            }
218
        }
219

    
220
        public string CachePath { get; set; }
221

    
222
        private List<string> _selectivePaths = new List<string>();
223
        public List<string> SelectivePaths
224
        {
225
            get { return _selectivePaths; }
226
            set { _selectivePaths = value; }
227
        }
228

    
229

    
230
        public void Post(WorkflowState workflowState)
231
        {
232
            if (workflowState == null)
233
                throw new ArgumentNullException("workflowState");
234
            Contract.EndContractBlock();
235

    
236
            _agent.Post(workflowState);
237
        }
238

    
239
        public void Stop()
240
        {
241
            if (_watcher != null)
242
            {
243
                _watcher.Dispose();
244
            }
245
            _watcher = null;
246

    
247
            if (_agent!=null)
248
                _agent.Stop();
249
        }
250

    
251
        // Enumerate all files in the Pithos directory except those in the Fragment folder
252
        // and files with a .ignore extension
253
        public IEnumerable<string> EnumerateFiles(string searchPattern="*")
254
        {
255
            var monitoredFiles = from filePath in Directory.EnumerateFileSystemEntries(RootPath, searchPattern, SearchOption.AllDirectories)
256
                                 where !Ignore(filePath)
257
                                 select filePath;
258
            return monitoredFiles;
259
        }
260

    
261
        public IEnumerable<FileInfo> EnumerateFileInfos(string searchPattern="*")
262
        {
263
            var rootDir = new DirectoryInfo(RootPath);
264
            var monitoredFiles = from file in rootDir.EnumerateFiles(searchPattern, SearchOption.AllDirectories)
265
                                 where !Ignore(file.FullName)
266
                                 select file;
267
            return monitoredFiles;
268
        }                
269

    
270
        public IEnumerable<string> EnumerateFilesAsRelativeUrls(string searchPattern="*")
271
        {
272
            var rootDir = new DirectoryInfo(RootPath);
273
            var monitoredFiles = from file in rootDir.EnumerateFiles(searchPattern, SearchOption.AllDirectories)
274
                                 where !Ignore(file.FullName)
275
                                 select file.AsRelativeUrlTo(RootPath);
276
            return monitoredFiles;
277
        }                
278

    
279
        public IEnumerable<string> EnumerateFilesSystemInfosAsRelativeUrls(string searchPattern="*")
280
        {
281
            var rootDir = new DirectoryInfo(RootPath);
282
            var monitoredFiles = from file in rootDir.EnumerateFileSystemInfos(searchPattern, SearchOption.AllDirectories)
283
                                 where !Ignore(file.FullName)
284
                                 select file.AsRelativeUrlTo(RootPath);
285
            return monitoredFiles;
286
        }                
287

    
288

    
289
        
290

    
291
        private bool Ignore(string filePath)
292
        {
293
            //Ignore all first-level directories and files (ie at the container folders level)
294
            if (FoundBelowRoot(filePath, RootPath,1))
295
                return true;
296

    
297
            //Ignore first-level items under the "others" folder (ie at the accounts folders level).
298
            var othersPath = Path.Combine(RootPath, FolderConstants.OthersFolder);
299
            if (FoundBelowRoot(filePath, othersPath,1))
300
                return true;
301

    
302
            //Ignore second-level (container) folders under the "others" folder (ie at the container folders level). 
303
            if (FoundBelowRoot(filePath, othersPath,2))
304
                return true;            
305

    
306

    
307
            //Ignore anything happening in the cache path
308
            if (filePath.StartsWith(CachePath))
309
                return true;
310
            if (_ignoreFiles.ContainsKey(filePath.ToLower()))
311
                return true;
312

    
313
            //Ignore if selective synchronization is defined, 
314
            return SelectivePaths.Count > 0 
315
                //And the target file is not below any of the selective paths
316
                && !SelectivePaths.Any(filePath.IsAtOrDirectlyBelow);
317
        }
318

    
319
/*        private static bool FoundInRoot(string filePath, string rootPath)
320
        {
321
            //var rootDirectory = new DirectoryInfo(rootPath);
322

    
323
            //If the paths are equal, return true
324
            if (filePath.Equals(rootPath, StringComparison.InvariantCultureIgnoreCase))
325
                return true;
326

    
327
            //If the filepath is below the root path
328
            if (filePath.StartsWith(rootPath,StringComparison.InvariantCulture))
329
            {
330
                //Get the relative path
331
                var relativePath = filePath.Substring(rootPath.Length + 1);
332
                //If the relativePath does NOT contains a path separator, we found a match
333
                return (!relativePath.Contains(@"\"));
334
            }
335

    
336
            //If the filepath is not under the root path, return false
337
            return false;            
338
        }*/
339

    
340

    
341
        private static bool FoundBelowRoot(string filePath, string rootPath,int level)
342
        {
343
            //var rootDirectory = new DirectoryInfo(rootPath);
344

    
345
            //If the paths are equal, return true
346
            if (filePath.Equals(rootPath, StringComparison.InvariantCultureIgnoreCase))
347
                return true;
348

    
349
            //If the filepath is below the root path
350
            if (filePath.StartsWith(rootPath,StringComparison.InvariantCulture))
351
            {
352
                //Get the relative path
353
                var relativePath = filePath.Substring(rootPath.Length + 1);
354
                //If the relativePath does NOT contains a path separator, we found a match
355
                var levels=relativePath.ToCharArray().Count(c=>c=='\\')+1;                
356
                return levels==level;
357
            }
358

    
359
            //If the filepath is not under the root path, return false
360
            return false;            
361
        }
362

    
363
        //Post a Change message for all events except rename
364
        void OnFileEvent(object sender, FileSystemEventArgs e)
365
        {
366
            //Ignore events that affect the cache folder
367
            var filePath = e.FullPath;
368
            if (Ignore(filePath)) 
369
                return;
370
            _eventIdleBatch.Post(e);
371
        }
372

    
373

    
374
/*
375
        //Post a Change message for renames containing the old and new names
376
        void OnRenameEvent(object sender, RenamedEventArgs e)
377
        {
378
            var oldFullPath = e.OldFullPath;
379
            var fullPath = e.FullPath;
380
            if (Ignore(oldFullPath) || Ignore(fullPath))
381
                return;
382

    
383
            _agent.Post(new WorkflowState
384
            {
385
                AccountInfo=AccountInfo,
386
                OldPath = oldFullPath,
387
                OldFileName = e.OldName,
388
                Path = fullPath,
389
                FileName = e.Name,
390
                TriggeringChange = e.ChangeType
391
            });
392
        }
393
*/
394

    
395
        //Post a Change message for moves containing the old and new names
396
        void OnMoveEvent(object sender, MovedEventArgs e)
397
        {
398
            var oldFullPath = e.OldFullPath;
399
            var fullPath = e.FullPath;
400
            if (Ignore(oldFullPath) || Ignore(fullPath))
401
                return;
402

    
403
            _eventIdleBatch.Post(e);
404
        }
405

    
406

    
407

    
408
        private Dictionary<WatcherChangeTypes, FileStatus> _statusDict = new Dictionary<WatcherChangeTypes, FileStatus>
409
                                                                             {
410
            {WatcherChangeTypes.Created,FileStatus.Created},
411
            {WatcherChangeTypes.Changed,FileStatus.Modified},
412
            {WatcherChangeTypes.Deleted,FileStatus.Deleted},
413
            {WatcherChangeTypes.Renamed,FileStatus.Renamed}
414
        };
415

    
416
        private Dictionary<string, string> _ignoreFiles=new Dictionary<string, string>();
417

    
418
        private WorkflowState UpdateFileStatus(WorkflowState state)
419
        {
420
            if (state==null)
421
                throw new ArgumentNullException("state");
422
            if (String.IsNullOrWhiteSpace(state.Path))
423
                throw new ArgumentException("The state's Path can't be empty","state");
424
            Contract.EndContractBlock();
425

    
426
            var path = state.Path;
427
            var status = _statusDict[state.TriggeringChange];
428
            var oldStatus = Workflow.StatusKeeper.GetFileStatus(path);
429
            if (status == oldStatus)
430
            {
431
                state.Status = status;
432
                state.Skip = true;
433
                return state;
434
            }
435
            if (state.Status == FileStatus.Renamed)
436
                Workflow.ClearFileStatus(path);
437

    
438
            state.Status = Workflow.SetFileStatus(path, status);
439
            return state;
440
        }
441

    
442
        private WorkflowState UpdateOverlayStatus(WorkflowState state)
443
        {
444
            if (state==null)
445
                throw new ArgumentNullException("state");
446
            Contract.EndContractBlock();
447

    
448
            if (state.Skip)
449
                return state;
450

    
451
            switch (state.Status)
452
            {
453
                case FileStatus.Created:
454
                    this.StatusKeeper.SetFileOverlayStatus(state.Path, FileOverlayStatus.Modified,state.ShortHash);
455
                    break;
456
                case FileStatus.Modified:
457
                    this.StatusKeeper.SetFileOverlayStatus(state.Path, FileOverlayStatus.Modified,state.ShortHash);
458
                    break;
459
                case FileStatus.Deleted:
460
                    //this.StatusAgent.RemoveFileOverlayStatus(state.Path);
461
                    break;
462
                case FileStatus.Renamed:
463
                    this.StatusKeeper.ClearFileStatus(state.OldPath);
464
                    this.StatusKeeper.SetFileOverlayStatus(state.Path, FileOverlayStatus.Modified,state.ShortHash);
465
                    break;
466
                case FileStatus.Unchanged:
467
                    this.StatusKeeper.SetFileOverlayStatus(state.Path, FileOverlayStatus.Normal,state.ShortHash);
468
                    break;
469
            }
470

    
471
            if (state.Status == FileStatus.Deleted)
472
                NativeMethods.RaiseChangeNotification(Path.GetDirectoryName(state.Path));
473
            else
474
                NativeMethods.RaiseChangeNotification(state.Path);
475
            return state;
476
        }
477

    
478

    
479
        private WorkflowState UpdateFileChecksum(WorkflowState state)
480
        {
481
            if (state.Skip)
482
                return state;
483

    
484
            if (state.Status == FileStatus.Deleted)
485
                return state;
486

    
487
            var path = state.Path;
488
            //Skip calculation for folders
489
            if (Directory.Exists(path))
490
                return state;
491

    
492

    
493
            var info = new FileInfo(path);
494
            StatusNotification.Notify(new StatusNotification(String.Format("Hashing [{0}]",info.Name)));
495

    
496
            var shortHash = info.ComputeShortHash(); 
497
            
498
            string merkleHash = info.CalculateHash(StatusKeeper.BlockSize,StatusKeeper.BlockHash);
499
            StatusKeeper.UpdateFileChecksum(path,shortHash, merkleHash);
500

    
501
            state.Hash = merkleHash;
502
            return state;
503
        }
504

    
505
        //Does the file exist in the container's local folder?
506
        public bool Exists(string relativePath)
507
        {
508
            if (String.IsNullOrWhiteSpace(relativePath))
509
                throw new ArgumentNullException("relativePath");
510
            //A RootPath must be set before calling this method
511
            if (String.IsNullOrWhiteSpace(RootPath))
512
                throw new InvalidOperationException("RootPath was not set");
513
            Contract.EndContractBlock();
514
            //Create the absolute path by combining the RootPath with the relativePath
515
            var absolutePath=Path.Combine(RootPath, relativePath);
516
            //Is this a valid file?
517
            if (File.Exists(absolutePath))
518
                return true;
519
            //Or a directory?
520
            if (Directory.Exists(absolutePath))
521
                return true;
522
            //Fail if it is neither
523
            return false;
524
        }
525

    
526
        public static FileAgent GetFileAgent(AccountInfo accountInfo)
527
        {
528
            return GetFileAgent(accountInfo.AccountPath);
529
        }
530

    
531
        public static FileAgent GetFileAgent(string rootPath)
532
        {
533
            return AgentLocator<FileAgent>.Get(rootPath.ToLower());
534
        }
535

    
536

    
537
        public FileSystemInfo GetFileSystemInfo(string relativePath)
538
        {
539
            if (String.IsNullOrWhiteSpace(relativePath))
540
                throw new ArgumentNullException("relativePath");
541
            //A RootPath must be set before calling this method
542
            if (String.IsNullOrWhiteSpace(RootPath))
543
                throw new InvalidOperationException("RootPath was not set");            
544
            Contract.EndContractBlock();            
545

    
546
            var absolutePath = Path.Combine(RootPath, relativePath);
547

    
548
            if (Directory.Exists(absolutePath))
549
                return new DirectoryInfo(absolutePath).WithProperCapitalization();
550
            else
551
                return new FileInfo(absolutePath).WithProperCapitalization();
552
            
553
        }
554

    
555
        public void Delete(string relativePath)
556
        {
557
            var absolutePath = Path.Combine(RootPath, relativePath).ToLower();
558
            if (File.Exists(absolutePath))
559
            {    
560
                try
561
                {
562
                    File.Delete(absolutePath);
563
                }
564
                //The file may have been deleted by another thread. Just ignore the relevant exception
565
                catch (FileNotFoundException) { }
566
            }
567
            else if (Directory.Exists(absolutePath))
568
            {
569
                try
570
                {
571
                    Directory.Delete(absolutePath, true);
572
                }
573
                //The directory may have been deleted by another thread. Just ignore the relevant exception
574
                catch (DirectoryNotFoundException){}                
575
            }
576
        
577
            //_ignoreFiles[absolutePath] = absolutePath;                
578
            StatusKeeper.ClearFileStatus(absolutePath);
579
        }
580
    }
581
}