horazont / aioxmpp / 303

Build # 303

Build Type

push

travis-ci

Committed by Jonas Wielicki

Commit Message

<a href="https://github.com/horazont/aioxmpp/commit/<a class=hub.com/horazont/aioxmpp/commit/48a5d04230d347cf421e832b5d5499872cca18ea">48a5d0423<a href="https://github.com/horazont/aioxmpp/commit/48a5d04230d347cf421e832b5d5499872cca18ea">&quot;&gt;Revert &amp;quot;entitycaps: rate-limit and delay local hash updates&quot;

This reverts commit </a><a class="double-link" href="https://github.com/horazont/aioxmpp/commit/<a class="double-link" href="https://github.com/horazont/aioxmpp/commit/fbeb89d86a20e2b99d5631076b89774bc94e3e1c">fbeb89d86</a>">fbeb89d86</a><a href="https://github.com/horazont/aioxmpp/commit/48a5d04230d347cf421e832b5d5499872cca18ea">.

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

10254 of 10498 relevant lines covered (97.68%)

0.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.71

/aioxmpp/entitycaps/service.py

########################################################################
# File name: service.py
# This file is part of: aioxmpp
#
# LICENSE
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program.  If not, see
# <http://www.gnu.org/licenses/>.
#
########################################################################
import asyncio
import collections
import copy
import functools
import logging
import os
import tempfile
import urllib.parse

import aioxmpp.callbacks
import aioxmpp.disco as disco
import aioxmpp.service
import aioxmpp.xml
import aioxmpp.xso

from aioxmpp.utils import namespaces

from . import caps115, caps390


logger = logging.getLogger("aioxmpp.entitycaps")


class Cache:
    """
    This provides a two-level cache for entity capabilities information. The
    idea is to have a trusted database, e.g. installed system-wide or shipped
    with :mod:`aioxmpp` and in addition a user-level database which is
    automatically filled with hashes which have been found by the
    :class:`Service`.

    The trusted database is taken as read-only and overrides the user-collected
    database. When a hash is in both databases, it is removed from the
    user-collected database (to save space).

    In addition to serving the databases, it provides deduplication for queries
    by holding a cache of futures looking up the same hash.

    Database management (user API):

    .. automethod:: set_system_db_path

    .. automethod:: set_user_db_path

    Queries (API intended for :class:`Service`):

    .. automethod:: create_query_future

    .. automethod:: lookup_in_database

    .. automethod:: lookup
    """

    def __init__(self):
        self._lookup_cache = {}
        self._memory_overlay = {}
        self._system_db_path = None
        self._user_db_path = None

    def _erase_future(self, key, fut):
        try:
            existing = self._lookup_cache[key]
        except KeyError:
            pass
        else:
            if existing is fut:
                del self._lookup_cache[key]

    def set_system_db_path(self, path):
        self._system_db_path = path

    def set_user_db_path(self, path):
        self._user_db_path = path

    def lookup_in_database(self, key):
        try:
            result = self._memory_overlay[key]
        except KeyError:
            pass
        else:
            logger.debug("memory cache hit: %s", key)
            return result

        key_path = key.path

        if self._system_db_path is not None:
            try:
                f = (
                    self._system_db_path / key_path
                ).open("rb")
            except OSError:
                pass
            else:
                logger.debug("system db hit: %s", key)
                with f:
                    return aioxmpp.xml.read_single_xso(f, disco.xso.InfoQuery)

        if self._user_db_path is not None:
            try:
                f = (
                    self._user_db_path / key_path
                ).open("rb")
            except OSError:
                pass
            else:
                logger.debug("user db hit: %s", key)
                with f:
                    return aioxmpp.xml.read_single_xso(f, disco.xso.InfoQuery)

        raise KeyError(key)

    @asyncio.coroutine
    def lookup(self, key):
        """
        Look up the given `node` URL using the given `hash_` first in the
        database and then by waiting on the futures created with
        :meth:`create_query_future` for that node URL and hash.

        If the hash is not in the database, :meth:`lookup` iterates as long as
        there are pending futures for the given `hash_` and `node`. If there
        are no pending futures, :class:`KeyError` is raised. If a future raises
        a :class:`ValueError`, it is ignored. If the future returns a value, it
        is used as the result.
        """
        try:
            result = self.lookup_in_database(key)
        except KeyError:
            pass
        else:
            return result

        while True:
            fut = self._lookup_cache[key]
            try:
                result = yield from fut
            except ValueError:
                continue
            else:
                return result

    def create_query_future(self, key):
        """
        Create and return a :class:`asyncio.Future` for the given `hash_`
        function and `node` URL. The future is referenced internally and used
        by any calls to :meth:`lookup` which are made while the future is
        pending. The future is removed from the internal storage automatically
        when a result or exception is set for it.

        This allows for deduplication of queries for the same hash.
        """
        fut = asyncio.Future()
        fut.add_done_callback(
            functools.partial(self._erase_future, key)
        )
        self._lookup_cache[key] = fut
        return fut

    def add_cache_entry(self, key, entry):
        """
        Add the given `entry` (which must be a :class:`~.disco.xso.InfoQuery`
        instance) to the user-level database keyed with the hash function type
        `hash_` and the `node` URL. The `entry` is **not** validated to
        actually map to `node` with the given `hash_` function, it is expected
        that the caller perfoms the validation.
        """
        copied_entry = copy.copy(entry)
        self._memory_overlay[key] = copied_entry
        if self._user_db_path is not None:
            asyncio.async(asyncio.get_event_loop().run_in_executor(
                None,
                writeback,
                self._user_db_path / key.path,
                entry.captured_events))


