/usr/lib/python3/dist-packages/nbxmpp/smacks.py is in python3-nbxmpp 0.5.3-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 133 134 135 136 137 138 139 | 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
self.failed_resume = False # If last resuming attempt failed
self.supports_sm = False # If server supports sm
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
if self.failed_resume:
self.con._discover_server_at_connection(self.con.connection)
self.failed_resume = False
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 = stanza.getAttr('h')
if not h:
log.error('Server did not send h attribute')
return
h = int(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)
self.failed_resume = True
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
|