core, plugins file, XEP-0234, bridge: progression api enhancement:
- progressStarted have a new metadata parameter, useful to know the kind of progression, direction, etc. Check bridge doc
- progressGetAllMetadata can be used to retrieve this data and discover on currently running progressions
- progressFinished also have a new metadata parameter, used to e.g. indicate that hash is checked
- core: fixed progressGetAll
- file, XEP-0234: implemented the API modifications, hash is returned on progressFinished
- file: SatFile.checkSize allows to check size independently of close (be sure that all the data have been transfered though)
This commit is contained in:
parent
9b2055bd3f
commit
55c75bc304
5 changed files with 161 additions and 29 deletions
|
|
@ -196,13 +196,13 @@ class DbusObject(dbus.service.Object):
|
|||
pass
|
||||
|
||||
@dbus.service.signal(const_INT_PREFIX+const_CORE_SUFFIX,
|
||||
signature='ss')
|
||||
def progressFinished(self, id, profile):
|
||||
signature='sa{ss}s')
|
||||
def progressFinished(self, id, metadata, profile):
|
||||
pass
|
||||
|
||||
@dbus.service.signal(const_INT_PREFIX+const_CORE_SUFFIX,
|
||||
signature='ss')
|
||||
def progressStarted(self, id, profile):
|
||||
signature='sa{ss}s')
|
||||
def progressStarted(self, id, metadata, profile):
|
||||
pass
|
||||
|
||||
@dbus.service.signal(const_INT_PREFIX+const_CORE_SUFFIX,
|
||||
|
|
@ -458,6 +458,12 @@ class DbusObject(dbus.service.Object):
|
|||
def progressGetAll(self, profile):
|
||||
return self._callback("progressGetAll", unicode(profile))
|
||||
|
||||
@dbus.service.method(const_INT_PREFIX+const_CORE_SUFFIX,
|
||||
in_signature='s', out_signature='a{sa{sa{ss}}}',
|
||||
async_callbacks=None)
|
||||
def progressGetAllMetadata(self, profile):
|
||||
return self._callback("progressGetAllMetadata", unicode(profile))
|
||||
|
||||
@dbus.service.method(const_INT_PREFIX+const_CORE_SUFFIX,
|
||||
in_signature='s', out_signature='b',
|
||||
async_callbacks=None)
|
||||
|
|
@ -638,11 +644,11 @@ class DBusBridge(Bridge):
|
|||
def progressError(self, error, profile, arg_2):
|
||||
self.dbus_bridge.progressError(error, profile, arg_2)
|
||||
|
||||
def progressFinished(self, id, profile):
|
||||
self.dbus_bridge.progressFinished(id, profile)
|
||||
def progressFinished(self, id, metadata, profile):
|
||||
self.dbus_bridge.progressFinished(id, metadata, profile)
|
||||
|
||||
def progressStarted(self, id, profile):
|
||||
self.dbus_bridge.progressStarted(id, profile)
|
||||
def progressStarted(self, id, metadata, profile):
|
||||
self.dbus_bridge.progressStarted(id, metadata, profile)
|
||||
|
||||
def subscribe(self, sub_type, entity_jid, profile):
|
||||
self.dbus_bridge.subscribe(sub_type, entity_jid, profile)
|
||||
|
|
|
|||
|
|
@ -170,18 +170,28 @@ doc_param_3=%(doc_profile)s
|
|||
[progressStarted]
|
||||
type=signal
|
||||
category=core
|
||||
sig_in=ss
|
||||
sig_in=sa{ss}s
|
||||
doc=A progressing operation has just started
|
||||
doc_param_0=id: id of the progression operation
|
||||
doc_param_1=%(doc_profile)s
|
||||
doc_param_1=metadata: dict of progress metadata, key can be:
|
||||
- name: name of the progression, full path for a file
|
||||
- direction: "in" for incoming data, "out" else
|
||||
- type: type of the progression:
|
||||
C.META_TYPE_FILE: file transfer
|
||||
doc_param_2=%(doc_profile)s
|
||||
|
||||
[progressFinished]
|
||||
type=signal
|
||||
category=core
|
||||
sig_in=ss
|
||||
sig_in=sa{ss}s
|
||||
doc=A progressing operation is finished
|
||||
doc_param_0=id: id of the progression operation
|
||||
doc_param_1=%(doc_profile)s
|
||||
doc_param_1=metadata: dict of progress status metadata, key can be:
|
||||
- hash: value of the computed hash
|
||||
- hash_algo: alrorithm used to compute hash
|
||||
- hash_verified: C.BOOL_TRUE if hash is verified and OK
|
||||
C.BOOL_FALSE if hash was not received ([progressError] will be used if there is a mismatch)
|
||||
doc_param_2=%(doc_profile)s
|
||||
|
||||
[progressError]
|
||||
type=signal
|
||||
|
|
@ -666,6 +676,17 @@ doc_return=dict with progress informations:
|
|||
- position: current position
|
||||
- size: end position
|
||||
|
||||
[progressGetAllMetadata]
|
||||
type=method
|
||||
category=core
|
||||
sig_in=s
|
||||
sig_out=a{sa{sa{ss}}}
|
||||
doc=Get all active progress informations
|
||||
doc_param_0=%(doc_profile)s or C.PROF_KEY_ALL for all profiles
|
||||
doc_return= a dict which map profile to progress_dict
|
||||
progress_dict map progress_id to progress_metadata
|
||||
progress_metadata is the same dict as sent by [progressStarted]
|
||||
|
||||
[progressGetAll]
|
||||
type=method
|
||||
category=core
|
||||
|
|
|
|||
|
|
@ -824,12 +824,14 @@ class SAT(service.Service):
|
|||
client = self.getClient(profile)
|
||||
return [action_tuple[:-1] for action_tuple in client.actions.itervalues()]
|
||||
|
||||
def registerProgressCb(self, progress_id, callback, profile):
|
||||
def registerProgressCb(self, progress_id, callback, metadata=None, profile=C.PROF_KEY_NONE):
|
||||
"""Register a callback called when progress is requested for id"""
|
||||
if metadata is None:
|
||||
metadata = {}
|
||||
client = self.getClient(profile)
|
||||
if progress_id in client._progress_cb:
|
||||
raise exceptions.ConflictError(u"Progress ID is not unique !")
|
||||
client._progress_cb[progress_id] = callback
|
||||
client._progress_cb[progress_id] = (callback, metadata)
|
||||
|
||||
def removeProgressCb(self, progress_id, profile):
|
||||
"""Remove a progress callback"""
|
||||
|
|
@ -855,7 +857,7 @@ class SAT(service.Service):
|
|||
"""
|
||||
client = self.getClient(profile)
|
||||
try:
|
||||
data = client._progress_cb[progress_id](progress_id, profile)
|
||||
data = client._progress_cb[progress_id][0](progress_id, profile)
|
||||
except KeyError:
|
||||
data = {}
|
||||
return data
|
||||
|
|
@ -868,11 +870,30 @@ class SAT(service.Service):
|
|||
data[key] = unicode(value)
|
||||
return progress_all
|
||||
|
||||
def progressGetAll(self, profile_key):
|
||||
"""Return all progress informations
|
||||
def progressGetAllMetadata(self, profile_key):
|
||||
"""Return all progress metadata at once
|
||||
|
||||
@param profile_key: %(doc_profile)s get all progress from this profile
|
||||
if C.PROF_KEY_ALL is used, all progress from all profiles are returned
|
||||
@param profile_key: %(doc_profile)s
|
||||
if C.PROF_KEY_ALL is used, all progress metadata from all profiles are returned
|
||||
@return (dict[dict[dict]]): a dict which map profile to progress_dict
|
||||
progress_dict map progress_id to progress_data
|
||||
progress_metadata is the same dict as sent by [progressStarted]
|
||||
"""
|
||||
clients = self.getClients(profile_key)
|
||||
progress_all = {}
|
||||
for client in clients:
|
||||
profile = client.profile
|
||||
progress_dict = {}
|
||||
progress_all[profile] = progress_dict
|
||||
for progress_id, (dummy, progress_metadata) in client._progress_cb.iteritems():
|
||||
progress_dict[progress_id] = progress_metadata
|
||||
return progress_all
|
||||
|
||||
def progressGetAll(self, profile_key):
|
||||
"""Return all progress status at once
|
||||
|
||||
@param profile_key: %(doc_profile)s
|
||||
if C.PROF_KEY_ALL is used, all progress status from all profiles are returned
|
||||
@return (dict[dict[dict]]): a dict which map profile to progress_dict
|
||||
progress_dict map progress_id to progress_data
|
||||
progress_data is the same dict as returned by [progressGet]
|
||||
|
|
@ -883,10 +904,8 @@ class SAT(service.Service):
|
|||
profile = client.profile
|
||||
progress_dict = {}
|
||||
progress_all[profile] = progress_dict
|
||||
for progress_id, progress_cb in client._progress_cb.iteritems():
|
||||
data = {}
|
||||
progress_dict[progress_id] = data
|
||||
progress_dict[progress_id] = progress_cb(progress_id, data, profile)
|
||||
for progress_id, (progress_cb, dummy) in client._progress_cb.iteritems():
|
||||
progress_dict[progress_id] = progress_cb(progress_id, profile)
|
||||
return progress_all
|
||||
|
||||
def registerCallback(self, callback, *args, **kwargs):
|
||||
|
|
|
|||
|
|
@ -71,16 +71,52 @@ class SatFile(object):
|
|||
self.size = size
|
||||
self.data_cb = data_cb
|
||||
self.profile = profile
|
||||
self.host.registerProgressCb(self.uid, self.getProgress, profile)
|
||||
self.host.bridge.progressStarted(self.uid, self.profile)
|
||||
metadata = self.getProgressMetadata()
|
||||
self.host.registerProgressCb(self.uid, self.getProgress, metadata, profile=profile)
|
||||
self.host.bridge.progressStarted(self.uid, metadata, self.profile)
|
||||
|
||||
def close(self):
|
||||
def checkSize(self):
|
||||
"""Check that current size correspond to given size
|
||||
|
||||
must be used when the transfer is supposed to be finished
|
||||
@return (bool): True if the position is the same as given size
|
||||
@raise exceptions.NotFound: size has not be specified
|
||||
"""
|
||||
position = self._file.tell()
|
||||
if self.size is None:
|
||||
raise exceptions.NotFound
|
||||
return position == self.size
|
||||
|
||||
|
||||
def close(self, progress_metadata=None, error=None):
|
||||
"""Close the current file
|
||||
|
||||
@param progress_metadata(None, dict): metadata to send with _onProgressFinished message
|
||||
@param error(None, unicode): set to an error message if progress was not successful
|
||||
mutually exclusive with progress_metadata
|
||||
error can happen even if error is None, if current size differ from given size
|
||||
"""
|
||||
if error is None:
|
||||
try:
|
||||
size_ok = self.checkSize()
|
||||
except exceptions.NotFound:
|
||||
size_ok = True
|
||||
finally:
|
||||
if not size_ok:
|
||||
error = u'declared size and actual mismatch'
|
||||
log.warning(u"successful close was requested, but there is a size mismatch")
|
||||
progress_metadata = None
|
||||
|
||||
self._file.close()
|
||||
if not self.size or self.size == position:
|
||||
self.host.bridge.progressFinished(self.uid, self.profile)
|
||||
|
||||
if error is None:
|
||||
if progress_metadata is None:
|
||||
progress_metadata = {}
|
||||
self.host.bridge.progressFinished(self.uid, progress_metadata, self.profile)
|
||||
else:
|
||||
self.host.bridge.progressError(self.uid, u"size doesn't match", self.profile)
|
||||
assert progress_metadata is None
|
||||
self.host.bridge.progressError(self.uid, error, self.profile)
|
||||
|
||||
self.host.removeProgressCb(self.uid, self.profile)
|
||||
|
||||
def flush(self):
|
||||
|
|
@ -103,6 +139,32 @@ class SatFile(object):
|
|||
def tell(self):
|
||||
return self._file.tell()
|
||||
|
||||
def mode(self):
|
||||
return self._file.mode()
|
||||
|
||||
def getProgressMetadata(self):
|
||||
"""Return progression metadata as given to progressStarted
|
||||
|
||||
@return (dict): metadata (check bridge for documentation)
|
||||
"""
|
||||
metadata = {'type': C.META_TYPE_FILE}
|
||||
|
||||
mode = self._file.mode
|
||||
if '+' in mode:
|
||||
pass # we have no direction in read/write modes
|
||||
elif mode in ('r', 'rb'):
|
||||
metadata['direction'] = 'out'
|
||||
elif mode in ('w', 'wb'):
|
||||
metadata['direction'] = 'in'
|
||||
elif 'U' in mode:
|
||||
metadata['direction'] = 'out'
|
||||
else:
|
||||
raise exceptions.InternalError
|
||||
|
||||
metadata['name'] = self._file.name
|
||||
|
||||
return metadata
|
||||
|
||||
def getProgress(self, progress_id, profile):
|
||||
ret = {'position': self._file.tell()}
|
||||
if self.size:
|
||||
|
|
|
|||
|
|
@ -283,11 +283,21 @@ class XEP_0234(object):
|
|||
|
||||
if hash_ == hash_given:
|
||||
log.info(u"Hash checked, file was successfully transfered: {}".format(hash_))
|
||||
progress_metadata = {'hash': hash_,
|
||||
'hash_algo': file_data['hash_algo'],
|
||||
'hash_verified': C.BOOL_TRUE
|
||||
}
|
||||
error = None
|
||||
else:
|
||||
log.warning(u"Hash mismatch, the file was not transfered correctly")
|
||||
progress_metadata=None
|
||||
error = u"Hash mismatch: given={algo}:{given}, calculated={algo}:{our}".format(
|
||||
algo = file_data['hash_algo'],
|
||||
given = hash_given,
|
||||
our = hash_)
|
||||
|
||||
self._j.delayedContentTerminate(session, content_name, profile=profile)
|
||||
content_data['file_obj'].close()
|
||||
content_data['file_obj'].close(progress_metadata, error)
|
||||
# we may have the last_try timer still active, so we try to cancel it
|
||||
try:
|
||||
content_data['last_try_timer'].cancel()
|
||||
|
|
|
|||
Loading…
Reference in a new issue