using System.Text;
using System.Threading.Tasks;
using Pithos.Interfaces;
+using Pithos.Network;
+using log4net;
namespace Pithos.Core.Agents
{
[Import]
public IStatusKeeper StatusKeeper { get; set; }
- //We should avoid processing files stored in the Fragments folder
- //The Full path to the fragments folder is stored in FragmentsPath
- public string FragmentsPath { get; set; }
-
[Import]
public NetworkAgent NetworkAgent { get; set; }
+ private static readonly ILog Log = LogManager.GetLogger("WorkflowAgent");
+
public void Start()
{
_agent = Agent<WorkflowState>.Start(inbox =>
var message = inbox.Receive();
var process = message.Then(Process, inbox.CancellationToken);
inbox.LoopAsync(process,loop,ex=>
- Trace.TraceError("[ERROR] Synch for {0}:\r{1}", message.Result.FileName, ex));
+ Log.ErrorFormat("[ERROR] Synch for {0}:\r{1}", message.Result.FileName, ex));
};
loop();
});
private Task<object> Process(WorkflowState state)
{
- if (state.Skip)
- return CompletedTask<object>.Default;
- string path = state.Path.ToLower();
- string fileName = Path.GetFileName(path);
-
- //Bypass deleted files, unless the status is Deleted
- if (!(File.Exists(path) || state.Status != FileStatus.Deleted))
+ var accountInfo = state.AccountInfo;
+ using (log4net.ThreadContext.Stacks["Workflow"].Push("Process"))
{
- state.Skip = true;
- this.StatusKeeper.RemoveFileOverlayStatus(path);
+ if (Log.IsDebugEnabled) Log.DebugFormat("State {0} {1} {2}", state.FileName,state.Status,state.TriggeringChange);
+
+ if (state.Skip)
+ {
+ if (Log.IsDebugEnabled) Log.DebugFormat("Skipping {0}",state.FileName);
+
+ return CompletedTask<object>.Default;
+ }
+ string path = state.Path.ToLower();
+
+ //Bypass deleted files, unless the status is Deleted
+ if (!File.Exists(path) && state.Status != FileStatus.Deleted)
+ {
+ state.Skip = true;
+ this.StatusKeeper.ClearFileStatus(path);
+
+ if (Log.IsDebugEnabled) Log.DebugFormat("Skipped missing {0}", state.FileName);
+
+ return CompletedTask<object>.Default;
+ }
+
+ var fileState = FileState.FindByFilePath(path);
+ var info = new FileInfo(path);
+
+
+ switch (state.Status)
+ {
+ case FileStatus.Created:
+ case FileStatus.Modified:
+ NetworkAgent.Post(new CloudUploadAction(accountInfo,info, fileState, accountInfo.BlockSize,
+ accountInfo.BlockHash));
+ break;
+ case FileStatus.Deleted:
+ NetworkAgent.Post(new CloudDeleteAction(accountInfo,info, fileState));
+ break;
+ case FileStatus.Renamed:
+ NetworkAgent.Post(new CloudMoveAction(accountInfo,CloudActionType.RenameCloud, new FileInfo(state.OldPath),
+ new FileInfo(state.Path)));
+ break;
+ }
+
return CompletedTask<object>.Default;
}
- var fileState = FileState.FindByFilePath(path);
- switch (state.Status)
- {
- case FileStatus.Created:
- case FileStatus.Modified:
- var info = new FileInfo(path);
- NetworkAgent.Post(new CloudAction(CloudActionType.UploadUnconditional, info, ObjectInfo.Empty, fileState));
- break;
- case FileStatus.Deleted:
- NetworkAgent.Post(new CloudAction(CloudActionType.DeleteCloud, null, new ObjectInfo { Name = fileName }, fileState));
- break;
- case FileStatus.Renamed:
- NetworkAgent.Post(new CloudAction(CloudActionType.RenameCloud, state.OldFileName, state.OldPath, state.FileName, state.Path));
- break;
- }
-
- return CompletedTask<object>.Default;
}
-
- public void RestartInterruptedFiles()
+ //Starts interrupted files for a specific account
+ public void RestartInterruptedFiles(AccountInfo accountInfo)
{
StatusNotification.NotifyChange("Restart processing interrupted files", TraceLevel.Verbose);
- var interruptedStates = new[] { FileOverlayStatus.Unversioned, FileOverlayStatus.Modified };
-
- var pendingEntries = (from state in FileState.Queryable
- where interruptedStates.Contains(state.OverlayStatus) &&
- !state.FilePath.StartsWith(FragmentsPath) &&
- !state.FilePath.EndsWith(".ignore")
- select state).ToList();
- var staleEntries = from state in pendingEntries
- where !File.Exists(state.FilePath)
- select state;
- var staleKeys = staleEntries.Select(state=>state.Id);
- FileState.DeleteAll(staleKeys);
-
- var validEntries = from state in pendingEntries.Except(staleEntries)
- where File.Exists(state.FilePath)
- select new WorkflowState
- {
- Path = state.FilePath.ToLower(),
- FileName = Path.GetFileName(state.FilePath).ToLower(),
- Hash = state.Checksum,
- Status = state.OverlayStatus == FileOverlayStatus.Unversioned ?
- FileStatus.Created :
- FileStatus.Modified,
- TriggeringChange = state.OverlayStatus == FileOverlayStatus.Unversioned ?
- WatcherChangeTypes.Created :
- WatcherChangeTypes.Changed
- };
- foreach (var entry in validEntries)
+
+ using (log4net.ThreadContext.Stacks["Workflow"].Push("Restart"))
{
- Post(entry);
- }
+ if (Log.IsDebugEnabled)
+ Log.Debug("Starting interrupted files");
+
+ var cachePath = Path.Combine(accountInfo.AccountPath, FolderConstants.CacheFolder)
+ .ToLower();
+
+ var pendingEntries = from state in FileState.Queryable
+ where state.FileStatus != FileStatus.Unchanged &&
+ !state.FilePath.StartsWith(cachePath) &&
+ !state.FilePath.EndsWith(".ignore") &&
+ state.FilePath.StartsWith(accountInfo.AccountPath)
+ select state;
+ if (Log.IsDebugEnabled)
+ Log.DebugFormat("Found {0} interrupted files",pendingEntries.Count());
+
+ var validEntries = from state in pendingEntries
+ select new WorkflowState
+ {
+ AccountInfo=accountInfo,
+ Path = state.FilePath.ToLower(),
+ FileName = Path.GetFileName(state.FilePath).ToLower(),
+ Hash = state.Checksum,
+ Status = state.OverlayStatus == FileOverlayStatus.Unversioned
+ ? FileStatus.Created
+ : state.FileStatus,
+ TriggeringChange =
+ state.OverlayStatus == FileOverlayStatus.Unversioned
+ ? WatcherChangeTypes.Created
+ : WatcherChangeTypes.Changed
+ };
+ foreach (var entry in validEntries)
+ {
+ Post(entry);
+ }
+ }
}
public void Post(WorkflowState workflowState)
{
+ if (Log.IsDebugEnabled)
+ Log.DebugFormat("Posted {0} {1} {2}", workflowState.Path, workflowState.Status, workflowState.TriggeringChange);
_agent.Post(workflowState);
}
}
projects / pithos-ms-client / blobdiff