connector: wanted(port) says whether anyone watches or reads a port

backend/fluksio/flow/connector.py

@@ -21,6 +21,8 @@ What a connector gets from the base class:
 * :meth:`ConnectorNode.save_artifact`, for a device whose readings are bytes —
   a camera frame, a recorded clip — which travel as a reference on a media
   port rather than as the message itself;
+* :meth:`ConnectorNode.wanted`, so a camera fetches frames only while someone
+  watches them or a running flow reads them;
 * :attr:`ConnectorNode.store`, for the few short strings that have to outlive
   a restart — a certificate pinned on first contact, a pairing token — shared
   by every node of the connector's package; :mod:`fluksio.flow.connector_tls`
@@ -40,6 +42,7 @@ from typing import TYPE_CHECKING, Any, ClassVar
 from pydantic import BaseModel, Field
 
 from fluksio.flow.connector_store import ConnectorStore, namespace_of
+from fluksio.flow.demand import demand
 from fluksio.flow.nodes import Node
 
 if TYPE_CHECKING:
@@ -173,6 +176,17 @@ class ConnectorNode(Node):
             [data], name=name, media_type=media_type, volatile=volatile
         )
 
+    def wanted(self, port: str) -> bool:
+        """Whether anyone wants what ``port`` publishes right now.
+
+        A dashboard tile on screen, a screen through the portal, or a node of a
+        running flow reading the port's message. A camera asks before each
+        frame, so it sends nothing while nobody watches. Always true on a
+        remote worker, which cannot see who watches.
+        """
+        names = {s.name for s in self.output_ports if s.port == port and s.name}
+        return demand.wants(names)
+
     @property
     def store(self) -> ConnectorStore:
         """Short strings this connector keeps across restarts.

backend/fluksio/flow/demand.py

@@ -58,6 +58,20 @@ class Demand:
         with self._lock:
             return set().union(*self._wanted.values())
 
+    def wants(self, names: set[str]) -> bool:
+        """Whether a socket views, or a running flow reads, any of ``names``.
+
+        What a connector's camera asks before it fetches a frame. True while
+        nothing is bound, which is a worker process: the sockets are in the
+        engine's, out of its sight.
+        """
+        controller = self._controller
+        if controller is None:
+            # ponytail: a remote worker always sends; forward demand to workers
+            # when a camera placed on one costs too much.
+            return True
+        return bool(names & (self.viewed() | controller.consumed_messages()))
+
     def ports(self, device_id: str) -> list[str]:
         """The device's ports that have demand, sorted."""
         return list(self._compute().get(device_id, ()))

backend/tests/flow/test_connector.py

@@ -306,3 +306,29 @@ def test_a_connector_publishes_bytes_as_a_media_reference(tmp_path):
     # point of a media dtype.
     MessageSpec(name="demo.frame", dtype=DType.IMAGE).check(reference)
     assert store.path(reference["digest"]).read_bytes() == b"\x89PNG..."
+
+
+def test_a_port_is_wanted_while_watched_or_read(monkeypatch):
+    """A camera fetches frames only while a screen shows them or a flow reads them."""
+    from fluksio.flow.demand import demand
+
+    node = a_sensor([])
+    # A worker process binds no controller and cannot see who watches.
+    assert node.wanted("reading")
+
+    reads: set[str] = set()
+    monkeypatch.setattr(
+        demand, "_controller", SimpleNamespace(consumed_messages=lambda: reads)
+    )
+    assert not node.wanted("reading")
+
+    reads.add("demo.reading")
+    assert node.wanted("reading")
+
+    reads.clear()
+    demand.want("a viewer", {"demo.reading"})
+    try:
+        assert node.wanted("reading")
+    finally:
+        demand.drop("a viewer")
+    assert not node.wanted("reading")

docs/reference/connector-contract.md

@@ -144,6 +144,20 @@ new frames arrive. Use durable storage for recordings.
 Local screens receive pushed live frames; portal screens fetch the bytes.
 Choose a modest poll rate for remote viewing.
 
+`self.wanted(port)` says whether anyone wants what a port publishes right
+now. That is a dashboard tile on screen, a portal screen, or a node of a
+running flow reading its message. Check it before fetching a frame so a camera
+sends nothing while nobody watches:
+
+```python
+async def poll(self):
+    if not self.wanted("frame"):
+        return None
+    ...
+```
+
+On a remote worker it is always true, since the worker cannot see who watches.
+
 ## Keeping state across restarts
 
 Keep connections and device handles in node attributes. Store short strings