/usr/lib/python3/dist-packages/checkbox_ng/commands/service.py is in python3-checkbox-ng 0.3.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | # This file is part of Checkbox.
#
# Copyright 2013 Canonical Ltd.
# Written by:
# Zygmunt Krynicki <zygmunt.krynicki@canonical.com>
#
# Checkbox is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3,
# as published by the Free Software Foundation.
#
# Checkbox is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.
"""
:mod:`checkbox_ng.commands.service` -- service sub-command
==========================================================
"""
import logging
import os
from dbus import StarterBus, SessionBus
from dbus.mainloop.glib import DBusGMainLoop, threads_init
from dbus.service import BusName
from gi.repository import GObject
from plainbox.impl.commands import PlainBoxCommand
from plainbox.impl.highlevel import Service
from checkbox_ng.service import ServiceWrapper
logger = logging.getLogger("checkbox.ng.commands.service")
def connect_to_session_bus():
"""
Connect to the session bus properly.
Returns a tuple (session_bus, loop) where loop is a GObject.MainLoop
instance. The loop is there so that you can listen to signals.
"""
# We'll need an event loop to observe signals. We will need the instance
# later below so let's keep it. Note that we're not passing it directly
# below as DBus needs specific API. The DBusGMainLoop class that we
# instantiate and pass is going to work with this instance transparently.
#
# NOTE: DBus tutorial suggests that we should create the loop _before_
# connecting to the bus.
logger.debug("Setting up glib-based event loop")
# Make sure gobject threads don't crash
GObject.threads_init()
threads_init()
loop = GObject.MainLoop()
# Let's get the system bus object.
logger.debug("Connecting to DBus session bus")
if os.getenv("DBUS_STARTER_ADDRESS"):
session_bus = StarterBus(mainloop=DBusGMainLoop())
else:
session_bus = SessionBus(mainloop=DBusGMainLoop())
return session_bus, loop
class ServiceInvocation:
def __init__(self, provider_list, config, ns):
self.provider_list = provider_list
self.config = config
self.ns = ns
def run(self):
bus, loop = connect_to_session_bus()
logger.info("Setting up DBus objects...")
session_list = [] # TODO: load sessions
logger.debug("Constructing Service object")
service_obj = Service(self.provider_list, session_list, self.config)
logger.debug("Constructing ServiceWrapper")
service_wrp = ServiceWrapper(service_obj, on_exit=lambda: loop.quit())
logger.info("Publishing all objects on DBus")
service_wrp.publish_related_objects(bus)
logger.info("Publishing all managed objects (events should fire there)")
service_wrp.publish_managed_objects()
logger.debug("Attempting to claim bus name: %s", self.ns.bus_name)
bus_name = BusName(self.ns.bus_name, bus)
logger.info(
"PlainBox DBus service ready, claimed name: %s",
bus_name.get_name())
try:
loop.run()
except KeyboardInterrupt:
logger.warning((
"Main loop interrupted!"
" It is recommended to call the Exit() method on the"
" exported service object instead"))
finally:
logger.debug("Releasing %s", bus_name)
# XXX: ugly but that's how one can reliably release a bus name
del bus_name
# Remove objects from the bus
service_wrp.remove_from_connection()
logger.debug("Closing %s", bus)
bus.close()
logger.debug("Main loop terminated, exiting...")
class ServiceCommand(PlainBoxCommand):
"""
DBus service for PlainBox
"""
# XXX: Maybe drop provider / config and handle them differently
def __init__(self, provider_list, config):
self.provider_list = provider_list
self.config = config
def invoked(self, ns):
return ServiceInvocation(self.provider_list, self.config, ns).run()
def register_parser(self, subparsers):
parser = subparsers.add_parser("service", help="spawn dbus service")
parser.add_argument(
'--bus-name', action="store",
default="com.canonical.certification.PlainBox1",
help="Use the specified DBus bus name")
parser.set_defaults(command=self)
|