Emit an event right before the event loop starts.

Author: gbeced

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Features
 
 * `fees.Percentage` now supports charging fees in base currency.
+* Both `BacktestingDispatcher` and `RealtimeDispatcher` now support `subscribe_event_loop_started` to register handlers that are called once when the dispatch loop starts.
 
 ## 1.10.2
 

basana/core/dispatcher/__init__.py

@@ -19,6 +19,7 @@
 from .base import (
     EventDispatcher,
     EventHandler,
+    LoopStartedHandler,
     SchedulerJob,
 )
 

basana/core/dispatcher/backtesting.py

@@ -53,6 +53,9 @@ def _set_now(self, now: datetime.datetime):
         self._last_dt = now
 
     async def _dispatch_loop(self):
+        self._last_dt = self._event_mux.peek_next_event_dt()
+        await self._notify_loop_started()
+
         while not self.stopped:
             # Time is driven by event timestamps, so "now" advances only when events are processed.
             next_dt = self._event_mux.peek_next_event_dt()

basana/core/dispatcher/base.py

@@ -33,6 +33,7 @@
 
 logger = logging.getLogger(__name__)
 EventHandler = Callable[[core_event.Event], Awaitable[Any]]
+LoopStartedHandler = Callable[[], Awaitable[Any]]
 
 
 # Indices for prefetched event tuples stored in the EventMultiplexer heap.
@@ -114,6 +115,7 @@ def __init__(self, max_concurrent: int):
         self._event_handlers: Dict[core_event.EventSource, List[EventHandler]] = defaultdict(list)
         self._sniffers_pre: List[EventHandler] = []
         self._sniffers_post: List[EventHandler] = []
+        self._loop_started_handlers: List[LoopStartedHandler] = []
         self._producers: Set[core_event.Producer] = set()
         # Task group for core tasks like producers and dispatch loop.
         self._core_tasks: Optional[helpers.TaskGroup] = None
@@ -179,6 +181,17 @@ def subscribe_all(self, event_handler: EventHandler, front_run: bool = False):
         if event_handler not in sniffers:
             sniffers.append(event_handler)
 
+    def subscribe_event_loop_started(self, handler: LoopStartedHandler):
+        """Registers an async callable that will be called once when the event loop starts.
+
+        :param handler: An async callable that receives no arguments.
+        """
+
+        assert not self._running, "Subscribing once we're running is not currently supported."
+
+        if handler not in self._loop_started_handlers:
+            self._loop_started_handlers.append(handler)
+
     def schedule(self, when: datetime.datetime, job: SchedulerJob):
         """Schedules a function to be executed at a given time.
 
@@ -248,6 +261,10 @@ async def _core_task_group(self):
         finally:
             self._core_tasks = None
 
+    async def _notify_loop_started(self):
+        if self._loop_started_handlers:
+            await gather_no_raise(*[handler() for handler in self._loop_started_handlers])
+
     async def _dispatch_event(self, event: core_event.Event, handlers: List[EventHandler]):
         if self._debug_enabled:
             logger.debug(logs.StructuredMessage(

basana/core/dispatcher/realtime.py

@@ -55,6 +55,7 @@ def subscribe_idle(self, idle_handler: IdleHandler):
             self._idle_handlers.append(idle_handler)
 
     async def _dispatch_loop(self):
+        await self._notify_loop_started()
         while not self.stopped:
             now = dt.utc_now()
             # Feed the task pool with scheduled jobs and events that are ready for processing.

tests/test_backtesting_dispatcher.py

@@ -219,6 +219,37 @@ def test_now_fails_if_no_events_were_processed(backtesting_dispatcher):
         backtesting_dispatcher.now()
 
 
+async def test_event_loop_started(backtesting_dispatcher):
+    started = []
+
+    async def on_started():
+        assert backtesting_dispatcher.now_available
+        started.append(True)
+
+    src = event.FifoQueueEventSource(events=[event.Event(dt.utc_now())])
+    backtesting_dispatcher.subscribe(src, lambda e: None)
+    backtesting_dispatcher.subscribe_event_loop_started(on_started)
+    await backtesting_dispatcher.run()
+
+    assert len(started) == 1
+
+
+async def test_duplicate_loop_started_subscription_is_ignored(backtesting_dispatcher):
+    started = []
+
+    async def on_started():
+        assert backtesting_dispatcher.now_available
+        started.append(True)
+
+    src = event.FifoQueueEventSource(events=[event.Event(dt.utc_now())])
+    backtesting_dispatcher.subscribe(src, lambda e: None)
+    backtesting_dispatcher.subscribe_event_loop_started(on_started)
+    backtesting_dispatcher.subscribe_event_loop_started(on_started)
+    await backtesting_dispatcher.run()
+
+    assert len(started) == 1
+
+
 async def test_recursive_schedule_bug(backtesting_dispatcher):
     jobs_processed = 0
 

tests/test_realtime_dispatcher.py

@@ -218,6 +218,24 @@ async def on_idle():
     assert handler_calls == 2
 
 
+async def test_event_loop_started(realtime_dispatcher):
+    started = []
+
+    async def on_started():
+        started.append(True)
+
+    async def on_idle():
+        realtime_dispatcher.stop()
+
+    async def test_main():
+        realtime_dispatcher.subscribe_event_loop_started(on_started)
+        realtime_dispatcher.subscribe_idle(on_idle)
+        await realtime_dispatcher.run()
+        assert len(started) == 1
+
+    await asyncio.wait_for(test_main(), 2)
+
+
 async def test_cancelation_is_forwarded(realtime_dispatcher):
     async def test_main():
         with pytest.raises(asyncio.CancelledError):