refactor: extract _collect_results and unify internal helpers
1. Extract _collect_results method — Deduplicated ~30 lines of identical result processing from route_event and process_client_event (Exception→error / WsResult→as_result / dict→wrap / None→strategy branch) into a private method with a skip_none parameter. * route_event calls _collect_results(skip_none=False) — None becomes ok=True (server-initiated, callers expect a result for every handler) * process_client_event calls _collect_results(skip_none=True) — None is skipped (client-initiated, matching legacy _dispatch fire-and-forget semantics) 2. Document None semantics difference — Added # NOTE: comment at the route_event call site explaining why skip_none=False differs from process_client_event. 3. Unify _timestamp() usage — Replaced inline timestamp formatting in _wrap_envelope and handle_connect with self._timestamp() method reuse.#
keyboardstaff committed
Mar 27, 2026 at 23:00 UTC
6e8c9d8224ff7a03c90e5680efe246ab995c38fe
1 file changed
+59
-80
helpers/ws_manager.py
+59
-80
@@ -582,6 +582,51 @@ class WsManager:
582
]
583
)
584
585
+ results = self._collect_results(
586
+ executions, event_type, correlation_id, skip_none=True,
587
+ )
588
+
589
+ await self._publish_diagnostic_event(
590
+ lambda: {
591
+ "kind": "inbound",
592
+ "sourceNamespace": namespace,
593
+ "namespace": namespace,
594
+ "eventType": event_type,
595
+ "sid": sid,
596
+ "correlationId": correlation_id,
597
+ "timestamp": self._timestamp(),
598
+ "handlerCount": len(handlers),
599
+ "durationMs": sum(
600
+ (exec.duration_ms or 0.0) for exec in executions
601
+ ),
602
+ "resultSummary": self._summarize_results(results),
603
+ "payloadSummary": self._summarize_payload(handler_payload),
604
+ }
605
+ )
606
+
607
+ response = {"correlationId": correlation_id, "results": results}
608
+ self._debug(
609
+ f"Completed client event namespace={namespace} '{event_type}' "
610
+ f"sid={sid} correlation={correlation_id}"
611
+ )
612
+ return response
613
+
614
+ def _collect_results(
615
+ self,
616
+ executions: list[_HandlerExecution],
617
+ event_type: str,
618
+ correlation_id: str,
619
+ *,
620
+ skip_none: bool = False,
621
+ ) -> List[dict[str, Any]]:
622
+ """Build a result list from handler executions.
623
+
624
+ Args:
625
+ skip_none: When ``True``, handlers that return ``None`` are omitted
626
+ from the results (fire-and-forget / opt-out semantics used by
627
+ ``process_client_event``). When ``False``, ``None`` is converted
628
+ to ``WsResult(ok=True)`` (default ``route_event`` behaviour).
629
+ """
630
results: List[dict[str, Any]] = []
631
for execution in executions:
632
handler = execution.handler
@@ -615,12 +660,11 @@ class WsManager:
660
)
661
continue
662
618
- # Skip handlers that return None — they opted out of contributing
619
- # a result (fire-and-forget semantics, matching legacy _dispatch).
663
if value is None:
621
- continue
622
-
623
- if isinstance(value, dict):
664
+ if skip_none:
665
+ continue
666
+ helper_result = WsResult(ok=True)
667
+ elif isinstance(value, dict):
668
helper_result = WsResult(ok=True, data=value)
669
else:
670
helper_result = WsResult(ok=True, data={"result": value})
@@ -632,31 +676,7 @@ class WsManager:
676
duration_ms=duration_ms,
677
)
678
)
635
-
636
- await self._publish_diagnostic_event(
637
- lambda: {
638
- "kind": "inbound",
639
- "sourceNamespace": namespace,
640
- "namespace": namespace,
641
- "eventType": event_type,
642
- "sid": sid,
643
- "correlationId": correlation_id,
644
- "timestamp": self._timestamp(),
645
- "handlerCount": len(handlers),
646
- "durationMs": sum(
647
- (exec.duration_ms or 0.0) for exec in executions
648
- ),
649
- "resultSummary": self._summarize_results(results),
650
- "payloadSummary": self._summarize_payload(handler_payload),
651
- }
652
- )
653
-
654
- response = {"correlationId": correlation_id, "results": results}
655
- self._debug(
656
- f"Completed client event namespace={namespace} '{event_type}' "
657
- f"sid={sid} correlation={correlation_id}"
658
- )
659
- return response
679
+ return results
680
681
async def _invoke_handler(
682
self,
@@ -707,9 +727,7 @@ class WsManager:
727
sid,
728
"server_restart",
729
{
710
- "emittedAt": _utcnow()
711
- .isoformat(timespec="milliseconds")
712
- .replace("+00:00", "Z"),
730
+ "emittedAt": self._timestamp(),
731
"runtimeId": runtime.get_runtime_id(),
732
},
733
handler_id=self._identifier,
@@ -942,52 +960,13 @@ class WsManager:
960
]
961
)
962
945
- results: List[dict[str, Any]] = []
946
- for execution in executions:
947
- handler = execution.handler
948
- value = execution.value
949
- duration_ms = execution.duration_ms
950
-
951
- if isinstance(value, Exception): # pragma: no cover - defensive logging
952
- PrintStyle.error(
953
- f"Error in handler {handler.identifier} for '{event_type}' (correlation {correlation_id}): {value}"
954
- )
955
- results.append(
956
- self._build_error_result(
957
- handler_id=handler.identifier,
958
- code="HANDLER_ERROR",
959
- message="Internal server error",
960
- details=str(value),
961
- correlation_id=correlation_id,
962
- duration_ms=duration_ms,
963
- )
964
- )
965
- continue
966
-
967
- if isinstance(value, WsResult):
968
- results.append(
969
- value.as_result(
970
- handler_id=handler.identifier,
971
- fallback_correlation_id=correlation_id,
972
- duration_ms=duration_ms,
973
- )
974
- )
975
- continue
976
-
977
- if value is None:
978
- helper_result = WsResult(ok=True)
979
- elif isinstance(value, dict):
980
- helper_result = WsResult(ok=True, data=value)
981
- else:
982
- helper_result = WsResult(ok=True, data={"result": value})
983
-
984
- results.append(
985
- helper_result.as_result(
986
- handler_id=handler.identifier,
987
- fallback_correlation_id=correlation_id,
988
- duration_ms=duration_ms,
989
- )
990
- )
963
+ # NOTE: skip_none=False here — route_event converts None to ok=True,
964
+ # unlike process_client_event which skips None (fire-and-forget).
965
+ # This is intentional: route_event is server-initiated and callers
966
+ # expect a result entry for every handler.
967
+ results = self._collect_results(
968
+ executions, event_type, correlation_id, skip_none=False,
969
+ )
970
971
await self._publish_diagnostic_event(
972
lambda: {
@@ -1221,7 +1200,7 @@ class WsManager:
1200
correlation_id: str | None = None,
1201
) -> dict[str, Any]:
1202
hid = handler_id or self._identifier
1224
- ts = _utcnow().isoformat(timespec="milliseconds").replace("+00:00", "Z")
1203
+ ts = self._timestamp()
1204
event_id = str(uuid.uuid4())
1205
correlation = correlation_id or str(uuid.uuid4())
1206
return {