+2005-05-24 John (J5) Palmieri <johnp@redhat.com>
+
+ * python/decorators.py: add explicitly_pass_message decorator
+ for passing in the dbus message as keyword for edge case signal
+ handling
+
+ * python/matchrules.py (SignalMatchRule.__repr__): fix output
+ to conform with what dbus expects for match rules
+ (SignalMatchRule.execute): add the dbus message as a keyword
+ if the signal handler has requested it
+
+ * python/examples/example/signal-recipient.py: added some more
+ examples on how to hook up to signals
+
2005-05-23 John (J5) Palmieri <johnp@redhat.com>
* python/decorators.py: import dbus_bindings
match_rule = SignalMatchRule(signal_name, dbus_interface, named_service, path)
- args = message.get_args_list()
-
- self._match_rule_tree.exec_matches(match_rule, *args)
+ self._match_rule_tree.exec_matches(match_rule, message)
def start_service_by_name(self, named_service):
return dbus_bindings.bus_start_service_by_name(self._connection, named_service)
import _util
import inspect
import dbus_bindings
-
+import new
def method(dbus_interface):
_util._validate_interface_or_name(dbus_interface)
return decorator
+def explicitly_pass_message(func):
+ func._dbus_pass_message = True
+ return func
object = bus.get_object("org.designfu.TestService","/org/designfu/TestService/object")
def hello_signal_handler(hello_string):
- print ("Received signal and it says: " + hello_string)
+ print ("Received signal and it says: " + hello_string)
+
+@dbus.explicitly_pass_message
+def catchall_signal_handler(*args, **keywords):
+ #The dbus.handler directive passes in the special __dbus_message__ variable
+ dbus_message = keywords["dbus_message"]
+ print "Caught signal " + dbus_message.get_member()
+ for arg in args:
+ print " " + str(arg)
+
+def catchall_hello_signals_handler(hello_string):
+ print ("Received a hello signal and it says ") + hello_string
+
+@dbus.explicitly_pass_message
+def catchall_testservice_interface_handler(hello_string, dbus_message):
+ print "org.designfu.TestService interface says " + hello_string + " when it sent signal " + dbus_message.get_member()
object.connect_to_signal("HelloSignal", hello_signal_handler, dbus_interface="org.designfu.TestService")
+#lets make a catchall
+bus.add_signal_receiver(catchall_signal_handler)
+bus.add_signal_receiver(catchall_hello_signals_handler, dbus_interface = "org.designfu.TestService", signal_name = "HelloSignal")
+bus.add_signal_receiver(catchall_testservice_interface_handler, dbus_interface = "org.designfu.TestService")
+
gobject.timeout_add(2000, emit_signal)
# Tell the remote object to emit the signal
path = signal.add(rule.signal_name)
path.add(rule.path, leaf=rule)
- def exec_matches(self, match_rule, *args):
+ def exec_matches(self, match_rule, message):
sender_matches = self._tree.get_matches(match_rule.sender)
for sender_node in sender_matches:
interface_matches = sender_node.get_matches(match_rule.dbus_interface)
for path_node in path_matches:
if(path_node.rules):
for rule in path_node.rules:
- rule.execute(*args)
+ rule.execute(message)
def remove(self, rule):
try:
self.sender = sender
self.path = path
- def execute(self, *args):
+ def execute(self, message):
+ args = message.get_args_list()
for handler in self.handler_functions:
- handler(*args)
+ if getattr(handler, "_dbus_pass_message", False):
+ keywords = {"dbus_message": message}
+ handler(*args, **keywords)
+ else:
+ handler(*args)
def add_handler(self, handler):
self.handler_functions.append(handler)
repr = "type='signal'"
if (self.dbus_interface):
repr = repr + ",interface='%s'" % (self.dbus_interface)
- else:
- repr = repr + ",interface='*'"
if (self.sender):
repr = repr + ",sender='%s'" % (self.sender)
- else:
- repr = repr + ",sender='*'"
if (self.path):
repr = repr + ",path='%s'" % (self.path)
- else:
- repr = repr + ",path='*'"
if (self.signal_name):
repr = repr + ",member='%s'" % (self.signal_name)
- else:
- repr = repr + ",member='*'"
return repr