Add useStreamListener hook

packages/flutter_hooks/lib/src/async.dart

@@ -250,6 +250,103 @@ class _StreamHookState<T> extends HookState<AsyncSnapshot<T>, _StreamHook<T>> {
   String get debugLabel => 'useStream';
 }
 
+/// Subscribes to a [Stream] and calls the [Stream.listen] to register the [onData],
+/// [onError], and [onDone].
+///
+/// See also:
+///   * [Stream], the object listened.
+///   * [Stream.listen], calls the provided handlers.
+///   * [useStream], subscribes to a [Stream] returns its current state as an [AsyncSnapshot].
+void useStreamListener<T>(
+  Stream<T>? stream, {
+  void Function(T event)? onData,
+  void Function(Object error, StackTrace stackTrace)? onError,
+  void Function()? onDone,
+  bool? cancelOnError,
+}) {
+  use(
+    _StreamListenerHook<T>(
+      stream,
+      onData: onData,
+      onError: onError,
+      onDone: onDone,
+      cancelOnError: cancelOnError,
+    ),
+  );
+}
+
+class _StreamListenerHook<T> extends Hook<void> {
+  const _StreamListenerHook(
+    this.stream, {
+    this.onData,
+    this.onError,
+    this.onDone,
+    this.cancelOnError,
+  });
+
+  final Stream<T>? stream;
+  final void Function(T event)? onData;
+  final void Function(Object error, StackTrace stackTrace)? onError;
+  final void Function()? onDone;
+  final bool? cancelOnError;
+
+  @override
+  _StreamListenerHookState<T> createState() => _StreamListenerHookState<T>();
+}
+
+class _StreamListenerHookState<T>
+    extends HookState<void, _StreamListenerHook<T>> {
+  StreamSubscription<T>? _subscription;
+
+  @override
+  void initHook() {
+    super.initHook();
+    _subscribe();
+  }
+
+  @override
+  void didUpdateHook(_StreamListenerHook<T> oldWidget) {
+    super.didUpdateHook(oldWidget);
+    if (oldWidget.stream != hook.stream ||
+        oldWidget.cancelOnError != hook.cancelOnError) {
+      if (_subscription != null) {
+        _unsubscribe();
+      }
+      _subscribe();
+    }
+  }
+
+  @override
+  void dispose() {
+    _unsubscribe();
+  }
+
+  void _subscribe() {
+    if (hook.stream != null) {
+      _subscription = hook.stream!.listen(
+        hook.onData?.call,
+        onError: hook.onError?.call,
+        onDone: hook.onDone?.call,
+        cancelOnError: hook.cancelOnError,
+      );
+    }
+  }
+
+  void _unsubscribe() {
+    _subscription?.cancel();
+    _subscription = null;
+  }
+
+  @override
+  void build(BuildContext context) {}
+
+  @override
+  String get debugLabel => 'useStreamListener';
+
+  @override
+  bool get debugSkipValue => true;
+}
+
 /// Creates a [StreamController] which is automatically disposed when necessary.
 ///
 /// See also:

packages/flutter_hooks/test/use_stream_listener_test.dart

@@ -0,0 +1,183 @@
+import 'dart:async';
+
+import 'package:flutter/foundation.dart';
+import 'package:flutter/widgets.dart';
+import 'package:flutter_hooks/flutter_hooks.dart';
+
+import 'mock.dart';
+
+void main() {
+  testWidgets(
+    'debugFillProperties',
+    (tester) async {
+      final stream = Stream.value(42);
+
+      await tester.pumpWidget(
+        HookBuilder(builder: (context) {
+          useStreamListener(stream);
+          return const SizedBox();
+        }),
+      );
+
+      await tester.pump();
+
+      final element = tester.element(find.byType(HookBuilder));
+
+      expect(
+        element
+            .toDiagnosticsNode(style: DiagnosticsTreeStyle.offstage)
+            .toStringDeep(),
+        equalsIgnoringHashCodes(
+          'HookBuilder\n'
+          ' │ useStreamListener\n'
+          ' └SizedBox(renderObject: RenderConstrainedBox#00000)\n',
+        ),
+      );
+    },
+  );
+
+  testWidgets(
+    'calls onData when data arrives',
+    (tester) async {
+      const data = 42;
+      final stream = Stream<int>.value(data);
+
+      late int value;
+
+      await tester.pumpWidget(
+        HookBuilder(builder: (context) {
+          useStreamListener<int>(
+            stream,
+            onData: (data) {
+              value = data;
+            },
+          );
+          return const SizedBox();
+        }),
+      );
+
+      expect(value, data);
+    },
+  );
+
+  testWidgets(
+    'calls onError when error occurs',
+    (tester) async {
+      final error = Exception();
+      final stream = Stream<int>.error(error);
+
+      late Object receivedError;
+
+      await tester.pumpWidget(
+        HookBuilder(builder: (context) {
+          useStreamListener<int>(
+            stream,
+            onError: (error, stackTrace) {
+              receivedError = error;
+            },
+          );
+          return const SizedBox();
+        }),
+      );
+
+      expect(receivedError, same(error));
+    },
+  );
+
+  testWidgets(
+    'calls onDone when stream is closed',
+    (tester) async {
+      final streamController = StreamController<int>.broadcast();
+
+      var onDoneCalled = false;
+
+      await tester.pumpWidget(
+        HookBuilder(builder: (context) {
+          useStreamListener<int>(
+            streamController.stream,
+            onDone: () {
+              onDoneCalled = true;
+            },
+          );
+          return const SizedBox();
+        }),
+      );
+
+      await streamController.close();
+
+      expect(onDoneCalled, isTrue);
+    },
+  );
+
+  testWidgets(
+    'cancels subscription when cancelOnError is true and error occurrs',
+    (tester) async {
+      // ignore: close_sinks
+      final streamController = StreamController<int>();
+
+      await tester.pumpWidget(
+        HookBuilder(builder: (context) {
+          useStreamListener<int>(
+            streamController.stream,
+            // onError needs to be set to prevent unhandled errors from propagating.
+            onError: (error, stackTrace) {},
+            cancelOnError: true,
+          );
+          return const SizedBox();
+        }),
+      );
+
+      expect(streamController.hasListener, isTrue);
+
+      streamController.addError(Exception());
+
+      await tester.pump();
+
+      expect(streamController.hasListener, isFalse);
+    },
+  );
+
+  testWidgets(
+    'listens new stream when stream is changed',
+    (tester) async {
+      const value1 = 42;
+      const value2 = 43;
+
+      final stream1 = Stream<int>.value(value1);
+      final stream2 = Stream<int>.value(value2);
+
+      late int receivedValue;
+
+      await tester.pumpWidget(
+        HookBuilder(
+          key: const Key('hook_builder'),
+          builder: (context) {
+            useStreamListener<int>(
+              stream1,
+              onData: (data) => receivedValue = data,
+            );
+            return const SizedBox();
+          },
+        ),
+      );
+
+      expect(receivedValue, value1);
+
+      // Listens to the stream2
+      await tester.pumpWidget(
+        HookBuilder(
+          key: const Key('hook_builder'),
+          builder: (context) {
+            useStreamListener<int>(
+              stream2,
+              onData: (data) => receivedValue = data,
+            );
+            return const SizedBox();
+          },
+        ),
+      );
+
+      expect(receivedValue, value2);
+    },
+  );
+}