@@ -197,6 +197,12 @@ async def __start_up(self, start_wait: float):
197197 # Start the big-segment status poll now that a loop is running.
198198 self .__big_segment_store_manager .start ()
199199
200+ # FDv2 builds its data sources from builders; wire the shared session into
201+ # them before starting (FDv1 pulls the session itself via its provider).
202+ datasystem_config = self ._config .datasystem_config
203+ if datasystem_config is not None and not self ._config .offline :
204+ self ._wire_data_source_sessions (datasystem_config )
205+
200206 if self ._config .offline :
201207 log .info ("Started LaunchDarkly Client in offline mode" )
202208
@@ -221,7 +227,7 @@ async def __start_up(self, start_wait: float):
221227 log .info ("Waiting up to " + str (start_wait ) + " seconds for LaunchDarkly client to initialize..." )
222228 await update_processor_ready .wait (start_wait )
223229
224- if self .is_initialized () is True :
230+ if await self .is_initialized () is True :
225231 log .info ("Started LaunchDarkly Client: OK" )
226232 else :
227233 log .warning ("Initialization timeout exceeded for LaunchDarkly Client or an error occurred. " "Feature Flags may not yet be available." )
@@ -243,7 +249,9 @@ def _make_data_system(self) -> AsyncDataSystem:
243249
244250 return AsyncFDv1 (self ._config , self ._select_feature_store (), self ._get_session )
245251
246- raise NotImplementedError ("FDv2 is not yet supported in the async client" )
252+ from ldclient .impl .datasystem .async_fdv2 import AsyncFDv2
253+
254+ return AsyncFDv2 (self ._config , datasystem_config )
247255
248256 def _select_feature_store (self ) -> AsyncFeatureStore :
249257 """Choose the async feature store for the v1 data system based on the
@@ -253,6 +261,34 @@ def _select_feature_store(self) -> AsyncFeatureStore:
253261 return AsyncInMemoryFeatureStore ()
254262 return feature_store
255263
264+ def _wire_data_source_sessions (self , data_system_config ) -> None :
265+ """Provide the client's aiohttp session to any async data source
266+ builders so the sources they build share the client's connection pool."""
267+ from ldclient .impl .datasourcev2 .async_polling import (
268+ AsyncFallbackToFDv1PollingDataSourceBuilder ,
269+ AsyncPollingDataSourceBuilder
270+ )
271+ from ldclient .impl .datasourcev2 .async_streaming import (
272+ AsyncStreamingDataSourceBuilder
273+ )
274+
275+ builders = list (data_system_config .initializers or []) + list (
276+ data_system_config .synchronizers or []
277+ )
278+ if data_system_config .fdv1_fallback_synchronizer is not None :
279+ builders .append (data_system_config .fdv1_fallback_synchronizer )
280+
281+ for builder in builders :
282+ if isinstance (
283+ builder ,
284+ (
285+ AsyncFallbackToFDv1PollingDataSourceBuilder ,
286+ AsyncPollingDataSourceBuilder ,
287+ AsyncStreamingDataSourceBuilder ,
288+ ),
289+ ):
290+ builder .session (self ._get_session ())
291+
256292 async def __register_plugins (self , environment_metadata : EnvironmentMetadata ):
257293 for plugin in self ._config .plugins :
258294 try :
@@ -343,18 +379,20 @@ def is_offline(self) -> bool:
343379 """Returns true if the client is in offline mode."""
344380 return self ._config .offline
345381
346- def is_initialized (self ) -> bool :
382+ async def is_initialized (self ) -> bool :
347383 """Returns true if the client has successfully connected to LaunchDarkly.
348384
349385 If this returns false, it means that the client has not yet successfully connected to LaunchDarkly.
350386 It might still be in the process of starting up, or it might be attempting to reconnect after an
351387 unsuccessful attempt, or it might have received an unrecoverable error (such as an invalid SDK key)
352388 and given up.
389+
390+ This is a coroutine because determining readiness may query a persistent store.
353391 """
354392 if self .is_offline () or self ._config .use_ldd :
355393 return True
356394
357- return self ._data_system .data_availability .at_least (DataAvailability .CACHED )
395+ return ( await self ._data_system .data_availability ()) .at_least (DataAvailability .CACHED )
358396
359397 async def flush (self ):
360398 """Flushes all pending analytics events.
@@ -455,8 +493,9 @@ async def _evaluate_internal(self, key: str, context: Context, default: Any, eve
455493 if self ._config .offline :
456494 return EvaluationDetail (default , None , error_reason ('CLIENT_NOT_READY' )), None
457495
458- if self ._data_system .data_availability != DataAvailability .REFRESHED :
459- if self ._data_system .data_availability == DataAvailability .CACHED :
496+ availability = await self ._data_system .data_availability ()
497+ if availability != DataAvailability .REFRESHED :
498+ if availability == DataAvailability .CACHED :
460499 log .warning ("Feature Flag evaluation attempted before client has initialized - using last known values from feature store for feature key: " + key )
461500 else :
462501 log .warning ("Feature Flag evaluation attempted before client has initialized! Feature store unavailable - returning default: " + str (default ) + " for feature key: " + key )
@@ -526,8 +565,9 @@ async def all_flags_state(self, context: Context, **kwargs) -> FeatureFlagsState
526565 log .warning ("all_flags_state() called, but client is in offline mode. Returning empty state" )
527566 return FeatureFlagsState (False )
528567
529- if self ._data_system .data_availability != DataAvailability .REFRESHED :
530- if self ._data_system .data_availability == DataAvailability .CACHED :
568+ availability = await self ._data_system .data_availability ()
569+ if availability != DataAvailability .REFRESHED :
570+ if availability == DataAvailability .CACHED :
531571 log .warning ("all_flags_state() called before client has finished initializing! Using last known values from feature store" )
532572 else :
533573 log .warning ("all_flags_state() called before client has finished initializing! Feature store unavailable - returning empty state" )
0 commit comments