class EntityCapsService(aioxmpp.service.Service):
    """
    This service implements :xep:`0115`, transparently. Besides loading the
    service, no interaction is required to get some of the benefits of
    :xep:`0115`.

    Two additional things need to be done by users to get full support and
    performance:

    1. To make sure that peers are always up-to-date with the current
       capabilities, it is required that users listen on the
       :meth:`on_ver_changed` signal and re-emit their current presence when it
       fires.

       The service takes care of attaching capabilities information on the
       outgoing stanza, using a stanza filter.

    2. Users should use a process-wide :class:`Cache` instance and assign it to
       the :attr:`cache` of each :class:`.entitycaps.Service` they use. This
       improves performance by sharing (verified) hashes among :class:`Service`
       instances.

       In addition, the hashes should be saved and restored on shutdown/start
       of the process. See the :class:`Cache` for details.

    .. signal:: on_ver_changed

       The signal emits whenever the Capability Hashset of the local client
       changes. This happens when the set of features or identities announced
       in the :class:`.DiscoServer` changes.

    .. autoattribute:: cache

    .. autoattribute:: xep115_support

    .. autoattribute:: xep390_support

    .. versionchanged:: 0.8

       This class was formerly known as :class:`aioxmpp.entitycaps.Service`. It
       is still available under that name, but the alias will be removed in
       1.0.

    .. versionchanged:: 0.9

        Support for :xep:`390` was added.

    """

    ORDER_AFTER = {
        disco.DiscoClient,
        disco.DiscoServer,
    }

    NODE = "http://aioxmpp.zombofant.net/"

    on_ver_changed = aioxmpp.callbacks.Signal()

    _xep115_feature = disco.register_feature(namespaces.xep0115_caps)
    _xep390_feature = disco.register_feature(namespaces.xep0390_caps)

    def __init__(self, node, **kwargs):
        super().__init__(node, **kwargs)

        self.__current_keys = {}
        self._cache = Cache()

        self.disco_server = self.dependencies[disco.DiscoServer]
        self.disco_client = self.dependencies[disco.DiscoClient]

        self.__115 = caps115.Implementation(self.NODE)
        self.__390 = caps390.Implementation(
            aioxmpp.hashes.default_hash_algorithms
        )

        self.__active_hashsets = []
        self.__key_users = collections.Counter()

    @property
    def xep115_support(self):
        """
        Boolean to control whether :xep:`115` support is enabled or not.

        Defaults to :data:`True`.

        If set to false, inbound :xep:`115` capabilities will not be processed
        and no :xep:`115` capabilities will be emitted.

        .. note::

            At some point, this will default to :data:`False` to save
            bandwidth. The exact release depends on the adoption of :xep:`390`
            and will be announced in time. If you depend on :xep:`115` support,
            set this boolean to :data:`True`.

            The attribute itself will not be removed until :xep:`115` support
            is removed from :mod:`aioxmpp` entirely, which is unlikely to
            happen any time soon.

        .. versionadded:: 0.9
        """

        return self._xep115_feature.enabled

    @xep115_support.setter
    def xep115_support(self, value):
        self._xep115_feature.enabled = value

    @property
    def xep390_support(self):
        """
        Boolean to control whether :xep:`390` support is enabled or not.

        Defaults to :data:`True`.

        If set to false, inbound :xep:`390` Capability Hash Sets will not be
        processed and no Capability Hash Sets or Capability Nodes will be
        generated.

        The hash algortihms used for generating Capability Hash Sets are those
        from :data:`aioxmpp.hashes.default_hash_algorithms`.
        """
        return self._xep390_feature.enabled

    @xep390_support.setter
    def xep390_support(self, value):
        self._xep390_feature.enabled = value

    @property
    def cache(self):
        """
        The :class:`Cache` instance used for this :class:`Service`. Deleting
        this attribute will automatically create a new :class:`Cache` instance.

        The attribute can be used to share a single :class:`Cache` among
        multiple :class:`Service` instances.
        """
        return self._cache

    @cache.setter
    def cache(self, v):
        self._cache = v

    @cache.deleter
    def cache(self):
        self._cache = Cache()

    @aioxmpp.service.depsignal(
        disco.DiscoServer,
        "on_info_changed")
    def _info_changed(self):
        self.logger.debug("info changed, scheduling re-calculation of version")
        asyncio.get_event_loop().call_soon(
            self.update_hash
        )

    @asyncio.coroutine
    def _shutdown(self):
        for group in self.__current_keys.values():
            for key in group:
                self.disco_server.unmount_node(key.node)

    @asyncio.coroutine
    def query_and_cache(self, jid, key, fut):
        data = yield from self.disco_client.query_info(
            jid,
            node=key.node,
            require_fresh=True)

        try:
            if key.verify(data):
                self.cache.add_cache_entry(key, data)
                fut.set_result(data)
            else:
                raise ValueError("hash mismatch")
        except ValueError as exc:
            fut.set_exception(exc)

        return data

    @asyncio.coroutine
    def lookup_info(self, jid, keys):
        for key in keys:
            try:
                info = yield from self.cache.lookup(key)
            except KeyError:
                continue

            self.logger.debug("found %s in cache", key)
            return info

        first_key = keys[0]
        self.logger.debug("using key %s to query peer", first_key)
        fut = self.cache.create_query_future(first_key)
        info = yield from self.query_and_cache(
            jid, first_key, fut
        )
        self.logger.debug("%s maps to %r", key, info)

        return info

    @aioxmpp.service.outbound_presence_filter
    def handle_outbound_presence(self, presence):
        if (presence.type_ == aioxmpp.structs.PresenceType.AVAILABLE
                and self.__active_hashsets):
            current_hashset = self.__active_hashsets[-1]

            try:
                keys = current_hashset[self.__115]
            except KeyError:
                pass
            else:
                self.__115.put_keys(keys, presence)

            try:
                keys = current_hashset[self.__390]
            except KeyError:
                pass
            else:
                self.__390.put_keys(keys, presence)

        return presence

    @aioxmpp.service.inbound_presence_filter
    def handle_inbound_presence(self, presence):
        keys = []

        if self.xep390_support:
            keys.extend(self.__390.extract_keys(presence))

        if self.xep115_support:
            keys.extend(self.__115.extract_keys(presence))

        if keys:
            lookup_task = asyncio.async(
                self.lookup_info(presence.from_, keys)
            )
            self.disco_client.set_info_future(
                presence.from_,
                None,
                lookup_task
            )

        return presence

    def _push_hashset(self, node, hashset):
        if self.__active_hashsets and hashset == self.__active_hashsets[-1]:
            return False

        for group in hashset.values():
            for key in group:
                if not self.__key_users[key.node]:
                    self.disco_server.mount_node(key.node, node)
                self.__key_users[key.node] += 1
        self.__active_hashsets.append(hashset)

        for expired in self.__active_hashsets[:-3]:
            for group in expired.values():
                for key in group:
                    self.__key_users[key.node] -= 1
                    if not self.__key_users[key.node]:
                        self.disco_server.unmount_node(key.node)
                        del self.__key_users[key.node]

        del self.__active_hashsets[:-3]

        return True

    def update_hash(self):
        node = disco.StaticNode.clone(self.disco_server)
        info = node.as_info_xso()

        new_hashset = {}

        if self.xep115_support:
            new_hashset[self.__115] = set(self.__115.calculate_keys(info))

        if self.xep390_support:
            new_hashset[self.__390] = set(self.__390.calculate_keys(info))

        self.logger.debug("new hashset=%r", new_hashset)

        if self._push_hashset(node, new_hashset):
            self.on_ver_changed()


