(
self,
type: Literal["flows/add", "flows/update"],
f: mitmproxy.flow.Flow,
flow_json: dict, # Passing the flow_json dictionary to avoid recalculating it for each client
)
| 447 | conn._broadcast_flow(type, f, flow_json) |
| 448 | |
| 449 | def _broadcast_flow( |
| 450 | self, |
| 451 | type: Literal["flows/add", "flows/update"], |
| 452 | f: mitmproxy.flow.Flow, |
| 453 | flow_json: dict, # Passing the flow_json dictionary to avoid recalculating it for each client |
| 454 | ) -> None: |
| 455 | filters = {name: bool(expr(f)) for name, expr in self.filters.items()} |
| 456 | message = self._json_dumps( |
| 457 | { |
| 458 | "type": type, |
| 459 | "payload": { |
| 460 | "flow": flow_json, |
| 461 | "matching_filters": filters, |
| 462 | }, |
| 463 | }, |
| 464 | ) |
| 465 | self.send(message) |
| 466 | |
| 467 | def update_filter(self, name: str, expr: str) -> None: |
| 468 | if expr: |
no test coverage detected