class FallbackObject(dbus.service.Object):
"""A base class for exporting your own Objects across the Bus.
Just inherit from Object and provide a list of methods to share
across the Bus
"""
def __init__(self, bus_name, fallback_object_path):
self._object_path = fallback_object_path
self._name = bus_name
self._bus = bus_name.get_bus()
self._connection = self._bus.get_connection()
self._connection._register_object_path(fallback_object_path, self._message_cb, self._unregister_cb, True)
def _message_cb(self, connection, message):
try:
# lookup candidate method and parent method
method_name = message.get_member()
interface_name = message.get_interface()
(candidate_method, parent_method) = dbus.service._method_lookup(self, method_name, interface_name)
# set up method call parameters
args = message.get_args_list()
keywords = {}
# iterate signature into list of complete types
if parent_method._dbus_out_signature:
signature = tuple(dbus.types.Signature(parent_method._dbus_out_signature))
else:
signature = None
# set up async callback functions
if parent_method._dbus_async_callbacks:
(return_callback, error_callback) = parent_method._dbus_async_callbacks
keywords[return_callback] = lambda *retval: dbus.service._method_reply_return(connection, message, method_name, signature, *retval)
keywords[error_callback] = lambda exception: dbus.service._method_reply_error(connection, message, exception)
# include the sender if desired
if parent_method._dbus_sender_keyword:
keywords[parent_method._dbus_sender_keyword] = message.get_sender()
if parent_method._dbus_message_keyword:
keywords[parent_method._dbus_message_keyword] = message
# call method
retval = candidate_method(self, *args, **keywords)
# we're done - the method has got callback functions to reply with
if parent_method._dbus_async_callbacks:
return
# otherwise we send the return values in a reply. if we have a
# signature, use it to turn the return value into a tuple as
# appropriate
if parent_method._dbus_out_signature:
# if we have zero or one return values we want make a tuple
# for the _method_reply_return function, otherwise we need
# to check we're passing it a sequence
if len(signature) == 0:
if retval == None:
retval = ()
else:
raise TypeError('%s has an empty output signature but did not return None' %
method_name)
elif len(signature) == 1:
retval = (retval,)
else:
if operator.isSequenceType(retval):
# multi-value signature, multi-value return... proceed unchanged
pass
else:
raise TypeError('%s has multiple output values in signature %s but did not return a sequence' %
(method_name, signature))
# no signature, so just turn the return into a tuple and send it as normal
else:
signature = None
if retval == None:
retval = ()
else:
retval = (retval,)
dbus.service._method_reply_return(connection, message, method_name, signature, *retval)
except Exception, exception:
# send error reply
dbus.service._method_reply_error(connection, message, exception)
@method('org.freedesktop.DBus.Introspectable', in_signature='', out_signature='s', dbus_message_keyword="dbus_message")
def Introspect(self, dbus_message=None):
reflection_data = '\n'
op = dbus_message.get_path()
reflection_data += '\n' % (op)
interfaces = self._dbus_class_table[self.__class__.__module__ + '.' + self.__class__.__name__]
for (name, funcs) in interfaces.iteritems():
reflection_data += ' \n' % (name)
for func in funcs.values():
if getattr(func, '_dbus_is_method', False):
reflection_data += self.__class__._reflect_on_method(func)
elif getattr(func, '_dbus_is_signal', False):
reflection_data += self.__class__._reflect_on_signal(func)
reflection_data += ' \n'
reflection_data += '\n'
return reflection_data
def __repr__(self):
return '' % (self._object_path, self._name, id(self))
__str__ = __repr__