github/SleekXMPP
1
0
Fork 0
mirror of https://github.com/correl/SleekXMPP.git synced 2024-11-24 03:00:15 +00:00
Code Issues Projects Releases Packages Wiki Activity

added send queueing to avoid mixed sending

Browse source
This commit is contained in:
Nathan Fritz 2010-02-15 02:13:44 -08:00
parent 06fa1fcf33
commit 58375955a9
2 changed files with 26 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
setup.py
View file

@ -44,10 +44,10 @@ packages = [ 'sleekxmpp',
'sleekxmpp/xmlstream/matcher', 'sleekxmpp/xmlstream/matcher',
'sleekxmpp/xmlstream/handler' ] 'sleekxmpp/xmlstream/handler' ]
if sys.version_info < (3, 0): #if sys.version_info < (3, 0):
packages.append('sleekxmpp/xmlstream/tostring26') # packages.append('sleekxmpp/xmlstream/tostring26')
else: #else:
packages.append('sleekxmpp/xmlstream/tostring') # packages.append('sleekxmpp/xmlstream/tostring')
setup( setup(
name = "sleekxmpp", name = "sleekxmpp",

34
sleekxmpp/xmlstream/xmlstream.py
View file

@ -66,6 +66,7 @@ class XMLStream(object):
self.stream_footer = "</stream>" self.stream_footer = "</stream>"
self.eventqueue = queue.Queue() self.eventqueue = queue.Queue()
self.sendqueue = queue.Queue()
self.namespace_map = {} self.namespace_map = {}
@ -139,12 +140,17 @@ class XMLStream(object):
for t in range(0, HANDLER_THREADS): for t in range(0, HANDLER_THREADS):
self.__thread['eventhandle%s' % t] = threading.Thread(name='eventhandle%s' % t, target=self._eventRunner) self.__thread['eventhandle%s' % t] = threading.Thread(name='eventhandle%s' % t, target=self._eventRunner)
self.__thread['eventhandle%s' % t].start() self.__thread['eventhandle%s' % t].start()
self.__thread['sendthread'] = threading.Thread(name='sendthread', target=self._sendThread)
self.__thread['sendthread'].start()
if threaded: if threaded:
self.__thread['process'] = threading.Thread(name='process', target=self._process) self.__thread['process'] = threading.Thread(name='process', target=self._process)
self.__thread['process'].start() self.__thread['process'].start()
else: else:
self._process() self._process()
def schedule(self, seconds, handler, args=None):
threading.Timer(seconds, handler, args).start()
def _process(self): def _process(self):
"Start processing the socket." "Start processing the socket."
firstrun = True firstrun = True
@ -222,19 +228,23 @@ class XMLStream(object):
if event == b'start': if event == b'start':
edepth += 1 edepth += 1
def _sendThread(self):
while True:
data = self.sendqueue.get(True)
logging.debug("SEND: %s" % data)
try:
self.socket.send(data.encode('utf-8'))
#self.socket.send(bytes(data, "utf-8"))
#except socket.error,(errno, strerror):
except:
self.state.set('connected', False)
if self.state.reconnect:
logging.error("Disconnected. Socket Error.")
traceback.print_exc()
self.disconnect(reconnect=True)
def sendRaw(self, data): def sendRaw(self, data):
logging.debug("SEND: %s" % data) self.sendqueue.put(data)
try:
self.socket.send(data.encode('utf-8'))
#self.socket.send(bytes(data, "utf-8"))
#except socket.error,(errno, strerror):
except:
self.state.set('connected', False)
if self.state.reconnect:
logging.error("Disconnected. Socket Error.")
traceback.print_exc()
self.disconnect(reconnect=True)
return False
return True return True
def disconnect(self, reconnect=False): def disconnect(self, reconnect=False):