--- 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)]