text: """Synchronous IO wrappers with thread safety
  """
  from concurrent.futures import Future
  from contextlib import contextmanager
  import functools
  import os
  from selectors import EVENT_READ
  import socket
  from queue import Queue, Full as QueueFull
  from threading import Lock, Thread
  from typing import Optional
  
  from jeepney import Message, MessageType
  from jeepney.bus import get_bus
  from jeepney.bus_messages import message_bus
  from jeepney.wrappers import ProxyBase, unwrap_msg
  from .blocking import (
      unwrap_read, prep_socket, DBusConnectionBase, timeout_to_deadline,
  )
  from .common import (
      MessageFilters, FilterHandle, ReplyMatcher, RouterClosed, check_replyable,
  )
  
  __all__ = [
      'open_dbus_connection',
      'open_dbus_router',
      'DBusConnection',
      'DBusRouter',
      'Proxy',
      'ReceiveStopped',
  ]
  
  
  class ReceiveStopped(Exception):
      pass
  
  
  class DBusConnection(DBusConnectionBase):
      def __init__(self, sock: socket.socket, enable_fds=False):
          super().__init__(sock, enable_fds=enable_fds)
          self._stop_r, self._stop_w = os.pipe()
          self.stop_key = self.selector.register(self._stop_r, EVENT_READ)
          self.send_lock = Lock()
          self.rcv_lock = Lock()
  
      def send(self, message: Message, serial=None):
          """Serialise and send a :class:`~.Message` object"""
          data, fds = self._serialise(message, serial)
          with self.send_lock:
              if fds:
                  self._send_with_fds(data, fds)
              else:
                  self.sock.sendall(data)
  
      def receive(self, *, timeout=None) -> Message:
          """Return the next available message from the connection
  
          If the data is ready, this will return immediately, even if timeout<=0.
          Otherwise, it will wait for up to timeout seconds, or indefinitely if
          timeout is None. If no message comes in time, it raises TimeoutError.
  
          If the connection is closed from another thread, this will raise
          ReceiveStopped.
          """
          deadline = timeout_to_deadline(timeout)
  
          if not self.rcv_lock.acquire(timeout=(timeout or -1)):
              raise TimeoutError(f"Did not get receive lock in {timeout} seconds")
          try:
              return self._receive(deadline)
          finally:
              self.rcv_lock.release()
  
      def _read_some_data(self, timeout=None):
          # Wait for data or a signal on the stop pipe
          for key, ev in self.selector.select(timeout):
              if key == self.select_key:
                  if self.enable_fds:
                      return self._read_with_fds()
                  else:
                      return unwrap_read(self.sock.recv(4096)), []
              elif key == self.stop_key:
                  raise ReceiveStopped("DBus receive stopped from another thread")
  
          raise TimeoutError
  
      def interrupt(self):
          """Make any threads waiting for a message raise ReceiveStopped"""
          os.write(self._stop_w, b'a')
  
      def reset_interrupt(self):
          """Allow calls to .receive() again after .interrupt()
  
          To avoid race conditions, you should typically wait for threads to
          respond (e.g. by joining them) between interrupting and resetting.
          """
          # Clear any data on the stop pipe
          while (self.stop_key, EVENT_READ) in self.selector.select(timeout=0):
              os.read(self._stop_r, 1024)
  
      def close(self):
          """Close the connection"""
          self.interrupt()
          super().close()
  
  
  def open_dbus_connection(bus='SESSION', enable_fds=False, auth_timeout=1.):
      """Open a plain D-Bus connection
  
      D-Bus has an authentication step before sending or receiving messages.
      This takes < 1 ms in normal operation, but there is a timeout so that client
      code won't get stuck if the server doesn't reply. *auth_timeout* configures
      this timeout in seconds.
  
      :return: :class:`DBusConnection`
      """
      bus_addr = get_bus(bus)
      sock = prep_socket(bus_addr, enable_fds, timeout=auth_timeout)
  
      conn = DBusConnection(sock, enable_fds)
  
      with DBusRouter(conn) as router:
          reply_body = Proxy(message_bus, router, timeout=10).Hello()
          conn.unique_name = reply_body[0]
  
      return conn
  
  
  class DBusRouter:
      """A client D-Bus connection which can wait for replies.
  
      This runs a separate receiver thread and dispatches received messages.
  
      It's possible to wrap a :class:`DBusConnection` in a router temporarily.
      Using the connection directly while it is wrapped is not supported,
      but you can use it again after the router is closed.
      """
      def __init__(self, conn: DBusConnection):
          self.conn = conn
          self._replies = ReplyMatcher()
          self._filters = MessageFilters()
          self._rcv_thread = Thread(target=self._receiver, daemon=True)
          self._rcv_thread.start()
  
      @property
      def unique_name(self):
          return self.conn.unique_name
  
      def send(self, message, *, serial=None):
          """Serialise and send a :class:`~.Message` object"""
          self.conn.send(message, serial=serial)
  
      def send_and_get_reply(self, msg: Message, *, timeout=None) -> Message:
          """Send a method call message, wait for and return a reply"""
          check_replyable(msg)
          if not self._rcv_thread.is_alive():
              raise RouterClosed("This D-Bus router has stopped")
  
          serial = next(self.conn.outgoing_serial)
  
          with self._replies.catch(serial, Future()) as reply_fut:
              self.conn.send(msg, serial=serial)
              return reply_fut.result(timeout=timeout)
  
      def close(self):
          """Close this router
  
          This does not close the underlying connection.
          """
          self.conn.interrupt()
          self._rcv_thread.join(timeout=10)
          self.conn.reset_interrupt()
  
      def __enter__(self):
          return self
  
      def __exit__(self, exc_type, exc_val, exc_tb):
          self.close()
          return False
  
      def filter(self, rule, *, queue: Optional[Queue] =None, bufsize=1):
          """Create a filter for incoming messages
  
          Usage::
  
              with router.filter(rule) as queue:
                  matching_msg = queue.get()
  
          :param jeepney.MatchRule rule: Catch messages matching this rule
          :param queue.Queue queue: Matched messages will be added to this
          :param int bufsize: If no queue is passed in, create one with this size
          """
          return FilterHandle(self._filters, rule, queue or Queue(maxsize=bufsize))
  
      # Code to run in receiver thread ------------------------------------
  
      def _dispatch(self, msg: Message):
          if self._replies.dispatch(msg):
              return
  
          for filter in self._filters.matches(msg):
              try:
                  filter.queue.put_nowait(msg)
              except QueueFull:
                  pass
  
      def _receiver(self):
          try:
              while True:
                  msg = self.conn.receive()
                  self._dispatch(msg)
          except ReceiveStopped:
              pass
          finally:
              # Send errors to any tasks still waiting for a message.
              self._replies.drop_all()
  
  class Proxy(ProxyBase):
      """A blocking proxy for calling D-Bus methods via a :class:`DBusRouter`.
  
      You can call methods on the proxy object, such as ``bus_proxy.Hello()``
      to make a method call over D-Bus and wait for a reply. It will either
      return a tuple of returned data, or raise :exc:`.DBusErrorResponse`.
      The methods available are defined by the message generator you wrap.
  
      You can set a time limit on a call by passing ``_timeout=`` in the method
      call, or set a default when creating the proxy. The ``_timeout`` argument
      is not passed to the message generator.
      All timeouts are in seconds, and :exc:`TimeoutErrror` is raised if it
      expires before a reply arrives.
  
      :param msggen: A message generator object
      :param ~threading.DBusRouter router: Router to send and receive messages
      :param float timeout: Default seconds to wait for a reply, or None for no limit
      """
      def __init__(self, msggen, router, *, timeout=None):
          super().__init__(msggen)
          self._router = router
          self._timeout = timeout
  
      def __repr__(self):
          extra = '' if (self._timeout is None) else f', timeout={self._timeout}'
          return f"Proxy({self._msggen}, {self._router}{extra})"
  
      def _method_call(self, make_msg):
          @functools.wraps(make_msg)
          def inner(*args, **kwargs):
              timeout = kwargs.pop('_timeout', self._timeout)
              msg = make_msg(*args, **kwargs)
              assert msg.header.message_type is MessageType.method_call
              reply = self._router.send_and_get_reply(msg, timeout=timeout)
              return unwrap_msg(reply)
  
          return inner
  
  @contextmanager
  def open_dbus_router(bus='SESSION', enable_fds=False):
      """Open a D-Bus 'router' to send and receive messages.
  
      Use as a context manager::
  
          with open_dbus_router() as router:
              ...
  
      On leaving the ``with`` block, the connection will be closed.
  
      :param str bus: 'SESSION' or 'SYSTEM' or a supported address.
      :param bool enable_fds: Whether to enable passing file descriptors.
      :return: :class:`DBusRouter`
      """
      with open_dbus_connection(bus=bus, enable_fds=enable_fds) as conn:
          with DBusRouter(conn) as router:
              yield router
  