fix(storage, web): clean up stream handlers on "hot restart" (#12927)

* fix(storage, web): clean up stream handlers on "hot restart"

* fix(storage, web): only create one stream for task, fix task.pause()

packages/firebase_storage/firebase_storage_web/lib/src/interop/storage.dart

@@ -343,8 +343,17 @@ class UploadTask extends JsObjectWrapper<storage_interop.UploadTaskJsImpl> {
   /// Returns [:true:] if it had an effect.
   bool cancel() => jsObject.cancel().toDart;
 
+  String _taskSnapshotWindowsKey(String appName, String bucket, String path) =>
+      'flutterfire-${appName}_${bucket}_${path}_storageTask';
+
   /// Stream for upload task state changed event.
-  Stream<UploadTaskSnapshot> get onStateChanged {
+  Stream<UploadTaskSnapshot> onStateChanged(
+    String appName,
+    String bucket,
+    String path,
+  ) {
+    final windowsKey = _taskSnapshotWindowsKey(appName, bucket, path);
+    unsubscribeWindowsListener(windowsKey);
     late StreamController<UploadTaskSnapshot> changeController;
     late JSFunction onStateChangedUnsubscribe;
 
@@ -367,11 +376,16 @@ class UploadTask extends JsObjectWrapper<storage_interop.UploadTaskJsImpl> {
         errorWrapper,
         onCompletion,
       );
+      setWindowsListener(
+        windowsKey,
+        onStateChangedUnsubscribe,
+      );
     }
 
     void stopListen() {
       onStateChangedUnsubscribe.callAsFunction();
       changeController.close();
+      removeWindowsListener(windowsKey);
     }
 
     changeController = StreamController<UploadTaskSnapshot>.broadcast(

packages/firebase_storage/firebase_storage_web/lib/src/task_web.dart

@@ -28,6 +28,8 @@ class TaskWeb extends TaskPlatform {
 
   final storage_interop.UploadTask _task;
 
+  Stream<TaskSnapshotPlatform>? _stream;
+
   /// Returns a [Stream] of [TaskSnapshot] events.
   ///
   /// If the task is canceled or fails, the stream will send an error event.
@@ -37,7 +39,7 @@ class TaskWeb extends TaskPlatform {
   /// wait for the stream to complete via [onComplete].
   @override
   Stream<TaskSnapshotPlatform> get snapshotEvents {
-    return guard(() {
+    _stream ??= guard(() {
       // The mobile version of the plugin pushes a "success" snapshot to the
       // onStateChanged stream, but the Firebase JS SDK does *not*.
       // We use a StreamGroup + Future.asStream to simulate that feature:
@@ -46,8 +48,13 @@ class TaskWeb extends TaskPlatform {
 
       // This stream converts the UploadTask Snapshots from JS to the plugins'
       // It can also throw a FirebaseError internally, so we handle it.
-      final onStateChangedStream =
-          _task.onStateChanged.map<TaskSnapshotPlatform>((snapshot) {
+      final onStateChangedStream = _task
+          .onStateChanged(
+        _reference.storage.app.name,
+        _reference.bucket,
+        _reference.fullPath,
+      )
+          .map<TaskSnapshotPlatform>((snapshot) {
         return fbUploadTaskSnapshotToTaskSnapshot(_reference, snapshot);
       });
 
@@ -66,6 +73,8 @@ class TaskWeb extends TaskPlatform {
 
       return group.stream;
     });
+
+    return _stream!;
   }
 
   /// Returns a [Future] once the task has completed.
@@ -101,8 +110,7 @@ class TaskWeb extends TaskPlatform {
     final paused = _task.pause();
     // Wait until the snapshot is paused, then return the value of paused...
     return snapshotEvents
-        .takeWhile((snapshot) => snapshot.state != TaskState.paused)
-        .last
+        .firstWhere((snapshot) => snapshot.state == TaskState.paused)
         .then<bool>((_) => paused);
   }