summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorThom Nichols <tmnichols@gmail.com>2010-06-01 22:51:49 -0400
committerThom Nichols <tmnichols@gmail.com>2010-06-01 22:51:49 -0400
commit4eb210bff5753143300e5bf6ae09617901e9a28e (patch)
treef8a9b69630792567ddae16ce0627ee2c2b1c6caa
parent1780ca900a9a78347745d95e7d8934ffae9d7594 (diff)
downloadslixmpp-4eb210bff5753143300e5bf6ae09617901e9a28e.tar.gz
slixmpp-4eb210bff5753143300e5bf6ae09617901e9a28e.tar.bz2
slixmpp-4eb210bff5753143300e5bf6ae09617901e9a28e.tar.xz
slixmpp-4eb210bff5753143300e5bf6ae09617901e9a28e.zip
fixed some major reconnection errors
-rw-r--r--sleekxmpp/__init__.py12
-rw-r--r--sleekxmpp/basexmpp.py2
-rw-r--r--sleekxmpp/xmlstream/handler/base.py2
-rw-r--r--sleekxmpp/xmlstream/handler/callback.py4
-rw-r--r--sleekxmpp/xmlstream/xmlstream.py242
5 files changed, 138 insertions, 124 deletions
diff --git a/sleekxmpp/__init__.py b/sleekxmpp/__init__.py
index 954ca996..6995f7c9 100644
--- a/sleekxmpp/__init__.py
+++ b/sleekxmpp/__init__.py
@@ -94,6 +94,8 @@ class ClientXMPP(basexmpp, XMLStream):
"""Connect to the Jabber Server. Attempts SRV lookup, and if it fails, uses
the JID server."""
+ if self.state['connected']: return True
+
if host:
self.server = host
if port is None: port = self.port
@@ -174,6 +176,7 @@ class ClientXMPP(basexmpp, XMLStream):
self._handleRoster(iq, request=True)
def _handleStreamFeatures(self, features):
+ logging.debug('handling stream features')
self.features = []
for sub in features.xml:
self.features.append(sub.tag)
@@ -181,12 +184,16 @@ class ClientXMPP(basexmpp, XMLStream):
for feature in self.registered_features:
if feature[0].match(subelement):
#if self.maskcmp(subelement, feature[0], True):
+ # This calls the feature handler & optionally breaks
if feature[1](subelement) and feature[2]: #if breaker, don't continue
return True
def handler_starttls(self, xml):
+ logging.debug( 'TLS start handler; SSL support: %s', self.ssl_support )
if not self.authenticated and self.ssl_support:
- self.add_handler("<proceed xmlns='urn:ietf:params:xml:ns:xmpp-tls' />", self.handler_tls_start, instream=True)
+ _stanza = "<proceed xmlns='urn:ietf:params:xml:ns:xmpp-tls' />"
+ if not self.event_handlers.get(_stanza,None): # don't add handler > once
+ self.add_handler( _stanza, self.handler_tls_start, instream=True )
self.sendXML(xml)
return True
else:
@@ -221,12 +228,13 @@ class ClientXMPP(basexmpp, XMLStream):
return True
def handler_auth_success(self, xml):
+ logging.debug("Authentication successful.")
self.authenticated = True
self.features = []
raise RestartStream()
def handler_auth_fail(self, xml):
- logging.info("Authentication failed.")
+ logging.warning("Authentication failed.")
self.disconnect()
self.event("failed_auth")
diff --git a/sleekxmpp/basexmpp.py b/sleekxmpp/basexmpp.py
index 6cd382ef..a916fe86 100644
--- a/sleekxmpp/basexmpp.py
+++ b/sleekxmpp/basexmpp.py
@@ -84,7 +84,7 @@ class basexmpp(object):
self.resource = self.getjidresource(jid)
self.jid = self.getjidbare(jid)
self.username = jid.split('@', 1)[0]
- self.server = jid.split('@',1)[-1].split('/', 1)[0]
+ self.domain = jid.split('@',1)[-1].split('/', 1)[0]
def process(self, *args, **kwargs):
for idx in self.plugin:
diff --git a/sleekxmpp/xmlstream/handler/base.py b/sleekxmpp/xmlstream/handler/base.py
index 5d55f4ee..a44edf0e 100644
--- a/sleekxmpp/xmlstream/handler/base.py
+++ b/sleekxmpp/xmlstream/handler/base.py
@@ -18,7 +18,7 @@ class BaseHandler(object):
def match(self, xml):
return self._matcher.match(xml)
- def prerun(self, payload):
+ def prerun(self, payload): # what's the point of this if the payload is called again in run??
self._payload = payload
def run(self, payload):
diff --git a/sleekxmpp/xmlstream/handler/callback.py b/sleekxmpp/xmlstream/handler/callback.py
index 49cfa14d..ea5acb5b 100644
--- a/sleekxmpp/xmlstream/handler/callback.py
+++ b/sleekxmpp/xmlstream/handler/callback.py
@@ -17,13 +17,15 @@ class Callback(base.BaseHandler):
self._once = once
self._instream = instream
- def prerun(self, payload):
+ def prerun(self, payload): # prerun actually calls run?!? WTF! Then it gets run AGAIN!
base.BaseHandler.prerun(self, payload)
if self._instream:
+ logging.debug('callback "%s" prerun', self.name)
self.run(payload, True)
def run(self, payload, instream=False):
if not self._instream or instream:
+ logging.debug('callback "%s" run', self.name)
base.BaseHandler.run(self, payload)
#if self._thread:
# x = threading.Thread(name="Callback_%s" % self.name, target=self._pointer, args=(payload,))
diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py
index 5650386e..fd307a5c 100644
--- a/sleekxmpp/xmlstream/xmlstream.py
+++ b/sleekxmpp/xmlstream/xmlstream.py
@@ -1,9 +1,9 @@
"""
- SleekXMPP: The Sleek XMPP Library
- Copyright (C) 2010 Nathanael C. Fritz
- This file is part of SleekXMPP.
+ SleekXMPP: The Sleek XMPP Library
+ Copyright (C) 2010 Nathanael C. Fritz
+ This file is part of SleekXMPP.
- See the file license.txt for copying permission.
+ See the file license.txt for copying permission.
"""
from __future__ import with_statement, unicode_literals
@@ -54,7 +54,7 @@ class XMLStream(object):
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, 'disconnecting':False}) #set initial states
+ self.state.addStates({'connected':False, 'is client':False, 'ssl':False, 'tls':False, 'reconnect':True, 'processing':False}) #set initial states
self.setSocket(socket)
self.address = (host, int(port))
@@ -101,30 +101,33 @@ class XMLStream(object):
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']:
- if host and port:
- self.address = (host, int(port))
- if use_ssl is not None:
- self.use_ssl = use_ssl
- if use_tls is not None:
- self.use_tls = use_tls
- self.state.set('is client', True)
- 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)
- 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)
+ logging.debug('connecting....')
+ try:
+ if host and port:
+ self.address = (host, int(port))
+ if use_ssl is not None:
+ self.use_ssl = use_ssl
+ if use_tls is not None:
+ 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)
self.filesocket = self.socket.makefile('rb', 0)
self.state.set('connected', True)
+ 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)
+ time.sleep(1) # TODO proper quiesce if connection attempt fails
def connectUnix(self, filepath):
"Connect to Unix file and create socket"
@@ -133,19 +136,19 @@ class XMLStream(object):
"Handshakes for TLS"
if self.ssl_support:
logging.info("Negotiating TLS")
- self.realsocket = self.socket
+# self.realsocket = self.socket # NOT USED
self.socket = ssl.wrap_socket(self.socket,
- ssl_version=ssl.PROTOCOL_TLSv1,
- do_handshake_on_connect=False,
- ca_certs=self.ca_certs)
- print "doing handshake..."
+ ssl_version=ssl.PROTOCOL_TLSv1,
+ do_handshake_on_connect=False,
+ ca_certs=self.ca_certs)
self.socket.do_handshake()
- print "got handshake..."
if sys.version_info < (3,0):
from . filesocket import filesocket
self.filesocket = filesocket(self.socket)
else:
self.filesocket = self.socket.makefile('rb', 0)
+
+ logging.debug("TLS negotitation successful")
return True
else:
logging.warning("Tried to enable TLS, but ssl module not found.")
@@ -154,8 +157,8 @@ class XMLStream(object):
def process(self, threaded=True):
self.scheduler.process(threaded=True)
+ self.run = True
for t in range(0, HANDLER_THREADS):
-<<<<<<< HEAD
th = threading.Thread(name='eventhandle%s' % t, target=self._eventRunner)
th.setDaemon(True)
self.__thread['eventhandle%s' % t] = th
@@ -164,13 +167,6 @@ class XMLStream(object):
th.setDaemon(True)
self.__thread['sendthread'] = th
th.start()
-=======
- logging.debug("Starting HANDLER THREAD")
- self.__thread['eventhandle%s' % t] = threading.Thread(name='eventhandle%s' % t, target=self._eventRunner)
- self.__thread['eventhandle%s' % t].start()
- self.__thread['sendthread'] = threading.Thread(name='sendthread', target=self._sendThread)
- self.__thread['sendthread'].start()
->>>>>>> master
if threaded:
th = threading.Thread(name='process', target=self._process)
th.setDaemon(True)
@@ -184,54 +180,54 @@ class XMLStream(object):
def _process(self):
"Start processing the socket."
- firstrun = True
- while self.run and (firstrun or self.state['reconnect']):
+ logging.debug('Process thread starting...')
+ while self.run:
self.state.set('processing', True)
- firstrun = False
try:
- if self.state['is client']:
- self.sendRaw(self.stream_header)
- while self.run and self.__readXML():
- if self.state['is client']:
- self.sendRaw(self.stream_header)
+ self.sendRaw(self.stream_header)
+ while self.run and self.__readXML(): pass
+ except socket.timeout:
+ logging.debug('socket rcv timeout')
+ pass
+ except CloseStream:
+ # TODO warn that the listener thread is exiting!!!
+ pass
+ except RestartStream:
+ 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()
- self.run = False
- self.scheduler.run = False
+ # TODO this is probably not necessary...
self.eventqueue.put(('quit', None, None))
return
- except CloseStream:
- return
except SystemExit:
+ # TODO shouldn't this be the same as KeyboardInterrupt????
self.eventqueue.put(('quit', None, None))
return
- except socket.error:
- if not self.state.reconnect:
- return
- else:
- self.state.set('processing', False)
- traceback.print_exc()
- self.disconnect(reconnect=True)
except:
+ logging.exception('Unexpected error in RCV thread')
if not self.state.reconnect:
return
else:
+ logging.debug('reconnecting...')
self.state.set('processing', False)
- traceback.print_exc()
self.disconnect(reconnect=True)
- if self.state['reconnect']:
- self.state.set('connected', False)
- self.state.set('processing', False)
- self.reconnect()
- else:
- self.eventqueue.put(('quit', None, None))
- #self.__thread['readXML'] = threading.Thread(name='readXML', target=self.__readXML)
- #self.__thread['readXML'].start()
- #self.__thread['spawnEvents'] = threading.Thread(name='spawnEvents', target=self.__spawnEvents)
- #self.__thread['spawnEvents'].start()
+ # 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')
def __readXML(self):
"Parses the incoming stream, adding to xmlin queue as it goes"
@@ -244,41 +240,50 @@ class XMLStream(object):
if edepth == 0: # and xmlobj.tag.split('}', 1)[-1] == self.basetag:
if event == b'start':
root = xmlobj
+ logging.debug('handling start stream')
self.start_stream_handler(root)
if event == b'end':
edepth += -1
if edepth == 0 and event == b'end':
- self.disconnect(reconnect=self.state['reconnect'])
+ # what is this case exactly? Premature EOF?
+ #self.disconnect(reconnect=self.state['reconnect'])
logging.debug("Ending readXML loop")
return False
elif edepth == 1:
#self.xmlin.put(xmlobj)
- try:
- self.__spawnEvent(xmlobj)
- except RestartStream:
- return True
- except CloseStream:
- logging.debug("Ending readXML loop")
- return False
- if root:
- root.clear()
+ self.__spawnEvent(xmlobj)
+ if root: root.clear()
if event == b'start':
edepth += 1
- logging.debug("Ending readXML loop")
+ logging.debug("Exiting readXML loop")
+ return False
def _sendThread(self):
+ logging.debug('send thread starting...')
while self.run:
- data = self.sendqueue.get(True)
- logging.debug("SEND: %s" % data)
+ if not self.state['connected']:
+ logging.warning("Not connected yet...")
+ time.sleep(1)
+ data = None
try:
- self.socket.send(data.encode('utf-8'))
- #self.socket.send(bytes(data, "utf-8"))
- #except socket.error,(errno, strerror):
+ data = self.sendqueue.get(True,10)
+ logging.debug("SEND: %s" % data)
+ self.socket.sendall(data.encode('utf-8'))
+ except queue.Empty:
+ logging.debug('nothing on send queue')
+ except socket.timeout:
+ # this is to prevent hanging
+ logging.debug('timeout sending packet data')
except:
logging.warning("Failed to send %s" % data)
- self.state.set('connected', False)
+ logging.exception("Socket error in SEND thread")
+ # TODO it's somewhat unsafe for the sender thread to assume it can just
+ # re-intitialize the connection, since the receiver thread could be doing
+ # 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.error("Disconnected. Socket Error.")
+ logging.debug('Reconnecting...')
traceback.print_exc()
self.disconnect(reconnect=True)
@@ -288,42 +293,40 @@ class XMLStream(object):
def disconnect(self, reconnect=False):
self.state.set('reconnect', reconnect)
- if self.state['disconnecting']:
+ if not self.state['connected']:
+ logging.warning("Already disconnected.")
return
- if not self.state['reconnect']:
- logging.debug("Disconnecting...")
- self.state.set('disconnecting', True)
- self.run = False
- self.scheduler.run = False
- if self.state['connected']:
- self.sendRaw(self.stream_footer)
- time.sleep(1)
- #send end of stream
- #wait for end of stream back
+ logging.debug("Disconnecting...")
+ self.sendRaw(self.stream_footer)
+ 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):
+ logging.exception("Error while disconnecting. Socket Error #%s: %s" % (errno, strerror))
+ try:
self.filesocket.close()
- self.socket.shutdown(socket.SHUT_RDWR)
- except socket.error as serr:
- #logging.warning("Error while disconnecting. Socket Error #%s: %s" % (errno, strerror))
- #thread.exit_thread()
- pass
- if self.state['processing']:
- #raise CloseStream
- pass
+ except socket.error as (errno,strerror):
+ logging.exception("Error closing filesocket.")
def reconnect(self):
self.state.set('tls',False)
self.state.set('ssl',False)
time.sleep(1)
- self.connect(self.server,self.port)
-
+ self.connect()
+
def incoming_filter(self, xmlobj):
return xmlobj
-
+
def __spawnEvent(self, xmlobj):
"watching xmlOut and processes handlers"
#convert XML into Stanza
+ # TODO surround this log statement with an if, it's expensive
logging.debug("RECV: %s" % cElementTree.tostring(xmlobj))
xmlobj = self.incoming_filter(xmlobj)
stanza = None
@@ -335,17 +338,21 @@ class XMLStream(object):
if stanza is None:
stanza = StanzaBase(self, xmlobj)
unhandled = True
+ # TODO inefficient linear search; performance might be improved by hashtable lookup
for handler in self.__handlers:
if handler.match(stanza):
+ logging.debug('matched stanza to handler %s', handler.name)
handler.prerun(stanza)
self.eventqueue.put(('stanza', handler, stanza))
- if handler.checkDelete(): self.__handlers.pop(self.__handlers.index(handler))
+ if handler.checkDelete():
+ logging.debug('deleting callback %s', handler.name)
+ self.__handlers.pop(self.__handlers.index(handler))
unhandled = False
if unhandled:
stanza.unhandled()
#loop through handlers and test match
#spawn threads as necessary, call handlers, sending Stanza
-
+
def _eventRunner(self):
logging.debug("Loading event runner")
while self.run:
@@ -353,34 +360,31 @@ class XMLStream(object):
event = self.eventqueue.get(True, timeout=5)
except queue.Empty:
event = None
- except KeyboardInterrupt:
- self.run = False
- self.scheduler.run = False
if event is not None:
etype = event[0]
handler = event[1]
args = event[2:]
- #etype, handler, *args = event #python 3.x way
+ #etype, handler, *args = event #python 3.x way
if etype == 'stanza':
try:
handler.run(args[0])
except Exception as e:
- traceback.print_exc()
+ logging.exception("Exception in event handler")
args[0].exception(e)
- elif etype == 'schedule':
+ elif etype == 'sched':
try:
- logging.debug(args)
- handler(*args[0])
+ #handler(*args[0])
+ handler.run(*args)
except:
logging.error(traceback.format_exc())
elif etype == 'quit':
logging.debug("Quitting eventRunner thread")
return False
-
+
def registerHandler(self, handler, before=None, after=None):
"Add handler with matcher class and parameters."
self.__handlers.append(handler)
-
+
def removeHandler(self, name):
"Removes the handler."
idx = 0
@@ -466,4 +470,4 @@ class XMLStream(object):
def start_stream_handler(self, xml):
"""Meant to be overridden"""
- pass
+ logging.warn("No start stream handler has been implemented.")