def writeback(base_path, hash_, node, captured_events):
    quoted = urllib.parse.quote(node, safe="")
    dest_path = base_path / "{}_{}.xml".format(hash_, quoted)
    with tempfile.NamedTemporaryFile(dir=str(base_path), delete=False) as tmpf:
        try:
            generator = aioxmpp.xml.XMPPXMLGenerator(
                tmpf,
                short_empty_elements=True)
            generator.startDocument()
            aioxmpp.xso.events_to_sax(captured_events, generator)
            generator.endDocument()
        except:
            os.unlink(tmpf.name)
            raise
        os.replace(tmpf.name, str(dest_path))

1	########################################################################
2	# File name: service.py
3	# This file is part of: aioxmpp
4	#
5	# LICENSE
6	#
7	# This program is free software: you can redistribute it and/or modify
8	# it under the terms of the GNU Lesser General Public License as
9	# published by the Free Software Foundation, either version 3 of the
10	# License, or (at your option) any later version.
11	#
12	# This program is distributed in the hope that it will be useful, but
13	# WITHOUT ANY WARRANTY; without even the implied warranty of
14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15	# Lesser General Public License for more details.
16	#
17	# You should have received a copy of the GNU Lesser General Public
18	# License along with this program. If not, see
19	# <http://www.gnu.org/licenses/>.
20	#
21	########################################################################
22	import asyncio	1✔
23	import collections	1✔
24	import copy	1✔
25	import functools	1✔
26	import logging	1✔
27	import os	1✔
28	import tempfile	1✔
29	import urllib.parse	1✔
30
31	import aioxmpp.callbacks	1✔
32	import aioxmpp.disco as disco	1✔
33	import aioxmpp.service	1✔
34	import aioxmpp.xml	1✔
35	import aioxmpp.xso	1✔
36
37	from aioxmpp.utils import namespaces	1✔
38
39	from . import caps115, caps390	1✔
40
41
42	logger = logging.getLogger("aioxmpp.entitycaps")	1✔
43
44
45	class Cache:	1✔
46	"""
47	This provides a two-level cache for entity capabilities information. The
48	idea is to have a trusted database, e.g. installed system-wide or shipped
49	with :mod:`aioxmpp` and in addition a user-level database which is
50	automatically filled with hashes which have been found by the
51	:class:`Service`.
52
53	The trusted database is taken as read-only and overrides the user-collected
54	database. When a hash is in both databases, it is removed from the
55	user-collected database (to save space).
56
57	In addition to serving the databases, it provides deduplication for queries
58	by holding a cache of futures looking up the same hash.
59
60	Database management (user API):
61
62	.. automethod:: set_system_db_path
63
64	.. automethod:: set_user_db_path
65
66	Queries (API intended for :class:`Service`):
67
68	.. automethod:: create_query_future
69
70	.. automethod:: lookup_in_database
71
72	.. automethod:: lookup
73	"""
74
75	def __init__(self):	1✔
76	self._lookup_cache = {}	1✔
77	self._memory_overlay = {}	1✔
78	self._system_db_path = None	1✔
79	self._user_db_path = None	1✔
80
81	def _erase_future(self, key, fut):	1✔
82	try:	1✔
83	existing = self._lookup_cache[key]	1✔
84	except KeyError:	×
85	pass	×
86	else:
87	if existing is fut:	1✔
88	del self._lookup_cache[key]	1✔
89
90	def set_system_db_path(self, path):	1✔
91	self._system_db_path = path	1✔
92
93	def set_user_db_path(self, path):	1✔
94	self._user_db_path = path	1✔
95
96	def lookup_in_database(self, key):	1✔
97	try:	1✔
98	result = self._memory_overlay[key]	1✔
99	except KeyError:	1✔
100	pass	1✔
101	else:
102	logger.debug("memory cache hit: %s", key)	1✔
103	return result	1✔
104
105	key_path = key.path	1✔
106
107	if self._system_db_path is not None:	1✔
108	try:	1✔
109	f = (	1✔
110	self._system_db_path / key_path
111	).open("rb")
112	except OSError:	1✔
113	pass	1✔
114	else:
115	logger.debug("system db hit: %s", key)	1✔
116	with f:	1✔
117	return aioxmpp.xml.read_single_xso(f, disco.xso.InfoQuery)	1✔
118
119	if self._user_db_path is not None:	1✔
120	try:	1✔
121	f = (	1✔
122	self._user_db_path / key_path
123	).open("rb")
124	except OSError:	×
125	pass	×
126	else:
127	logger.debug("user db hit: %s", key)	1✔
128	with f:	1✔
129	return aioxmpp.xml.read_single_xso(f, disco.xso.InfoQuery)	1✔
130
131	raise KeyError(key)	1✔
132
133	@asyncio.coroutine	1✔
134	def lookup(self, key):
135	"""
136	Look up the given `node` URL using the given `hash_` first in the
137	database and then by waiting on the futures created with
138	:meth:`create_query_future` for that node URL and hash.
139
140	If the hash is not in the database, :meth:`lookup` iterates as long as
141	there are pending futures for the given `hash_` and `node`. If there
142	are no pending futures, :class:`KeyError` is raised. If a future raises
143	a :class:`ValueError`, it is ignored. If the future returns a value, it
144	is used as the result.
145	"""
146	try:	1✔
147	result = self.lookup_in_database(key)	1✔
148	except KeyError:	1✔
149	pass	1✔
150	else:
151	return result	1✔
152
153	while True:	1✔
154	fut = self._lookup_cache[key]	1✔
155	try:	1✔
156	result = yield from fut	1✔
157	except ValueError:	1✔
158	continue	1✔
159	else:
160	return result	1✔
161
162	def create_query_future(self, key):	1✔
163	"""
164	Create and return a :class:`asyncio.Future` for the given `hash_`
165	function and `node` URL. The future is referenced internally and used
166	by any calls to :meth:`lookup` which are made while the future is
167	pending. The future is removed from the internal storage automatically
168	when a result or exception is set for it.
169
170	This allows for deduplication of queries for the same hash.
171	"""
172	fut = asyncio.Future()	1✔
173	fut.add_done_callback(	1✔
174	functools.partial(self._erase_future, key)
175	)
176	self._lookup_cache[key] = fut	1✔
177	return fut	1✔
178
179	def add_cache_entry(self, key, entry):	1✔
180	"""
181	Add the given `entry` (which must be a :class:`~.disco.xso.InfoQuery`
182	instance) to the user-level database keyed with the hash function type
183	`hash_` and the `node` URL. The `entry` is not validated to
184	actually map to `node` with the given `hash_` function, it is expected
185	that the caller perfoms the validation.
186	"""
187	copied_entry = copy.copy(entry)	1✔
188	self._memory_overlay[key] = copied_entry	1✔
189	if self._user_db_path is not None:	1✔
190	asyncio.async(asyncio.get_event_loop().run_in_executor(	1✔
191	None,
192	writeback,
193	self._user_db_path / key.path,
194	entry.captured_events))
195
196
197	class EntityCapsService(aioxmpp.service.Service):	1✔
198	"""
199	This service implements :xep:`0115`, transparently. Besides loading the
200	service, no interaction is required to get some of the benefits of
201	:xep:`0115`.
202
203	Two additional things need to be done by users to get full support and
204	performance:
205
206	1. To make sure that peers are always up-to-date with the current
207	capabilities, it is required that users listen on the
208	:meth:`on_ver_changed` signal and re-emit their current presence when it
209	fires.
210
211	The service takes care of attaching capabilities information on the
212	outgoing stanza, using a stanza filter.
213
214	2. Users should use a process-wide :class:`Cache` instance and assign it to
215	the :attr:`cache` of each :class:`.entitycaps.Service` they use. This
216	improves performance by sharing (verified) hashes among :class:`Service`
217	instances.
218
219	In addition, the hashes should be saved and restored on shutdown/start
220	of the process. See the :class:`Cache` for details.
221
222	.. signal:: on_ver_changed
223
224	The signal emits whenever the Capability Hashset of the local client
225	changes. This happens when the set of features or identities announced
226	in the :class:`.DiscoServer` changes.
227
228	.. autoattribute:: cache
229
230	.. autoattribute:: xep115_support
231
232	.. autoattribute:: xep390_support
233
234	.. versionchanged:: 0.8
235
236	This class was formerly known as :class:`aioxmpp.entitycaps.Service`. It
237	is still available under that name, but the alias will be removed in
238	1.0.
239
240	.. versionchanged:: 0.9
241
242	Support for :xep:`390` was added.
243
244	"""
245
246	ORDER_AFTER = {	1✔
247	disco.DiscoClient,
248	disco.DiscoServer,
249	}
250
251	NODE = "http://aioxmpp.zombofant.net/"	1✔
252
253	on_ver_changed = aioxmpp.callbacks.Signal()	1✔
254
255	_xep115_feature = disco.register_feature(namespaces.xep0115_caps)	1✔
256	_xep390_feature = disco.register_feature(namespaces.xep0390_caps)	1✔
257
258	def __init__(self, node, **kwargs):	1✔
259	super().__init__(node, **kwargs)	1✔
260
261	self.__current_keys = {}	1✔
262	self._cache = Cache()	1✔
263
264	self.disco_server = self.dependencies[disco.DiscoServer]	1✔
265	self.disco_client = self.dependencies[disco.DiscoClient]	1✔
266
267	self.__115 = caps115.Implementation(self.NODE)	1✔
268	self.__390 = caps390.Implementation(	1✔
269	aioxmpp.hashes.default_hash_algorithms
270	)
271
272	self.__active_hashsets = []	1✔
273	self.__key_users = collections.Counter()	1✔
274
275	@property	1✔
276	def xep115_support(self):
277	"""
278	Boolean to control whether :xep:`115` support is enabled or not.
279
280	Defaults to :data:`True`.
281
282	If set to false, inbound :xep:`115` capabilities will not be processed
283	and no :xep:`115` capabilities will be emitted.
284
285	.. note::
286
287	At some point, this will default to :data:`False` to save
288	bandwidth. The exact release depends on the adoption of :xep:`390`
289	and will be announced in time. If you depend on :xep:`115` support,
290	set this boolean to :data:`True`.
291
292	The attribute itself will not be removed until :xep:`115` support
293	is removed from :mod:`aioxmpp` entirely, which is unlikely to
294	happen any time soon.
295
296	.. versionadded:: 0.9
297	"""
298
299	return self._xep115_feature.enabled	1✔
300
301	@xep115_support.setter	1✔
302	def xep115_support(self, value):
303	self._xep115_feature.enabled = value	1✔
304
305	@property	1✔
306	def xep390_support(self):
307	"""
308	Boolean to control whether :xep:`390` support is enabled or not.
309
310	Defaults to :data:`True`.
311
312	If set to false, inbound :xep:`390` Capability Hash Sets will not be
313	processed and no Capability Hash Sets or Capability Nodes will be
314	generated.
315
316	The hash algortihms used for generating Capability Hash Sets are those
317	from :data:`aioxmpp.hashes.default_hash_algorithms`.
318	"""
319	return self._xep390_feature.enabled	1✔
320
321	@xep390_support.setter	1✔
322	def xep390_support(self, value):
323	self._xep390_feature.enabled = value	1✔
324
325	@property	1✔
326	def cache(self):
327	"""
328	The :class:`Cache` instance used for this :class:`Service`. Deleting
329	this attribute will automatically create a new :class:`Cache` instance.
330
331	The attribute can be used to share a single :class:`Cache` among
332	multiple :class:`Service` instances.
333	"""
334	return self._cache	1✔
335
336	@cache.setter	1✔
337	def cache(self, v):
338	self._cache = v	1✔
339
340	@cache.deleter	1✔
341	def cache(self):
342	self._cache = Cache()	1✔
343
344	@aioxmpp.service.depsignal(	1✔
345	disco.DiscoServer,
346	"on_info_changed")
347	def _info_changed(self):
348	self.logger.debug("info changed, scheduling re-calculation of version")	1✔
349	asyncio.get_event_loop().call_soon(	1✔
350	self.update_hash
351	)
352
353	@asyncio.coroutine	1✔
354	def _shutdown(self):
355	for group in self.__current_keys.values():	×
356	for key in group:	×
357	self.disco_server.unmount_node(key.node)	×
358
359	@asyncio.coroutine	1✔
360	def query_and_cache(self, jid, key, fut):
361	data = yield from self.disco_client.query_info(	1✔
362	jid,
363	node=key.node,
364	require_fresh=True)
365
366	try:	1✔
367	if key.verify(data):	1✔
368	self.cache.add_cache_entry(key, data)	1✔
369	fut.set_result(data)	1✔
370	else:
371	raise ValueError("hash mismatch")	1✔
372	except ValueError as exc:	1✔
373	fut.set_exception(exc)	1✔
374
375	return data	1✔
376
377	@asyncio.coroutine	1✔
378	def lookup_info(self, jid, keys):
379	for key in keys:	1✔
380	try:	1✔
381	info = yield from self.cache.lookup(key)	1✔
382	except KeyError:	1✔
383	continue	1✔
384
385	self.logger.debug("found %s in cache", key)	1✔
386	return info	1✔
387
388	first_key = keys[0]	1✔
389	self.logger.debug("using key %s to query peer", first_key)	1✔
390	fut = self.cache.create_query_future(first_key)	1✔
391	info = yield from self.query_and_cache(	1✔
392	jid, first_key, fut
393	)
394	self.logger.debug("%s maps to %r", key, info)	1✔
395
396	return info	1✔
397
398	@aioxmpp.service.outbound_presence_filter	1✔
399	def handle_outbound_presence(self, presence):
400	if (presence.type_ == aioxmpp.structs.PresenceType.AVAILABLE	1✔
401	and self.__active_hashsets):
402	current_hashset = self.__active_hashsets[-1]	1✔
403
404	try:	1✔
405	keys = current_hashset[self.__115]	1✔
406	except KeyError:	×
407	pass	×
408	else:
409	self.__115.put_keys(keys, presence)	1✔
410
411	try:	1✔
412	keys = current_hashset[self.__390]	1✔
413	except KeyError:	×
414	pass	×
415	else:
416	self.__390.put_keys(keys, presence)	1✔
417
418	return presence	1✔
419
420	@aioxmpp.service.inbound_presence_filter	1✔
421	def handle_inbound_presence(self, presence):
422	keys = []	1✔
423
424	if self.xep390_support:	1✔
425	keys.extend(self.__390.extract_keys(presence))	1✔
426
427	if self.xep115_support:	1✔
428	keys.extend(self.__115.extract_keys(presence))	1✔
429
430	if keys:	1✔
431	lookup_task = asyncio.async(	1✔
432	self.lookup_info(presence.from_, keys)
433	)
434	self.disco_client.set_info_future(	1✔
435	presence.from_,
436	None,
437	lookup_task
438	)
439
440	return presence	1✔
441
442	def _push_hashset(self, node, hashset):	1✔
443	if self.__active_hashsets and hashset == self.__active_hashsets[-1]:	1✔
444	return False	1✔
445
446	for group in hashset.values():	1✔
447	for key in group:	1✔
448	if not self.__key_users[key.node]:	1✔
449	self.disco_server.mount_node(key.node, node)	1✔
450	self.__key_users[key.node] += 1	1✔
451	self.__active_hashsets.append(hashset)	1✔
452
453	for expired in self.__active_hashsets[:-3]:	1✔
454	for group in expired.values():	1✔
455	for key in group:	1✔
456	self.__key_users[key.node] -= 1	1✔
457	if not self.__key_users[key.node]:	1✔
458	self.disco_server.unmount_node(key.node)	1✔
459	del self.__key_users[key.node]	1✔
460
461	del self.__active_hashsets[:-3]	1✔
462
463	return True	1✔
464
465	def update_hash(self):	1✔
466	node = disco.StaticNode.clone(self.disco_server)	1✔
467	info = node.as_info_xso()	1✔
468
469	new_hashset = {}	1✔
470
471	if self.xep115_support:	1✔
472	new_hashset[self.__115] = set(self.__115.calculate_keys(info))	1✔
473
474	if self.xep390_support:	1✔
475	new_hashset[self.__390] = set(self.__390.calculate_keys(info))	1✔
476
477	self.logger.debug("new hashset=%r", new_hashset)	1✔
478
479	if self._push_hashset(node, new_hashset):	1✔
480	self.on_ver_changed()	1✔
481
482
483	def writeback(base_path, hash_, node, captured_events):	1✔
484	quoted = urllib.parse.quote(node, safe="")	1✔
485	dest_path = base_path / "{}_{}.xml".format(hash_, quoted)	1✔
486	with tempfile.NamedTemporaryFile(dir=str(base_path), delete=False) as tmpf:	1✔
487	try:	1✔
488	generator = aioxmpp.xml.XMPPXMLGenerator(	1✔
489	tmpf,
490	short_empty_elements=True)
491	generator.startDocument()	1✔
492	aioxmpp.xso.events_to_sax(captured_events, generator)	1✔
493	generator.endDocument()	1✔
494	except:	1✔
495	os.unlink(tmpf.name)	1✔
496	raise	1✔
497	os.replace(tmpf.name, str(dest_path))	1✔

horazont / aioxmpp / 303

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous