watchStream<T> method

StreamSubscription<T> watchStream<T>(ZenithNode<AsyncValue<T>> node, Stream<T> stream)

Subscribes to stream and writes each event to node as AsyncData.

Sets node to AsyncLoading immediately (preserving any previous data), then maps each stream event to AsyncData and each error to AsyncError. The subscription is automatically cancelled via ZenithRef.onDispose, so no manual cleanup is needed.

Returns the underlying StreamSubscription in case you need to pause, resume, or cancel it earlier than disposal.

class ChatController {
  final messagesNode = ZenithNode<AsyncValue<List<Message>>>(
    const AsyncLoading(),
  );

  void listen(ZenithRef ref, Stream<List<Message>> stream) {
    ref.watchStream(messagesNode, stream);
  }
}

Implementation

StreamSubscription<T> watchStream<T>(
  ZenithNode<AsyncValue<T>> node,
  Stream<T> stream,
) {
  if (!isMounted) {
    // Ref already disposed — subscribe and immediately cancel to return
    // a valid, inert StreamSubscription without blocking.
    final subscription = stream.listen(null);
    subscription.cancel();
    return subscription;
  }

  set<AsyncValue<T>>(node, AsyncLoading<T>(node.value.valueOrNull));

  final subscription = stream.listen(
    (data) {
      if (!isMounted) return;
      set<AsyncValue<T>>(node, AsyncData<T>(data));
    },
    onError: (Object error, StackTrace stackTrace) {
      if (!isMounted) return;
      set<AsyncValue<T>>(node, AsyncError<T>(error, stackTrace));
    },
  );

  onDisposeAsync(subscription.cancel);

  return subscription;
}