smacks.py 4.42 KB
Newer Older
Yann Leboulanger's avatar
Yann Leboulanger committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132
from protocol import Acks
from protocol import NS_STREAM_MGMT
import logging
log = logging.getLogger('nbxmpp.smacks')

class Smacks():
    '''
    This is Smacks is the Stream Management class. It takes care of requesting
    and sending acks. Also, it keeps track of the unhandled outgoing stanzas.

    The dispatcher has to be able to access this class to increment the
    number of handled stanzas
    '''

    def __init__(self, con):
        self.con = con # Connection object
        self.out_h = 0 # Outgoing stanzas handled
        self.in_h = 0  # Incoming stanzas handled
        self.uqueue = [] # Unhandled stanzas queue
        self.session_id = None
        self.resumption = False # If server supports resume
        # Max number of stanzas in queue before making a request
        self.max_queue = 5
        self._owner = None
        self.resuming = False
        self.enabled = False # If SM is enabled
        self.location = None

    def set_owner(self, owner):
        self._owner = owner

        # Register handlers
        owner.Dispatcher.RegisterNamespace(NS_STREAM_MGMT)
        owner.Dispatcher.RegisterHandler('enabled', self._neg_response,
            xmlns=NS_STREAM_MGMT)
        owner.Dispatcher.RegisterHandler('r', self.send_ack,
            xmlns=NS_STREAM_MGMT)
        owner.Dispatcher.RegisterHandler('a', self.check_ack,
            xmlns=NS_STREAM_MGMT)
        owner.Dispatcher.RegisterHandler('resumed', self.check_ack,
            xmlns=NS_STREAM_MGMT)
        owner.Dispatcher.RegisterHandler('failed', self.error_handling,
            xmlns=NS_STREAM_MGMT)

    def _neg_response(self, disp, stanza):
        r = stanza.getAttr('resume')
        if r == 'true' or r == 'True' or r == '1':
            self.resumption = True
            self.session_id = stanza.getAttr('id')

        if r == 'false' or r == 'False' or r == '0':
            self.negociate(False)

        l = stanza.getAttr('location')
        if l:
            self.location = l

    def negociate(self, resume=True):
        # Every time we attempt to negociate, we must erase all previous info
        # about any previous session
        self.uqueue = []
        self.in_h = 0
        self.out_h = 0
        self.session_id = None
        self.enabled = True

        stanza = Acks()
        stanza.buildEnable(resume)
        self._owner.Connection.send(stanza, now=True)

    def resume_request(self):
        if not self.session_id:
            self.resuming = False
            log.error('Attempted to resume without a valid session id ')
            return
        resume = Acks()
        resume.buildResume(self.in_h, self.session_id)
        self._owner.Connection.send(resume, False)

    def send_ack(self, disp, stanza):
        ack = Acks()
        ack.buildAnswer(self.in_h)
        self._owner.Connection.send(ack, False)

    def request_ack(self):
        r = Acks()
        r.buildRequest()
        self._owner.Connection.send(r, False)

    def check_ack(self, disp, stanza):
        '''
        Checks if the number of stanzas sent are the same as the
        number of stanzas received by the server. Pops stanzas that were
        handled by the server from the queue.
        '''
        h = int(stanza.getAttr('h'))
        diff = self.out_h - h

        if len(self.uqueue) < diff or diff < 0:
            log.error('Server and client number of stanzas handled mismatch ')
        else:
            while (len(self.uqueue) > diff):
                self.uqueue.pop(0)

        if stanza.getName() == 'resumed':
            self.enabled = True
            self.resuming = True
            self.con.set_oldst()
            if self.uqueue != []:
                for i in self.uqueue:
                    self._owner.Connection.send(i, False)

    def error_handling(self, disp, stanza):
        # If the server doesn't recognize previd, forget about resuming
        # Ask for service discovery, etc..
        if stanza.getTag('item-not-found'):
            self.resuming = False
            self.enabled = False
            # we need to bind a resource
            self._owner.NonBlockingBind.resuming = False
            self._owner._on_auth_bind(None)
            return

        # Doesn't support resumption
        if stanza.getTag('feature-not-implemented'):
            self.negociate(False)
            return

        if stanza.getTag('unexpected-request'):
            self.enabled = False
            log.error('Gajim failed to negociate Stream Management')
            return