Asynchronously handles a batch of events in the module. Args: None Returns: bool: True if events were submitted for processing, False otherwise. Notes: - The method is wrapped in a task counter to monitor asynchronous operations
(self)
| 456 | return self._watched_events |
| 457 | |
| 458 | async def _handle_batch(self): |
| 459 | """ |
| 460 | Asynchronously handles a batch of events in the module. |
| 461 | |
| 462 | Args: |
| 463 | None |
| 464 | |
| 465 | Returns: |
| 466 | bool: True if events were submitted for processing, False otherwise. |
| 467 | |
| 468 | Notes: |
| 469 | - The method is wrapped in a task counter to monitor asynchronous operations. |
| 470 | - Checks if there are any events in the incoming queue and module is not in an error state. |
| 471 | - Invokes '_events_waiting()' to fetch a batch of events. |
| 472 | - Calls the module's 'handle_batch()' method to process these events. |
| 473 | - If a "FINISHED" event is found, invokes 'finish()' method of the module. |
| 474 | """ |
| 475 | finish = False |
| 476 | submitted = False |
| 477 | if self.batch_size <= 1: |
| 478 | return |
| 479 | if self.num_incoming_events > 0: |
| 480 | events, finish = await self._events_waiting() |
| 481 | if events and not self.errored: |
| 482 | self.verbose(f"Handling batch of {len(events):,} events") |
| 483 | event_types = {} |
| 484 | for e in events: |
| 485 | event_types[e.type] = event_types.get(e.type, 0) + 1 |
| 486 | event_types_sorted = sorted(event_types.items(), key=lambda x: x[1], reverse=True) |
| 487 | event_types_str = ", ".join(f"{k}: {v}" for k, v in event_types_sorted) |
| 488 | submitted = True |
| 489 | context = f"{self.name}.handle_batch({event_types_str})" |
| 490 | try: |
| 491 | await self.run_task(self.handle_batch(*events), context, n=len(events)) |
| 492 | except asyncio.CancelledError: |
| 493 | self.debug(f"{context} was cancelled") |
| 494 | self.verbose(f"Finished handling batch of {len(events):,} events") |
| 495 | if finish: |
| 496 | context = f"{self.name}.finish()" |
| 497 | await self.run_task(self.finish(), context) |
| 498 | return submitted |
| 499 | |
| 500 | def make_event(self, *args, **kwargs): |
| 501 | """Create an event for the scan. |
no test coverage detected