Signals

typed events, Connection handles, once-listeners and weak cleanup

▶ Run in browser

Tags: basics signals events

An Emitter node declares a typed Signal(int) class attribute. Each SPACE press emits a tick to its listeners and every delivery is appended to an on-screen log built from Text2D lines. One listener is connected with once=True and spends itself on the first emit; the root listener can be detached and re-attached live via its Connection handle; and an Echo node holds a bound-method connection that cleans itself up when the node is freed.

What it demonstrates

  • Signal(int) as a class attribute: each instance lazily gets its own signal, so listeners on one emitter never hear another.

  • connect(fn) returns a Connection; conn.disconnect() detaches it and conn.connected reports the state.

  • connect(fn, once=True): the listener auto-disconnects after one emit.

  • Bound methods on Nodes are held weakly: freeing the node cleans the connection up and a later emit simply skips it, with no crash.

Controls: SPACE - emit ticked(n) D - disconnect / reconnect the root listener via its Connection F - free the Echo node (its bound-method connection auto-cleans) ESC - quit

Run: uv run python examples/features/basics/signals.py Headless self-check: uv run python examples/features/basics/signals.py –test

Source

  1"""Signals: typed events, Connection handles, once-listeners and weak cleanup
  2
  3An Emitter node declares a typed `Signal(int)` class attribute. Each SPACE
  4press emits a tick to its listeners and every delivery is appended to an
  5on-screen log built from Text2D lines. One listener is connected with
  6`once=True` and spends itself on the first emit; the root listener can be
  7detached and re-attached live via its `Connection` handle; and an Echo node
  8holds a bound-method connection that cleans itself up when the node is freed.
  9
 10# /// simvx
 11# tags = ["basics", "signals", "events"]
 12# web = { root = "SignalsDemo", width = 960, height = 540, responsive = true }
 13# ///
 14
 15## What it demonstrates
 16
 17- `Signal(int)` as a class attribute: each instance lazily gets its own
 18  signal, so listeners on one emitter never hear another.
 19- `connect(fn)` returns a `Connection`; `conn.disconnect()` detaches it and
 20  `conn.connected` reports the state.
 21- `connect(fn, once=True)`: the listener auto-disconnects after one emit.
 22- Bound methods on Nodes are held weakly: freeing the node cleans the
 23  connection up and a later emit simply skips it, with no crash.
 24
 25Controls:
 26  SPACE - emit ticked(n)
 27  D     - disconnect / reconnect the root listener via its Connection
 28  F     - free the Echo node (its bound-method connection auto-cleans)
 29  ESC   - quit
 30
 31Run: uv run python examples/features/basics/signals.py
 32Headless self-check: uv run python examples/features/basics/signals.py --test
 33"""
 34
 35from simvx.core import Input, Key, Node2D, Signal, Text2D, Vec2
 36from simvx.graphics import App
 37
 38WIDTH, HEIGHT = 960, 540
 39LOG_LINES = 12
 40
 41
 42class Emitter(Node2D):
 43    """Owns the signal. It knows nothing about who listens."""
 44
 45    ticked = Signal(int)  # typed: every emit carries the tick number
 46
 47    def on_ready(self):
 48        self._tick = 0
 49
 50    def tick(self):
 51        self._tick += 1
 52        self.ticked.emit(self._tick)  # `self.ticked(self._tick)` is the same call
 53
 54
 55class Echo(Node2D):
 56    """A listener node. Its bound method is held weakly by the signal."""
 57
 58    def __init__(self, log, **kwargs):
 59        super().__init__(**kwargs)
 60        self._log = log
 61
 62    def on_tick(self, n: int):
 63        self._log(f"Echo.on_tick heard {n}")
 64
 65
 66class SignalsDemo(Node2D):
 67    input_actions = {
 68        "emit": [Key.SPACE],
 69        "toggle": [Key.D],
 70        "free": [Key.F],
 71        "quit": [Key.ESCAPE],
 72    }
 73
 74    def on_ready(self):
 75        self._emitter = self.add_child(Emitter(name="Emitter"))
 76        self._lines: list[str] = []
 77
 78        # HUD: a title, a live status line, the log, and the controls.
 79        self.add_child(Text2D(text="Signals", position=Vec2(20, 16), font_scale=1.6))
 80        self._status = self.add_child(Text2D(position=Vec2(20, 52), colour=(0.55, 0.85, 1.0, 1.0)))
 81        self._log_nodes = [
 82            self.add_child(Text2D(position=Vec2(40, 96 + i * 24), colour=(0.8, 0.8, 0.8, 1.0)))
 83            for i in range(LOG_LINES)
 84        ]
 85        self.add_child(
 86            Text2D(
 87                text="SPACE emit   D disconnect/reconnect root listener   F free Echo   ESC quit",
 88                position=Vec2(20, HEIGHT - 32),
 89                colour=(0.55, 0.55, 0.55, 1.0),
 90            )
 91        )
 92
 93        # Listener 1: a bound method on this node. The Connection handle is the
 94        # way back out: keep it if you ever want to disconnect.
 95        self._root_conn = self._emitter.ticked.connect(self._on_tick)
 96
 97        # Listener 2: once=True spends itself on the first emit.
 98        self._once_conn = self._emitter.ticked.connect(self._once_heard, once=True)
 99
