diff options
Diffstat (limited to 'sleekxmpp')
| -rw-r--r-- | sleekxmpp/__init__.py | 18 | ||||
| -rw-r--r-- | sleekxmpp/xmlstream/scheduler.py | 2 | ||||
| -rw-r--r-- | sleekxmpp/xmlstream/statemachine.py | 9 | ||||
| -rw-r--r-- | sleekxmpp/xmlstream/xmlstream.py | 53 |
4 files changed, 64 insertions, 18 deletions
diff --git a/sleekxmpp/__init__.py b/sleekxmpp/__init__.py index 6995f7c9..74eb290a 100644 --- a/sleekxmpp/__init__.py +++ b/sleekxmpp/__init__.py @@ -69,6 +69,8 @@ class ClientXMPP(basexmpp, XMLStream): #TODO: Use stream state here self.authenticated = False self.sessionstarted = False + self.bound = False + self.bindfail = False self.registerHandler(Callback('Stream Features', MatchXPath('{http://etherx.jabber.org/streams}features'), self._handleStreamFeatures, thread=True)) self.registerHandler(Callback('Roster Update', MatchXPath('{%s}iq/{jabber:iq:roster}query' % self.default_ns), self._handleRoster, thread=True)) #self.registerHandler(Callback('Roster Update', MatchXMLMask("<presence xmlns='%s' type='subscribe' />" % self.default_ns), self._handlePresenceSubscribe, thread=True)) @@ -146,13 +148,9 @@ class ClientXMPP(basexmpp, XMLStream): # overriding reconnect and disconnect so that we can get some events # should events be part of or required by xmlstream? Maybe that would be cleaner def reconnect(self): - logging.info("Reconnecting") - self.event("disconnected") - self.authenticated = False - self.sessionstarted = False - XMLStream.reconnect(self) + self.disconnect(reconnect=True) - def disconnect(self, init=True, close=False, reconnect=False): + def disconnect(self, reconnect=False): self.event("disconnected") self.authenticated = False self.sessionstarted = False @@ -248,19 +246,23 @@ class ClientXMPP(basexmpp, XMLStream): response = iq.send() #response = self.send(iq, self.Iq(sid=iq['id'])) self.set_jid(response.xml.find('{urn:ietf:params:xml:ns:xmpp-bind}bind/{urn:ietf:params:xml:ns:xmpp-bind}jid').text) + self.bound = True logging.info("Node set to: %s" % self.fulljid) - if "{urn:ietf:params:xml:ns:xmpp-session}session" not in self.features: + if "{urn:ietf:params:xml:ns:xmpp-session}session" not in self.features or self.bindfail: logging.debug("Established Session") self.sessionstarted = True self.event("session_start") def handler_start_session(self, xml): - if self.authenticated: + if self.authenticated and self.bound: iq = self.makeIqSet(xml) response = iq.send() logging.debug("Established Session") self.sessionstarted = True self.event("session_start") + else: + #bind probably hasn't happened yet + self.bindfail = True def _handleRoster(self, iq, request=False): if iq['type'] == 'set' or (iq['type'] == 'result' and request): diff --git a/sleekxmpp/xmlstream/scheduler.py b/sleekxmpp/xmlstream/scheduler.py index 945d9fad..40aaf695 100644 --- a/sleekxmpp/xmlstream/scheduler.py +++ b/sleekxmpp/xmlstream/scheduler.py @@ -76,7 +76,7 @@ class Scheduler(object): if updated: self.schedule = sorted(self.schedule, key=lambda task: task.next) except KeyboardInterrupt: self.run = False - logging.debug("Qutting Scheduler thread") + logging.debug("Quitting Scheduler thread") if self.parentqueue is not None: self.parentqueue.put(('quit', None, None)) diff --git a/sleekxmpp/xmlstream/statemachine.py b/sleekxmpp/xmlstream/statemachine.py index c5f51765..065d5794 100644 --- a/sleekxmpp/xmlstream/statemachine.py +++ b/sleekxmpp/xmlstream/statemachine.py @@ -110,7 +110,14 @@ class StateMachine(object): def reset(self): # TODO need to lock before calling this? self.transition(self.__current_state, self._default_state) - + + + def current_state(self): + ''' + Return the current state name. + ''' + return self.__current_state + def __getitem__(self, state): ''' diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index 3bcb3412..76aecee4 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -16,6 +16,7 @@ from . stanzabase import StanzaBase from xml.etree import cElementTree from xml.parsers import expat import logging +import random import socket import threading import time @@ -46,6 +47,10 @@ class CloseStream(Exception): stanza_extensions = {} +RECONNECT_MAX_DELAY = 3600 +RECONNECT_QUIESCE_FACTOR = 1.6180339887498948 # Phi +RECONNECT_QUIESCE_JITTER = 0.11962656472 # molar Planck constant times c, joule meter/mole + class XMLStream(object): "A connection manager with XML events." @@ -96,12 +101,29 @@ class XMLStream(object): def connect(self, host='', port=0, use_ssl=None, use_tls=None): "Link to connectTCP" - if self.state.transition('disconnected', 'connecting'): - return self.connectTCP(host, port, use_ssl, use_tls) + if not self.state.transition('disconnected','connecting'): + logging.warning("Can't connect now; Already in state %s", self.state.current_state()) + return False + + if not self.connectTCP(host, port, use_ssl, use_tls): + # return to the 'disconnected' state if connect failed: + # otherwise the connect method is not reentrant + if not self.state.transition('connecting','disconnected'): + logging.error("Couldn't transition to the 'disconnected' state!") + return False + return True + + # TODO currently a caller can't distinguish between "connection failed" and + # "we're already trying to connect from another thread" def connectTCP(self, host='', port=0, use_ssl=None, use_tls=None, reattempt=True): "Connect and create socket" - while reattempt and not self.state['connected']: # the self.state part is redundant. + + # Note that this is thread-safe by merit of being called solely from connect() which + # holds the state lock. + + delay = 1.0 # reconnection delay + while self.run: logging.debug('connecting....') try: if host and port: @@ -109,27 +131,42 @@ class XMLStream(object): if use_ssl is not None: self.use_ssl = use_ssl if use_tls is not None: + # TODO this variable doesn't seem to be used for anything! self.use_tls = use_tls if sys.version_info < (3, 0): self.socket = filesocket.Socket26(socket.AF_INET, socket.SOCK_STREAM) else: self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.settimeout(None) #10) + if self.use_ssl and self.ssl_support: logging.debug("Socket Wrapped for SSL") self.socket = ssl.wrap_socket(self.socket,ca_certs=self.ca_certs) - except: - logging.exception("Connection error") - try: + self.socket.connect(self.address) self.filesocket = self.socket.makefile('rb', 0) + if not self.state.transition('connecting','connected'): logging.error( "State transition error!!!! Shouldn't have happened" ) logging.debug('connect complete.') return True + except socket.error as serr: - logging.error("Could not connect. Socket Error #%s: %s" % (serr.errno, serr.strerror)) - time.sleep(1) # TODO proper quiesce if connection attempt fails + logging.exception("Socket Error #%s: %s", serr.errno, serr.strerror) + if not reattempt: return False + except: + logging.exception("Connection error") + if not reattempt: return False + + # quiesce if rconnection fails: + # This algorithm based loosely on Twisted internet.protocol + # http://twistedmatrix.com/trac/browser/trunk/twisted/internet/protocol.py#L310 + delay = min(delay * RECONNECT_QUIESCE_FACTOR, RECONNECT_MAX_DELAY) + delay = random.normalvariate(delay, delay * RECONNECT_QUIESCE_JITTER) + logging.debug('Waiting %fs until next reconnect attempt...', delay) + time.sleep(delay) + + def connectUnix(self, filepath): "Connect to Unix file and create socket" |