Mercurial > libervia-backend
changeset 1524:7b0fcefd52d4
plugin XEP-0047, XEP-0096: In-Band Bystream plugin cleaning:
- some renaming, comments improvments, etc
- progress callback is no more managed here, as it will be managed by application
- no more file data is used, beside file_obj
- a proper Deferred is used instead of success and error callbacks
- more clean error sending method
plugin XEP-0096 has been updated to handle changes. Its temporarily partially broken though
| author | Goffi <goffi@goffi.org> |
|---|---|
| date | Fri, 25 Sep 2015 19:19:12 +0200 |
| parents | 0209f8d35873 |
| children | 49d33cb48207 |
| files | src/plugins/plugin_xep_0047.py src/plugins/plugin_xep_0096.py |
| diffstat | 2 files changed, 248 insertions(+), 222 deletions(-) [+] |
line wrap: on
line diff
--- a/src/plugins/plugin_xep_0047.py Fri Sep 25 19:19:12 2015 +0200 +++ b/src/plugins/plugin_xep_0047.py Fri Sep 25 19:19:12 2015 +0200 @@ -20,9 +20,14 @@ from sat.core.i18n import _ from sat.core.log import getLogger log = getLogger(__name__) -from twisted.words.protocols.jabber import client as jabber_client, jid -from twisted.words.xish import domish +from sat.core import exceptions +from twisted.words.protocols.jabber import client as jabber_client +from twisted.words.protocols.jabber import jid +from twisted.words.protocols.jabber import xmlstream +from twisted.words.protocols.jabber import error from twisted.internet import reactor +from twisted.internet import defer +from twisted.python import failure from wokkel import disco, iwokkel @@ -39,11 +44,11 @@ IQ_SET = '/iq[@type="set"]' NS_IBB = 'http://jabber.org/protocol/ibb' IBB_OPEN = IQ_SET + '/open[@xmlns="' + NS_IBB + '"]' -IBB_CLOSE = IQ_SET + '/close[@xmlns="' + NS_IBB + '" and @sid="%s"]' -IBB_IQ_DATA = IQ_SET + '/data[@xmlns="' + NS_IBB + '" and @sid="%s"]' -IBB_MESSAGE_DATA = MESSAGE + '/data[@xmlns="' + NS_IBB + '" and @sid="%s"]' +IBB_CLOSE = IQ_SET + '/close[@xmlns="' + NS_IBB + '" and @sid="{}"]' +IBB_IQ_DATA = IQ_SET + '/data[@xmlns="' + NS_IBB + '" and @sid="{}"]' +IBB_MESSAGE_DATA = MESSAGE + '/data[@xmlns="' + NS_IBB + '" and @sid="{}"]' TIMEOUT = 60 # timeout for workflow -BLOCK_SIZE = 4096 +DEFER_KEY = 'finished' # key of the deferred used to track session end PLUGIN_INFO = { "name": "In-Band Bytestream Plugin", @@ -58,6 +63,7 @@ class XEP_0047(object): NAMESPACE = NS_IBB + BLOCK_SIZE = 4096 def __init__(self, host): log.info(_("In-Band Bytestreams plugin initialization")) @@ -70,275 +76,277 @@ client = self.host.getClient(profile) client.xep_0047_current_stream = {} # key: stream_id, value: data(dict) - def _timeOut(self, sid, profile): - """Delecte current_stream id, called after timeout - @param id: id of client.xep_0047_current_stream""" - log.info(_("In-Band Bytestream: TimeOut reached for id %(sid)s [%(profile)s]") - % {"sid": sid, "profile": profile}) - self._killId(sid, False, "TIMEOUT", profile) + def _timeOut(self, sid, client): + """Delete current_stream id, called after timeout + + @param sid(unicode): session id of client.xep_0047_current_stream + @param client: %(doc_client)s + """ + log.info(_("In-Band Bytestream: TimeOut reached for id {sid} [{profile}]") + .format(sid=sid, profile=client.profile)) + self._killSession(sid, client, "TIMEOUT") - def _killId(self, sid, success=False, failure_reason="UNKNOWN", profile=None): - """Delete an current_stream id, clean up associated observers - @param sid: id of client.xep_0047_current_stream""" - assert(profile) - client = self.host.getClient(profile) - if sid not in client.xep_0047_current_stream: + def _killSession(self, sid, client, failure_reason=None): + """Delete a current_stream id, clean up associated observers + + @param sid(unicode): session id + @param client: %(doc_client)s + @param failure_reason(None, unicode): if None the session is successful + else, will be used to call failure_cb + """ + try: + session = client.xep_0047_current_stream[sid] + except KeyError: log.warning(_("kill id called on a non existant id")) return - if "observer_cb" in client.xep_0047_current_stream[sid]: - client.xmlstream.removeObserver(client.xep_0047_current_stream[sid]["event_data"], client.xep_0047_current_stream[sid]["observer_cb"]) - if client.xep_0047_current_stream[sid]['timer'].active(): - client.xep_0047_current_stream[sid]['timer'].cancel() - if "size" in client.xep_0047_current_stream[sid]: - self.host.removeProgressCB(sid, profile) - file_obj = client.xep_0047_current_stream[sid]['file_obj'] - success_cb = client.xep_0047_current_stream[sid]['success_cb'] - failure_cb = client.xep_0047_current_stream[sid]['failure_cb'] + try: + observer_cb = session['observer_cb'] + except KeyError: + pass + else: + client.xmlstream.removeObserver(session["event_data"], observer_cb) + + if session['timer'].active(): + session['timer'].cancel() del client.xep_0047_current_stream[sid] + success = failure_reason is None + stream_d = session[DEFER_KEY] + if success: - success_cb(sid, file_obj, NS_IBB, profile) + stream_d.callback(None) else: - failure_cb(sid, file_obj, NS_IBB, failure_reason, profile) + stream_d.errback(failure.Failure(exceptions.DataError(failure_reason))) - def getProgress(self, sid, data, profile): - """Fill data with position of current transfer""" - client = self.host.getClient(profile) - try: - file_obj = client.xep_0047_current_stream[sid]["file_obj"] - data["position"] = str(file_obj.tell()) - data["size"] = str(client.xep_0047_current_stream[sid]["size"]) - except: - pass + def createSession(self, *args, **kwargs): + """like [_createSession] but return the session deferred instead of the whole session - def prepareToReceive(self, from_jid, sid, file_obj, size, success_cb, failure_cb, profile): + session deferred is fired when transfer is finished + """ + return self._createSession(*args, **kwargs)[DEFER_KEY] + + def _createSession(self, file_obj, to_jid, sid, profile): """Called when a bytestream is imminent - @param from_jid: jid of the sender - @param sid: Stream id - @param file_obj: File object where data will be written - @param size: full size of the data, or None if unknown - @param success_cb: method to call when successfuly finished - @param failure_cb: method to call when something goes wrong - @param profile: %(doc_profile)s""" + + @param file_obj(file): File object where data will be written + @param to_jid(jid.JId): jid of the other peer + @param sid(unicode): session id + @param profile: %(doc_profile)s + @return (dict): session data + """ client = self.host.getClient(profile) - data = client.xep_0047_current_stream[sid] = {} - data["from"] = from_jid - data["file_obj"] = file_obj - data["seq"] = -1 - if size: - data["size"] = size - self.host.registerProgressCB(sid, self.getProgress, profile) - data["timer"] = reactor.callLater(TIMEOUT, self._timeOut, sid) - data["success_cb"] = success_cb - data["failure_cb"] = failure_cb + if sid in client.xep_0047_current_stream: + raise exceptions.ConflictError(u'A session with this id already exists !') + session_data = client.xep_0047_current_stream[sid] = \ + {'id': sid, + DEFER_KEY: defer.Deferred(), + 'to': to_jid, + 'file_obj': file_obj, + 'seq': -1, + 'timer': reactor.callLater(TIMEOUT, self._timeOut, sid, client), + } - def streamOpening(self, IQ, profile): - log.debug(_("IBB stream opening")) - IQ.handled = True + return session_data + + def _onIBBOpen(self, iq_elt, profile): + """"Called when an IBB <open> element is received + + @param iq_elt(domish.Element): the whole <iq> stanza + @param profile: %(doc_profile)s + """ + log.debug(_(u"IBB stream opening")) + iq_elt.handled = True client = self.host.getClient(profile) - open_elt = IQ.firstChildElement() + open_elt = iq_elt.elements(NS_IBB, 'open').next() block_size = open_elt.getAttribute('block-size') sid = open_elt.getAttribute('sid') stanza = open_elt.getAttribute('stanza', 'iq') if not sid or not block_size or int(block_size) > 65535: - log.warning(_(u"malformed IBB transfer: %s" % IQ['id'])) - self.sendNotAcceptableError(IQ['id'], IQ['from'], client.xmlstream) - return + return self._sendError('not-acceptable', sid or None, iq_elt, client) if not sid in client.xep_0047_current_stream: log.warning(_(u"Ignoring unexpected IBB transfer: %s" % sid)) - self.sendNotAcceptableError(IQ['id'], IQ['from'], client.xmlstream) - return - if client.xep_0047_current_stream[sid]["from"] != jid.JID(IQ['from']): + return self._sendError('not-acceptable', sid or None, iq_elt, client) + session_data = client.xep_0047_current_stream[sid] + if session_data["to"] != jid.JID(iq_elt['from']): log.warning(_("sended jid inconsistency (man in the middle attack attempt ?)")) - self.sendNotAcceptableError(IQ['id'], IQ['from'], client.xmlstream) - self._killId(sid, False, "PROTOCOL_ERROR", profile=profile) - return + return self._sendError('not-acceptable', sid, iq_elt, client) - #at this stage, the session looks ok and will be accepted + # at this stage, the session looks ok and will be accepted - #we reset the timeout: - client.xep_0047_current_stream[sid]["timer"].reset(TIMEOUT) + # we reset the timeout: + session_data["timer"].reset(TIMEOUT) - #we save the xmlstream, events and observer data to allow observer removal - client.xep_0047_current_stream[sid]["event_data"] = event_data = (IBB_MESSAGE_DATA if stanza == 'message' else IBB_IQ_DATA) % sid - client.xep_0047_current_stream[sid]["observer_cb"] = observer_cb = self.messageData if stanza == 'message' else self.iqData - event_close = IBB_CLOSE % sid - #we now set the stream observer to look after data packet + # we save the xmlstream, events and observer data to allow observer removal + session_data["event_data"] = event_data = (IBB_MESSAGE_DATA if stanza == 'message' else IBB_IQ_DATA).format(sid) + session_data["observer_cb"] = observer_cb = self._onIBBData + event_close = IBB_CLOSE.format(sid) + # we now set the stream observer to look after data packet + # FIXME: if we never get the events, the observers stay. + # would be better to have generic observer and check id once triggered client.xmlstream.addObserver(event_data, observer_cb, profile=profile) - client.xmlstream.addOnetimeObserver(event_close, self.streamClosing, profile=profile) - #finally, we send the accept stanza - result = domish.Element((None, 'iq')) - result['type'] = 'result' - result['id'] = IQ['id'] - result['to'] = IQ['from'] - client.xmlstream.send(result) + client.xmlstream.addOnetimeObserver(event_close, self._onIBBClose, profile=profile) + # finally, we send the accept stanza + iq_result_elt = xmlstream.toResponse(iq_elt, 'result') + client.xmlstream.send(iq_result_elt) - def streamClosing(self, IQ, profile): - IQ.handled = True + def _onIBBClose(self, iq_elt, profile): + """"Called when an IBB <close> element is received + + @param iq_elt(domish.Element): the whole <iq> stanza + @param profile: %(doc_profile)s + """ + iq_elt.handled = True client = self.host.getClient(profile) log.debug(_("IBB stream closing")) - data_elt = IQ.firstChildElement() - sid = data_elt.getAttribute('sid') - result = domish.Element((None, 'iq')) - result['type'] = 'result' - result['id'] = IQ['id'] - result['to'] = IQ['from'] - client.xmlstream.send(result) - self._killId(sid, success=True, profile=profile) + close_elt = iq_elt.elements(NS_IBB, 'close').next() + # XXX: this observer is only triggered on valid sid, so we don't need to check it + sid = close_elt['sid'] + + iq_result_elt = xmlstream.toResponse(iq_elt, 'result') + client.xmlstream.send(iq_result_elt) + self._killSession(sid, client) - def iqData(self, IQ, profile): - IQ.handled = True + def _onIBBData(self, element, profile): + """Observer called on <iq> or <message> stanzas with data element + + Manage the data elelement (check validity and write to the file_obj) + @param element(domish.Element): <iq> or <message> stanza + @param profile: %(doc_profile)s + """ + element.handled = True client = self.host.getClient(profile) - data_elt = IQ.firstChildElement() - - if self._manageDataElt(data_elt, 'iq', IQ['id'], jid.JID(IQ['from']), profile): - #and send a success answer - result = domish.Element((None, 'iq')) - result['type'] = 'result' - result['id'] = IQ['id'] - result['to'] = IQ['from'] - - client.xmlstream.send(result) - - def messageData(self, message_elt, profile): - sid = message_elt.getAttribute('id', '') - self._manageDataElt(message_elt, 'message', sid, jid.JID(message_elt['from']), profile) + data_elt = element.elements(NS_IBB, 'data').next() + sid = data_elt['sid'] - def _manageDataElt(self, data_elt, stanza, sid, stanza_from_jid, profile): - """Manage the data elelement (check validity and write to the file_obj) - @param data_elt: "data" domish element - @return: True if success""" - client = self.host.getClient(profile) - sid = data_elt.getAttribute('sid') - if sid not in client.xep_0047_current_stream: - log.error(_("Received data for an unknown session id")) - return False + try: + session_data = client.xep_0047_current_stream[sid] + except KeyError: + log.warning(_(u"Received data for an unknown session id")) + return self._sendError('item-not-found', None, element, client) - from_jid = client.xep_0047_current_stream[sid]["from"] - file_obj = client.xep_0047_current_stream[sid]["file_obj"] + from_jid = session_data["to"] + file_obj = session_data["file_obj"] - if stanza_from_jid != from_jid: - log.warning(_("sended jid inconsistency (man in the middle attack attempt ?)")) - if stanza == 'iq': - self.sendNotAcceptableError(sid, from_jid, client.xmlstream) - return False + if from_jid.full() != element['from']: + log.warning(_(u"sended jid inconsistency (man in the middle attack attempt ?)\ninitial={initial}\ngiven={given}").format(initial=from_jid, given=element['from'])) + if element.name == 'iq': + self._sendError('not-acceptable', sid, element, client) + return - client.xep_0047_current_stream[sid]["seq"] += 1 - if int(data_elt.getAttribute("seq", -1)) != client.xep_0047_current_stream[sid]["seq"]: - log.warning(_("Sequence error")) - if stanza == 'iq': - self.sendNotAcceptableError(data_elt["id"], from_jid, client.xmlstream) - return False + session_data["seq"] = (session_data["seq"] + 1) % 65535 + if int(data_elt.getAttribute("seq", -1)) != session_data["seq"]: + log.warning(_(u"Sequence error")) + if element.name == 'iq': + reason = 'not-acceptable' + self._sendError(reason, sid, element, client) + self.terminateStream(session_data, client, reason) + return - #we reset the timeout: - client.xep_0047_current_stream[sid]["timer"].reset(TIMEOUT) + # we reset the timeout: + session_data["timer"].reset(TIMEOUT) - #we can now decode the data + # we can now decode the data try: file_obj.write(base64.b64decode(str(data_elt))) except TypeError: - #The base64 data is invalid - log.warning(_("Invalid base64 data")) - if stanza == 'iq': - self.sendNotAcceptableError(data_elt["id"], from_jid, client.xmlstream) - return False - return True + # The base64 data is invalid + log.warning(_(u"Invalid base64 data")) + if element.name == 'iq': + self._sendError('not-acceptable', sid, element, client) + self.terminateStream(session_data, client, reason) + return - def sendNotAcceptableError(self, iq_id, to_jid, xmlstream): - """Not acceptable error used when the stream is not expected or something is going wrong - @param iq_id: IQ id - @param to_jid: addressee - @param xmlstream: XML stream to use to send the error""" - result = domish.Element((None, 'iq')) - result['type'] = 'result' - result['id'] = iq_id - result['to'] = to_jid - error_el = result.addElement('error') - error_el['type'] = 'cancel' - error_el.addElement(('urn:ietf:params:xml:ns:xmpp-stanzas', 'not-acceptable')) - xmlstream.send(result) + # we can now ack success + if element.name == 'iq': + iq_result_elt = xmlstream.toResponse(element, 'result') + client.xmlstream.send(iq_result_elt) + + def _sendError(self, error_condition, sid, iq_elt, client): + """Send error stanza - def startStream(self, file_obj, to_jid, sid, length, successCb, failureCb, size=None, profile=None): + @param error_condition: one of twisted.words.protocols.jabber.error.STANZA_CONDITIONS keys + @param sid(unicode,None): jingle session id, or None, if session must not be destroyed + @param iq_elt(domish.Element): full <iq> stanza + @param client: %(doc_client)s + """ + iq_elt = error.StanzaError(error_condition).toResponse(iq_elt) + log.warning(u"Error while managing in-band bytestream session, cancelling: {}".format(error_condition)) + if sid is not None: + self._killSession(sid, client, error_condition) + client.xmlstream.send(iq_elt) + + def startStream(self, file_obj, to_jid, sid, block_size=None, profile=None): """Launch the stream workflow - @param file_obj: file_obj to send - @param to_jid: JID of the recipient - @param sid: Stream session id - @param length: number of byte to send, or None to send until the end - @param successCb: method to call when stream successfuly finished - @param failureCb: method to call when something goes wrong - @param profile: %(doc_profile)s""" + + @param file_obj(file): file_obj to send + @param to_jid(jid.JID): JID of the recipient + @param sid(unicode): Stream session id + @param block_size(int, None): size of the block (or None for default) + @param profile: %(doc_profile)s + """ + session_data = self._createSession(file_obj, to_jid, sid, profile) + session_defer = session_data[DEFER_KEY] client = self.host.getClient(profile) - if length is not None: - log.error(_('stream length not managed yet')) - return - data = client.xep_0047_current_stream[sid] = {} - data["timer"] = reactor.callLater(TIMEOUT, self._timeOut, sid) - data["file_obj"] = file_obj - data["to"] = to_jid - data["success_cb"] = successCb - data["failure_cb"] = failureCb - data["block_size"] = BLOCK_SIZE - if size: - data["size"] = size - self.host.registerProgressCB(sid, self.getProgress, profile) + + if block_size is None: + block_size = XEP_0047.BLOCK_SIZE + assert block_size <= 65535 + session_data["block_size"] = block_size + iq_elt = jabber_client.IQ(client.xmlstream, 'set') iq_elt['from'] = client.jid.full() iq_elt['to'] = to_jid.full() open_elt = iq_elt.addElement('open', NS_IBB) - open_elt['block-size'] = str(BLOCK_SIZE) + open_elt['block-size'] = str(block_size) open_elt['sid'] = sid - open_elt['stanza'] = 'iq' - iq_elt.addCallback(self.iqResult, sid, 0, length, profile) + open_elt['stanza'] = 'iq' # TODO: manage <message> stanza ? + iq_elt.addCallback(self._IQDataStream, session_data, client) iq_elt.send() + return session_defer - def iqResult(self, sid, seq, length, profile, iq_elt): - """Called when the result of open iq is received""" - client = self.host.getClient(profile) - data = client.xep_0047_current_stream[sid] - if iq_elt["type"] == "error": - log.warning(_("Transfer failed")) - self.terminateStream(sid, "IQ_ERROR") + def _IQDataStream(self, session_data, client, iq_elt): + """Called during the whole data streaming + + @param session_data(dict): data of this streaming session + @param client: %(doc_client)s + @param iq_elt(domish.Element): iq result + """ + if iq_elt['type'] == 'error': + log.warning(_(u"IBB transfer failed: {}").format(iq_elt)) + self.terminateStream(session_data, client, "IQ_ERROR") return - if data['timer'].active(): - data['timer'].cancel() + session_data["timer"].reset(TIMEOUT) - buffer = data["file_obj"].read(data["block_size"]) - if buffer: + buffer_ = session_data["file_obj"].read(session_data["block_size"]) + if buffer_: next_iq_elt = jabber_client.IQ(client.xmlstream, 'set') - next_iq_elt['to'] = data["to"].full() + next_iq_elt['to'] = session_data["to"].full() data_elt = next_iq_elt.addElement('data', NS_IBB) - data_elt['seq'] = str(seq) - data_elt['sid'] = sid - data_elt.addContent(base64.b64encode(buffer)) - next_iq_elt.addCallback(self.iqResult, sid, seq + 1, length, profile) + seq = session_data['seq'] = (session_data['seq'] + 1) % 65535 + data_elt['seq'] = unicode(seq) + data_elt['sid'] = session_data['id'] + data_elt.addContent(base64.b64encode(buffer_)) + next_iq_elt.addCallback(self._IQDataStream, session_data, client) next_iq_elt.send() else: - self.terminateStream(sid, profile=profile) + self.terminateStream(session_data, client) - def terminateStream(self, sid, failure_reason=None, profile=None): + def terminateStream(self, session_data, client, failure_reason=None): """Terminate the stream session - @param to_jid: recipient - @param sid: Session id - @param file_obj: file object used - @param xmlstream: XML stream used with this session - @param progress_cb: True if we have to remove the progress callback - @param callback: method to call after finishing - @param failure_reason: reason of the failure, or None if steam was successful""" - client = self.host.getClient(profile) - data = client.xep_0047_current_stream[sid] + + @param session_data(dict): data of this streaming session + @param client: %(doc_client)s + @param failure_reason(unicode, None): reason of the failure, or None if steam was successful + """ iq_elt = jabber_client.IQ(client.xmlstream, 'set') - iq_elt['to'] = data["to"].full() + iq_elt['to'] = session_data["to"].full() close_elt = iq_elt.addElement('close', NS_IBB) - close_elt['sid'] = sid + close_elt['sid'] = session_data['id'] iq_elt.send() - self.host.removeProgressCB(sid, profile) - if failure_reason: - self._killId(sid, False, failure_reason, profile=profile) - else: - self._killId(sid, True, profile=profile) + self._killSession(session_data['id'], client, failure_reason) class XEP_0047_handler(XMPPHandler): @@ -348,7 +356,7 @@ self.plugin_parent = parent def connectionInitialized(self): - self.xmlstream.addObserver(IBB_OPEN, self.plugin_parent.streamOpening, profile=self.parent.profile) + self.xmlstream.addObserver(IBB_OPEN, self.plugin_parent._onIBBOpen, profile=self.parent.profile) def getDiscoInfo(self, requestor, target, nodeIdentifier=''): return [disco.DiscoFeature(NS_IBB)]
--- a/src/plugins/plugin_xep_0096.py Fri Sep 25 19:19:12 2015 +0200 +++ b/src/plugins/plugin_xep_0096.py Fri Sep 25 19:19:12 2015 +0200 @@ -26,6 +26,7 @@ from twisted.words.protocols import jabber import os from twisted.internet import reactor +from twisted.python import failure from wokkel import data_form @@ -163,7 +164,9 @@ elif stream_method == self.host.plugins["XEP-0047"].NAMESPACE: file_obj = self._getFileObject(dest_path, can_range) range_offset = file_obj.tell() - self.host.plugins["XEP-0047"].prepareToReceive(jid.JID(data['from']), sid, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed, profile) + d = self.host.plugins["XEP-0047"].createSession(file_obj, jid.JID(data['from']), sid, int(data["size"]), profile) + d.addCallback(self._transferSucceeded, sid, file_obj, self.host.plugins["XEP-0047"].NAMESPACE, profile) + d.addErrback(self._transferFailed, sid, file_obj, self.host.plugins["XEP-0047"].NAMESPACE, profile) else: log.error(_("Unknown stream method, this should not happen at this stage, cancelling transfer")) del(client._xep_0096_waiting_for_approval[sid]) @@ -184,7 +187,10 @@ self.host.plugins["XEP-0095"].sendRejectedError(data["id"], data['from'], profile=profile) del(client._xep_0096_waiting_for_approval[sid]) - def _transferSucceeded(self, sid, file_obj, stream_method, profile): + def _transferSucceeded(self, dummy, sid, file_obj, stream_method, profile): + self.transferSucceeded(sid, file_obj, stream_method, profile) + + def transferSucceeded(self, dummy, sid, file_obj, stream_method, profile): """Called by the stream method when transfer successfuly finished @param id: stream id""" client = self.host.getClient(profile) @@ -193,15 +199,19 @@ del(client._xep_0096_waiting_for_approval[sid]) def _transferFailed(self, sid, file_obj, stream_method, reason, profile): + self.transferFailed(failure.Failure(Exception(reason)), sid, file_obj, stream_method, profile) + + def transferFailed(self, failure, sid, file_obj, stream_method, profile): """Called when something went wrong with the transfer + @param id: stream id - @param reason: can be TIMEOUT, IO_ERROR, PROTOCOL_ERROR""" + """ client = self.host.getClient(profile) data, timeout, stream_method, failed_methods = client._xep_0096_waiting_for_approval[sid] log.warning(_(u'Transfer %(id)s failed with stream method %(s_method)s: %(reason)s') % { 'id': sid, 's_method': stream_method, - 'reason': reason}) + 'reason': unicode(failure)}) filepath = file_obj.name file_obj.close() os.remove(filepath) @@ -250,12 +260,14 @@ file_obj = open(filepath, 'r') if range_offset: file_obj.seek(range_offset) - self.host.plugins["XEP-0065"].startStream(file_obj, jid.JID(IQ['from']), sid, range_length, self.sendSuccessCb, self.sendFailureCb, size, profile) + self.host.plugins["XEP-0065"].startStream(file_obj, jid.JID(IQ['from']), sid, range_length, self._sendSuccessCb, self._sendFailureCb, size, profile) elif stream_method == self.host.plugins["XEP-0047"].NAMESPACE: file_obj = open(filepath, 'r') if range_offset: file_obj.seek(range_offset) - self.host.plugins["XEP-0047"].startStream(file_obj, jid.JID(IQ['from']), sid, range_length, self.sendSuccessCb, self.sendFailureCb, size, profile) + d = self.host.plugins["XEP-0047"].startStream(file_obj, jid.JID(IQ['from']), sid, profile=profile) + d.addCallback(self.sendSuccessCb, sid, file_obj, self.host.plugins["XEP-0047"].NAMESPACE, profile) + d.addErrback(self.sendFailureCb, sid, file_obj, self.host.plugins["XEP-0047"].NAMESPACE, profile) else: log.warning(_("Invalid stream method received")) @@ -288,11 +300,17 @@ offer.addCallback(self.fileCb, filepath, sid, size, profile) return sid - def sendSuccessCb(self, sid, file_obj, stream_method, profile): + def _sendSuccessCb(self, sid, file_obj, stream_method, profile): + self.sendSuccessCb(sid, file_obj, stream_method, profile) + + def sendSuccessCb(self, dummy, sid, file_obj, stream_method, profile): log.info(_(u'Transfer %(sid)s successfuly finished [%(profile)s]') % {"sid": sid, "profile": profile}) file_obj.close() - def sendFailureCb(self, sid, file_obj, stream_method, reason, profile): + def _sendFailureCb(self, sid, file_obj, stream_method, reason, profile): + self.sendFailureCb(failure.Failure(Exception(reason)), sid, file_obj, stream_method, profile) + + def sendFailureCb(self, failure, sid, file_obj, stream_method, profile): file_obj.close() - log.warning(_(u'Transfer %(id)s failed with stream method %(s_method)s: %(reason)s [%(profile)s]') % {'id': sid, "s_method": stream_method, 'reason': reason, 'profile': profile}) + log.warning(_(u'Transfer %(id)s failed with stream method %(s_method)s: %(reason)s [%(profile)s]') % {'id': sid, "s_method": stream_method, 'reason': unicode(failure), 'profile': profile})