libervia-backend: sat/plugins/plugin_xep

comparison sat/plugins/plugin_xep_0384.py @ 3715:b9718216a1c0 0.9

merge bookmark 0.9

author	Goffi <goffi@goffi.org>
date	Wed, 01 Dec 2021 16:13:31 +0100
parents	09f5ac48ffe3
children	11f7ca8afd15

comparison

equal deleted inserted replaced

-:af09b5aaa5d7
+:b9718216a1c0
 @param promise_(promise.Promise): promise to convert
 @return (defer.Deferred): deferred instance linked to the promise
 """
 d = defer.Deferred()
-promise_.then(d.callback, d.errback)
+promise_.then(
+lambda result: reactor.callFromThread(d.callback, result),
+lambda exc: reactor.callFromThread(d.errback, exc)
+)
 return d
 class OmemoStorage(omemo.Storage):
 This method use partial to call callback with boolean then result when
 Deferred is called
 """
 deferred.addCallback(partial(callback, True))
 deferred.addErrback(partial(callback, False))
+def _callMainThread(self, callback, method, *args, check_jid=None):
+if check_jid is None:
+d = method(*args)
+else:
+check_jid_d = self._checkJid(check_jid)
+check_jid_d.addCallback(lambda __: method(*args))
+d = check_jid_d
+if callback is not None:
+d.addCallback(partial(callback, True))
+d.addErrback(partial(callback, False))
+def _call(self, callback, method, *args, check_jid=None):
+"""Create Deferred and add Promise callback to it
+This method use reactor.callLater to launch Deferred in main thread
+@param check_jid: run self._checkJid before method
+"""
+reactor.callFromThread(
+self._callMainThread, callback, method, *args, check_jid=check_jid
+)
 def _checkJid(self, bare_jid):
 """Check if jid is known, and store it if not
 @param bare_jid(unicode): bare jid to check
 if own_bare_jid != self.own_bare_jid_s or own_device_id != self.device_id:
 raise exceptions.InternalError('bare jid or device id inconsistency!')
 callback(True, None)
 def loadState(self, callback):
-d = self.data.get(KEY_STATE)
+self._call(callback, self.data.get, KEY_STATE)
-self.setCb(d, callback)
 def storeState(self, callback, state):
-d = self.data.force(KEY_STATE, state)
+self._call(callback, self.data.force, KEY_STATE, state)
-self.setCb(d, callback)
 def loadSession(self, callback, bare_jid, device_id):
 key = '\n'.join([KEY_SESSION, bare_jid, str(device_id)])
-d = self.data.get(key)
+self._call(callback, self.data.get, key)
-self.setCb(d, callback)
 def storeSession(self, callback, bare_jid, device_id, session):
 key = '\n'.join([KEY_SESSION, bare_jid, str(device_id)])
-d = self.data.force(key, session)
+self._call(callback, self._data.force, key, session)
-self.setCb(d, callback)
 def deleteSession(self, callback, bare_jid, device_id):
 key = '\n'.join([KEY_SESSION, bare_jid, str(device_id)])
-d = self.data.remove(key)
+self._call(callback, self.data.remove, key)
-self.setCb(d, callback)
 def loadActiveDevices(self, callback, bare_jid):
 key = '\n'.join([KEY_ACTIVE_DEVICES, bare_jid])
-d = self.data.get(key, {})
+self._call(callback, self.data.get, key, {})
-if callback is not None:
-self.setCb(d, callback)
-return d
 def loadInactiveDevices(self, callback, bare_jid):
 key = '\n'.join([KEY_INACTIVE_DEVICES, bare_jid])
-d = self.data.get(key, {})
+self._call(callback, self.data.get, key, {})
-if callback is not None:
-self.setCb(d, callback)
-return d
 def storeActiveDevices(self, callback, bare_jid, devices):
 key = '\n'.join([KEY_ACTIVE_DEVICES, bare_jid])
-d = self._checkJid(bare_jid)
+self._call(callback, self.data.force, key, devices, check_jid=bare_jid)
-d.addCallback(lambda _: self.data.force(key, devices))
-self.setCb(d, callback)
 def storeInactiveDevices(self, callback, bare_jid, devices):
 key = '\n'.join([KEY_INACTIVE_DEVICES, bare_jid])
-d = self._checkJid(bare_jid)
+self._call(callback, self.data.force, key, devices, check_jid=bare_jid)
-d.addCallback(lambda _: self.data.force(key, devices))
-self.setCb(d, callback)
 def storeTrust(self, callback, bare_jid, device_id, trust):
 key = '\n'.join([KEY_TRUST, bare_jid, str(device_id)])
-d = self.data.force(key, trust)
+self._call(callback, self.data.force, key, trust)
-self.setCb(d, callback)
 def loadTrust(self, callback, bare_jid, device_id):
 key = '\n'.join([KEY_TRUST, bare_jid, str(device_id)])
-d = self.data.get(key)
+self._call(callback, self.data.get, key)
+def listJIDs(self, callback):
 if callback is not None:
-self.setCb(d, callback)
+callback(True, self.all_jids)
-return d
-def listJIDs(self, callback):
-d = defer.succeed(self.all_jids)
-if callback is not None:
-self.setCb(d, callback)
-return d
 def _deleteJID_logResults(self, results):
 failed = [success for success, __ in results if not success]
 if failed:
 log.warning(
 d_list.append(lambda __: self.data.force(KEY_ALL_JIDS, self.all_jids))
 d = defer.DeferredList(d_list)
 d.addCallback(self._deleteJID_logResults)
 return d
-def deleteJID(self, callback, bare_jid):
+def _deleteJID(self, callback, bare_jid):
-"""Retrieve all (in)actives devices of bare_jid, and delete all related keys"""
 d_list = []
 key = '\n'.join([KEY_ACTIVE_DEVICES, bare_jid])
 d_list.append(self.data.get(key, []))
 d_list.append(d_inactive)
 d = defer.DeferredList(d_list)
 d.addCallback(self._deleteJID_gotDevices, bare_jid)
 if callback is not None:
 self.setCb(d, callback)
-return d
+def deleteJID(self, callback, bare_jid):
+"""Retrieve all (in)actives devices of bare_jid, and delete all related keys"""
+reactor.callFromThread(self._deleteJID, callback, bare_jid)
 class SatOTPKPolicy(omemo.DefaultOTPKPolicy):
 pass
 device_id = random.randint(1, 2**31-1)
 # we check that it's really unique
 while device_id in devices:
 device_id = random.randint(1, 2**31-1)
 # and we save it
-persistent_dict[KEY_DEVICE_ID] = device_id
+await persistent_dict.aset(KEY_DEVICE_ID, device_id)
 log.debug(f"our OMEMO device id is {device_id}")
 if device_id not in devices:
 log.debug(f"our device id ({device_id}) is not in the list, adding it")
 device_id=device_elt['id']))
 else:
 devices.add(device_id)
 return devices
-@defer.inlineCallbacks
+async def getDevices(self, client, entity_jid=None):
-def getDevices(self, client, entity_jid=None):
 """Retrieve list of registered OMEMO devices
 @param entity_jid(jid.JID, None): get devices from this entity
 None to get our own devices
 @return (set(int)): list of devices
 """
 if entity_jid is not None:
 assert not entity_jid.resource
 try:
-items, metadata = yield self._p.getItems(client, entity_jid, NS_OMEMO_DEVICES)
+items, metadata = await self._p.getItems(client, entity_jid, NS_OMEMO_DEVICES)
 except exceptions.NotFound:
 log.info(_("there is no node to handle OMEMO devices"))
-defer.returnValue(set())
+return set()
 devices = self.parseDevices(items)
-defer.returnValue(devices)
+return devices
 async def setDevices(self, client, devices):
 log.debug(f"setting devices with {', '.join(str(d) for d in devices)}")
 list_elt = domish.Element((NS_OMEMO, 'list'))
 for device in devices:
 except Exception as e:
 log.warning(_("Can't set devices: {reason}").format(reason=e))
 # bundles
-@defer.inlineCallbacks
+async def getBundles(self, client, entity_jid, devices_ids):
-def getBundles(self, client, entity_jid, devices_ids):
 """Retrieve public bundles of an entity devices
 @param entity_jid(jid.JID): bare jid of entity
 @param devices_id(iterable[int]): ids of the devices bundles to retrieve
 @return (tuple(dict[int, ExtendedPublicBundle], list(int))):
 bundles = {}
 missing = set()
 for device_id in devices_ids:
 node = NS_OMEMO_BUNDLE.format(device_id=device_id)
 try:
-items, metadata = yield self._p.getItems(client, entity_jid, node)
+items, metadata = await self._p.getItems(client, entity_jid, node)
 except exceptions.NotFound:
 log.warning(_("Bundle missing for device {device_id}")
 .format(device_id=device_id))
 missing.add(device_id)
 continue
 continue
 bundles[device_id] = ExtendedPublicBundle.parse(omemo_backend, ik, spk,
 spkSignature, otpks)
-defer.returnValue((bundles, missing))
+return (bundles, missing)
 async def setBundle(self, client, bundle, device_id):
 """Set public bundle for this device.
 @param bundle(ExtendedPublicBundle): bundle to publish

Mercurial > libervia-backend

comparison sat/plugins/plugin_xep_0384.py @ 3715:b9718216a1c0 0.9