简单（但具体）的侦听器和发送器 Python 3 DBus 示例

我想制作一个由两部分组成的程序。一个监听者（服务器，如果你愿意的话）和一个发送者（客户端）。我做了一些研究，了解到这是通过程序员调用 IPC（进程间通信）的方法来完成的；我相信您知道它的意思，我只是扩展了缩写词，以便您知道我不认为它意味着“互联网宠物食人者”（或其他一些不相关的令人不快的事情）。

我读到实现这一点的一个好方法是使用 dbus。所以我对dbus做了一些研究，现在我很困惑。显然，您可以使用 dbus 做很多事情，例如向 Gnome Shell 发送通知或与网络管理器交谈。我不想做那些事！我只想制作两个相互交谈的简单程序。除此之外，一些教程和文档显示了使用 python 2 的示例，一些使用 3，一些导入 dbus 和一些导入 Gio！我发现的很多信息超出了我的能力范围，这也损害了我的努力。

有人可以向我展示一个简单而优雅的示例，说明如何制作一个基本上可以实现此目的的程序吗：

$ ./server
Server is not running yet. Putting on listening ears.
$ ./client Hi
server: a client said "Hi"
$ ./server
Server is already running.
$ ./server stop
Server exiting...
$ ./client Do a barrel roll
client: No one can hear me!!

这就是一个简单会话的方式（当然使用 bash shell）。我想使用 Python 3 以及目前最合适的任何 dbus 绑定（我猜测是 gi.repository）。澄清一下，这适用于 Linux。

python3 中关于 dbus 的文档并不多，但我设法弄清楚了，所以我将在这里记录它：与所有 python2 示例的主要区别是替换import gobject with import gi.repository.GLib.

您可以在以下位置找到更多示例（其中使用的功能比我需要的更多）dbus-python 示例目录 http://cgit.freedesktop.org/dbus/dbus-python/tree/examples.

我没有在服务器中实现自我背景，因为这种风格的守护进程最近已经过时了。

常见.py：

# well-known name for our program
ECHO_BUS_NAME = 'com.stackoverflow.question_21793826.EchoService'

# interfaces implemented by some objects in our program
ECHO_INTERFACE = 'com.stackoverflow.question_21793826.EchoInterface'
QUIT_INTERFACE = 'com.stackoverflow.question_21793826.QuitInterface'

# paths to some objects in our program
ECHO_OBJECT_PATH = '/EchoServerObject'

服务器.py：

#!/usr/bin/env python3

# standard includes
import sys

# dbus includes
import gi.repository.GLib
import dbus
import dbus.service
import dbus.mainloop.glib

# project includes
import common


class EchoServerObject(dbus.service.Object):

    # TODO it would be nice to make a better decorator using annotations:
    #   def foo(self, a: 's', b: 's') -> '': pass
    # but the existing dbus decorator does its own reflection which
    # fails if there are any annotations (or keyword-only arguments)
    @dbus.service.method(common.ECHO_INTERFACE,
            in_signature='s', out_signature='')
    def echo(self, message):
        message = str(message) # get rid of subclass for repr
        print('server: a client said %r' % message)

    @dbus.service.method(common.QUIT_INTERFACE,
            in_signature='', out_signature='')
    def quit(self):
        # this should be a separate object, but I'm
        # showing how one object can have multiple interfaces
        self.mainloop.quit()

def stop():
    bus = dbus.SessionBus()

    proxy = bus.get_object(common.ECHO_BUS_NAME, common.ECHO_OBJECT_PATH)
    iface = dbus.Interface(proxy, common.QUIT_INTERFACE)

    iface.quit()

def server():
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

    bus = dbus.SessionBus()
    try:
        name = dbus.service.BusName(common.ECHO_BUS_NAME, bus, do_not_queue=True)
    except dbus.NameExistsException:
        sys.exit('Server is already running.')
    else:
        print('Server is not running yet. Putting on listening ears.')
    echo = EchoServerObject(bus, common.ECHO_OBJECT_PATH)

    mainloop = gi.repository.GLib.MainLoop()
    echo.mainloop = mainloop
    mainloop.run()

def main(exe, args):
    if args == ['stop']:
        stop()
    elif not args:
        server()
    else:
        sys.exit('Usage: %s [stop]' % exe)

if __name__ == '__main__':
    main(sys.argv[0], sys.argv[1:])

客户端.py:

#!/usr/bin/env python3

# standard includes
import sys

# dbus includes
import dbus

# project includes
import common


def client(mes):
    bus = dbus.SessionBus()

    try:
        proxy = bus.get_object(common.ECHO_BUS_NAME, common.ECHO_OBJECT_PATH)
    except dbus.DBusException as e:
        # There are actually two exceptions thrown:
        # 1: org.freedesktop.DBus.Error.NameHasNoOwner
        #   (when the name is not registered by any running process)
        # 2: org.freedesktop.DBus.Error.ServiceUnknown
        #   (during auto-activation since there is no .service file)
        # TODO figure out how to suppress the activation attempt
        # also, there *has* to be a better way of managing exceptions
        if e._dbus_error_name != 'org.freedesktop.DBus.Error.ServiceUnknown':
            raise
        if e.__context__._dbus_error_name != 'org.freedesktop.DBus.Error.NameHasNoOwner':
            raise
        print('client: No one can hear me!!')
    else:
        iface = dbus.Interface(proxy, common.ECHO_INTERFACE)
        iface.echo(mes)

def main(exe, args):
    if args:
        client(' '.join(args))
    else:
        sys.exit('Usage: %s message...' % exe)

if __name__ == '__main__':
    main(sys.argv[0], sys.argv[1:])