Mercurial > libervia-backend

comparison libervia/backend/memory/disco.py @ 4306:94e0968987cd

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
plugin XEP-0033: code modernisation, improve delivery, data validation: - Code has been rewritten using Pydantic models and `async` coroutines for data validation and cleaner element parsing/generation. - Delivery has been completely rewritten. It now works even if server doesn't support multicast, and send to local multicast service first. Delivering to local multicast service first is due to bad support of XEP-0033 in server (notably Prosody which has an incomplete implementation), and the current impossibility to detect if a sub-domain service handles fully multicast or only for local domains. This is a workaround to have a good balance between backward compatilibity and use of bandwith, and to make it work with the incoming email gateway implementation (the gateway will only deliver to entities of its own domain). - disco feature checking now uses `async` corountines. `host` implementation still use Deferred return values for compatibility with legacy code. rel 450
author Goffi <goffi@goffi.org>
date Thu, 26 Sep 2024 16:12:01 +0200
parents 0d7bb4df2343
children
comparison
equal deleted inserted replaced
4305:4cd4922de876 4306:94e0968987cd
1 #!/usr/bin/env python3 1 #!/usr/bin/env python3
2 2
3 3
4 # SAT: a jabber client 4 # Libervia XMPP client
5 # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) 5 # Copyright (C) 2009-2024 Jérôme Poisson (goffi@goffi.org)
6 6
7 # This program is free software: you can redistribute it and/or modify 7 # This program is free software: you can redistribute it and/or modify
8 # it under the terms of the GNU Affero General Public License as published by 8 # it under the terms of the GNU Affero General Public License as published by
9 # the Free Software Foundation, either version 3 of the License, or 9 # the Free Software Foundation, either version 3 of the License, or
10 # (at your option) any later version. 10 # (at your option) any later version.
15 # GNU Affero General Public License for more details. 15 # GNU Affero General Public License for more details.
16 16
17 # You should have received a copy of the GNU Affero General Public License 17 # You should have received a copy of the GNU Affero General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>. 18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
19 19
20 from typing import Optional 20 from typing import Iterable, Optional, cast
21
22 from twisted.internet.interfaces import IReactorCore
21 from libervia.backend.core.i18n import _ 23 from libervia.backend.core.i18n import _
22 from libervia.backend.core import exceptions 24 from libervia.backend.core import exceptions
23 from libervia.backend.core.log import getLogger 25 from libervia.backend.core.log import getLogger
24 from libervia.backend.core.core_types import SatXMPPEntity 26 from libervia.backend.core.core_types import SatXMPPEntity
25 27
120 def load(self): 122 def load(self):
121 """Load persistent hashes""" 123 """Load persistent hashes"""
122 self.hashes = HashManager(persistent.PersistentDict("disco")) 124 self.hashes = HashManager(persistent.PersistentDict("disco"))
123 return self.hashes.load() 125 return self.hashes.load()
124 126
125 @defer.inlineCallbacks 127 async def has_feature(
126 def hasFeature(self, client, feature, jid_=None, node=""): 128 self,
129 client: SatXMPPEntity,
130 feature: str,
131 jid_: jid.JID | None = None,
132 node: str = "",
133 ) -> bool:
127 """Tell if an entity has the required feature 134 """Tell if an entity has the required feature
128 135
129 @param feature: feature namespace 136 @param feature: feature namespace
130 @param jid_: jid of the target, or None for profile's server 137 @param jid_: jid of the target, or None for profile's server
131 @param node(unicode): optional node to use for disco request 138 @param node(unicode): optional node to use for disco request
132 @return: a Deferred which fire a boolean (True if feature is available) 139 @return: a Deferred which fire a boolean (True if feature is available)
133 """ 140 """
134 disco_infos = yield self.get_infos(client, jid_, node) 141 disco_infos = await self.get_infos(client, jid_, node)
135 defer.returnValue(feature in disco_infos.features) 142 return feature in disco_infos.features
136 143
137 @defer.inlineCallbacks 144 async def check_feature(
138 def check_feature(self, client, feature, jid_=None, node=""): 145 self,
139 """Like hasFeature, but raise an exception is feature is not Found 146 client: SatXMPPEntity,
147 feature: str,
148 jid_: jid.JID | None = None,
149 node: str = "",
150 ) -> None:
151 """Like has_feature, but raise an exception is feature is not Found
140 152
141 @param feature: feature namespace 153 @param feature: feature namespace
142 @param jid_: jid of the target, or None for profile's server 154 @param jid_: jid of the target, or None for profile's server
143 @param node(unicode): optional node to use for disco request 155 @param node: optional node to use for disco request
144 156
145 @raise: exceptions.FeatureNotFound 157 @raise: exceptions.FeatureNotFound
146 """ 158 """
147 disco_infos = yield self.get_infos(client, jid_, node) 159 disco_infos = await self.get_infos(client, jid_, node)
148 if not feature in disco_infos.features: 160 if not feature in disco_infos.features:
149 raise failure.Failure(exceptions.FeatureNotFound()) 161 raise exceptions.FeatureNotFound()
150 162
151 @defer.inlineCallbacks 163 async def check_features(
152 def check_features(self, client, features, jid_=None, identity=None, node=""): 164 self,
165 client: SatXMPPEntity,
166 features: Iterable[str],
167 jid_: jid.JID | None = None,
168 identity: tuple[str, str] | None = None,
169 node: str = "",
170 ) -> None:
153 """Like check_feature, but check several features at once, and check also identity 171 """Like check_feature, but check several features at once, and check also identity
154 172
155 @param features(iterable[unicode]): features to check 173 @param features: features to check
156 @param jid_(jid.JID): jid of the target, or None for profile's server 174 @param jid_: jid of the target, or None for profile's server
157 @param node(unicode): optional node to use for disco request 175 @param node: optional node to use for disco request
158 @param identity(None, tuple(unicode, unicode): if not None, the entity must have an identity with this (category, type) tuple 176 @param identity: if not None, the entity must have an identity with this
177 (category, type) tuple
159 178
160 @raise: exceptions.FeatureNotFound 179 @raise: exceptions.FeatureNotFound
161 """ 180 """
162 disco_infos = yield self.get_infos(client, jid_, node) 181 disco_infos = await self.get_infos(client, jid_, node)
163 if not set(features).issubset(disco_infos.features): 182 if not set(features).issubset(disco_infos.features):
164 raise failure.Failure(exceptions.FeatureNotFound()) 183 raise exceptions.FeatureNotFound()
165 184
166 if identity is not None and identity not in disco_infos.identities: 185 if identity is not None and identity not in disco_infos.identities:
167 raise failure.Failure(exceptions.FeatureNotFound()) 186 raise exceptions.FeatureNotFound()
168 187
169 async def has_identity( 188 async def has_identity(
170 self, 189 self,
171 client: SatXMPPEntity, 190 client: SatXMPPEntity,
172 category: str, 191 category: str,
336 reactor.callLater( 355 reactor.callLater(
337 TIMEOUT, d.cancel 356 TIMEOUT, d.cancel
338 ) # FIXME: one bad service make a general timeout 357 ) # FIXME: one bad service make a general timeout
339 return d 358 return d
340 359
341 def find_features_set(self, client, features, identity=None, jid_=None): 360 async def find_features_set(
361 self,
362 client: SatXMPPEntity,
363 features: Iterable[str],
364 identity: tuple[str, str] | None = None,
365 jid_: jid.JID | None = None,
366 ) -> set[jid.JID]:
342 """Return entities (including jid_ and its items) offering features 367 """Return entities (including jid_ and its items) offering features
343 368
369 @param client: Client session.
344 @param features: iterable of features which must be present 370 @param features: iterable of features which must be present
345 @param identity(None, tuple(unicode, unicode)): if not None, accept only this 371 @param identity: if not None, accept only this (category/type) identity
346 (category/type) identity
347 @param jid_: the jid of the target server (None for profile's server) 372 @param jid_: the jid of the target server (None for profile's server)
348 @param profile: %(doc_profile)s
349 @return: a set of found entities 373 @return: a set of found entities
350 """ 374 """
351 if jid_ is None: 375 if jid_ is None:
352 jid_ = jid.JID(client.jid.host) 376 jid_ = jid.JID(client.jid.host)
353 features = set(features) 377 features = set(features)
354 found_entities = set() 378 found_entities = set()
355 379
356 def infos_cb(infos, entity): 380 def infos_cb(infos: disco.DiscoInfo, entity: jid.JID) -> None:
357 if entity is None: 381 if entity is None:
358 log.warning(_("received an item without jid")) 382 log.warning(_("Received an item without JID"))
359 return 383 return
360 if identity is not None and identity not in infos.identities: 384 if identity is not None and identity not in infos.identities:
361 return 385 return
362 if features.issubset(infos.features): 386 if features.issubset(infos.features):
363 found_entities.add(entity) 387 found_entities.add(entity)
364 388
365 def got_items(items): 389 items = await self.get_items(client, jid_)
366 defer_list = [] 390 defer_list = []
367 for entity in [jid_] + [item.entity for item in items]: 391 for entity in [jid_] + [item.entity for item in items]:
368 infos_d = self.get_infos(client, entity) 392 infos_d = self.get_infos(client, entity)
369 infos_d.addCallbacks(infos_cb, self._infos_eb, [entity], None, [entity]) 393 infos_d.addCallbacks(infos_cb, self._infos_eb, [entity], None, [entity])
370 defer_list.append(infos_d) 394 infos_d.addTimeout(TIMEOUT, cast(IReactorCore, reactor))
371 return defer.DeferredList(defer_list) 395 defer_list.append(infos_d)
372 396 await defer.DeferredList(defer_list)
373 d = self.get_items(client, jid_) 397 return found_entities
374 d.addCallback(got_items)
375 d.addCallback(lambda __: found_entities)
376 reactor.callLater(
377 TIMEOUT, d.cancel
378 ) # FIXME: one bad service make a general timeout
379 return d
380 398
381 def generate_hash(self, services): 399 def generate_hash(self, services):
382 """Generate a unique hash for given service 400 """Generate a unique hash for given service
383 401
384 hash algorithm is the one described in XEP-0115 402 hash algorithm is the one described in XEP-0115