4 from twisted.internet import defer
5 from twisted.python.filepath import FilePath
6 from twisted.application import service
7 from foolscap.api import eventually
9 from allmydata.interfaces import IDirectoryNode
11 from allmydata.util.encodingutil import quote_output, get_filesystem_encoding
12 from allmydata.immutable.upload import FileName
15 class DropUploader(service.MultiService):
16 def __init__(self, client, upload_dircap, local_dir_utf8, inotify=None):
17 service.MultiService.__init__(self)
20 local_dir_u = os.path.expanduser(local_dir_utf8.decode('utf-8'))
21 if sys.platform == "win32":
22 local_dir = local_dir_u
24 local_dir = local_dir_u.encode(get_filesystem_encoding())
25 except (UnicodeEncodeError, UnicodeDecodeError):
26 raise AssertionError("The '[drop_upload] local.directory' parameter %s was not valid UTF-8 or "
27 "could not be represented in the filesystem encoding."
28 % quote_output(local_dir_utf8))
31 self._stats_provider = client.stats_provider
32 self._convergence = client.convergence
33 self._local_path = FilePath(local_dir)
36 from twisted.internet import inotify
37 self._inotify = inotify
39 if not self._local_path.isdir():
40 raise AssertionError("The drop-upload local path %r was not an existing directory." % quote_output(local_dir_u))
42 # TODO: allow a path rather than a cap URI.
43 self._parent = self._client.create_node_from_uri(upload_dircap)
44 if not IDirectoryNode.providedBy(self._parent):
45 raise AssertionError("The '[drop_upload] upload.dircap' parameter does not refer to a directory.")
46 if self._parent.is_unknown() or self._parent.is_readonly():
47 raise AssertionError("The '[drop_upload] upload.dircap' parameter is not a writecap to a directory.")
49 self._uploaded_callback = lambda ign: None
51 self._notifier = inotify.INotify()
53 # We don't watch for IN_CREATE, because that would cause us to read and upload a
54 # possibly-incomplete file before the application has closed it. There should always
55 # be an IN_CLOSE_WRITE after an IN_CREATE (I think).
56 # TODO: what about IN_MOVE_SELF or IN_UNMOUNT?
57 mask = inotify.IN_CLOSE_WRITE | inotify.IN_MOVED_TO | inotify.IN_ONLYDIR
58 self._notifier.watch(self._local_path, mask=mask, callbacks=[self._notify])
61 d = self._notifier.startReading()
62 self._stats_provider.count('drop_upload.dirs_monitored', 1)
65 def _notify(self, opaque, path, events_mask):
66 self._log("inotify event %r, %r, %r\n" % (opaque, path, ', '.join(self._inotify.humanReadableMask(events_mask))))
68 self._stats_provider.count('drop_upload.files_queued', 1)
69 eventually(self._process, opaque, path, events_mask)
71 def _process(self, opaque, path, events_mask):
72 d = defer.succeed(None)
74 # FIXME: if this already exists as a mutable file, we replace the directory entry,
75 # but we should probably modify the file (as the SFTP frontend does).
77 name = path.basename()
78 # on Windows the name is already Unicode
79 if not isinstance(name, unicode):
80 name = name.decode(get_filesystem_encoding())
82 u = FileName(path.path, self._convergence)
83 return self._parent.add_file(name, u)
84 d.addCallback(_add_file)
87 self._stats_provider.count('drop_upload.files_queued', -1)
88 self._stats_provider.count('drop_upload.files_uploaded', 1)
90 self._stats_provider.count('drop_upload.files_queued', -1)
92 self._log("drop-upload: %r failed to upload due to %r" % (path.path, f))
93 self._stats_provider.count('drop_upload.files_failed', 1)
96 self._log("drop-upload: notified file %r disappeared "
97 "(this is normal for temporary files): %r" % (path.path, f))
98 self._stats_provider.count('drop_upload.files_disappeared', 1)
100 d.addCallbacks(_succeeded, _failed)
101 d.addBoth(self._uploaded_callback)
104 def set_uploaded_callback(self, callback):
105 """This sets a function that will be called after a file has been uploaded."""
106 self._uploaded_callback = callback
108 def finish(self, for_tests=False):
109 self._notifier.stopReading()
110 self._stats_provider.count('drop_upload.dirs_monitored', -1)
111 if for_tests and hasattr(self._notifier, 'wait_until_stopped'):
112 return self._notifier.wait_until_stopped()
114 return defer.succeed(None)
117 self._client.log(msg)
118 #open("events", "ab+").write(msg)