[analyzer] Ensure watchers can be subscribed to more than once

The stream created by memory resources was a single-subscriber stream. The stream from the physical file system watchers was a broadcast stream, but the code to translate FileSystemException to the abstractions was written in a way that made it single-subscriber.

This fixes both to ensure that a `ResourceWatcher` can support multiple subscribers that can subscribe and cancel at different times.

We don't currently use this, so this change should effectively be a no-op right now, but it will allow us to reuse the temporary catchers we create when set up analysis roots instead of creating new ones.

There's an issue at https://github.com/dart-lang/sdk/issues/54274 discussing changing how we watch here, but it may still be worth picking the low-hanging fruit in the short-term.

Change-Id: I3bda77ddaee676c573021aa671cb01bae48bd56f
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/341740
Reviewed-by: Brian Wilkerson <brianwilkerson@google.com>
Commit-Queue: Brian Wilkerson <brianwilkerson@google.com>
Reviewed-by: Konstantin Shcheglov <scheglov@google.com>

pkg/analyzer/lib/file_system/file_system.dart

@@ -255,16 +255,30 @@ abstract class ResourceProvider {
 /// The [ready] event will not fire until a listener has been set up on
 /// [changes] and the watcher initialization is complete.
 class ResourceWatcher {
+  /// A broadcast stream of changes from this watcher.
+  ///
+  /// This stream can be subscribed to by multiple listeners, but each listener
+  /// should await the latest [Future] returned by [ready] after subscribing to
+  /// ensure the watcher is set up. The internal watcher may be closed when
+  /// there are no subscribers and re-created when another subscriber appears.
   final Stream<WatchEvent> changes;
 
+  /// A function to obtain the to current [ready] [Future] from the underlying
+  /// watcher.
+  final Future<void> Function() _ready;
+
+  ResourceWatcher(this.changes, this._ready);
+
   /// An event that fires when the watcher is fully initialized and ready to
   /// produce events.
   ///
   /// This event will not fire until a listener has been set up on [changes] and
   /// the watcher initialization is complete.
-  final Future<void> ready;
-
-  ResourceWatcher(this.changes, this.ready);
+  ///
+  /// This Future may change over time because the internal watcher may be
+  /// closed when there are no subscribers and re-created when another
+  /// subscriber appears.
+  Future<void> get ready => _ready();
 }
 
 extension FolderExtension on Folder {

pkg/analyzer/lib/file_system/memory_file_system.dart

@@ -614,7 +614,7 @@ abstract class _MemoryResource implements Resource {
   /// watcher.
   @override
   ResourceWatcher watch() {
-    final streamController = StreamController<WatchEvent>();
+    final streamController = StreamController<WatchEvent>.broadcast();
     final ready = Completer<void>();
 
     /// A helper that sets up the watcher that may be called synchronously
@@ -650,7 +650,7 @@ abstract class _MemoryResource implements Resource {
       setupWatcher();
     }
 
-    return ResourceWatcher(streamController.stream, ready.future);
+    return ResourceWatcher(streamController.stream, () => ready.future);
   }
 }
 

pkg/analyzer/lib/file_system/physical_file_system.dart

@@ -184,7 +184,10 @@ class _PhysicalFile extends _PhysicalResource implements File {
   @override
   ResourceWatcher watch() {
     final watcher = FileWatcher(_entry.path);
-    return ResourceWatcher(_wrapWatcherStream(watcher.events), watcher.ready);
+    return ResourceWatcher(
+      watcher.events.transform(_exceptionTransformer),
+      () => watcher.ready,
+    );
   }
 
   @override
@@ -320,7 +323,10 @@ class _PhysicalFolder extends _PhysicalResource implements Folder {
             // Don't suppress "Directory watcher closed," so the outer
             // listener can see the interruption & act on it.
             !error.message.startsWith("Directory watcher closed unexpectedly"));
-    return ResourceWatcher(_wrapWatcherStream(events), watcher.ready);
+    return ResourceWatcher(
+      events.transform(_exceptionTransformer),
+      () => watcher.ready,
+    );
   }
 }
 
@@ -328,6 +334,17 @@ class _PhysicalFolder extends _PhysicalResource implements Folder {
 abstract class _PhysicalResource implements Resource {
   final io.FileSystemEntity _entry;
 
+  /// Wraps [FileSystemException]s in the stream through [_wrapException].
+  late final _exceptionTransformer =
+      StreamTransformer<WatchEvent, WatchEvent>.fromHandlers(
+    handleError: (error, stackTrace, sink) {
+      if (error is io.FileSystemException) {
+        error = _wrapException(error);
+      }
+      sink.addError(error);
+    },
+  );
+
   _PhysicalResource(this._entry);
 
   @override
@@ -415,22 +432,4 @@ abstract class _PhysicalResource implements Resource {
       return FileSystemException(e.path ?? path, e.message);
     }
   }
-
-  /// Wraps a `Stream<WatchEvent>` to map all known errors through
-  /// [_wrapException] into server types.
-  Stream<WatchEvent> _wrapWatcherStream(Stream<WatchEvent> original) {
-    /// Helper to map thrown `FileSystemException`s to servers abstraction.
-    Object mapException(Object e) {
-      return e is io.FileSystemException ? _wrapException(e) : e;
-    }
-
-    final mappedEventsController = StreamController<WatchEvent>();
-    final subscription = original.listen(
-      mappedEventsController.add,
-      onError: (Object e) => mappedEventsController.addError(mapException(e)),
-      onDone: () => mappedEventsController.close(),
-    );
-    mappedEventsController.onCancel = subscription.cancel;
-    return mappedEventsController.stream;
-  }
 }

pkg/analyzer/test/file_system/file_system_test_support.dart

@@ -2,13 +2,17 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
+import 'dart:async';
+import 'dart:io';
 import 'dart:typed_data';
 
 import 'package:analyzer/file_system/file_system.dart';
+import 'package:analyzer/file_system/physical_file_system.dart';
 import 'package:analyzer/source/source.dart';
 import 'package:path/path.dart' as path;
 import 'package:test/test.dart';
 import 'package:test_reflective_loader/test_reflective_loader.dart';
+import 'package:watcher/watcher.dart' show ChangeType;
 
 final isFile = TypeMatcher<File>();
 final isFileSystemException = TypeMatcher<FileSystemException>();
@@ -16,6 +20,11 @@ final isFolder = TypeMatcher<Folder>();
 
 final throwsFileSystemException = throwsA(isFileSystemException);
 
+typedef _WatcherChanges = ({
+  List<ChangeType> changes,
+  Future<void> Function() cancel,
+});
+
 abstract class FileSystemTestSupport {
   /// The content used for the file at the [defaultFilePath] if it is created
   /// and no other content is provided.
@@ -65,6 +74,38 @@ abstract class FileSystemTestSupport {
           String? part5,
           String? part6]) =>
       provider.pathContext.join(part1, part2, part3, part4, part5, part6);
+
+  /// Subscribes to [watcher], waits for it to become ready, and returns
+  /// a record containing a list of events received (which updates as events
+  /// arrive) and a function to cancel.
+  Future<_WatcherChanges> _subscribeToWatcher(ResourceWatcher watcher) async {
+    var changes = <ChangeType>[];
+    var subscription =
+        watcher.changes.listen((event) => changes.add(event.type));
+
+    await watcher.ready;
+
+    return (changes: changes, cancel: subscription.cancel);
+  }
+
+  /// Waits up to [maxDuration] for [condition] to return `true`.
+  ///
+  /// Throws if condition is not `true` before the time expires.
+  Future<void> _waitFor(
+    FutureOr<bool> Function() condition, [
+    // Set the default max high enough to ensure no flakes on slow bots. We
+    // check periodically and will exit early if the condition is met.
+    Duration maxDuration = const Duration(seconds: 5),
+  ]) async {
+    var endTime = DateTime.now().add(maxDuration);
+    while (DateTime.now().isBefore(endTime)) {
+      if (await condition()) {
+        return;
+      }
+      await pumpEventQueue(times: 1000);
+    }
+    throw 'Condition was not true within $maxDuration';
+  }
 }
 
 /// Unlike most test mixins, this mixin defines some abstract test methods.
