Allow coroutines in service methods

dbus_next/aio/message_bus.py

@@ -2,6 +2,7 @@
 from .._private.unmarshaller import Unmarshaller
 from ..message import Message
 from ..constants import BusType, NameFlag, RequestNameReply, ReleaseNameReply, MessageType, MessageFlag
+from ..service import ServiceInterface
 from .._private.auth import auth_external, auth_parse_line, auth_begin, AuthResponse
 from ..errors import AuthError, DBusError
 from .proxy_object import ProxyObject
@@ -33,7 +34,6 @@ class MessageBus(BaseMessageBus):
         be :class:`None` until the message bus connects.
     :vartype unique_name: str
     """
-
     def __init__(self, bus_address: str = None, bus_type: BusType = BusType.SESSION):
         super().__init__(bus_address, bus_type, ProxyObject)
         self._loop = asyncio.get_event_loop()
@@ -257,6 +257,23 @@ def send(self, msg: Message):
     def get_proxy_object(self, bus_name: str, path: str, introspection: intr.Node) -> ProxyObject:
         return super().get_proxy_object(bus_name, path, introspection)
 
+    @classmethod
+    def _make_method_handler(cls, interface, method):
+        if not asyncio.iscoroutinefunction(method.fn):
+            return super()._make_method_handler(interface, method)
+
+        def handler(msg, send_reply):
+            def done(fut):
+                with send_reply:
+                    result = fut.result()
+                    body = ServiceInterface._fn_result_to_body(result, method.out_signature_tree)
+                    send_reply(Message.new_method_return(msg, method.out_signature, body))
+
+            fut = asyncio.ensure_future(method.fn(interface, *msg.body))
+            fut.add_done_callback(done)
+
+        return handler
+
     def _message_reader(self):
         try:
             while True:

dbus_next/aio/proxy_object.py

@@ -65,7 +65,6 @@ class ProxyInterface(BaseProxyInterface):
     If the service returns an error for a DBus call, a :class:`DBusError
     <dbus_next.DBusError>` will be raised with information about the error.
     """
