using System.IO;
using System.Linq;
using System.Text;
+using System.Threading.Tasks;
using Pithos.Interfaces;
using Pithos.Network;
+using log4net;
+using log4net.Core;
namespace Pithos.Core.Agents
{
- [Export]
+// [Export]
public class FileAgent
{
Agent<WorkflowState> _agent;
private FileSystemWatcher _watcher;
- [Import]
+ //[Import]
public IStatusKeeper StatusKeeper { get; set; }
- [Import]
+ //[Import]
public IPithosWorkflow Workflow { get; set; }
- [Import]
+ //[Import]
public WorkflowAgent WorkflowAgent { get; set; }
- public string RootPath { get; private set; }
+ private AccountInfo AccountInfo { get; set; }
- public void Start(string rootPath)
+ private string RootPath { get; set; }
+
+ private static readonly ILog Log = LogManager.GetLogger("FileAgent");
+
+ public void Start(AccountInfo accountInfo,string rootPath)
{
+ if (accountInfo==null)
+ throw new ArgumentNullException("accountInfo");
+ if (String.IsNullOrWhiteSpace(rootPath))
+ throw new ArgumentNullException("rootPath");
+ if (!Path.IsPathRooted(rootPath))
+ throw new ArgumentException("rootPath must be an absolute path","rootPath");
+ Contract.EndContractBlock();
+
+ AccountInfo = accountInfo;
RootPath = rootPath;
_watcher = new FileSystemWatcher(rootPath);
+ _watcher.IncludeSubdirectories = true;
_watcher.Changed += OnFileEvent;
_watcher.Created += OnFileEvent;
_watcher.Deleted += OnFileEvent;
loop = () =>
{
var message = inbox.Receive();
- var process = message.ContinueWith(t =>
- {
- var state = t.Result;
- Process(state);
- inbox.DoAsync(loop);
- });
-
- process.ContinueWith(t =>
- {
- inbox.DoAsync(loop);
- if (t.IsFaulted)
- {
- var ex = t.Exception.InnerException;
- if (ex is OperationCanceledException)
- inbox.Stop();
- Trace.TraceError("[ERROR] File Event Processing:\r{0}", ex);
- }
- });
+ var process=message.Then(Process,inbox.CancellationToken);
+ inbox.LoopAsync(process,loop,ex=>
+ Log.ErrorFormat("[ERROR] File Event Processing:\r{0}", ex));
};
loop();
});
}
+ private Task<object> Process(WorkflowState state)
+ {
+ if (state==null)
+ throw new ArgumentNullException("state");
+ Contract.EndContractBlock();
+
+ if (Ignore(state.Path))
+ return CompletedTask<object>.Default;
+
+ var networkState = NetworkGate.GetNetworkState(state.Path);
+ //Skip if the file is already being downloaded or uploaded and
+ //the change is create or modify
+ if (networkState != NetworkOperation.None &&
+ (
+ state.TriggeringChange == WatcherChangeTypes.Created ||
+ state.TriggeringChange == WatcherChangeTypes.Changed
+ ))
+ return CompletedTask<object>.Default;
+
+ try
+ {
+ UpdateFileStatus(state);
+ UpdateOverlayStatus(state);
+ UpdateFileChecksum(state);
+ WorkflowAgent.Post(state);
+ }
+ catch (IOException exc)
+ {
+ if (File.Exists(state.Path))
+ {
+ Log.WarnFormat("File access error occured, retrying {0}\n{1}", state.Path, exc);
+ _agent.Post(state);
+ }
+ else
+ {
+ Log.WarnFormat("File {0} does not exist. Will be ignored\n{1}", state.Path, exc);
+ }
+ }
+ catch (Exception exc)
+ {
+ Log.WarnFormat("Error occured while indexing{0}. The file will be skipped\n{1}",
+ state.Path, exc);
+ }
+ return CompletedTask<object>.Default;
+ }
+
public bool Pause
{
get { return _watcher == null || !_watcher.EnableRaisingEvents; }
}
}
- public string FragmentsPath { get; set; }
+ public string CachePath { get; set; }
+
+ private List<string> _selectivePaths = new List<string>();
+ public List<string> SelectivePaths
+ {
+ get { return _selectivePaths; }
+ set { _selectivePaths = value; }
+ }
+
public void Post(WorkflowState workflowState)
{
+ if (workflowState == null)
+ throw new ArgumentNullException("workflowState");
+ Contract.EndContractBlock();
+
_agent.Post(workflowState);
}
}
_watcher = null;
- _agent.Stop();
+ if (_agent!=null)
+ _agent.Stop();
}
// Enumerate all files in the Pithos directory except those in the Fragment folder
private bool Ignore(string filePath)
{
- if (filePath.StartsWith(FragmentsPath))
+ var pithosPath = Path.Combine(RootPath, "pithos");
+ if (pithosPath.Equals(filePath, StringComparison.InvariantCultureIgnoreCase))
+ return true;
+ if (filePath.StartsWith(CachePath))
+ return true;
+ if (_ignoreFiles.ContainsKey(filePath.ToLower()))
return true;
return false;
}
//Post a Change message for all events except rename
void OnFileEvent(object sender, FileSystemEventArgs e)
{
- //Ignore events that affect the Fragments folder
+ //Ignore events that affect the cache folder
var filePath = e.FullPath;
if (Ignore(filePath))
return;
- _agent.Post(new WorkflowState { Path = filePath, FileName = e.Name, TriggeringChange = e.ChangeType });
+
+ _agent.Post(new WorkflowState{AccountInfo=AccountInfo, Path = filePath, FileName = e.Name, TriggeringChange = e.ChangeType });
}
_agent.Post(new WorkflowState
{
+ AccountInfo=AccountInfo,
OldPath = oldFullPath,
OldFileName = e.OldName,
Path = fullPath,
}
- private void Process(WorkflowState state)
- {
- Debug.Assert(!Ignore(state.Path));
-
- var networkState = NetworkGate.GetNetworkState(state.Path);
- //Skip if the file is already being downloaded or uploaded and
- //the change is create or modify
- if (networkState != NetworkOperation.None &&
- (
- state.TriggeringChange == WatcherChangeTypes.Created ||
- state.TriggeringChange == WatcherChangeTypes.Changed
- ))
- return;
-
- try
- {
- UpdateFileStatus(state);
- UpdateOverlayStatus(state);
- UpdateFileChecksum(state);
- WorkflowAgent.Post(state);
- }
- catch (IOException exc)
- {
- Trace.TraceWarning("File access error occured, retrying {0}\n{1}", state.Path, exc);
- _agent.Post(state);
- }
- catch (Exception exc)
- {
- Trace.TraceWarning("Error occured while indexing{0. The file will be skipped}\n{1}", state.Path, exc);
- }
- }
private Dictionary<WatcherChangeTypes, FileStatus> _statusDict = new Dictionary<WatcherChangeTypes, FileStatus>
{
{WatcherChangeTypes.Renamed,FileStatus.Renamed}
};
+ private Dictionary<string,string> _ignoreFiles=new Dictionary<string, string>();
+
private WorkflowState UpdateFileStatus(WorkflowState state)
{
- Debug.Assert(!state.Path.Contains("fragments"));
- Debug.Assert(!state.Path.EndsWith(".ignore", StringComparison.InvariantCultureIgnoreCase));
+ if (state==null)
+ throw new ArgumentNullException("state");
+ if (String.IsNullOrWhiteSpace(state.Path))
+ throw new ArgumentException("The state's Path can't be empty","state");
+ Contract.EndContractBlock();
- string path = state.Path;
- FileStatus status = _statusDict[state.TriggeringChange];
+ var path = state.Path;
+ var status = _statusDict[state.TriggeringChange];
var oldStatus = Workflow.StatusKeeper.GetFileStatus(path);
if (status == oldStatus)
{
private WorkflowState UpdateOverlayStatus(WorkflowState state)
{
+ if (state==null)
+ throw new ArgumentNullException("state");
+ Contract.EndContractBlock();
+
if (state.Skip)
return state;
this.StatusKeeper.SetFileOverlayStatus(state.Path, FileOverlayStatus.Modified);
break;
case FileStatus.Deleted:
- this.StatusKeeper.RemoveFileOverlayStatus(state.Path);
+ //this.StatusAgent.RemoveFileOverlayStatus(state.Path);
break;
case FileStatus.Renamed:
- this.StatusKeeper.RemoveFileOverlayStatus(state.OldPath);
+ this.StatusKeeper.ClearFileStatus(state.OldPath);
this.StatusKeeper.SetFileOverlayStatus(state.Path, FileOverlayStatus.Modified);
break;
case FileStatus.Unchanged:
if (Directory.Exists(path))
return state;
- string hash = Signature.CalculateMD5(path);
-
+ var info = new FileInfo(path);
+ string hash = info.CalculateHash(StatusKeeper.BlockSize,StatusKeeper.BlockHash);
StatusKeeper.UpdateFileChecksum(path, hash);
state.Hash = hash;
if (File.Exists(absolutePath))
return true;
//Or a directory?
- if (Directory.Exists(RootPath))
+ if (Directory.Exists(absolutePath))
return true;
//Fail if it is neither
return false;
}
- public FileInfo GetFileInfo(string relativePath)
+ public FileSystemInfo GetFileSystemInfo(string relativePath)
{
if (String.IsNullOrWhiteSpace(relativePath))
throw new ArgumentNullException("relativePath");
Contract.EndContractBlock();
var absolutePath = Path.Combine(RootPath, relativePath);
- Debug.Assert(File.Exists(absolutePath),String.Format("Path {0} doesn't exist",absolutePath));
- return new FileInfo(absolutePath);
+ if (Directory.Exists(absolutePath))
+ return new DirectoryInfo(absolutePath).WithProperCapitalization();
+ else
+ return new FileInfo(absolutePath).WithProperCapitalization();
}
+ public void Delete(string relativePath)
+ {
+ var absolutePath = Path.Combine(RootPath, relativePath).ToLower();
+ if (File.Exists(absolutePath))
+ {
+ try
+ {
+ File.Delete(absolutePath);
+ }
+ //The file may have been deleted by another thread. Just ignore the relevant exception
+ catch (FileNotFoundException) { }
+ }
+ else if (Directory.Exists(absolutePath))
+ {
+ try
+ {
+ Directory.Delete(absolutePath, true);
+ }
+ //The directory may have been deleted by another thread. Just ignore the relevant exception
+ catch (DirectoryNotFoundException){}
+ }
+
+ //_ignoreFiles[absolutePath] = absolutePath;
+ StatusKeeper.ClearFileStatus(absolutePath);
+ }
}
}
projects / pithos-ms-client / blobdiff