-
- All Implemented Interfaces:
-
io.getstream.android.core.api.observers.StreamStartableComponent,io.getstream.android.core.api.subscribe.StreamObservable
public interface StreamWatcher<T extends Object> implements StreamObservable<StreamRewatchListener<T>>, StreamStartableComponent
Manages a registry of watched resources (channels, conversations, streams) and automatically triggers re-watching when the connection state changes.
This component is critical for maintaining active subscriptions across reconnection cycles. When the WebSocket disconnects and reconnects, all active watches must be re-established on the server. StreamWatcher tracks which items of type T are currently being watched and invokes a callback on every connection state change, allowing the product SDK to re-subscribe.
Product SDK watches a channel:
watcher.watch("messaging:general")Watcher adds the identifier to its internal registry
Product SDK registers a rewatch listener:
watcher.subscribe(StreamRewatchListener { ids, connectionId -> resubscribe(ids, connectionId) })Call
watcher.start()to begin monitoring connection state changesConnection state changes (e.g., reconnection after network loss)
Watcher invokes listener with all watched entries:
["messaging:general", "livestream:sports"]Product SDK re-establishes server-side watches for each entry
All methods are thread-safe and can be called from any thread
The rewatch callback is invoked asynchronously on an internal coroutine scope
Callback invocation happens asynchronously on connection state changes
This component implements StreamStartableComponent and requires explicit lifecycle management:
Call start to begin monitoring connection state changes
Call stopWatching when a resource is no longer needed
Call stop when shutting down to cancel monitoring and clean up resources
Call clear to remove all watched entries from the registry
-
-
Method Summary
-
-
Method Detail
-
watch
abstract Result<T> watch(T item)
Registers a channel or resource as actively watched.
Adds the given item to the internal registry of watched resources. When the connection state changes (especially during reconnection), the rewatch callback will include this item in the list of resources that need to be re-subscribed.
Multiple calls with the same item are idempotent—only one entry is maintained per unique identifier.
val cid = "messaging:general" watcher.watch(cid) .onSuccess { logger.i { "Now watching: $it" } } .onFailure { error -> logger.e(error) { "Failed to watch" } }- Parameters:
item- The identifier to watch (e.g.
-
stopWatching
abstract Result<T> stopWatching(T item)
Unregisters a channel or resource from the watch registry.
Removes the given item from the internal registry. Subsequent connection state changes will no longer include this entry in the rewatch callback list.
Calling this method with an identifier that is not being watched is a no-op (not an error).
val cid = "messaging:general" watcher.stopWatching(cid) .onSuccess { logger.i { "Stopped watching: $it" } } .onFailure { error -> logger.e(error) { "Failed to stop watching" } }- Parameters:
item- The identifier to stop watching
-
clear
abstract Result<Unit> clear()
Removes all entries from the watch registry.
After calling this method, the rewatch callback will NOT be invoked on subsequent connection state changes (since the registry is empty) until new resources are watched.
This method is typically called during cleanup or logout to ensure no stale watches remain.
// During logout watcher.clear() .onSuccess { logger.i { "All watches cleared" } } .onFailure { error -> logger.e(error) { "Failed to clear watches" } }
-
-
-
-