Method Detail

• StreamWatcher

   final static <T extends Any> StreamWatcher<T> StreamWatcher(CoroutineScope scope, StreamLogger logger, StateFlow<StreamConnectionState> connectionState)

  Creates a new StreamWatcher instance that observes connection state changes.

  This factory method instantiates the default implementation (StreamWatcherImpl) and creates the necessary internal subscription manager. The watcher observes the provided connection state flow and triggers rewatch callbacks when the connection state changes to Connected.

  • scope: The CoroutineScope used for launching async operations (collecting state flow and invoking rewatch callbacks). This scope should typically use SupervisorJob + Dispatchers. Default to ensure callback failures don't cancel the scope and to avoid blocking the main thread. The scope is NOT cancelled when StreamWatcher.stop is called, allowing the component to be restarted.

  • logger: A StreamLogger instance for diagnostic output, tagged appropriately (e.g., "Watcher"). Used for logging state changes, rewatch events, and errors.

  • connectionState: A StateFlow providing connection state updates. Typically obtained from streamClient.connectionState. The watcher will collect from this flow when started.

  The returned watcher is created in a stopped state. You must call StreamWatcher.start to begin monitoring connection state changes. The component can be started and stopped multiple times throughout its lifetime.

  val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
  val logger = StreamLogger.getLogger("MyApp.Watcher")
  
  // Create watcher - only needs the connection state flow
  val watcher = StreamWatcher<String>(
      scope = scope,
      logger = logger,
      connectionState = streamClient.connectionState
  )
  
  // Register rewatch listener
  watcher.subscribe(StreamRewatchListener { ids, connectionId ->
      println("Re-watching ${ids.size} channels on connection $connectionId")
      // Re-establish server-side watches...
  })
  
  // Start monitoring
  watcher.start()
  
  // Watch channels
  watcher.watch("messaging:general")
  watcher.watch("livestream:sports")

  Parameters:

  scope - Coroutine scope for async operations (state collection and rewatch callback invocations)

  logger - Logger for diagnostic output and error reporting

  connectionState - StateFlow providing connection state updates