Server IP : 104.21.38.3 / Your IP : 162.158.106.186 Web Server : Apache System : Linux krdc-ubuntu-s-2vcpu-4gb-amd-blr1-01.localdomain 5.15.0-142-generic #152-Ubuntu SMP Mon May 19 10:54:31 UTC 2025 x86_64 User : www ( 1000) PHP Version : 7.4.33 Disable Function : passthru,exec,system,putenv,chroot,chgrp,chown,shell_exec,popen,proc_open,pcntl_exec,ini_alter,ini_restore,dl,openlog,syslog,readlink,symlink,popepassthru,pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,imap_open,apache_setenv MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /usr/lib/python3/dist-packages/jeepney/integrate/ |
Upload File : |
"""Deprecated: use jeepney.io.asyncio instead""" import asyncio from warnings import warn from jeepney import Parser, MessageType from jeepney.auth import AuthenticationError, BEGIN, make_auth_external, SASLParser from jeepney.bus import get_bus from jeepney.bus_messages import message_bus from jeepney.routing import Router from jeepney.wrappers import ProxyBase warn("jeepney.integrate.asyncio is deprecated: please use jeepney.io.asyncio " "instead.", stacklevel=2) class DBusProtocol(asyncio.Protocol): def __init__(self): self.auth_parser = SASLParser() self.parser = Parser() self.router = Router(asyncio.Future) self.authentication = asyncio.Future() self.unique_name = None def connection_made(self, transport): self.transport = transport self.transport.write(b'\0' + make_auth_external()) def _authenticated(self): self.transport.write(BEGIN) self.authentication.set_result(True) self.data_received = self.data_received_post_auth self.data_received(self.auth_parser.buffer) def data_received(self, data): self.auth_parser.feed(data) if self.auth_parser.authenticated: self._authenticated() elif self.auth_parser.error: self.authentication.set_exception(AuthenticationError(self.auth_parser.error)) def data_received_post_auth(self, data): for msg in self.parser.feed(data): self.router.incoming(msg) def send_message(self, message): if not self.authentication.done(): raise RuntimeError("Wait for authentication before sending messages") future = self.router.outgoing(message) data = message.serialise() self.transport.write(data) return future async def send_and_get_reply(self, message): if message.header.message_type != MessageType.method_call: raise TypeError("Only method call messages have replies") return await self.send_message(message) class Proxy(ProxyBase): """An asyncio proxy for calling D-Bus methods :param msggen: A message generator object. :param DBusProtocol proto: Protocol object to send and receive messages. """ def __init__(self, msggen, protocol): super().__init__(msggen) self._protocol = protocol def __repr__(self): return 'Proxy({}, {})'.format(self._msggen, self._protocol) def _method_call(self, make_msg): async def inner(*args, **kwargs): msg = make_msg(*args, **kwargs) assert msg.header.message_type is MessageType.method_call return await self._protocol.send_and_get_reply(msg) return inner async def connect_and_authenticate(bus='SESSION', loop=None): if loop is None: loop = asyncio.get_event_loop() (t, p) = await loop.create_unix_connection(DBusProtocol, path=get_bus(bus)) await p.authentication bus = Proxy(message_bus, p) hello_reply = await bus.Hello() p.unique_name = hello_reply[0] return (t, p)