[analyzer] Ensure wrapped watcher streams are cancelled when subscriber cancels

Fixes #51746.

Change-Id: If5e98eef60c9404378c7bbb050fcfbe6a6d807f2
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/289080
Commit-Queue: Brian Wilkerson <[email protected]>
Reviewed-by: Brian Wilkerson <[email protected]>

Author: DanTup

pkg/analyzer/lib/file_system/physical_file_system.dart

@@ -425,10 +425,12 @@ abstract class _PhysicalResource implements Resource {
     }
 
     final mappedEventsController = StreamController<WatchEvent>();
-    original.listen(
+    final subscription = original.listen(
       mappedEventsController.add,
       onError: (Object e) => mappedEventsController.addError(mapException(e)),
+      onDone: () => mappedEventsController.close(),
     );
+    mappedEventsController.onCancel = subscription.cancel;
     return mappedEventsController.stream;
   }
 }