Package 

Interface StreamRewatchListener

  • 
public interface StreamRewatchListener<T extends Object>

    Callback interface invoked when watched resources need to be re-subscribed after a connection state change.

    This functional interface is used with StreamWatcher to receive notifications when the WebSocket connection state changes (especially during reconnections), allowing the product SDK to re-establish server-side watches for all actively monitored resources.

    The callback is triggered on every StreamConnectionState.Connected event when the watched CID registry is non-empty. This ensures that all active subscriptions are restored after:

    • Initial connection establishment

    • Network reconnection after temporary disconnection

    • WebSocket recovery after connection loss

    • The callback is invoked asynchronously on an internal coroutine scope

    • Implementations should be thread-safe as concurrent invocations are possible

    • Long-running operations should be dispatched to an appropriate dispatcher

    • Exceptions thrown by this callback are caught and logged by the watcher

    • Errors are also surfaced via StreamClientListener.onError for user-level handling

    • A failing callback will not prevent other listeners from being notified

    val watcher = StreamWatcher<String>(scope, logger, rewatchManager, clientManager)

// Register a rewatch listener
watcher.subscribe(
    listener = StreamRewatchListener { ids, connectionId ->
        logger.i { "Re-watching ${ids.size} channels on connection $connectionId" }
        ids.forEach { id ->
            // Re-establish server-side watch for each identifier
            channelClient.watch(id, connectionId)
        }
    }
)

watcher.start()

    • Method Summary

      Modifier and Type Method Description
      abstract Unit onRewatch(Set<T> items, String connectionId) Called when watched resources need to be re-subscribed.

      • Methods inherited from class java.lang.Object

        clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

    • Method Detail

      • onRewatch

         abstract Unit onRewatch(Set<T> items, String connectionId)

        Called when watched resources need to be re-subscribed.

        This suspend method is invoked with the complete set of entries currently in the watch registry whenever the connection state changes to StreamConnectionState.Connected and the registry is non-empty.

        Implementations should iterate over the provided list and re-establish server-side watches for each entry to maintain active subscriptions across reconnection cycles. Since this is a suspend function, you can directly call suspend API functions without launching coroutines.

        Rewatch callbacks are executed sequentially using a serial processing queue. This ensures:

        • Callbacks execute in order, one at a time

        • No race conditions when re-establishing watches

        • Predictable execution order across reconnections

        • The set is never empty when this method is called

        • The set represents a snapshot of the watch registry at the time of invocation

        • Modifications to the set do not affect the internal registry

        • Multiple calls may occur for the same set of entries during reconnection scenarios

        • The connectionId reflects the current active connection at the moment of invocation

        Exceptions thrown by this method are caught, logged, and surfaced via error callbacks. The failure of one listener does not prevent other listeners from being notified.

        watcher.subscribe(StreamRewatchListener { channelIds, connectionId ->
    // Direct suspend calls - no need to launch!
    channelIds.forEach { channelId ->
        channelApi.watch(channelId, connectionId) // suspend function
    }
})
        Parameters:
        items - A non-empty set of entries that require re-watching
        connectionId - The connection ID of the current active WebSocket connection