byp4a55 professor6T9

Linux sonniq.ru 5.15.0-161-generic #171-Ubuntu SMP Sat Oct 11 08:17:01 UTC 2025 x86_64
Apache/2.4.52 (Ubuntu)
127.0.0.1

File Name : routing.py

from warnings import warn

from .low_level import MessageType, HeaderFields
from .wrappers import DBusErrorResponse

class Router:
    """Routing for messages coming back to a client application.
    
    :param handle_factory: Constructor for an object like asyncio.Future,
        with methods *set_result* and *set_exception*. Outgoing method call
        messages will get a handle associated with them.
    :param on_unhandled: Callback for messages not otherwise dispatched.
    """
    def __init__(self, handle_factory, on_unhandled=None):
        self.handle_factory = handle_factory
        self._on_unhandled = on_unhandled
        self.outgoing_serial = 0
        self.awaiting_reply = {}
        self.signal_callbacks = {}

@property
    def on_unhandled(self):
        return self._on_unhandled

@on_unhandled.setter
    def on_unhandled(self, value):
        warn("Setting on_unhandled is deprecated. Please use the filter() "
             "method or simple receive() calls instead.", stacklevel=2)
        self._on_unhandled = value

def outgoing(self, msg):
        """Set the serial number in the message & make a handle if a method call
        """
        self.outgoing_serial += 1
        msg.header.serial = self.outgoing_serial

if msg.header.message_type is MessageType.method_call:
            self.awaiting_reply[msg.header.serial] = handle = self.handle_factory()
            return handle

def subscribe_signal(self, callback, path, interface, member):
        """Add a callback for a signal.
        """
        warn("The subscribe_signal() method is deprecated. "
             "Please use the filter() API instead.", stacklevel=2)
        self.signal_callbacks[(path, interface, member)] = callback

def incoming(self, msg):
        """Route an incoming message.
        """
        hdr = msg.header

# Signals:
        if hdr.message_type is MessageType.signal:
            key = (hdr.fields.get(HeaderFields.path, None),
                   hdr.fields.get(HeaderFields.interface, None),
                   hdr.fields.get(HeaderFields.member, None)
                  )
            cb = self.signal_callbacks.get(key, None)
            if cb is not None:
                cb(msg.body)
                return

# Method returns & errors
        reply_serial = hdr.fields.get(HeaderFields.reply_serial, -1)
        reply_handle = self.awaiting_reply.pop(reply_serial, None)
        if reply_handle is not None:
            if hdr.message_type is MessageType.method_return:
                reply_handle.set_result(msg.body)
                return
            elif hdr.message_type is MessageType.error:
                reply_handle.set_exception(DBusErrorResponse(msg))
                return

if self.on_unhandled:
            self.on_unhandled(msg)

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	0/0	0755	2022-04-21 04:01:10
..	dir	-	0/0	0755	2025-12-29 06:26:53
__pycache__	dir	-	0/0	0755	2022-04-21 04:01:10
integrate	dir	-	0/0	0755	2022-04-21 04:01:10
io	dir	-	0/0	0755	2022-04-21 04:01:10
tests	dir	-	0/0	0755	2022-04-21 04:01:10
__init__.py	text/plain	408 B	0/0	0644	2021-07-28 09:14:49
auth.py	text/x-script.python	4.82 KB	0/0	0644	2021-07-10 06:19:49
bindgen.py	text/plain	3.96 KB	0/0	0644	2021-07-10 06:19:49
bus.py	text/plain	1.77 KB	0/0	0644	2017-02-26 07:42:05
bus_messages.py	text/plain	7.95 KB	0/0	0644	2021-07-10 06:19:49
fds.py	text/x-script.python	4.94 KB	0/0	0644	2021-07-10 06:19:49
low_level.py	text/x-script.python	18.67 KB	0/0	0644	2021-07-10 06:19:49
routing.py	text/x-script.python	2.76 KB	0/0	0644	2021-07-10 06:28:19
wrappers.py	text/x-script.python	7.79 KB	0/0	0644	2021-07-10 06:19:49

Team Anon Force

https://t.me/Professor6T9x