dbus-python async_callbacks

dbus_async_callbacks.py

import dbus
import dbus.service
from dbus.mainloop.glib import DBusGMainLoop
import time
from gi.repository import GLib

DBusGMainLoop(set_as_default=True)


class MyTestClass(dbus.service.Object):
    def __init__(self, name, path):
        super(MyTestClass, self).__init__(
            bus_name=dbus.service.BusName(name),
            object_path=path)

    @dbus.service.method(dbus_interface='org.Example.Test', in_signature='i',
                         out_signature='')
    def long_running(self, seconds):
        time.sleep(seconds)

    @dbus.service.method(dbus_interface='org.Example.Test', in_signature='i',
                         out_signature='', async_callbacks=('ok', 'err'))
    def long_running_async_ok(self, seconds, ok, err):
        time.sleep(seconds)
        ok('Done waiting')

    @dbus.service.method(dbus_interface='org.Example.Test', in_signature='i',
                         out_signature='', async_callbacks=('ok', 'err'))
    def long_running_async_err(self, seconds, ok, err):
        time.sleep(seconds)
        err(dbus.exceptions.DBusException('Something went wrong'))

    @dbus.service.method(dbus_interface='org.Example.Test', in_signature='i',
                         out_signature='s', async_callbacks=('ok', 'err'))
    def long_running_async_return_ok(self, seconds, ok, err):
        time.sleep(seconds)
        ok('Slept for %s seconds' % seconds)

    @dbus.service.method(dbus_interface='org.Example.Test', in_signature='i',
                         out_signature='s', async_callbacks=('ok', 'err'))
    def long_running_async_return_err(self, seconds, ok, err):
        time.sleep(seconds)
        err(dbus.exceptions.DBusException('Something went wrong'))
        
MyTest = MyTestClass('org.example.Test', '/org/example/Test')

loop = GLib.MainLoop()
loop.run()