100        # Listener 3: a separate node. Freeing it (F) shows that a bound-method
101        # connection dies with its node instead of dangling.
102        self._echo = self.add_child(Echo(self._log, name="Echo"))
103        self._emitter.ticked.connect(self._echo.on_tick)
104
105        self._log("connected: root listener, once listener, Echo.on_tick")
106        self._refresh()
107
108    # -- listeners ----------------------------------------------------------
109
110    def _on_tick(self, n: int):
111        self._log(f"root._on_tick heard {n}")
112
113    def _once_heard(self, n: int):
114        self._log(f"once listener heard {n} and auto-disconnected")
115
116    # -- log + status -------------------------------------------------------
117
118    def _log(self, line: str):
119        self._lines = (self._lines + [line])[-LOG_LINES:]
120        for node, text in zip(self._log_nodes, self._lines, strict=False):
121            node.text = text
122
123    def _refresh(self):
124        root = "connected" if self._root_conn.connected else "disconnected"
125        once = "armed" if self._once_conn.connected else "spent"
126        echo = "alive" if self._echo is not None else "freed"
127        self._status.text = f"root listener: {root}   |   once listener: {once}   |   Echo: {echo}"
128
129    # -- input --------------------------------------------------------------
130
131    def on_update(self, dt: float):
132        if Input.is_action_just_pressed("emit"):
133            self._log(f"emit ticked({self._emitter._tick + 1})")
134            self._emitter.tick()
135        if Input.is_action_just_pressed("toggle"):
136            if self._root_conn.connected:
137                self._root_conn.disconnect()
138                self._log("root listener disconnected via Connection.disconnect()")
139            else:
140                self._root_conn = self._emitter.ticked.connect(self._on_tick)
141                self._log("root listener reconnected")
142        if Input.is_action_just_pressed("free") and self._echo is not None:
143            self._echo.destroy()
144            self._echo = None
145            self._log("Echo freed: its connection cleans up, later emits are safe")
146        if Input.is_action_just_pressed("quit"):
147            self.app.quit()
148        self._refresh()
149
150
151def _selftest() -> bool:
152    """Logic-level checks of the exact behaviours the demo shows on screen."""
153    import gc
154
155    ok = True
156
157    def check(label: str, passed: bool, detail: str):
158        nonlocal ok
159        ok = ok and passed
160        print(f"{'ok  ' if passed else 'FAIL'} {label}: {detail}")
161
162    # Instance-scoped signals: listeners on one emitter never hear another.
163    e1, e2 = Emitter(name="e1"), Emitter(name="e2")
164    got: list[int] = []
165    conn = e1.ticked.connect(got.append)
166    e1.ticked.emit(1)
167    e2.ticked.emit(99)
168    check("signals are instance-scoped", got == [1], f"e1 listener saw {got}")
169
170    # once=True fires exactly once, then reports disconnected.
171    once_got: list[int] = []
172    once_conn = e1.ticked.connect(once_got.append, once=True)
173    e1.ticked(2)
174    e1.ticked(3)
175    check("once listener fired exactly once", once_got == [2], f"saw {once_got}")
176    check("once connection reports spent", not once_conn.connected, f"connected={once_conn.connected}")
177    check("the ordinary listener kept hearing", got == [1, 2, 3], f"saw {got}")
178
179    # Connection.disconnect() detaches; emits after it deliver nothing.
180    conn.disconnect()
181    e1.ticked(4)
182    check("disconnect stops delivery", got == [1, 2, 3], f"saw {got}")
183    check("connection reports disconnected", not conn.connected, f"connected={conn.connected}")
184
185    # A freed node's bound-method connection auto-cleans (weak reference):
186    # the node is garbage-collected without destroy() and the next emit skips
187    # and prunes the dead connection instead of crashing.
188    echoed: list[str] = []
189    echo = Echo(echoed.append, name="echo")
190    e1.ticked.connect(echo.on_tick)
191    e1.ticked(5)
192    check("Echo heard while alive", len(echoed) == 1, f"{len(echoed)} deliveries")
193    del echo
194    gc.collect()
195    e1.ticked(6)  # must not raise
196    check("freed Echo hears nothing, no crash", len(echoed) == 1, f"{len(echoed)} deliveries")
197    check("dead connections are pruned", "connections=0" in repr(e1.ticked), repr(e1.ticked))
198
199    # destroy() proactively disconnects a node's outgoing bound-method
200    # connections, even while other references keep the node object alive.
201    echo2 = Echo(echoed.append, name="echo2")
202    e1.ticked.connect(echo2.on_tick)
203    echo2.destroy()
204    e1.ticked(7)
205    check("destroy() disconnects bound methods", len(echoed) == 1, f"{len(echoed)} deliveries")
206
207    print("SELFTEST:", "PASS" if ok else "FAIL")
208    return ok
209
210
211if __name__ == "__main__":
212    import sys
213
214    if "--test" in sys.argv:
215        sys.exit(0 if _selftest() else 1)
216    App(title="Signals", width=WIDTH, height=HEIGHT).run(SignalsDemo())