withRequestScopeStream<S> static method

Stream<S> withRequestScopeStream<S>(
  1. Stream<S> body(
    1. BloomQueryScope scope
    ), {
  2. BloomQueryScope? scope,
  3. String? debugLabel,
})

Variant of withRequestScope for Stream results (streaming SSR).

Keeps the request scope alive until the stream closes, errors, or the subscriber cancels. Disposes owned scopes exactly once on done/error/cancel.

Implementation

static Stream<S> withRequestScopeStream<S>(
  Stream<S> Function(BloomQueryScope scope) body, {
  BloomQueryScope? scope,
  String? debugLabel,
}) {
  BloomQueryScope? current;
  if (scope != null) {
    current = scope;
  } else {
    final zoned = Zone.current[scopeZoneKey] as BloomQueryScope?;
    if (zoned != null) {
      current = zoned;
    }
  }
  final effective =
      current ?? BloomQueryScope(debugLabel: debugLabel ?? 'ssr-stream');
  final owned = current == null;
  late final Stream<S> raw;
  try {
    raw = runZoned(() => body(effective),
        zoneValues: {scopeZoneKey: effective});
  } catch (_) {
    if (owned) effective.dispose();
    rethrow;
  }
  if (!owned) return raw;
  final controller = StreamController<S>();
  StreamSubscription<S>? sub;
  var cleaned = false;
  void cleanup() {
    if (cleaned) return;
    cleaned = true;
    effective.dispose();
  }

  sub = raw.listen(
    controller.add,
    onError: (Object e, StackTrace s) {
      controller.addError(e, s);
    },
    onDone: () async {
      await controller.close();
      cleanup();
    },
    cancelOnError: false,
  );
  controller.onCancel = () async {
    await sub?.cancel();
    cleanup();
  };
  controller.onPause = () => sub?.pause();
  controller.onResume = () => sub?.resume();
  return controller.stream;
}