(self, name: str, expr: str)
| 465 | self.send(message) |
| 466 | |
| 467 | def update_filter(self, name: str, expr: str) -> None: |
| 468 | if expr: |
| 469 | filt = flowfilter.parse(expr) |
| 470 | self.filters[name] = filt |
| 471 | matching_flow_ids = [f.id for f in self.application.master.view if filt(f)] |
| 472 | else: |
| 473 | self.filters.pop(name, None) |
| 474 | matching_flow_ids = None |
| 475 | |
| 476 | message = self._json_dumps( |
| 477 | { |
| 478 | "type": "flows/filterUpdate", |
| 479 | "payload": { |
| 480 | "name": name, |
| 481 | "matching_flow_ids": matching_flow_ids, |
| 482 | }, |
| 483 | }, |
| 484 | ) |
| 485 | self.send(message=message) |
| 486 | |
| 487 | async def on_message(self, message: str | bytes): |
| 488 | try: |
no test coverage detected