SImple (but specific) Python 3 listener and sender DBus example

I want to create a two-part program. Listener (server, if you want) and sender (client). I did some research and found out that this is done using a method that programmers call IPC (interprocess communication); I’m sure that you know what this means, I’m just expanding the abbreviation so that you know that I don’t think it means that these are online cannibal pets (or some other unpleasant things related to this).

I read that a good way to achieve this is to use dbus. So I did some research on dbus, and now I'm just confused. There are apparently a lot of things you can do with dbus, like sending notifications to Gnome Shell or talking to Network Manager. I do not want to do this! I just want to make two simple programs that talk to each other. To add to this, some tutorials and documentation show examples with python 2, some use 3, some import dbus and import Gio! Most of the information I found is above my head, which also worsens my efforts.

Would anyone be so kind as to show me a simple, elegant example of how to get a program that essentially does this:

$ ./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!! 

Here's how a simple session goes (using the bash shell, of course). I would like to use Python 3 and any dbus bindings are most suitable at the moment (I assume it will be gi.repository). To clarify, this will be for Linux.

+2
source share
1 answer

There is not much documentation for dbus in python3, but I managed to figure it out, so I will write it here: The main difference from all python2 examples is replacing import gobject with import gi.repository.GLib .

You can find more examples (which use more features than I needed) in the dbus-python example directory .

I did not implement self-phoning on the server, because this daemon style has recently gone out of fashion.

common.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' 

server.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:]) 

client.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:]) 
+5
source

Source: https://habr.com/ru/post/1489058/


All Articles