#
# Client support for DBus observers
#
# Copyright (C) 2019 Red Hat, Inc. All rights reserved.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301
# USA
#
import logging
from functools import partial
from dasbus.constants import DBUS_FLAG_NONE
from dasbus.signal import Signal
import gi
gi.require_version("Gio", "2.0")
from gi.repository import Gio
log = logging.getLogger(__name__)
__all__ = [
"DBusObserverError",
"DBusObserver",
"GLibMonitoring"
]
[docs]class DBusObserverError(Exception):
"""Exception class for the DBus observers."""
pass
[docs]class GLibMonitoring(object):
"""The low-level DBus monitoring library based on GLib."""
[docs] @classmethod
def watch_name(cls, connection, name, flags=DBUS_FLAG_NONE,
name_appeared=None, name_vanished=None):
"""Watch a service name on the DBus connection."""
name_appeared_closure = None
name_vanished_closure = None
if name_appeared:
name_appeared_closure = partial(
cls._name_appeared_callback,
user_data=(name_appeared, ())
)
if name_vanished:
name_vanished_closure = partial(
cls._name_vanished_callback,
user_data=(name_vanished, ())
)
registration_id = Gio.bus_watch_name_on_connection(
connection,
name,
flags,
name_appeared_closure,
name_vanished_closure
)
return partial(
cls._unwatch_name,
connection,
registration_id
)
@classmethod
def _name_appeared_callback(cls, connection, name, name_owner, user_data):
"""Callback for watch_name.."""
# Prepare the user's callback.
callback, callback_args = user_data
# Call user's callback.
callback(name_owner, *callback_args)
@classmethod
def _name_vanished_callback(cls, connection, name, user_data):
"""Callback for watch_name."""
# Prepare the user's callback.
callback, callback_args = user_data
# Call user's callback.
callback(*callback_args)
@classmethod
def _unwatch_name(cls, connection, registration_id):
"""Stops watching a service name on the DBus connection."""
Gio.bus_unwatch_name(registration_id)
[docs]class DBusObserver(object):
"""Base class for DBus observers.
This class is recommended to use only to watch the availability
of a service on DBus. It doesn't provide any support for accessing
objects provided by the service.
Usage:
.. code-block:: python
# Create the observer and connect to its signals.
observer = DBusObserver(SystemBus, "org.freedesktop.NetworkManager")
def callback1(observer):
print("Service is available!")
def callback2(observer):
print("Service is unavailable!")
observer.service_available.connect(callback1)
observer.service_unavailable.connect(callback2)
# Connect to the service once it is available.
observer.connect_once_available()
# Disconnect the observer.
observer.disconnect()
"""
def __init__(self, message_bus, service_name, monitoring=GLibMonitoring):
"""Creates a DBus service observer.
:param message_bus: a message bus
:param service_name: a DBus name of a service
"""
self._message_bus = message_bus
self._service_name = service_name
self._is_service_available = False
self._service_available = Signal()
self._service_unavailable = Signal()
self._monitoring = monitoring
self._subscriptions = []
@property
def service_name(self):
"""Returns a DBus name."""
return self._service_name
@property
def is_service_available(self):
"""The proxy can be accessed."""
return self._is_service_available
@property
def service_available(self):
"""Signal that emits when the service is available.
Signal emits this class as an argument. You have to
call the watch method to activate the signals.
"""
return self._service_available
@property
def service_unavailable(self):
"""Signal that emits when the service is unavailable.
Signal emits this class as an argument. You have to
call the watch method to activate the signals.
"""
return self._service_unavailable
[docs] def connect_once_available(self):
"""Connect to the service once it is available.
The observer is not connected to the service until it
emits the service_available signal.
"""
self._watch()
[docs] def disconnect(self):
"""Disconnect from the service.
Disconnect from the service if it is connected and stop
watching its availability.
"""
self._unwatch()
if self.is_service_available:
self._disable_service()
def _watch(self):
"""Watch the service name on DBus."""
subscription = self._monitoring.watch_name(
self._message_bus.connection,
self.service_name,
DBUS_FLAG_NONE,
self._service_name_appeared_callback,
self._service_name_vanished_callback
)
self._subscriptions.append(subscription)
def _unwatch(self):
"""Stop to watch the service name on DBus."""
while self._subscriptions:
callback = self._subscriptions.pop()
callback()
def _enable_service(self):
"""Enable the service."""
self._is_service_available = True
self._service_available.emit(self)
def _disable_service(self):
"""Disable the service."""
self._is_service_available = False
self._service_unavailable.emit(self)
def _service_name_appeared_callback(self, *args):
"""Callback for the watch method."""
if not self.is_service_available:
self._enable_service()
def _service_name_vanished_callback(self, *args):
"""Callback for the watch method."""
if self.is_service_available:
self._disable_service()
def __str__(self):
"""Returns a string version of this object."""
return self._service_name
def __repr__(self):
"""Returns a string representation."""
return "{}({})".format(
self.__class__.__name__,
self._service_name
)