feat: Add flush_and_wait to DefaultAsyncEventProcessor

Triggers a flush and awaits delivery via a new inbox message, returning
whether it completed within the timeout.

Author: jsonbailey

ldclient/impl/events/async_event_processor.py

@@ -132,6 +132,10 @@ async def _run_main_loop(self):
                     self._context_keys.clear()
                 elif message.type == 'diagnostic':
                     self._send_and_reset_diagnostics()
+                elif message.type == 'flush_and_wait':
+                    self._trigger_flush()
+                    await self._flush_workers.wait()
+                    message.param.set()
                 elif message.type == 'test_sync':
                     await self._flush_workers.wait()
                     if self._diagnostic_flush_workers is not None:
@@ -204,6 +208,13 @@ def send_event(self, event: EventInput):
     def flush(self):
         self._post_to_inbox(EventProcessorMessage('flush', None))
 
+    async def flush_and_wait(self, timeout: float) -> bool:
+        try:
+            await asyncio.wait_for(self._post_message_and_wait('flush_and_wait'), timeout)
+            return True
+        except asyncio.TimeoutError:
+            return False
+
     async def stop(self):
         async with self._close_lock:
             if self._closed:

ldclient/testing/impl/events/test_async_event_processor.py

@@ -174,6 +174,35 @@ async def test_custom_event_is_queued():
         assert output[1]['metricValue'] == 1.5
 
 
+# ---------------------------------------------------------------------------
+# flush_and_wait
+# ---------------------------------------------------------------------------
+
+async def test_flush_and_wait_delivers_events_and_returns_true():
+    mock_http = MockAioHttp()
+    async with make_processor(mock_http) as ep:
+        ep.send_event(EventInputIdentify(timestamp, context))
+
+        delivered = await ep.flush_and_wait(5)
+
+        assert delivered is True
+        assert mock_http.request_data is not None
+        output = json.loads(mock_http.request_data)
+        assert len(output) == 1
+        assert output[0]['kind'] == 'identify'
+
+
+async def test_flush_and_wait_returns_false_on_timeout():
+    mock_http = MockAioHttp()
+    async with make_processor(mock_http) as ep:
+        ep.send_event(EventInputIdentify(timestamp, context))
+
+        # A zero timeout can't complete the flush round-trip, so it reports False.
+        delivered = await ep.flush_and_wait(0)
+
+        assert delivered is False
+
+
 async def test_two_events_for_same_context_only_produce_one_index_event():
     mock_http = MockAioHttp()
     async with make_processor(mock_http) as ep: