using System.Diagnostics;\r
using System.Diagnostics.Contracts;\r
using System.IO;\r
+using System.Reflection;\r
using System.Threading;\r
using System.Threading.Tasks;\r
-using System.Threading.Tasks.Dataflow;\r
using Castle.ActiveRecord;\r
using Pithos.Interfaces;\r
using Pithos.Network;\r
using System;\r
using System.Collections.Generic;\r
using System.Linq;\r
- using System.Text;\r
\r
/// <summary>\r
/// PollAgent periodically polls the server to detect object changes. The agent retrieves a listing of all\r
[Export]\r
public class PollAgent\r
{\r
- private static readonly ILog Log = LogManager.GetLogger("PollAgent");\r
+ private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);\r
\r
[System.ComponentModel.Composition.Import]\r
public IStatusKeeper StatusKeeper { get; set; }\r
//The Sync Event signals a manual synchronisation\r
private readonly AsyncManualResetEvent _syncEvent = new AsyncManualResetEvent();\r
\r
- private ConcurrentDictionary<string, DateTime> _lastSeen = new ConcurrentDictionary<string, DateTime>();\r
- private readonly ConcurrentBag<AccountInfo> _accounts = new ConcurrentBag<AccountInfo>();\r
+ private readonly ConcurrentDictionary<string, DateTime> _lastSeen = new ConcurrentDictionary<string, DateTime>();\r
+ private readonly ConcurrentDictionary<Uri, AccountInfo> _accounts = new ConcurrentDictionary<Uri,AccountInfo>();\r
\r
\r
/// <summary>\r
_syncEvent.Set();\r
}\r
\r
- //Remote files are polled periodically. Any changes are processed\r
+ /// <summary>\r
+ /// Remote files are polled periodically. Any changes are processed\r
+ /// </summary>\r
+ /// <param name="since"></param>\r
+ /// <returns></returns>\r
public async Task PollRemoteFiles(DateTime? since = null)\r
{\r
- Debug.Assert(Thread.CurrentThread.IsBackground, "Polling Ended up in the main thread!");\r
+ if (Log.IsDebugEnabled)\r
+ Log.DebugFormat("Polling changes after [{0}]",since);\r
\r
- UpdateStatus(PithosStatus.Syncing);\r
- StatusNotification.Notify(new PollNotification());\r
+ Debug.Assert(Thread.CurrentThread.IsBackground, "Polling Ended up in the main thread!");\r
+ \r
\r
- using (log4net.ThreadContext.Stacks["Retrieve Remote"].Push("All accounts"))\r
+ using (ThreadContext.Stacks["Retrieve Remote"].Push("All accounts"))\r
{\r
//If this poll fails, we will retry with the same since value\r
var nextSince = since;\r
try\r
{\r
- //Next time we will check for all changes since the current check minus 1 second\r
- //This is done to ensure there are no discrepancies due to clock differences\r
- var current = DateTime.Now.AddSeconds(-1);\r
+ UpdateStatus(PithosStatus.PollSyncing);\r
\r
- var tasks = from accountInfo in _accounts\r
+ var tasks = from accountInfo in _accounts.Values\r
select ProcessAccountFiles(accountInfo, since);\r
\r
- await TaskEx.WhenAll(tasks.ToList());\r
+ var nextTimes=await TaskEx.WhenAll(tasks.ToList());\r
\r
_firstPoll = false;\r
//Reschedule the poll with the current timestamp as a "since" value\r
- nextSince = current;\r
+\r
+ if (nextTimes.Length>0)\r
+ nextSince = nextTimes.Min();\r
+ if (Log.IsDebugEnabled)\r
+ Log.DebugFormat("Next Poll at [{0}]",nextSince);\r
}\r
catch (Exception ex)\r
{\r
//In case of failure retry with the same "since" value\r
}\r
\r
- UpdateStatus(PithosStatus.InSynch);\r
- //Wait for the polling interval to pass or the Sync event to be signalled\r
- nextSince = await WaitForScheduledOrManualPoll(nextSince);\r
-\r
- TaskEx.Run(()=>PollRemoteFiles(nextSince));\r
-\r
+ UpdateStatus(PithosStatus.PollComplete);\r
+ //The multiple try blocks are required because we can't have an await call\r
+ //inside a finally block\r
+ //TODO: Find a more elegant solution for reschedulling in the event of an exception\r
+ try\r
+ {\r
+ //Wait for the polling interval to pass or the Sync event to be signalled\r
+ nextSince = await WaitForScheduledOrManualPoll(nextSince);\r
+ }\r
+ finally\r
+ {\r
+ //Ensure polling is scheduled even in case of error\r
+ TaskEx.Run(() => PollRemoteFiles(nextSince)); \r
+ }\r
}\r
}\r
\r
return since;\r
}\r
\r
- public async Task ProcessAccountFiles(AccountInfo accountInfo, DateTime? since = null)\r
+ public async Task<DateTime?> ProcessAccountFiles(AccountInfo accountInfo, DateTime? since = null)\r
{\r
if (accountInfo == null)\r
throw new ArgumentNullException("accountInfo");\r
Contract.EndContractBlock();\r
\r
\r
- using (log4net.ThreadContext.Stacks["Retrieve Remote"].Push(accountInfo.UserName))\r
+ using (ThreadContext.Stacks["Retrieve Remote"].Push(accountInfo.UserName))\r
{\r
+\r
await NetworkAgent.GetDeleteAwaiter();\r
\r
Log.Info("Scheduled");\r
var client = new CloudFilesClient(accountInfo);\r
\r
- var containers = client.ListContainers(accountInfo.UserName);\r
+ //We don't need to check the trash container\r
+ var containers = client.ListContainers(accountInfo.UserName)\r
+ .Where(c=>c.Name!="trash")\r
+ .ToList();\r
\r
\r
CreateContainerFolders(accountInfo, containers);\r
\r
+ //The nextSince time fallback time is the same as the current.\r
+ //If polling succeeds, the next Since time will be the smallest of the maximum modification times\r
+ //of the shared and account objects\r
+ var nextSince = since;\r
+\r
try\r
{\r
//Wait for any deletions to finish\r
await NetworkAgent.GetDeleteAwaiter();\r
//Get the poll time now. We may miss some deletions but it's better to keep a file that was deleted\r
//than delete a file that was created while we were executing the poll \r
- var pollTime = DateTime.Now;\r
\r
//Get the list of server objects changed since the last check\r
//The name of the container is passed as state in order to create a dictionary of tasks in a subsequent step\r
select Task<IList<ObjectInfo>>.Factory.StartNew(_ =>\r
client.ListObjects(accountInfo.UserName, container.Name, since), container.Name)).ToList();\r
\r
- var listShared = Task<IList<ObjectInfo>>.Factory.StartNew(_ => client.ListSharedObjects(since), "shared");\r
+ var listShared = Task<IList<ObjectInfo>>.Factory.StartNew(_ => \r
+ client.ListSharedObjects(since), "shared");\r
listObjects.Add(listShared);\r
var listTasks = await Task.Factory.WhenAll(listObjects.ToArray());\r
\r
- using (log4net.ThreadContext.Stacks["SCHEDULE"].Push("Process Results"))\r
+ using (ThreadContext.Stacks["SCHEDULE"].Push("Process Results"))\r
{\r
var dict = listTasks.ToDictionary(t => t.AsyncState);\r
\r
//Get all non-trash objects. Remember, the container name is stored in AsyncState\r
- var remoteObjects = from objectList in listTasks\r
+ var remoteObjects = (from objectList in listTasks\r
where (string)objectList.AsyncState != "trash"\r
from obj in objectList.Result\r
- select obj;\r
+ select obj).ToList();\r
+ \r
+ //Get the latest remote object modification date, only if it is after\r
+ //the original since date\r
+ nextSince = GetLatestDateAfter(nextSince, remoteObjects);\r
\r
- var trashObjects = dict["trash"].Result;\r
var sharedObjects = dict["shared"].Result;\r
+ nextSince = GetLatestDateBefore(nextSince, sharedObjects);\r
\r
//DON'T process trashed files\r
//If some files are deleted and added again to a folder, they will be deleted\r
*/\r
\r
var cleanRemotes = (from info in remoteObjects.Union(sharedObjects)\r
- let name = info.Name\r
+ let name = info.Name??""\r
where !name.EndsWith(".ignore", StringComparison.InvariantCultureIgnoreCase) &&\r
!name.StartsWith(FolderConstants.CacheFolder + "/",\r
StringComparison.InvariantCultureIgnoreCase)\r
\r
var differencer = _differencer.PostSnapshot(accountInfo, cleanRemotes);\r
\r
- ProcessDeletedFiles(accountInfo, differencer.Deleted.FilterBelow(SelectiveUris), pollTime);\r
+ var filterUris = SelectiveUris[accountInfo.AccountKey];\r
+\r
+ ProcessDeletedFiles(accountInfo, differencer.Deleted.FilterDirectlyBelow(filterUris));\r
\r
// @@@ NEED To add previous state here as well, To compare with previous hash\r
\r
\r
\r
//Create a list of actions from the remote files\r
- var allActions = ChangesToActions(accountInfo, differencer.Changed.FilterBelow(SelectiveUris))\r
+ \r
+ var allActions = MovesToActions(accountInfo,differencer.Moved.FilterDirectlyBelow(filterUris))\r
+ .Union(\r
+ ChangesToActions(accountInfo, differencer.Changed.FilterDirectlyBelow(filterUris)))\r
.Union(\r
- CreatesToActions(accountInfo, differencer.Created.FilterBelow(SelectiveUris)));\r
+ CreatesToActions(accountInfo, differencer.Created.FilterDirectlyBelow(filterUris)));\r
\r
//And remove those that are already being processed by the agent\r
var distinctActions = allActions\r
- .Except(NetworkAgent.GetEnumerable(), new PithosMonitor.LocalFileComparer())\r
+ .Except(NetworkAgent.GetEnumerable(), new LocalFileComparer())\r
.ToList();\r
\r
//Queue all the actions\r
catch (Exception ex)\r
{\r
Log.ErrorFormat("[FAIL] ListObjects for{0} in ProcessRemoteFiles with {1}", accountInfo.UserName, ex);\r
- return;\r
+ return nextSince;\r
}\r
\r
Log.Info("[LISTENER] Finished");\r
-\r
+ return nextSince;\r
}\r
}\r
\r
- AccountsDifferencer _differencer = new AccountsDifferencer();\r
- private List<Uri> _selectiveUris=new List<Uri>();\r
+ /// <summary>\r
+ /// Returns the latest LastModified date from the list of objects, but only if it is before\r
+ /// than the threshold value\r
+ /// </summary>\r
+ /// <param name="threshold"></param>\r
+ /// <param name="cloudObjects"></param>\r
+ /// <returns></returns>\r
+ private static DateTime? GetLatestDateBefore(DateTime? threshold, IList<ObjectInfo> cloudObjects)\r
+ {\r
+ DateTime? maxDate = null;\r
+ if (cloudObjects!=null && cloudObjects.Count > 0)\r
+ maxDate = cloudObjects.Max(obj => obj.Last_Modified);\r
+ if (maxDate == null || maxDate == DateTime.MinValue)\r
+ return threshold;\r
+ if (threshold == null || threshold == DateTime.MinValue || threshold > maxDate)\r
+ return maxDate;\r
+ return threshold;\r
+ }\r
+\r
+ /// <summary>\r
+ /// Returns the latest LastModified date from the list of objects, but only if it is after\r
+ /// the threshold value\r
+ /// </summary>\r
+ /// <param name="threshold"></param>\r
+ /// <param name="cloudObjects"></param>\r
+ /// <returns></returns>\r
+ private static DateTime? GetLatestDateAfter(DateTime? threshold, IList<ObjectInfo> cloudObjects)\r
+ {\r
+ DateTime? maxDate = null;\r
+ if (cloudObjects!=null && cloudObjects.Count > 0)\r
+ maxDate = cloudObjects.Max(obj => obj.Last_Modified);\r
+ if (maxDate == null || maxDate == DateTime.MinValue)\r
+ return threshold;\r
+ if (threshold == null || threshold == DateTime.MinValue || threshold < maxDate)\r
+ return maxDate;\r
+ return threshold;\r
+ }\r
+\r
+ readonly AccountsDifferencer _differencer = new AccountsDifferencer();\r
+ private Dictionary<Uri, List<Uri>> _selectiveUris = new Dictionary<Uri, List<Uri>>();\r
\r
/// <summary>\r
/// Deletes local files that are not found in the list of cloud files\r
/// </summary>\r
/// <param name="accountInfo"></param>\r
/// <param name="cloudFiles"></param>\r
- /// <param name="pollTime"></param>\r
- private void ProcessDeletedFiles(AccountInfo accountInfo, IEnumerable<ObjectInfo> cloudFiles, DateTime pollTime)\r
+ private void ProcessDeletedFiles(AccountInfo accountInfo, IEnumerable<ObjectInfo> cloudFiles)\r
{\r
if (accountInfo == null)\r
throw new ArgumentNullException("accountInfo");\r
var deletedFiles = new List<FileSystemInfo>();\r
foreach (var objectInfo in cloudFiles)\r
{\r
+ if (Log.IsDebugEnabled)\r
+ Log.DebugFormat("Handle deleted [{0}]",objectInfo.Uri);\r
var relativePath = objectInfo.RelativeUrlToFilePath(accountInfo.UserName);\r
var item = FileAgent.GetFileAgent(accountInfo).GetFileSystemInfo(relativePath);\r
+ if (Log.IsDebugEnabled)\r
+ Log.DebugFormat("Will delete [{0}] for [{1}]", item.FullName,objectInfo.Uri);\r
if (item.Exists)\r
{\r
if ((item.Attributes & FileAttributes.ReadOnly) == FileAttributes.ReadOnly)\r
item.Attributes = item.Attributes & ~FileAttributes.ReadOnly;\r
\r
}\r
- item.Delete();\r
+ \r
+ \r
+ Log.DebugFormat("Deleting {0}", item.FullName);\r
+\r
+ var directory = item as DirectoryInfo;\r
+ if (directory!=null)\r
+ directory.Delete(true);\r
+ else\r
+ item.Delete();\r
+ Log.DebugFormat("Deleted [{0}] for [{1}]", item.FullName, objectInfo.Uri);\r
DateTime lastDate;\r
_lastSeen.TryRemove(item.FullName, out lastDate);\r
deletedFiles.Add(item);\r
}\r
StatusKeeper.SetFileState(item.FullName, FileStatus.Deleted, FileOverlayStatus.Deleted);\r
}\r
+ Log.InfoFormat("[{0}] files were deleted",deletedFiles.Count);\r
StatusNotification.NotifyForFiles(deletedFiles, String.Format("{0} files were deleted", deletedFiles.Count), TraceLevel.Info);\r
}\r
\r
}\r
\r
- //Creates an appropriate action for each server file\r
+ /// <summary>\r
+ /// Creates a Sync action for each changed server file\r
+ /// </summary>\r
+ /// <param name="accountInfo"></param>\r
+ /// <param name="changes"></param>\r
+ /// <returns></returns>\r
private IEnumerable<CloudAction> ChangesToActions(AccountInfo accountInfo, IEnumerable<ObjectInfo> changes)\r
{\r
if (changes == null)\r
foreach (var objectInfo in changes)\r
{\r
var relativePath = objectInfo.RelativeUrlToFilePath(accountInfo.UserName);\r
- //and remove any matching objects from the list, adding them to the commonObjects list\r
+ //If a directory object already exists, we may need to sync it\r
if (fileAgent.Exists(relativePath))\r
{\r
- //If a directory object already exists, we don't need to perform any other action \r
var localFile = fileAgent.GetFileSystemInfo(relativePath);\r
+ //We don't need to sync directories\r
if (objectInfo.Content_Type == @"application/directory" && localFile is DirectoryInfo)\r
continue;\r
using (new SessionScope(FlushAction.Never))\r
}\r
}\r
\r
+ /// <summary>\r
+ /// Creates a Local Move action for each moved server file\r
+ /// </summary>\r
+ /// <param name="accountInfo"></param>\r
+ /// <param name="moves"></param>\r
+ /// <returns></returns>\r
+ private IEnumerable<CloudAction> MovesToActions(AccountInfo accountInfo, IEnumerable<ObjectInfo> moves)\r
+ {\r
+ if (moves == null)\r
+ throw new ArgumentNullException();\r
+ Contract.EndContractBlock();\r
+ var fileAgent = FileAgent.GetFileAgent(accountInfo);\r
+\r
+ //In order to avoid multiple iterations over the files, we iterate only once\r
+ //over the remote files\r
+ foreach (var objectInfo in moves)\r
+ {\r
+ var previousRelativepath = objectInfo.Previous.RelativeUrlToFilePath(accountInfo.UserName);\r
+ //If the previous file already exists, we can execute a Move operation\r
+ if (fileAgent.Exists(previousRelativepath))\r
+ {\r
+ var previousFile = fileAgent.GetFileSystemInfo(previousRelativepath);\r
+ using (new SessionScope(FlushAction.Never))\r
+ {\r
+ var state = StatusKeeper.GetStateByFilePath(previousFile.FullName);\r
+ _lastSeen[previousFile.FullName] = DateTime.Now;\r
+\r
+ //For each moved object we need to move both the local file and update \r
+ yield return new CloudAction(accountInfo, CloudActionType.RenameLocal,\r
+ previousFile, objectInfo, state, accountInfo.BlockSize,\r
+ accountInfo.BlockHash);\r
+ //For modified files, we need to download the changes as well\r
+ if (objectInfo.Hash!=objectInfo.PreviousHash)\r
+ yield return new CloudDownloadAction(accountInfo,objectInfo);\r
+ }\r
+ }\r
+ //If the previous file does not exist, we need to download it in the new location\r
+ else\r
+ {\r
+ //Remote files should be downloaded\r
+ yield return new CloudDownloadAction(accountInfo, objectInfo);\r
+ }\r
+ }\r
+ }\r
+\r
+\r
+ /// <summary>\r
+ /// Creates a download action for each new server file\r
+ /// </summary>\r
+ /// <param name="accountInfo"></param>\r
+ /// <param name="creates"></param>\r
+ /// <returns></returns>\r
private IEnumerable<CloudAction> CreatesToActions(AccountInfo accountInfo, IEnumerable<ObjectInfo> creates)\r
{\r
if (creates == null)\r
//over the remote files\r
foreach (var objectInfo in creates)\r
{\r
+ if (Log.IsDebugEnabled)\r
+ Log.DebugFormat("[NEW INFO] {0}",objectInfo.Uri);\r
+\r
var relativePath = objectInfo.RelativeUrlToFilePath(accountInfo.UserName);\r
- //and remove any matching objects from the list, adding them to the commonObjects list\r
+ //If the object already exists, we probably have a conflict\r
if (fileAgent.Exists(relativePath))\r
{\r
- //If the object already exists, we probably have a conflict\r
+ Log.DebugFormat("[SKIP EXISTING] {0}", objectInfo.Uri);\r
//If a directory object already exists, we don't need to perform any other action \r
var localFile = fileAgent.GetFileSystemInfo(relativePath);\r
StatusKeeper.SetFileState(localFile.FullName, FileStatus.Conflict, FileOverlayStatus.Conflict);\r
}\r
}\r
\r
- private void ProcessTrashedFiles(AccountInfo accountInfo, IEnumerable<ObjectInfo> trashObjects)\r
+ /// <summary>\r
+ /// Notify the UI to update the visual status\r
+ /// </summary>\r
+ /// <param name="status"></param>\r
+ private void UpdateStatus(PithosStatus status)\r
{\r
- var fileAgent = FileAgent.GetFileAgent(accountInfo);\r
- foreach (var trashObject in trashObjects)\r
+ try\r
{\r
- var barePath = trashObject.RelativeUrlToFilePath(accountInfo.UserName);\r
- //HACK: Assume only the "pithos" container is used. Must find out what happens when\r
- //deleting a file from a different container\r
- var relativePath = Path.Combine("pithos", barePath);\r
- fileAgent.Delete(relativePath);\r
+ StatusNotification.SetPithosStatus(status);\r
+ //StatusNotification.Notify(new Notification());\r
+ }\r
+ catch (Exception exc)\r
+ {\r
+ //Failure is not critical, just log it\r
+ Log.Warn("Error while updating status", exc);\r
}\r
- }\r
-\r
- private void UpdateStatus(PithosStatus status)\r
- {\r
- StatusKeeper.SetPithosStatus(status);\r
- StatusNotification.Notify(new Notification());\r
}\r
\r
private static void CreateContainerFolders(AccountInfo accountInfo, IEnumerable<ContainerInfo> containers)\r
}\r
}\r
\r
- public void SetSyncUris(string[] uris)\r
- {\r
- var selectiveUris = uris.Select(uri => new Uri(uri));\r
- SelectiveUris=selectiveUris.ToList();\r
+ public void SetSyncUris(Uri accountKey, Uri[] uris)\r
+ { \r
+ SelectiveUris[accountKey]=uris.ToList();\r
}\r
\r
- protected List<Uri> SelectiveUris\r
+ protected Dictionary<Uri,List<Uri>> SelectiveUris\r
{\r
get { return _selectiveUris;}\r
set { _selectiveUris = value; }\r
}\r
+\r
+ public void AddAccount(AccountInfo accountInfo)\r
+ {\r
+ //Avoid adding a duplicate accountInfo\r
+ _accounts.TryAdd(accountInfo.AccountKey, accountInfo);\r
+ }\r
+\r
+ public void RemoveAccount(AccountInfo accountInfo)\r
+ {\r
+ AccountInfo account;\r
+ _accounts.TryRemove(accountInfo.AccountKey, out account);\r
+ SnapshotDifferencer differencer;\r
+ _differencer.Differencers.TryRemove(accountInfo.AccountKey, out differencer);\r
+ }\r
}\r
}\r
projects / pithos-ms-client / blobdiff