jshiffer/slixmpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

XEP-0047: replace threading events with simple booleans.

Browse Source
This commit is contained in:
Emmanuel Gil Peyrot 2015-04-14 17:54:16 +02:00
parent eb4e09b0ca
commit d34ddf33db
2 changed files with 13 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
slixmpp/plugins/xep_0047/ibb.py
View File

@ -142,7 +142,7 @@ class XEP_0047(BasePlugin):
log.debug('IBB stream (%s) accepted by %s', stream.sid, iq['from']) log.debug('IBB stream (%s) accepted by %s', stream.sid, iq['from'])
stream.self_jid = iq['to'] stream.self_jid = iq['to']
stream.peer_jid = iq['from'] stream.peer_jid = iq['from']
stream.stream_started.set() stream.stream_started = True
self.api['set_stream'](stream.self_jid, stream.sid, stream.peer_jid, stream) self.api['set_stream'](stream.self_jid, stream.sid, stream.peer_jid, stream)
self.xmpp.event('ibb_stream_start', stream) self.xmpp.event('ibb_stream_start', stream)
self.xmpp.event('stream:%s:%s' % (stream.sid, stream.peer_jid), stream) self.xmpp.event('stream:%s:%s' % (stream.sid, stream.peer_jid), stream)
@ -167,7 +167,7 @@ class XEP_0047(BasePlugin):
stream = IBBytestream(self.xmpp, sid, size, stream = IBBytestream(self.xmpp, sid, size,
iq['to'], iq['from']) iq['to'], iq['from'])
stream.stream_started.set() stream.stream_started = True
self.api['set_stream'](stream.self_jid, stream.sid, stream.peer_jid, stream) self.api['set_stream'](stream.self_jid, stream.sid, stream.peer_jid, stream)
iq.reply().send() iq.reply().send()

24
slixmpp/plugins/xep_0047/stream.py
View File

@ -1,5 +1,4 @@
import socket import socket
import threading
import logging import logging
from queue import Queue from queue import Queue
@ -26,15 +25,14 @@ class IBBytestream(object):
self.send_seq = -1 self.send_seq = -1
self.recv_seq = -1 self.recv_seq = -1
self.stream_started = threading.Event() self.stream_started = False
self.stream_in_closed = threading.Event() self.stream_in_closed = False
self.stream_out_closed = threading.Event() self.stream_out_closed = False
self.recv_queue = Queue() self.recv_queue = Queue()
def send(self, data): def send(self, data):
if not self.stream_started.is_set() or \ if not self.stream_started or self.stream_out_closed:
self.stream_out_closed.is_set():
raise socket.error raise socket.error
data = data[0:self.block_size] data = data[0:self.block_size]
self.send_seq = (self.send_seq + 1) % 65535 self.send_seq = (self.send_seq + 1) % 65535
@ -90,8 +88,7 @@ class IBBytestream(object):
return self.read(block=True) return self.read(block=True)
def read(self, block=True, timeout=None, **kwargs): def read(self, block=True, timeout=None, **kwargs):
if not self.stream_started.is_set() or \ if not self.stream_started or self.stream_in_closed:
self.stream_in_closed.is_set():
raise socket.error raise socket.error
if timeout is not None: if timeout is not None:
block = True block = True
@ -106,14 +103,15 @@ class IBBytestream(object):
iq['to'] = self.peer_jid iq['to'] = self.peer_jid
iq['from'] = self.self_jid iq['from'] = self.self_jid
iq['ibb_close']['sid'] = self.sid iq['ibb_close']['sid'] = self.sid
self.stream_out_closed.set() self.stream_out_closed = True
iq.send(block=False, def _close_stream(_):
callback=lambda x: self.stream_in_closed.set()) self.stream_in_closed = True
iq.send(block=False, callback=_close_stream)
self.xmpp.event('ibb_stream_end', self) self.xmpp.event('ibb_stream_end', self)
def _closed(self, iq): def _closed(self, iq):
self.stream_in_closed.set() self.stream_in_closed = True
self.stream_out_closed.set() self.stream_out_closed = True
iq.reply().send() iq.reply().send()
self.xmpp.event('ibb_stream_end', self) self.xmpp.event('ibb_stream_end', self)