-
     def _add_method(self, intr_method):
         async def method_fn(*args):
             msg = await self.bus.call(
@@ -128,7 +127,6 @@ class ProxyObject(BaseProxyObject):
 
     For more information, see the :class:`BaseProxyObject <dbus_next.proxy_object.BaseProxyObject>`.
     """
-
     def __init__(self, bus_name: str, path: str, introspection: Union[intr.Node, str, ET.Element],
                  bus: BaseMessageBus):
         super().__init__(bus_name, path, introspection, bus, ProxyInterface)

dbus_next/glib/message_bus.py

@@ -140,7 +140,6 @@ class MessageBus(BaseMessageBus):
         be :class:`None` until the message bus connects.
     :vartype unique_name: str
     """
-
     def __init__(self, bus_address: str = None, bus_type: BusType = BusType.SESSION):
         if _import_error:
             raise _import_error

dbus_next/glib/proxy_object.py

@@ -102,7 +102,6 @@ def set_callback(error: Exception)
     :class:`DBusError <dbus_next.DBusError>` will be raised with information
     about the error.
     """
-
     def _add_method(self, intr_method):
         in_len = len(intr_method.in_args)
         out_len = len(intr_method.out_args)
@@ -270,7 +269,6 @@ class ProxyObject(BaseProxyObject):
 
     For more information, see the :class:`BaseProxyObject <dbus_next.proxy_object.BaseProxyObject>`.
     """
-
     def __init__(self, bus_name: str, path: str, introspection: Union[intr.Node, str, ET.Element],
                  bus: BaseMessageBus):
         super().__init__(bus_name, path, introspection, bus, ProxyInterface)

dbus_next/introspection.py

@@ -28,7 +28,6 @@ class Arg:
         - :class:`InvalidSignatureError <dbus_next.InvalidSignatureError>` - If the signature is not valid.
         - :class:`InvalidIntrospectionError <dbus_next.InvalidIntrospectionError>` - If the signature is not a single complete type.
     """
-
     def __init__(self,
                  signature: Union[SignatureType, str],
                  direction: List[ArgDirection] = None,
@@ -101,7 +100,6 @@ class Signal:
     :raises:
         - :class:`InvalidMemberNameError <dbus_next.InvalidMemberNameError>` - If the name of the signal is not a valid member name.
     """
-
     def __init__(self, name: str, args: List[Arg] = None):
         if name is not None:
             assert_member_name_valid(name)
@@ -165,7 +163,6 @@ class Method:
     :raises:
         - :class:`InvalidMemberNameError <dbus_next.InvalidMemberNameError>` - If the name of this method is not valid.
     """
-
     def __init__(self, name: str, in_args: List[Arg] = [], out_args: List[Arg] = []):
         assert_member_name_valid(name)
 
@@ -238,8 +235,9 @@ class Property:
         - :class `InvalidSignatureError <dbus_next.InvalidSignatureError>` - If the given signature is not valid.
         - :class: `InvalidMemberNameError <dbus_next.InvalidMemberNameError>` - If the member name is not valid.
     """
-
-    def __init__(self, name: str, signature: str,
+    def __init__(self,
+                 name: str,
+                 signature: str,
                  access: PropertyAccess = PropertyAccess.READWRITE):
         assert_member_name_valid(name)
 
@@ -303,7 +301,6 @@ class Interface:
     :raises:
         - :class:`InvalidInterfaceNameError <dbus_next.InvalidInterfaceNameError>` - If the name is not a valid interface name.
     """
-
     def __init__(self,
                  name: str,
                  methods: List[Method] = None,
@@ -387,7 +384,6 @@ class Node:
     :raises:
         - :class:`InvalidIntrospectionError <dbus_next.InvalidIntrospectionError>` - If the name is not a valid node name.
     """
-
     def __init__(self, name: str = None, interfaces: List[Interface] = None, is_root: bool = True):
         if not is_root and not name:
             raise InvalidIntrospectionError('child nodes must have a "name" attribute')

dbus_next/message.py

@@ -54,7 +54,6 @@ class Message:
         - :class:`InvalidMemberNameError` - If ``member`` is not a valid member name.
         - :class:`InvalidInterfaceNameError` - If ``error_name`` or ``interface`` is not a valid interface name.
     """
-
     def __init__(self,
                  destination: str = None,
                  path: str = None,

dbus_next/message_bus.py

@@ -43,7 +43,6 @@ class BaseMessageBus:
         be :class:`None` until the message bus connects.
     :vartype unique_name: str
     """
-
     def __init__(self,
                  bus_address: Optional[str] = None,
                  bus_type: BusType = BusType.SESSION,
@@ -184,8 +183,8 @@ def reply_notify(reply, err):
     def request_name(self,
                      name: str,
                      flags: NameFlag = NameFlag.NONE,
-                     callback: Optional[
-                         Callable[[Optional[RequestNameReply], Optional[Exception]], None]] = None):
+                     callback: Optional[Callable[[Optional[RequestNameReply], Optional[Exception]],
+                                                 None]] = None):
         """Request that this message bus owns the given name.
 
         :param name: The name to request.
@@ -227,8 +226,8 @@ def reply_notify(reply, err):
 
     def release_name(self,
                      name: str,
-                     callback: Optional[
-                         Callable[[Optional[ReleaseNameReply], Optional[Exception]], None]] = None):
+                     callback: Optional[Callable[[Optional[ReleaseNameReply], Optional[Exception]],
+                                                 None]] = None):
         """Request that this message bus release the given name.
 
         :param name: The name to release.
@@ -497,6 +496,32 @@ def _on_message(self, msg):
             logging.error(
                 f'got unexpected error processing a message: {e}.\n{traceback.format_exc()}')
 
+    def _send_reply(self, msg):
+        bus = self
+
+        class SendReply:
+            def __enter__(self):
+                return self
+
+            def __call__(self, reply):
+                if msg.flags & MessageFlag.NO_REPLY_EXPECTED:
+                    return
+
+                bus.send(reply)
+
+            def __exit__(self, type, value, tb):
+                if type == DBusError:
+                    self(value._as_message(msg))
+
+                if type == Exception:
+                    self(
+                        Message.new_error(
+                            msg, ErrorType.SERVICE_ERROR,
+                            f'The service interface raised an error: {value}.\n{traceback.format_tb(tb)}'
+                        ))
+
+        return SendReply()
+
     def _process_message(self, msg):
         handled = False
 
@@ -542,32 +567,18 @@ def _process_message(self, msg):
             if not handled:
                 handler = self._find_message_handler(msg)
 
-                send_reply = self.send
+                send_reply = self._send_reply(msg)
 
-                if msg.flags & MessageFlag.NO_REPLY_EXPECTED:
-                    send_reply = lambda msg: None
-
-                if handler:
-                    try:
-                        result = handler(msg)
-                        if type(result) is Message:
-                            send_reply(result)
-                    except DBusError as e:
-                        send_reply(e._as_message(msg))
-                    except Exception as e:
+                with send_reply:
+                    if handler:
+                        handler(msg, send_reply)
+                    else:
                         send_reply(
                             Message.new_error(
-                                msg, ErrorType.SERVICE_ERROR,
-                                f'The service interface raised an error: {e}.\n{traceback.format_exc()}'
+                                msg, ErrorType.UNKNOWN_METHOD,
+                                f'{msg.interface}.{msg.member} with signature "{msg.signature}" could not be found'
                             ))
 
-                else:
-                    send_reply(
-                        Message.new_error(
-                            msg, ErrorType.UNKNOWN_METHOD,
-                            f'{msg.interface}.{msg.member} with signature "{msg.signature}" could not be found'
-                        ))
-
         else:
             # An ERROR or a METHOD_RETURN
             if msg.reply_serial in self._method_return_handlers:
@@ -576,6 +587,15 @@ def _process_message(self, msg):
                     handler(msg, None)
                 del self._method_return_handlers[msg.reply_serial]
 
+    @classmethod
+    def _make_method_handler(cls, interface, method):
+        def handler(msg, send_reply):
+            result = method.fn(interface, *msg.body)
+            body = ServiceInterface._fn_result_to_body(result, method.out_signature_tree)
+            send_reply(Message.new_method_return(msg, method.out_signature, body))
+
+        return handler
+
     def _find_message_handler(self, msg):
         handler = None
 
@@ -594,7 +614,7 @@ def _find_message_handler(self, msg):
                 handler = self._default_get_machine_id_handler
         elif msg._matches(interface='org.freedesktop.DBus.ObjectManager',
                           member='GetManagedObjects'):
-            handler = self._default_object_manager
+            handler = self._default_get_managed_objects_handler
 
         else:
             for interface in self._path_exports.get(msg.path, []):
@@ -604,23 +624,23 @@ def _find_message_handler(self, msg):
                     if msg._matches(interface=interface.name,
                                     member=method.name,
                                     signature=method.in_signature):
-                        handler = ServiceInterface._make_method_handler(interface, method)
+                        handler = self._make_method_handler(interface, method)
                         break
                 if handler:
                     break
 
         return handler
 
-    def _default_introspect_handler(self, msg):
+    def _default_introspect_handler(self, msg, send_reply):
         introspection = self._introspect_export_path(msg.path).tostring()
-        return Message.new_method_return(msg, 's', [introspection])
+        send_reply(Message.new_method_return(msg, 's', [introspection]))
 
-    def _default_ping_handler(self, msg):
-        return Message.new_method_return(msg)
+    def _default_ping_handler(self, msg, send_reply):
+        send_reply(Message.new_method_return(msg))
 
-    def _default_get_machine_id_handler(self, msg):
+    def _default_get_machine_id_handler(self, msg, send_reply):
         if self._machine_id:
-            self.send(Message.new_method_return(msg, 's', self._machine_id))
+            send_reply(Message.new_method_return(msg, 's', self._machine_id))
             return
 
         def reply_handler(reply, err):
@@ -630,19 +650,19 @@ def reply_handler(reply, err):
 
             if reply.message_type == MessageType.METHOD_RETURN:
                 self._machine_id = reply.body[0]
-                self.send(Message.new_method_return(msg, 's', [self._machine_id]))
+                send_reply(Message.new_method_return(msg, 's', [self._machine_id]))
             elif reply.message_type == MessageType.ERROR:
-                self.send(Message.new_error(msg, reply.error_name, reply.body))
+                send_reply(Message.new_error(msg, reply.error_name, reply.body))
             else:
-                self.send(Message.new_error(msg, ErrorType.FAILED, 'could not get machine_id'))
+                send_reply(Message.new_error(msg, ErrorType.FAILED, 'could not get machine_id'))
 
         self._call(
             Message(destination='org.freedesktop.DBus',
                     path='/org/freedesktop/DBus',
                     interface='org.freedesktop.DBus.Peer',
                     member='GetMachineId'), reply_handler)
 
-    def _default_object_manager(self, msg):
+    def _default_get_managed_objects_handler(self, msg, send_reply):
         result = {}
 
         for node in self._path_exports:
@@ -653,9 +673,9 @@ def _default_object_manager(self, msg):
             for interface in self._path_exports[node]:
                 result[node][interface.name] = self._get_all_properties(interface)
 
-        return Message.new_method_return(msg, 'a{oa{sa{sv}}}', [result])
+        send_reply(Message.new_method_return(msg, 'a{oa{sa{sv}}}', [result]))
 
-    def _default_properties_handler(self, msg):
+    def _default_properties_handler(self, msg, send_reply):
         methods = {'Get': 'ss', 'Set': 'ssv', 'GetAll': 's'}
         if msg.member not in methods or methods[msg.member] != msg.signature:
             raise DBusError(
@@ -695,7 +715,8 @@ def _default_properties_handler(self, msg):
                     raise DBusError(ErrorType.UNKNOWN_PROPERTY,
                                     'the property does not have read access')
                 prop_value = getattr(interface, prop.prop_getter.__name__)
-                return Message.new_method_return(msg, 'v', [Variant(prop.signature, prop_value)])
+                send_reply(
+                    Message.new_method_return(msg, 'v', [Variant(prop.signature, prop_value)]))
             elif msg.member == 'Set':
                 if not prop.access.writable():
                     raise DBusError(ErrorType.PROPERTY_READ_ONLY, 'the property is readonly')
@@ -705,11 +726,11 @@ def _default_properties_handler(self, msg):
                                     f'wrong signature for property. expected "{prop.signature}"')
                 assert prop.prop_setter
                 setattr(interface, prop.prop_setter.__name__, value.value)
-                return Message.new_method_return(msg)
+                send_reply(Message.new_method_return(msg))
 
         elif msg.member == 'GetAll':
             result = self._get_all_properties(interface)
-            return Message.new_method_return(msg, 'a{sv}', [result])
+            send_reply(Message.new_method_return(msg, 'a{sv}', [result]))
         else:
             assert False
 

dbus_next/proxy_object.py

@@ -37,7 +37,6 @@ class BaseProxyInterface:
     :ivar bus: The message bus this proxy interface is connected to.
     :vartype bus: :class:`BaseMessageBus <dbus_next.message_bus.BaseMessageBus>`
     """
-
     def __init__(self, bus_name, path, introspection, bus):
 
         self.bus_name = bus_name
@@ -104,7 +103,6 @@ class BaseProxyObject:
         - :class:`InvalidObjectPathError <dbus_next.InvalidObjectPathError>` - If the given object path is not valid.
         - :class:`InvalidIntrospectionError <dbus_next.InvalidIntrospectionError>` - If the introspection data for the node is not valid.
     """
-
     def __init__(self, bus_name: str, path: str, introspection: Union[intr.Node, str, ET.Element],
                  bus: 'message_bus.BaseMessageBus', ProxyInterface: Type[BaseProxyInterface]):
         assert_object_path_valid(path)