Best-effort one-way message send to the Visualizer extension host. - Never throws (returns False on failure) - Uses the same WS channel as runtime tracing - Messages are sent regardless of subscription state (run mode)
(
self,
payload: dict[str, object],
*,
require_connection: bool = True,
connect_timeout_s: float = 2.0,
enqueue_timeout_s: float = 0.2,
)
| 592 | self._record_history(payload) |
| 593 | |
| 594 | def send_message( |
| 595 | self, |
| 596 | payload: dict[str, object], |
| 597 | *, |
| 598 | require_connection: bool = True, |
| 599 | connect_timeout_s: float = 2.0, |
| 600 | enqueue_timeout_s: float = 0.2, |
| 601 | ) -> bool: |
| 602 | """ |
| 603 | Best-effort one-way message send to the Visualizer extension host. |
| 604 | |
| 605 | - Never throws (returns False on failure) |
| 606 | - Uses the same WS channel as runtime tracing |
| 607 | - Messages are sent regardless of subscription state (run mode) |
| 608 | """ |
| 609 | if not isinstance(payload, dict) or not payload: |
| 610 | return False |
| 611 | |
| 612 | try: |
| 613 | self.start() |
| 614 | except Exception: |
| 615 | return False |
| 616 | |
| 617 | if require_connection: |
| 618 | try: |
| 619 | if not self.wait_connected(timeout_s=max(0.0, float(connect_timeout_s))): |
| 620 | return False |
| 621 | except Exception: |
| 622 | return False |
| 623 | |
| 624 | try: |
| 625 | self._outq.put(payload, timeout=max(0.0, float(enqueue_timeout_s))) |
| 626 | return True |
| 627 | except Exception: |
| 628 | return False |
| 629 | |
| 630 | def _enqueue(self, msg: dict[str, Any]) -> None: |
| 631 | try: |
no test coverage detected