Commits

jason kirtland  committed 21438d4

Added Signal.connect_via() decorator.

  • Participants
  • Parent commits 230b3b3

Comments (0)

Files changed (3)

File blinker/base.py

                 raise
         return receiver
 
+    def connect_via(self, sender, weak=False):
+        """Connect the decorated function as a receiver for *sender*.
+
+        :param sender: Any object or :obj:`ANY`.  The decorated function
+          will only receive :meth:`send` emissions sent by *sender*.  If
+          ``ANY``, the receiver will always be notified.  A function may be
+          decorated multiple times with differing *sender* values.
+
+        :param weak: If true, the Signal will hold a weakref to the
+          decorated function and automatically disconnect when *receiver*
+          goes out of scope or is garbage collected.  Unlike
+          :meth:`connect`, this defaults to False.
+
+        The decorated function will be invoked by :meth:`send` with
+          `sender=` as a single positional argument and any \*\*kwargs that
+          were provided to the call to :meth:`send`.
+
+
+        .. versionadded:: 1.1
+
+        """
+        def decorator(fn):
+            self.connect(fn, sender, weak)
+            return fn
+        return decorator
+
     @contextmanager
     def connected_to(self, receiver, sender=ANY):
         """Execute a block with the signal temporarily connected to *receiver*.

File docs/source/signals.rst

 
 While convenient, this form unfortunately does not allow the
 ``sender`` or ``weak`` arguments to be customized for the connected
-function.
+function.  For this, :meth:`~Signal.connect_via` can be used:
+
+.. doctest::
+
+  >>> dice_roll = signal('dice_roll')
+  >>> @dice_roll.connect_via(1)
+  ... @dice_roll.connect_via(3)
+  ... @dice_roll.connect_via(5)
+  ... def odd_subscriber(sender):
+  ...     print("Observed dice roll %r." % sender)
+  ...
+  >>> result = dice_roll.send(3)
+  Observed dice roll 3.
 
 
 Optimizing Signal Sending

File tests/test_signals.py

     values_are_empty_sets_(sig._by_sender)
 
 
+def test_decorated_receiver():
+    sentinel = []
+
+    class Object(object):
+        pass
+    obj = Object()
+
+    sig = blinker.Signal()
+
+    @sig.connect_via(obj)
+    def receiver(sender, **kw):
+        sentinel.append(kw)
+
+    assert not sentinel
+    sig.send()
+    assert not sentinel
+    sig.send(1)
+    assert not sentinel
+    sig.send(obj)
+    assert sig.receivers
+
+    del receiver
+    collect()
+    assert sig.receivers
+
+
 def test_no_double_send():
     sentinel = []
     def received(sender):