Synnefo » Pithos MS Client

using System;

using System.Collections.Generic;

using System.ComponentModel.Composition;

using System.Diagnostics;

using System.Diagnostics.Contracts;

using System.IO;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

using Pithos.Interfaces;

using Pithos.Network;

using log4net;

namespace Pithos.Core.Agents

{

    [Export]

    public class WorkflowAgent

    {

        Agent<WorkflowState> _agent;

        public IStatusNotification StatusNotification { get; set; }

        [Import]

        public IStatusKeeper StatusKeeper { get; set; }

        [Import]

        public NetworkAgent NetworkAgent { get; set; }

        private static readonly ILog Log = LogManager.GetLogger("WorkflowAgent");

        public void Start()

        {

            _agent = Agent<WorkflowState>.Start(inbox =>

            {

                Action loop = null;

                loop = () =>

                {

                    var message = inbox.Receive();

                    var process = message.Then(Process, inbox.CancellationToken);                        

                    inbox.LoopAsync(process,loop,ex=>

                            Log.ErrorFormat("[ERROR] Synch for {0}:\r{1}", message.Result.FileName, ex));

                };

                loop();

            });

        }

        private Task<object> Process(WorkflowState state)

        {

            var accountInfo = state.AccountInfo;

            using (log4net.ThreadContext.Stacks["Workflow"].Push("Process"))

            {

                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:

                        string fileName = info.AsRelativeUrlTo(accountInfo.AccountPath);

                        NetworkAgent.Post(new CloudDeleteAction(accountInfo,fileName, fileState));

                        break;

                    case FileStatus.Renamed:

                        NetworkAgent.Post(new CloudMoveAction(accountInfo,CloudActionType.RenameCloud, state.OldFileName,

                                                              state.OldPath, state.FileName, state.Path));

                        break;

                }

                return CompletedTask<object>.Default;

            }

        }

        //Starts interrupted files for a specific account

        public void RestartInterruptedFiles(AccountInfo accountInfo)

        {

            StatusNotification.NotifyChange("Restart processing interrupted files", TraceLevel.Verbose);

            using (log4net.ThreadContext.Stacks["Workflow"].Push("Restart"))

            {

                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);

        }

    }

}