pep/slixmpp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Run the send queue in a separate coroutine

Browse source 
To be able to run async stream filters
This commit is contained in:
mathieui 2019-08-21 21:18:24 +02:00
parent 9b3874b5df
commit 31f6ef6814
No known key found for this signature in database
GPG key ID: C59F84CEEFD616E3
1 changed files with 42 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
slixmpp/xmlstream/xmlstream.py
View file

@ -21,6 +21,8 @@ import ssl
import weakref import weakref
import uuid import uuid
from asyncio import iscoroutinefunction
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
from slixmpp.xmlstream.asyncio import asyncio from slixmpp.xmlstream.asyncio import asyncio
@ -83,6 +85,8 @@ class XMLStream(asyncio.BaseProtocol):
self.force_starttls = None self.force_starttls = None
self.disable_starttls = None self.disable_starttls = None
self.waiting_queue = asyncio.Queue()
# A dict of {name: handle} # A dict of {name: handle}
self.scheduled_events = {} self.scheduled_events = {}
@ -263,6 +267,10 @@ class XMLStream(asyncio.BaseProtocol):
localhost localhost
""" """
asyncio.ensure_future(
self.run_filters(),
loop=self.loop,
)
self.disconnect_reason = None self.disconnect_reason = None
self.cancel_connection_attempt() self.cancel_connection_attempt()
if host and port: if host and port:
@ -789,7 +797,7 @@ class XMLStream(asyncio.BaseProtocol):
# If the callback is a coroutine, schedule it instead of # If the callback is a coroutine, schedule it instead of
# running it directly # running it directly
if asyncio.iscoroutinefunction(handler_callback): if iscoroutinefunction(handler_callback):
async def handler_callback_routine(cb): async def handler_callback_routine(cb):
try: try:
await cb(data) await cb(data)
@ -888,11 +896,41 @@ class XMLStream(asyncio.BaseProtocol):
""" """
return xml return xml
async def run_filters(self):
"""
Background loop that processes stanzas to send.
"""
while True:
(data, use_filters) = await self.waiting_queue.get()
if isinstance(data, ElementBase):
if use_filters:
for filter in self.__filters['out']:
if iscoroutinefunction(filter):
data = await filter(data)
else:
data = filter(data)
if data is None:
return
if isinstance(data, ElementBase):
if use_filters:
for filter in self.__filters['out_sync']:
if iscoroutinefunction(filter):
data = await filter(data)
else:
data = filter(data)
if data is None:
return
str_data = tostring(data.xml, xmlns=self.default_ns,
stream=self, top_level=True)
self.send_raw(str_data)
else:
self.send_raw(data)
self.waiting_queue.task_done()
def send(self, data, use_filters=True): def send(self, data, use_filters=True):
"""A wrapper for :meth:`send_raw()` for sending stanza objects. """A wrapper for :meth:`send_raw()` for sending stanza objects.
May optionally block until an expected response is received.
:param data: The :class:`~slixmpp.xmlstream.stanzabase.ElementBase` :param data: The :class:`~slixmpp.xmlstream.stanzabase.ElementBase`
stanza to send on the stream. stanza to send on the stream.
:param bool use_filters: Indicates if outgoing filters should be :param bool use_filters: Indicates if outgoing filters should be
@ -900,24 +938,7 @@ class XMLStream(asyncio.BaseProtocol):
filters is useful when resending stanzas. filters is useful when resending stanzas.
Defaults to ``True``. Defaults to ``True``.
""" """
if isinstance(data, ElementBase): self.waiting_queue.put_nowait((data, use_filters))
if use_filters:
for filter in self.__filters['out']:
data = filter(data)
if data is None:
return
if isinstance(data, ElementBase):
if use_filters:
for filter in self.__filters['out_sync']:
data = filter(data)
if data is None:
return
str_data = tostring(data.xml, xmlns=self.default_ns,
stream=self, top_level=True)
self.send_raw(str_data)
else:
self.send_raw(data)
def send_xml(self, data): def send_xml(self, data):
"""Send an XML object on the stream """Send an XML object on the stream