mirror of
https://github.com/tahoe-lafs/tahoe-lafs.git
synced 2025-06-19 07:48:11 +00:00
refactor/white-space MagicFolder creation
This commit is contained in:
@ -561,18 +561,18 @@ class Client(node.Node, pollmixin.PollMixin):
|
|||||||
|
|
||||||
if self.get_config("magic_folder", "enabled", False, boolean=True):
|
if self.get_config("magic_folder", "enabled", False, boolean=True):
|
||||||
#print "magic folder enabled"
|
#print "magic folder enabled"
|
||||||
upload_dircap = self.get_private_config("magic_folder_dircap")
|
|
||||||
collective_dircap = self.get_private_config("collective_dircap")
|
|
||||||
|
|
||||||
local_dir_config = self.get_config("magic_folder", "local.directory").decode("utf-8")
|
|
||||||
local_dir = abspath_expanduser_unicode(local_dir_config, base=self.basedir)
|
|
||||||
|
|
||||||
dbfile = os.path.join(self.basedir, "private", "magicfolderdb.sqlite")
|
|
||||||
dbfile = abspath_expanduser_unicode(dbfile)
|
|
||||||
|
|
||||||
from allmydata.frontends import magic_folder
|
from allmydata.frontends import magic_folder
|
||||||
umask = self.get_config("magic_folder", "download.umask", 0077)
|
|
||||||
s = magic_folder.MagicFolder(self, upload_dircap, collective_dircap, local_dir, dbfile, umask)
|
db_filename = os.path.join(self.basedir, "private", "magicfolderdb.sqlite")
|
||||||
|
local_dir_config = self.get_config("magic_folder", "local.directory").decode("utf-8")
|
||||||
|
s = magic_folder.MagicFolder(
|
||||||
|
client=self,
|
||||||
|
upload_dircap=self.get_private_config("magic_folder_dircap"),
|
||||||
|
collective_dircap=self.get_private_config("collective_dircap"),
|
||||||
|
local_path_u=abspath_expanduser_unicode(local_dir_config, base=self.basedir),
|
||||||
|
dbfile=abspath_expanduser_unicode(db_filename),
|
||||||
|
umask=self.get_config("magic_folder", "download.umask", 0077),
|
||||||
|
)
|
||||||
self._magic_folder = s
|
self._magic_folder = s
|
||||||
s.setServiceParent(self)
|
s.setServiceParent(self)
|
||||||
s.startService()
|
s.startService()
|
||||||
|
@ -63,7 +63,7 @@ class MagicFolder(service.MultiService):
|
|||||||
name = 'magic-folder'
|
name = 'magic-folder'
|
||||||
|
|
||||||
def __init__(self, client, upload_dircap, collective_dircap, local_path_u, dbfile, umask,
|
def __init__(self, client, upload_dircap, collective_dircap, local_path_u, dbfile, umask,
|
||||||
pending_delay=1.0, clock=None):
|
pending_delay=1.0, clock=None, poll_interval=3):
|
||||||
precondition_abspath(local_path_u)
|
precondition_abspath(local_path_u)
|
||||||
|
|
||||||
service.MultiService.__init__(self)
|
service.MultiService.__init__(self)
|
||||||
@ -83,7 +83,7 @@ class MagicFolder(service.MultiService):
|
|||||||
self.uploader = Uploader(client, local_path_u, db, upload_dirnode, pending_delay, clock)
|
self.uploader = Uploader(client, local_path_u, db, upload_dirnode, pending_delay, clock)
|
||||||
self.downloader = Downloader(client, local_path_u, db, collective_dirnode,
|
self.downloader = Downloader(client, local_path_u, db, collective_dirnode,
|
||||||
upload_dirnode.get_readonly_uri(), clock, self.uploader.is_pending, umask,
|
upload_dirnode.get_readonly_uri(), clock, self.uploader.is_pending, umask,
|
||||||
self.set_public_status)
|
self.set_public_status, poll_interval=poll_interval)
|
||||||
self._public_status = (False, ['Magic folder has not yet started'])
|
self._public_status = (False, ['Magic folder has not yet started'])
|
||||||
|
|
||||||
def enable_debug_log(self, enabled=True):
|
def enable_debug_log(self, enabled=True):
|
||||||
@ -131,9 +131,8 @@ class MagicFolder(service.MultiService):
|
|||||||
|
|
||||||
|
|
||||||
class QueueMixin(HookMixin):
|
class QueueMixin(HookMixin):
|
||||||
scan_interval = 0
|
|
||||||
|
|
||||||
def __init__(self, client, local_path_u, db, name, clock, delay=0):
|
def __init__(self, client, local_path_u, db, name, clock):
|
||||||
self._client = client
|
self._client = client
|
||||||
self._local_path_u = local_path_u
|
self._local_path_u = local_path_u
|
||||||
self._local_filepath = to_filepath(local_path_u)
|
self._local_filepath = to_filepath(local_path_u)
|
||||||
@ -163,9 +162,6 @@ class QueueMixin(HookMixin):
|
|||||||
# do we also want to bound on "maximum age"?
|
# do we also want to bound on "maximum age"?
|
||||||
self._process_history = deque(maxlen=20)
|
self._process_history = deque(maxlen=20)
|
||||||
self._stopped = False
|
self._stopped = False
|
||||||
# XXX pass in an initial value for this; it seems like .10 broke this and it's always 0
|
|
||||||
self._turn_delay = delay
|
|
||||||
self._log('delay is %f' % self._turn_delay)
|
|
||||||
|
|
||||||
# a Deferred to wait for the _do_processing() loop to exit
|
# a Deferred to wait for the _do_processing() loop to exit
|
||||||
# (gets set to the return from _do_processing() if we get that
|
# (gets set to the return from _do_processing() if we get that
|
||||||
@ -212,35 +208,31 @@ class QueueMixin(HookMixin):
|
|||||||
(processing each item). After that we yield for _turn_deque
|
(processing each item). After that we yield for _turn_deque
|
||||||
seconds.
|
seconds.
|
||||||
"""
|
"""
|
||||||
# we subtract here so there's a scan on the very first iteration
|
|
||||||
last_scan = self._clock.seconds() - self.scan_interval
|
|
||||||
while not self._stopped:
|
while not self._stopped:
|
||||||
self._log("doing iteration")
|
self._log("doing iteration")
|
||||||
d = task.deferLater(self._clock, self._turn_delay, lambda: None)
|
d = task.deferLater(self._clock, self._next_scan_delay(), lambda: None)
|
||||||
# ">=" is important here if scan scan_interval is 0
|
|
||||||
if self._clock.seconds() - last_scan >= self.scan_interval:
|
# adds items to our deque
|
||||||
# XXX can't we unify the "_full_scan" vs what
|
yield self._when_queue_is_empty()
|
||||||
# Downloader does...
|
|
||||||
last_scan = self._clock.seconds()
|
|
||||||
yield self._when_queue_is_empty() # (this no-op for us, only Downloader uses it...)
|
|
||||||
self._log("did scan; now %d" % last_scan)
|
|
||||||
else:
|
|
||||||
self._log("skipped scan")
|
|
||||||
|
|
||||||
# process anything in our queue
|
# process anything in our queue
|
||||||
yield self._process_deque()
|
yield self._process_deque()
|
||||||
self._log("one loop; call_hook iteration %r" % self)
|
|
||||||
self._call_hook(None, 'iteration')
|
|
||||||
# we want to have our callLater queued in the reactor
|
# we want to have our callLater queued in the reactor
|
||||||
# *before* we trigger the 'iteration' hook, so that hook
|
# *before* we trigger the 'iteration' hook, so that hook
|
||||||
# can successfully advance the Clock and bypass the delay
|
# can successfully advance the Clock and bypass the delay
|
||||||
# if required (e.g. in the tests).
|
# if required (e.g. in the tests).
|
||||||
|
self._log("one loop; call_hook iteration %r" % self)
|
||||||
|
self._call_hook(None, 'iteration')
|
||||||
if not self._stopped:
|
if not self._stopped:
|
||||||
self._log("waiting... %r" % d)
|
self._log("waiting... %r" % d)
|
||||||
yield d
|
yield d
|
||||||
|
|
||||||
self._log("stopped")
|
self._log("stopped")
|
||||||
|
|
||||||
|
def _next_scan_delay(self):
|
||||||
|
return self._turn_delay
|
||||||
|
|
||||||
def _when_queue_is_empty(self):
|
def _when_queue_is_empty(self):
|
||||||
return
|
return
|
||||||
|
|
||||||
@ -346,7 +338,7 @@ class UploadItem(QueuedItem):
|
|||||||
class Uploader(QueueMixin):
|
class Uploader(QueueMixin):
|
||||||
|
|
||||||
def __init__(self, client, local_path_u, db, upload_dirnode, pending_delay, clock):
|
def __init__(self, client, local_path_u, db, upload_dirnode, pending_delay, clock):
|
||||||
QueueMixin.__init__(self, client, local_path_u, db, 'uploader', clock, delay=pending_delay)
|
QueueMixin.__init__(self, client, local_path_u, db, 'uploader', clock)
|
||||||
|
|
||||||
self.is_ready = False
|
self.is_ready = False
|
||||||
|
|
||||||
@ -364,6 +356,7 @@ class Uploader(QueueMixin):
|
|||||||
|
|
||||||
self._periodic_full_scan_duration = 10 * 60 # perform a full scan every 10 minutes
|
self._periodic_full_scan_duration = 10 * 60 # perform a full scan every 10 minutes
|
||||||
self._periodic_callid = None
|
self._periodic_callid = None
|
||||||
|
self._turn_delay = pending_delay
|
||||||
|
|
||||||
if hasattr(self._notifier, 'set_pending_delay'):
|
if hasattr(self._notifier, 'set_pending_delay'):
|
||||||
self._notifier.set_pending_delay(pending_delay)
|
self._notifier.set_pending_delay(pending_delay)
|
||||||
@ -753,12 +746,11 @@ class DownloadItem(QueuedItem):
|
|||||||
|
|
||||||
|
|
||||||
class Downloader(QueueMixin, WriteFileMixin):
|
class Downloader(QueueMixin, WriteFileMixin):
|
||||||
scan_interval = 3
|
|
||||||
|
|
||||||
def __init__(self, client, local_path_u, db, collective_dirnode,
|
def __init__(self, client, local_path_u, db, collective_dirnode,
|
||||||
upload_readonly_dircap, clock, is_upload_pending, umask,
|
upload_readonly_dircap, clock, is_upload_pending, umask,
|
||||||
status_reporter):
|
status_reporter, poll_interval=3):
|
||||||
QueueMixin.__init__(self, client, local_path_u, db, 'downloader', clock, delay=self.scan_interval)
|
QueueMixin.__init__(self, client, local_path_u, db, 'downloader', clock)
|
||||||
|
|
||||||
if not IDirectoryNode.providedBy(collective_dirnode):
|
if not IDirectoryNode.providedBy(collective_dirnode):
|
||||||
raise AssertionError("The URI in '%s' does not refer to a directory."
|
raise AssertionError("The URI in '%s' does not refer to a directory."
|
||||||
@ -772,11 +764,12 @@ class Downloader(QueueMixin, WriteFileMixin):
|
|||||||
self._is_upload_pending = is_upload_pending
|
self._is_upload_pending = is_upload_pending
|
||||||
self._umask = umask
|
self._umask = umask
|
||||||
self._status_reporter = status_reporter
|
self._status_reporter = status_reporter
|
||||||
|
self._poll_interval = poll_interval
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def start_downloading(self):
|
def start_downloading(self):
|
||||||
self._log("start_downloading")
|
self._log("start_downloading")
|
||||||
self._turn_delay = self.scan_interval
|
self._turn_delay = self._poll_interval
|
||||||
files = self._db.get_all_relpaths()
|
files = self._db.get_all_relpaths()
|
||||||
self._log("all files %s" % files)
|
self._log("all files %s" % files)
|
||||||
|
|
||||||
@ -794,7 +787,7 @@ class Downloader(QueueMixin, WriteFileMixin):
|
|||||||
"Last tried at %s" % self.nice_current_time(),
|
"Last tried at %s" % self.nice_current_time(),
|
||||||
)
|
)
|
||||||
twlog.msg("Magic Folder failed initial scan: %s" % (e,))
|
twlog.msg("Magic Folder failed initial scan: %s" % (e,))
|
||||||
yield task.deferLater(self._clock, self.scan_interval, lambda: None)
|
yield task.deferLater(self._clock, self._poll_interval, lambda: None)
|
||||||
|
|
||||||
def nice_current_time(self):
|
def nice_current_time(self):
|
||||||
return format_time(datetime.fromtimestamp(self._clock.seconds()).timetuple())
|
return format_time(datetime.fromtimestamp(self._clock.seconds()).timetuple())
|
||||||
|
@ -122,8 +122,8 @@ class MagicFolderCLITestMixin(CLITestMixin, GridTestMixin, NonASCIIPathMixin):
|
|||||||
d = defer.succeed(None)
|
d = defer.succeed(None)
|
||||||
def _clean(ign):
|
def _clean(ign):
|
||||||
d = self.magicfolder.finish()
|
d = self.magicfolder.finish()
|
||||||
self.magicfolder.uploader._clock.advance(self.magicfolder.uploader.scan_interval + 1)
|
self.magicfolder.uploader._clock.advance(self.magicfolder.uploader._turn_delay + 1)
|
||||||
self.magicfolder.downloader._clock.advance(self.magicfolder.downloader.scan_interval + 1)
|
self.magicfolder.downloader._clock.advance(self.magicfolder.downloader._turn_delay + 1)
|
||||||
return d
|
return d
|
||||||
|
|
||||||
d.addCallback(_clean)
|
d.addCallback(_clean)
|
||||||
|
@ -111,13 +111,13 @@ def iterate_downloader(magic):
|
|||||||
# can do either of these:
|
# can do either of these:
|
||||||
#d = magic.downloader._process_deque()
|
#d = magic.downloader._process_deque()
|
||||||
d = magic.downloader.set_hook('iteration')
|
d = magic.downloader.set_hook('iteration')
|
||||||
magic.downloader._clock.advance(magic.downloader.scan_interval + 1)
|
magic.downloader._clock.advance(magic.downloader._turn_delay + 1)
|
||||||
return d
|
return d
|
||||||
|
|
||||||
|
|
||||||
def iterate_uploader(magic):
|
def iterate_uploader(magic):
|
||||||
d = magic.uploader.set_hook('iteration')
|
d = magic.uploader.set_hook('iteration')
|
||||||
magic.uploader._clock.advance(magic.uploader.scan_interval + 1)
|
magic.uploader._clock.advance(magic.uploader._turn_delay + 1)
|
||||||
return d
|
return d
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
@ -308,7 +308,7 @@ class MagicFolderAliceBobTestMixin(MagicFolderCLITestMixin, ShouldFailMixin, Rea
|
|||||||
self.alice_fileops = FileOperationsHelper(self.alice_magicfolder.uploader, self.inject_inotify)
|
self.alice_fileops = FileOperationsHelper(self.alice_magicfolder.uploader, self.inject_inotify)
|
||||||
d0 = self.alice_magicfolder.uploader.set_hook('iteration')
|
d0 = self.alice_magicfolder.uploader.set_hook('iteration')
|
||||||
d1 = self.alice_magicfolder.downloader.set_hook('iteration')
|
d1 = self.alice_magicfolder.downloader.set_hook('iteration')
|
||||||
self.alice_clock.advance(self.alice_magicfolder.uploader.scan_interval + 1)
|
self.alice_clock.advance(self.alice_magicfolder.uploader._turn_delay + 1)
|
||||||
d0.addCallback(lambda ign: d1)
|
d0.addCallback(lambda ign: d1)
|
||||||
d0.addCallback(lambda ign: result)
|
d0.addCallback(lambda ign: result)
|
||||||
return d0
|
return d0
|
||||||
@ -332,7 +332,7 @@ class MagicFolderAliceBobTestMixin(MagicFolderCLITestMixin, ShouldFailMixin, Rea
|
|||||||
self.bob_fileops = FileOperationsHelper(self.bob_magicfolder.uploader, self.inject_inotify)
|
self.bob_fileops = FileOperationsHelper(self.bob_magicfolder.uploader, self.inject_inotify)
|
||||||
d0 = self.bob_magicfolder.uploader.set_hook('iteration')
|
d0 = self.bob_magicfolder.uploader.set_hook('iteration')
|
||||||
d1 = self.bob_magicfolder.downloader.set_hook('iteration')
|
d1 = self.bob_magicfolder.downloader.set_hook('iteration')
|
||||||
self.bob_clock.advance(self.alice_magicfolder.uploader.scan_interval + 1)
|
self.bob_clock.advance(self.alice_magicfolder.uploader._turn_delay + 1)
|
||||||
d0.addCallback(lambda ign: d1)
|
d0.addCallback(lambda ign: d1)
|
||||||
d0.addCallback(lambda ign: result)
|
d0.addCallback(lambda ign: result)
|
||||||
return d0
|
return d0
|
||||||
@ -347,7 +347,7 @@ class MagicFolderAliceBobTestMixin(MagicFolderCLITestMixin, ShouldFailMixin, Rea
|
|||||||
|
|
||||||
for mf in [self.alice_magicfolder, self.bob_magicfolder]:
|
for mf in [self.alice_magicfolder, self.bob_magicfolder]:
|
||||||
for loader in [mf.uploader, mf.downloader]:
|
for loader in [mf.uploader, mf.downloader]:
|
||||||
loader._clock.advance(loader.scan_interval + 1)
|
loader._clock.advance(loader._turn_delay + 1)
|
||||||
|
|
||||||
yield d0
|
yield d0
|
||||||
yield d1
|
yield d1
|
||||||
|
Reference in New Issue
Block a user