diff --git a/src/radical/orbit/gateway.py b/src/radical/orbit/gateway.py index 2feb42e..1b9ede6 100644 --- a/src/radical/orbit/gateway.py +++ b/src/radical/orbit/gateway.py @@ -255,8 +255,17 @@ def detach(self) -> None: def _sse_frame(topic: str, data: Any) -> str: """Format one SSE frame as HTTP clients expect it: ``data: {"topic": , "data": }\\n\\n`` (no keepalive - comments).""" - return "data: %s\n\n" % json.dumps({"topic": topic, "data": data}) + comments). + + The tap forwards whatever a plugin published, and plugin payloads + are not required to be JSON -- a ``bytes`` payload (e.g. a raw + stream event) used to kill the frame with a ``TypeError``, one + dropped notification per event. The SSE feed is a monitoring + tier: it must render every event, so anything JSON cannot encode + falls back to ``str()`` rather than raising. + """ + return "data: %s\n\n" % json.dumps({"topic": topic, "data": data}, + default=str) def _on_event(self, event: Dict[str, Any]) -> None: """Raw event tap callback — runs on the **plugin-host loop**. diff --git a/tests/unittests/test_gateway.py b/tests/unittests/test_gateway.py index 543cd23..c229850 100644 --- a/tests/unittests/test_gateway.py +++ b/tests/unittests/test_gateway.py @@ -455,6 +455,25 @@ def test_sse_topology_on_connect_and_disconnect(harness): sse.close() +def test_sse_frame_survives_non_json_payloads(): + # The tap forwards plugin payloads verbatim, and those are not + # required to be JSON: a bytes payload (a raw stream event, say) used + # to raise TypeError inside json.dumps and silently drop the frame. + # The SSE feed is monitoring -- it must render every event. + import json as _json + from radical.orbit.gateway import Gateway + + frame = Gateway._sse_frame('notification', {'data': b'\x80\x05binary'}) + assert frame.startswith('data: ') and frame.endswith('\n\n') + parsed = _json.loads(frame[len('data: '):]) + assert parsed['topic'] == 'notification' + assert 'binary' in parsed['data']['data'] # str() fallback + + # JSON-clean payloads are untouched by the fallback + frame = Gateway._sse_frame('notification', {'data': {'hello': 'world'}}) + assert _json.loads(frame[len('data: '):])['data'] == {'data': {'hello': 'world'}} + + def test_sse_queue_drop_oldest_counter(): # The gateway now uses the shared bounded, drop-oldest queue. from radical.orbit.queues import BoundedDropOldestQueue