direct_server_ravelled

#!/usr/bin/python3
#+
# DBussy example -- a server which accepts direct socket connections
# not going through the D-Bus daemon. Use the accompanying direct_client
# script to test requests to this server.
#
# Copyright 2021 by Lawrence D'Oliveiro <ldo@geek-central.gen.nz>. This
# script is licensed CC0
# <https://creativecommons.org/publicdomain/zero/1.0/>; do with it
# what you will.
#-

import sys
import signal
import time
import dbussy as dbus
from dbussy import \
    DBUS
import ravel

my_socket_name = "unix:path=/tmp/direct_server_demo"
my_interface_name = "com.example.direct_server"

idle_timeout = 15 # seconds, short value for testing

@ravel.interface(ravel.INTERFACE.SERVER, name = my_interface_name)
class DirectConnectServer :

    __slots__ = ("bus", "_last_request")

    def __init__(self, bus) :
        self.bus = bus
    #end __init__

    @ravel.method \
      (
        name = "add",
        in_signature = "ii",
        out_signature = "i",
        arg_keys = ["addend", "augend"],
        result_keys = ["sum"],
        result_keyword = "result",
      )
    def add(self, addend, augend, result) :
        result["sum"] = addend + augend
    #end add

#end DirectConnectServer

loop = dbus.get_event_loop()

signal.signal(signal.SIGINT, signal.SIG_DFL) # abort without raising KeyboardInterrupt on CTRL/C

async def mainline() :
    listen = ravel.Server(address = my_socket_name)
    clients = {}

    @ravel.signal \
      (
        in_signature = "",
        bus_keyword = "conn"
      )
    def connection_terminated(conn) :
        # handler for signal sent by libdbus when client disconnects.
        sys.stdout.write("connection from PID %s terminated\n" % conn.connection.unix_process_id)
        del clients[id(conn)]
    #end connection_terminated

#begin mainline
    last_conn = time.time()
    while True :
        if len(clients) == 0 and time.time() - last_conn > idle_timeout :
            sys.stdout.write("no clients in a while, exiting\n")
            listen.server.disconnect()
            break
        #end if
        conn = await listen.await_new_connection(timeout = idle_timeout / 3)
        if conn != None :
            sys.stdout.write("new connection from PID %s\n" % conn.connection.unix_process_id)
              # unix_process_id always seems to be None at this point, but does
              # get correct value at some point later.
            last_conn = time.time()
            conn.register \
              (
                path = "/",
                fallback = True,
                interface = DirectConnectServer(conn)
              )
            conn.listen_signal \
              (
                path = "/",
                fallback = True,
                interface = DBUS.INTERFACE_LOCAL,
                name = "Disconnected",
                func = connection_terminated
              )
            clients[id(conn)] = conn
        #end if
    #end while
#end mainline

loop.run_until_complete(mainline())