summaryrefslogtreecommitdiff
path: root/sleekxmpp/xmlstream/xmlstream.py
diff options
context:
space:
mode:
Diffstat (limited to 'sleekxmpp/xmlstream/xmlstream.py')
-rw-r--r--sleekxmpp/xmlstream/xmlstream.py124
1 files changed, 62 insertions, 62 deletions
diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py
index fd307a5c..e0b3a23c 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."
@@ -53,8 +58,9 @@ class XMLStream(object):
global ssl_support
self.ssl_support = ssl_support
self.escape_quotes = escape_quotes
- self.state = statemachine.StateMachine()
- self.state.addStates({'connected':False, 'is client':False, 'ssl':False, 'tls':False, 'reconnect':True, 'processing':False}) #set initial states
+ self.state = statemachine.StateMachine(('disconnected','connecting',
+ 'connected'))
+ self.should_reconnect = True
self.setSocket(socket)
self.address = (host, int(port))
@@ -86,21 +92,26 @@ class XMLStream(object):
def setSocket(self, socket):
"Set the socket"
self.socket = socket
- if socket is not None:
+ if socket is not None and self.state.transition('disconnected','connecting'):
self.filesocket = socket.makefile('rb', 0) # ElementTree.iterparse requires a file. 0 buffer files have to be binary
- self.state.set('connected', True)
-
+ self.state.transition('connecting','connected')
def setFileSocket(self, filesocket):
self.filesocket = filesocket
- def connect(self, host='', port=0, use_ssl=False, use_tls=True):
+ def connect(self, host='', port=0, use_ssl=None, use_tls=None):
"Link to connectTCP"
- return self.connectTCP(host, port, use_ssl, use_tls)
+ if self.state.transition('disconnected', 'connecting'):
+ return self.connectTCP(host, port, use_ssl, use_tls)
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']:
+
+ # 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:
@@ -114,20 +125,35 @@ class XMLStream(object):
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)
- self.state.set('connected', True)
+
+ 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 code based loosely on Twisted internet.protocol
+ # http://twistedmatrix.com/trac/browser/trunk/twisted/internet/protocol.py#L310
+ delay = min(delay * RECONNECT_QUIESCE_JITTER, 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"
@@ -182,7 +208,7 @@ class XMLStream(object):
"Start processing the socket."
logging.debug('Process thread starting...')
while self.run:
- self.state.set('processing', True)
+ if not self.state.ensure('connected',wait=2): continue
try:
self.sendRaw(self.stream_header)
while self.run and self.__readXML(): pass
@@ -196,38 +222,16 @@ class XMLStream(object):
logging.debug("Restarting stream...")
continue # DON'T re-initialize the stream -- this exception is sent
# specifically when we've initialized TLS and need to re-send the <stream> header.
- except KeyboardInterrupt:
- logging.debug("Keyboard Escape Detected")
- self.state.set('processing', False)
- self.state.set('reconnect', False)
- self.disconnect()
- # TODO this is probably not necessary...
- self.eventqueue.put(('quit', None, None))
- return
- except SystemExit:
- # TODO shouldn't this be the same as KeyboardInterrupt????
+ except (KeyboardInterrupt, SystemExit):
+ logging.debug("System interrupt detected")
+ self.shutdown()
self.eventqueue.put(('quit', None, None))
- return
except:
logging.exception('Unexpected error in RCV thread')
- if not self.state.reconnect:
- return
- else:
- logging.debug('reconnecting...')
- self.state.set('processing', False)
+ if self.should_reconnect:
self.disconnect(reconnect=True)
- # TODO the individual exception handlers above already handle reconnect!
- # Why are we attempting to do it again down here???
-# if self.state['reconnect']:
-# self.state.set('connected', False)
- self.state.set('processing', False)
-# self.reconnect()
-# else:
-# TODO I think this is getting queued, and when the eventRunner comes back online after
-# reconnect, it immediately processes a 'quit' event and exits again, meanwhile the
-# rest of the client is just starting to connect and process the incoming event stream!!!
-# self.eventqueue.put(('quit', None, None))
- logging.debug('Quitting Process thread')
+
+ logging.debug('Quitting Process thread')
def __readXML(self):
"Parses the incoming stream, adding to xmlin queue as it goes"
@@ -246,7 +250,6 @@ class XMLStream(object):
edepth += -1
if edepth == 0 and event == b'end':
# what is this case exactly? Premature EOF?
- #self.disconnect(reconnect=self.state['reconnect'])
logging.debug("Ending readXML loop")
return False
elif edepth == 1:
@@ -261,9 +264,8 @@ class XMLStream(object):
def _sendThread(self):
logging.debug('send thread starting...')
while self.run:
- if not self.state['connected']:
- logging.warning("Not connected yet...")
- time.sleep(1)
+ if not self.state.ensure('connected',wait=2): continue
+
data = None
try:
data = self.sendqueue.get(True,10)
@@ -272,7 +274,7 @@ class XMLStream(object):
except queue.Empty:
logging.debug('nothing on send queue')
except socket.timeout:
- # this is to prevent hanging
+ # this is to prevent a thread blocked indefinitely
logging.debug('timeout sending packet data')
except:
logging.warning("Failed to send %s" % data)
@@ -282,9 +284,7 @@ class XMLStream(object):
# the same thing concurrently. Oops! The safer option would be to throw
# some sort of event that could be handled by a common thread or the reader
# thread to perform reconnect and then re-initialize the handler threads as well.
- if self.state.reconnect:
- logging.debug('Reconnecting...')
- traceback.print_exc()
+ if self.should_reconnect:
self.disconnect(reconnect=True)
def sendRaw(self, data):
@@ -292,8 +292,7 @@ class XMLStream(object):
return True
def disconnect(self, reconnect=False):
- self.state.set('reconnect', reconnect)
- if not self.state['connected']:
+ if not self.state.transition('connected','disconnected'):
logging.warning("Already disconnected.")
return
logging.debug("Disconnecting...")
@@ -301,10 +300,7 @@ class XMLStream(object):
time.sleep(5)
#send end of stream
#wait for end of stream back
- self.run = False
- self.scheduler.run = False
try:
- self.state.set('connected',False)
# self.socket.shutdown(socket.SHUT_RDWR)
self.socket.close()
except socket.error as (errno,strerror):
@@ -312,13 +308,17 @@ class XMLStream(object):
try:
self.filesocket.close()
except socket.error as (errno,strerror):
- logging.exception("Error closing filesocket.")
+ logging.exception("Error closing filesocket.")
+
+ if reconnect: self.connect()
- def reconnect(self):
- self.state.set('tls',False)
- self.state.set('ssl',False)
- time.sleep(1)
- self.connect()
+ def shutdown(self):
+ '''
+ Disconnects and shuts down all event threads.
+ '''
+ self.disconnect()
+ self.run = False
+ self.scheduler.run = False
def incoming_filter(self, xmlobj):
return xmlobj