watchLocalWithRealtime<TResult> method

@protected Stream<TResult> watchLocalWithRealtime<TResult>(Stream<TResult> localStream, { required String scope, String? id, QueryParams? queryParams, Map<String, String>? headers, Map<String, dynamic>? extra, bool retryOnFail = true, })

Wraps a local watch stream and ties a remote realtime subscription to the returned stream's lifecycle.

Implementation

@protected
Stream<TResult> watchLocalWithRealtime<TResult>(
  Stream<TResult> localStream, {
  required String scope,
  String? id,
  QueryParams? queryParams,
  Map<String, String>? headers,
  Map<String, dynamic>? extra,
  bool retryOnFail = true,
}) {
  _realtimeAdapterOrThrow();

  late final StreamController<TResult> controller;
  StreamSubscription<TResult>? localSubscription;
  _RealtimeSubscriptionKey? key;

  controller = StreamController<TResult>(
    onListen: () {
      key = _retainRealtimeSubscription(
        scope: scope,
        id: id,
        queryParams: queryParams,
        headers: headers,
        extra: extra,
        retryOnFail: retryOnFail,
      );

      localSubscription = localStream.listen(
        controller.add,
        onError: controller.addError,
        onDone: controller.close,
      );
    },
    onPause: () {
      localSubscription?.pause();
    },
    onResume: () {
      localSubscription?.resume();
    },
    onCancel: () async {
      await localSubscription?.cancel();
      final activeKey = key;
      if (activeKey != null) {
        await _releaseRealtimeSubscription(activeKey,
            retryOnFail: retryOnFail);
      }
    },
  );

  return controller.stream;
}