@@ -229,6 +270,77 @@ mixin FileTestMixin implements FileSystemTestSupport {
     expect(() => file.modificationStamp, throwsA(isFileSystemException));
   }
 
+  /// Test that we can reuse a [ResourceWatcher] to create a second subscription
+  /// and cancel it at a different time.
+  test_multipleWatchers_staggeredCancel() async {
+    var filePath = path.join(tempPath, 'foo');
+    var file = getFile(filePath: filePath, exists: true);
+
+    // On Windows, package:watcher may miss modification events for files that
+    // were created just before we started watching (see
+    // https://github.com/dart-lang/watcher/issues/159), so add an artificial
+    // delay there to ensure we don't miss events until that issue is resolved.
+    if (provider is PhysicalResourceProvider && Platform.isWindows) {
+      await Future<void>.delayed(const Duration(seconds: 1));
+    }
+
+    var resource = provider.getResource(filePath);
+    var watcher = resource.watch();
+
+    var watch1 = await _subscribeToWatcher(watcher);
+    var watch2 = await _subscribeToWatcher(watcher);
+    file.writeAsStringSync('first');
+    await _waitFor(
+      () => watch1.changes.isNotEmpty && watch2.changes.isNotEmpty,
+    );
+
+    // Cancel the second watcher before causing the second event.
+    await watch2.cancel();
+
+    file.writeAsStringSync('second');
+    await _waitFor(() => watch1.changes.length >= 2);
+
+    await watch1.cancel();
+
+    // Watcher 1 should have both events, and watcher 2 should have only the
+    // first one.
+    expect(watch1.changes.length, 2);
+    expect(watch2.changes.length, 1);
+  }
+
+  /// Test that we can reuse a [ResourceWatcher] to create a second subscription
+  /// at a different time.
+  test_multipleWatchers_staggeredSubscribe() async {
+    var filePath = path.join(tempPath, 'foo');
+    var file = getFile(filePath: filePath, exists: true);
+
+    // On Windows, package:watcher may miss modification events for files that
+    // were created just before we started watching (see
+    // https://github.com/dart-lang/watcher/issues/159), so add an artificial
+    // delay there to ensure we don't miss events until that issue is resolved.
+    if (provider is PhysicalResourceProvider && Platform.isWindows) {
+      await Future<void>.delayed(const Duration(seconds: 1));
+    }
+
+    var resource = provider.getResource(filePath);
+    var watcher = resource.watch();
+
+    var watch1 = await _subscribeToWatcher(watcher);
+    file.writeAsStringSync('first');
+    await _waitFor(() => watch1.changes.isNotEmpty);
+
+    var watch2 = await _subscribeToWatcher(watcher);
+    file.writeAsStringSync('second');
+    await _waitFor(() => watch2.changes.isNotEmpty);
+
+    await Future.wait([watch1.cancel(), watch2.cancel()]);
+
+    // Watcher 1 should have both events, and watcher 2 should have only the
+    // second.
+    expect(watch1.changes.length, 2);
+    expect(watch2.changes.length, 1);
+  }
+
   test_parent2() {
     File file = getFile(exists: true);
 
@@ -778,6 +890,61 @@ mixin FolderTestMixin implements FileSystemTestSupport {
     expect(folder.isOrContains(defaultFolderPath), isTrue);
   }
 
+  /// Test that we can reuse a [ResourceWatcher] to create a second subscription
+  /// and cancel it at a different time.
+  test_multipleWatchers_staggeredCancel() async {
+    var folderPath = path.join(tempPath, 'foo');
+    var fileInFolder =
+        getFile(filePath: path.join(folderPath, 'file'), exists: true);
+    var resource = provider.getResource(folderPath);
+    var watcher = resource.watch();
+
+    var watch1 = await _subscribeToWatcher(watcher);
+    var watch2 = await _subscribeToWatcher(watcher);
+    fileInFolder.writeAsStringSync('first');
+    await _waitFor(
+      () => watch1.changes.isNotEmpty && watch2.changes.isNotEmpty,
+    );
+
+    // Cancel the second watcher before causing the second event.
+    await watch2.cancel();
+
+    fileInFolder.writeAsStringSync('second');
+    await _waitFor(() => watch1.changes.length >= 2);
+
+    await watch1.cancel();
+
+    // Watcher 1 should have both events, and watcher 2 should have only the
+    // first one.
+    expect(watch1.changes.length, 2);
+    expect(watch2.changes.length, 1);
+  }
+
+  /// Test that we can reuse a [ResourceWatcher] to create a second subscription
+  /// at a different time.
+  test_multipleWatchers_staggeredSubscribe() async {
+    var folderPath = path.join(tempPath, 'foo');
+    var fileInFolder =
+        getFile(filePath: path.join(folderPath, 'file'), exists: true);
+    var resource = provider.getResource(folderPath);
+    var watcher = resource.watch();
+
+    var watch1 = await _subscribeToWatcher(watcher);
+    fileInFolder.writeAsStringSync('first');
+    await _waitFor(() => watch1.changes.isNotEmpty);
+
+    var watch2 = await _subscribeToWatcher(watcher);
+    fileInFolder.writeAsStringSync('second');
+    await _waitFor(() => watch2.changes.isNotEmpty);
+
+    await Future.wait([watch1.cancel(), watch2.cancel()]);
+
+    // Watcher 1 should have both events, and watcher 2 should have only the
+    // second.
+    expect(watch1.changes.length, 2);
+    expect(watch2.changes.length, 1);
+  }
+
   test_parent() {
     Folder folder = getFolder(exists: true);
 

pkg/analyzer/test/file_system/physical_resource_provider_watch_test.dart

@@ -2,6 +2,7 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
+import 'dart:async';
 import 'dart:io' as io;
 
 import 'package:analyzer/file_system/file_system.dart';