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

   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.Text;
   9 using System.Threading.Tasks;
  10 using Pithos.Interfaces;
  11
  12 namespace Pithos.Core.Agents
  13 {
  14     [Export]
  15     public class WorkflowAgent
  16     {
  17         Agent<WorkflowState> _agent;
  18
  19         public IStatusNotification StatusNotification { get; set; }
  20         [Import]
  21         public IStatusKeeper StatusKeeper { get; set; }
  22
  23         //We should avoid processing files stored in the Fragments folder
  24         //The Full path to the fragments folder is stored in FragmentsPath
  25         public string FragmentsPath { get; set; }
  26
  27         [Import]
  28         public NetworkAgent NetworkAgent { get; set; }
  29
  30         public void Start()
  31         {
  32             _agent = Agent<WorkflowState>.Start(inbox =>
  33             {
  34                 Action loop = null;
  35                 loop = () =>
  36                 {
  37                     var message = inbox.Receive();
  38                     var process = message.Then(Process, inbox.CancellationToken);
  39                     inbox.LoopAsync(process,loop,ex=>
  40                             Trace.TraceError("[ERROR] Synch for {0}:\r{1}", message.Result.FileName, ex));
  41                 };
  42                 loop();
  43             });
  44         }
  45
  46         private Task<object> Process(WorkflowState state)
  47         {
  48             if (state.Skip)
  49                 return CompletedTask<object>.Default;
  50             string path = state.Path.ToLower();
  51             string fileName = Path.GetFileName(path);
  52
  53             //Bypass deleted files, unless the status is Deleted
  54             if (!File.Exists(path) && state.Status != FileStatus.Deleted)
  55             {
  56                 state.Skip = true;
  57                 this.StatusKeeper.RemoveFileOverlayStatus(path);
  58                 return CompletedTask<object>.Default;
  59             }
  60             var fileState = FileState.FindByFilePath(path);
  61             var blockHash = NetworkAgent.BlockHash;
  62             var blockSize = NetworkAgent.BlockSize;
  63
  64             switch (state.Status)
  65             {
  66                 case FileStatus.Created:
  67                 case FileStatus.Modified:
  68                     var info = new FileInfo(path);
  69                     NetworkAgent.Post(new CloudAction(CloudActionType.UploadUnconditional, info, ObjectInfo.Empty, fileState,blockSize,blockHash));
  70                     break;
  71                 case FileStatus.Deleted:
  72                     NetworkAgent.Post(new CloudAction(CloudActionType.DeleteCloud, null, new ObjectInfo { Name = fileName }, fileState, blockSize, blockHash));
  73                     break;
  74                 case FileStatus.Renamed:
  75                     NetworkAgent.Post(new CloudAction(CloudActionType.RenameCloud, state.OldFileName, state.OldPath, state.FileName, state.Path));
  76                     break;
  77             }
  78
  79             return CompletedTask<object>.Default;
  80         }
  81
  82
  83
  84         public void RestartInterruptedFiles()
  85         {
  86
  87             StatusNotification.NotifyChange("Restart processing interrupted files", TraceLevel.Verbose);
  88
  89             var pendingEntries = from state in FileState.Queryable
  90                                    where state.FileStatus != FileStatus.Unchanged &&
  91                                          !state.FilePath.StartsWith(FragmentsPath.ToLower()) &&
  92                                          !state.FilePath.EndsWith(".ignore")
  93                                    select state;
  94
  95             var validEntries = from state in pendingEntries
  96                              select new WorkflowState
  97                              {
  98                                  Path = state.FilePath.ToLower(),
  99                                  FileName = Path.GetFileName(state.FilePath).ToLower(),
 100                                  Hash = state.Checksum,
 101                                  Status = state.OverlayStatus == FileOverlayStatus.Unversioned ?
 102                                                    FileStatus.Created :
 103                                                    state.FileStatus,
 104                                  TriggeringChange = state.OverlayStatus == FileOverlayStatus.Unversioned ?
 105                                                    WatcherChangeTypes.Created :
 106                                                    WatcherChangeTypes.Changed
 107                              };
 108             foreach (var entry in validEntries)
 109             {
 110                 Post(entry);
 111             }
 112
 113         }
 114
 115
 116
 117         public void Post(WorkflowState workflowState)
 118         {
 119             _agent.Post(workflowState);
 120         }
 121     }
 122 }