diff --git a/pkg/analyzer/lib/file_system/file_system.dart b/pkg/analyzer/lib/file_system/file_system.dart index 588e17e49f3..306034b82c3 100644 --- a/pkg/analyzer/lib/file_system/file_system.dart +++ b/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 changes; + /// A function to obtain the to current [ready] [Future] from the underlying + /// watcher. + final Future 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 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 get ready => _ready(); } extension FolderExtension on Folder { diff --git a/pkg/analyzer/lib/file_system/memory_file_system.dart b/pkg/analyzer/lib/file_system/memory_file_system.dart index 09c61eb0683..c963949e5b2 100644 --- a/pkg/analyzer/lib/file_system/memory_file_system.dart +++ b/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(); + final streamController = StreamController.broadcast(); final ready = Completer(); /// 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); } } diff --git a/pkg/analyzer/lib/file_system/physical_file_system.dart b/pkg/analyzer/lib/file_system/physical_file_system.dart index c87286a88b3..9aeee9e9ceb 100644 --- a/pkg/analyzer/lib/file_system/physical_file_system.dart +++ b/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.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` to map all known errors through - /// [_wrapException] into server types. - Stream _wrapWatcherStream(Stream 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(); - final subscription = original.listen( - mappedEventsController.add, - onError: (Object e) => mappedEventsController.addError(mapException(e)), - onDone: () => mappedEventsController.close(), - ); - mappedEventsController.onCancel = subscription.cancel; - return mappedEventsController.stream; - } } diff --git a/pkg/analyzer/test/file_system/file_system_test_support.dart b/pkg/analyzer/test/file_system/file_system_test_support.dart index 0a8035b49f2..6441f435492 100644 --- a/pkg/analyzer/test/file_system/file_system_test_support.dart +++ b/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(); final isFileSystemException = TypeMatcher(); @@ -16,6 +20,11 @@ final isFolder = TypeMatcher(); final throwsFileSystemException = throwsA(isFileSystemException); +typedef _WatcherChanges = ({ + List changes, + Future 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 = []; + 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 _waitFor( + FutureOr 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.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.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); diff --git a/pkg/analyzer/test/file_system/physical_resource_provider_watch_test.dart b/pkg/analyzer/test/file_system/physical_resource_provider_watch_test.dart index 08584bd66d9..3d0512169b8 100644 --- a/pkg/analyzer/test/file_system/physical_resource_provider_watch_test.dart +++ b/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';