diff options
Diffstat (limited to 'slixmpp/plugins/xep_0047/ibb.py')
| -rw-r--r-- | slixmpp/plugins/xep_0047/ibb.py | 89 |
1 files changed, 62 insertions, 27 deletions
diff --git a/slixmpp/plugins/xep_0047/ibb.py b/slixmpp/plugins/xep_0047/ibb.py index bd96eca2..ec08a8b3 100644 --- a/slixmpp/plugins/xep_0047/ibb.py +++ b/slixmpp/plugins/xep_0047/ibb.py @@ -1,8 +1,17 @@ +# Slixmpp: The Slick XMPP Library +# This file is part of Slixmpp +# See the file LICENSE for copying permission import asyncio import uuid import logging -from slixmpp import Message, Iq +from typing import ( + Optional, + Union, +) + +from slixmpp import JID +from slixmpp.stanza import Message, Iq from slixmpp.exceptions import XMPPError from slixmpp.xmlstream.handler import Callback from slixmpp.xmlstream.matcher import StanzaPath @@ -15,9 +24,27 @@ log = logging.getLogger(__name__) class XEP_0047(BasePlugin): + """ + XEP-0047: In-Band Bytestreams + + Events registered by this plugin: + + - :term:`ibb_stream_start` + - :term:`ibb_stream_end` + - :term:`ibb_stream_data` + - :term:`stream:[stream id]:[peer jid]` + + Plugin Parameters: + + - ``block_size`` (default: ``4096``): default block size to negociate + - ``max_block_size`` (default: ``8192``): max block size to accept + - ``auto_accept`` (default: ``False``): if incoming streams should be + accepted automatically. + + """ name = 'xep_0047' - description = 'XEP-0047: In-band Bytestreams' + description = 'XEP-0047: In-Band Bytestreams' dependencies = {'xep_0030'} stanza = stanza default_config = { @@ -105,17 +132,29 @@ class XEP_0047(BasePlugin): def _preauthorize_sid(self, jid, sid, ifrom, data): self._preauthed_sids[(jid, sid, ifrom)] = True - def open_stream(self, jid, block_size=None, sid=None, use_messages=False, - ifrom=None, timeout=None, callback=None): + async def open_stream(self, jid: JID, *, block_size: Optional[int] = None, + sid: Optional[str] = None, use_messages: bool = False, + ifrom: Optional[JID] = None, + **iqkwargs) -> IBBytestream: + """Open an IBB stream with a peer JID. + + .. versionchanged:: 1.8.0 + This function is now a coroutine and must be awaited. + All parameters except ``jid`` are keyword-args only. + + :param jid: The remote JID to initiate the stream with. + :param block_size: The block size to advertise. + :param sid: The IBB stream id (if not provided, will be auto-generated). + :param use_messages: If the stream should use message stanzas instead of iqs. + :returns: The opened byte stream with the remote JID + :raises .IqError: When the remote entity denied the stream. + """ if sid is None: sid = str(uuid.uuid4()) if block_size is None: block_size = self.block_size - iq = self.xmpp.Iq() - iq['type'] = 'set' - iq['to'] = jid - iq['from'] = ifrom + iq = self.xmpp.make_iq_set(ito=jid, ifrom=ifrom) iq['ibb_open']['block_size'] = block_size iq['ibb_open']['sid'] = sid iq['ibb_open']['stanza'] = 'message' if use_messages else 'iq' @@ -123,25 +162,21 @@ class XEP_0047(BasePlugin): stream = IBBytestream(self.xmpp, sid, block_size, iq['from'], iq['to'], use_messages) - stream_future = asyncio.Future() + callback = iqkwargs.pop('callback', None) + result = await iq.send(**iqkwargs) - def _handle_opened_stream(iq): - log.debug('IBB stream (%s) accepted by %s', stream.sid, iq['from']) - stream.self_jid = iq['to'] - stream.peer_jid = iq['from'] - stream.stream_started = True - self.api['set_stream'](stream.self_jid, stream.sid, stream.peer_jid, stream) - stream_future.set_result(stream) - if callback is not None: - callback(stream) - self.xmpp.event('ibb_stream_start', stream) - self.xmpp.event('stream:%s:%s' % (stream.sid, stream.peer_jid), stream) - - iq.send(timeout=timeout, callback=_handle_opened_stream) - - return stream_future + log.debug('IBB stream (%s) accepted by %s', stream.sid, result['from']) + stream.self_jid = result['to'] + stream.peer_jid = result['from'] + stream.stream_started = True + self.api['set_stream'](stream.self_jid, stream.sid, stream.peer_jid, stream) + if callback is not None: + self.xmpp.add_event_handler('ibb_stream_start', callback, disposable=True) + self.xmpp.event('ibb_stream_start', stream) + self.xmpp.event('stream:%s:%s' % (stream.sid, stream.peer_jid), stream) + return stream - def _handle_open_request(self, iq): + def _handle_open_request(self, iq: Iq): sid = iq['ibb_open']['sid'] size = iq['ibb_open']['block_size'] or self.block_size @@ -165,7 +200,7 @@ class XEP_0047(BasePlugin): self.xmpp.event('ibb_stream_start', stream) self.xmpp.event('stream:%s:%s' % (sid, stream.peer_jid), stream) - def _handle_data(self, stanza): + def _handle_data(self, stanza: Union[Iq, Message]): sid = stanza['ibb_data']['sid'] stream = self.api['get_stream'](stanza['to'], sid, stanza['from']) if stream is not None and stanza['from'] == stream.peer_jid: @@ -173,7 +208,7 @@ class XEP_0047(BasePlugin): else: raise XMPPError('item-not-found') - def _handle_close(self, iq): + def _handle_close(self, iq: Iq): sid = iq['ibb_close']['sid'] stream = self.api['get_stream'](iq['to'], sid, iq['from']) if stream is not None and iq['from'] == stream.peer_jid: |