Source Code for Module nbxmpp.smacks

  1  from protocol import Acks 
  2  from protocol import NS_STREAM_MGMT 
  3  import logging 
  4  log = logging.getLogger('nbxmpp.smacks') 
  5   
  6 -class Smacks(): 
  7      ''' 
  8      This is Smacks is the Stream Management class. It takes care of requesting 
  9      and sending acks. Also, it keeps track of the unhandled outgoing stanzas. 
 10   
 11      The dispatcher has to be able to access this class to increment the 
 12      number of handled stanzas 
 13      ''' 
 14   
 15 -    def __init__(self, con): 
 16          self.con = con # Connection object 
 17          self.out_h = 0 # Outgoing stanzas handled 
 18          self.in_h = 0  # Incoming stanzas handled 
 19          self.uqueue = [] # Unhandled stanzas queue 
 20          self.session_id = None 
 21          self.resumption = False # If server supports resume 
 22          # Max number of stanzas in queue before making a request 
 23          self.max_queue = 5 
 24          self._owner = None 
 25          self.resuming = False 
 26          self.enabled = False # If SM is enabled 
 27          self.location = None 
 28   
 29 -    def set_owner(self, owner): 
 30          self._owner = owner 
 31   
 32          # Register handlers 
 33          owner.Dispatcher.RegisterNamespace(NS_STREAM_MGMT) 
 34          owner.Dispatcher.RegisterHandler('enabled', self._neg_response, 
 35              xmlns=NS_STREAM_MGMT) 
 36          owner.Dispatcher.RegisterHandler('r', self.send_ack, 
 37              xmlns=NS_STREAM_MGMT) 
 38          owner.Dispatcher.RegisterHandler('a', self.check_ack, 
 39              xmlns=NS_STREAM_MGMT) 
 40          owner.Dispatcher.RegisterHandler('resumed', self.check_ack, 
 41              xmlns=NS_STREAM_MGMT) 
 42          owner.Dispatcher.RegisterHandler('failed', self.error_handling, 
 43              xmlns=NS_STREAM_MGMT) 
 44   
 45 -    def _neg_response(self, disp, stanza): 
 46          r = stanza.getAttr('resume') 
 47          if r == 'true' or r == 'True' or r == '1': 
 48              self.resumption = True 
 49              self.session_id = stanza.getAttr('id') 
 50   
 51          if r == 'false' or r == 'False' or r == '0': 
 52              self.negociate(False) 
 53   
 54          l = stanza.getAttr('location') 
 55          if l: 
 56              self.location = l 
 57   
 58 -    def negociate(self, resume=True): 
 59          # Every time we attempt to negociate, we must erase all previous info 
 60          # about any previous session 
 61          self.uqueue = [] 
 62          self.in_h = 0 
 63          self.out_h = 0 
 64          self.session_id = None 
 65          self.enabled = True 
 66   
 67          stanza = Acks() 
 68          stanza.buildEnable(resume) 
 69          self._owner.Connection.send(stanza, now=True) 
 70   
 71 -    def resume_request(self): 
 72          if not self.session_id: 
 73              self.resuming = False 
 74              log.error('Attempted to resume without a valid session id ') 
 75              return 
 76          resume = Acks() 
 77          resume.buildResume(self.in_h, self.session_id) 
 78          self._owner.Connection.send(resume, False) 
 79   
 80 -    def send_ack(self, disp, stanza): 
 81          ack = Acks() 
 82          ack.buildAnswer(self.in_h) 
 83          self._owner.Connection.send(ack, False) 
 84   
 85 -    def request_ack(self): 
 86          r = Acks() 
 87          r.buildRequest() 
 88          self._owner.Connection.send(r, False) 
 89   
 90 -    def check_ack(self, disp, stanza): 
 91          ''' 
 92          Checks if the number of stanzas sent are the same as the 
 93          number of stanzas received by the server. Pops stanzas that were 
 94          handled by the server from the queue. 
 95          ''' 
 96          h = int(stanza.getAttr('h')) 
 97          diff = self.out_h - h 
 98   
 99          if len(self.uqueue) < diff or diff < 0: 
100              log.error('Server and client number of stanzas handled mismatch ') 
101          else: 
102              while (len(self.uqueue) > diff): 
103                  self.uqueue.pop(0) 
104   
105          if stanza.getName() == 'resumed': 
106              self.enabled = True 
107              self.resuming = True 
108              self.con.set_oldst() 
109              if self.uqueue != []: 
110                  for i in self.uqueue: 
111                      self._owner.Connection.send(i, False) 
112   
113 -    def error_handling(self, disp, stanza): 
114          # If the server doesn't recognize previd, forget about resuming 
115          # Ask for service discovery, etc.. 
116          if stanza.getTag('item-not-found'): 
117              self.resuming = False 
118              self.enabled = False 
119              # we need to bind a resource 
120              self._owner.NonBlockingBind.resuming = False 
121              self._owner._on_auth_bind(None) 
122              return 
123   
124          # Doesn't support resumption 
125          if stanza.getTag('feature-not-implemented'): 
126              self.negociate(False) 
127              return 
128   
129          if stanza.getTag('unexpected-request'): 
130              self.enabled = False 
131              log.error('Gajim failed to negociate Stream Management') 
132              return 
133