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
This commit is contained in:
Goffi
parent
0c5ec666c2
commit
7121b58961
2 changed files with 257 additions and 231 deletions
|
|
@ -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 @@ MESSAGE = '/message'
|
|||
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 @@ PLUGIN_INFO = {
|
|||
|
||||
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 @@ class XEP_0047(object):
|
|||
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
|
||||
|
||||
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:
|
||||
@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 _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"""
|
||||
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
|
||||
|
||||
def streamOpening(self, IQ, profile):
|
||||
log.debug(_("IBB stream opening"))
|
||||
IQ.handled = True
|
||||
@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)
|
||||
open_elt = IQ.firstChildElement()
|
||||
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),
|
||||
}
|
||||
|
||||
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_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']
|
||||
|
||||
def iqData(self, IQ, profile):
|
||||
IQ.handled = True
|
||||
iq_result_elt = xmlstream.toResponse(iq_elt, 'result')
|
||||
client.xmlstream.send(iq_result_elt)
|
||||
self._killSession(sid, client)
|
||||
|
||||
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()
|
||||
data_elt = element.elements(NS_IBB, 'data').next()
|
||||
sid = data_elt['sid']
|
||||
|
||||
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']
|
||||
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)
|
||||
|
||||
client.xmlstream.send(result)
|
||||
from_jid = session_data["to"]
|
||||
file_obj = session_data["file_obj"]
|
||||
|
||||
def messageData(self, message_elt, profile):
|
||||
sid = message_elt.getAttribute('id', '')
|
||||
self._manageDataElt(message_elt, 'message', sid, jid.JID(message_elt['from']), profile)
|
||||
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
|
||||
|
||||
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
|
||||
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
|
||||
|
||||
from_jid = client.xep_0047_current_stream[sid]["from"]
|
||||
file_obj = client.xep_0047_current_stream[sid]["file_obj"]
|
||||
# we reset the timeout:
|
||||
session_data["timer"].reset(TIMEOUT)
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
|
||||
#we reset the timeout:
|
||||
client.xep_0047_current_stream[sid]["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
|
||||
|
||||
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)
|
||||
|
||||
def startStream(self, file_obj, to_jid, sid, length, successCb, failureCb, 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"""
|
||||
client = self.host.getClient(profile)
|
||||
if length is not None:
|
||||
log.error(_('stream length not managed yet'))
|
||||
# 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
|
||||
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)
|
||||
|
||||
# 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
|
||||
|
||||
@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): 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 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 @@ class XEP_0047_handler(XMPPHandler):
|
|||
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)]
|
||||
|
|
|
|||
|
|
@ -26,6 +26,7 @@ from twisted.words.protocols.jabber import jid
|
|||
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 @@ class XEP_0096(object):
|
|||
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 @@ class XEP_0096(object):
|
|||
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 @@ class XEP_0096(object):
|
|||
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 @@ class XEP_0096(object):
|
|||
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 @@ class XEP_0096(object):
|
|||
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})
|
||||
|
|
|
|||
Loading…
Reference in a new issue