Work around dart-lang/sdk#45645

Author: nex3

lib/src/executable/repl.dart

@@ -14,6 +14,7 @@ import '../import_cache.dart';
 import '../importer/filesystem.dart';
 import '../logger/tracking.dart';
 import '../parse/parser.dart';
+import '../utils.dart';
 import '../visitor/evaluate.dart';
 
 /// Runs an interactive SassScript shell according to [options].
@@ -24,12 +25,14 @@ Future<void> repl(ExecutableOptions options) async {
       importer: FilesystemImporter('.'),
       importCache: ImportCache(loadPaths: options.loadPaths, logger: logger),
       logger: logger);
-  await for (String line in repl.runAsync()) {
-    if (line.trim().isEmpty) continue;
+
+  // Use forEachCancelable to work around dart-lang/sdk#45645.
+  await repl.runAsync().forEachCancelable((line) {
+    if (line.trim().isEmpty) return;
     try {
       if (line.startsWith("@")) {
         evaluator.use(UseRule.parse(line, logger: logger));
-        continue;
+        return;
       }
 
       if (Parser.isVariableDeclarationLike(line)) {
@@ -44,7 +47,7 @@ Future<void> repl(ExecutableOptions options) async {
     } on SassException catch (error, stackTrace) {
       _logError(error, stackTrace, line, repl, options, logger);
     }
-  }
+  }).value;
 }
 
 /// Logs an error from the interactive shell.

lib/src/executable/watch.dart

@@ -4,6 +4,7 @@
 
 import 'dart:collection';
 
+import 'package:async/async.dart';
 import 'package:path/path.dart' as p;
 import 'package:stack_trace/stack_trace.dart';
 import 'package:stream_transform/stream_transform.dart';
@@ -14,6 +15,7 @@ import '../importer/filesystem.dart';
 import '../io.dart';
 import '../stylesheet_graph.dart';
 import '../util/multi_dir_watcher.dart';
+import '../utils.dart';
 import 'compile_stylesheet.dart';
 import 'options.dart';
 
@@ -123,30 +125,33 @@ class _Watcher {
   /// Listens to `watcher.events` and updates the filesystem accordingly.
   ///
   /// Returns a future that will only complete if an unexpected error occurs.
-  Future<void> watch(MultiDirWatcher watcher) async {
-    await for (var event in _debounceEvents(watcher.events)) {
+  Future<void> watch(MultiDirWatcher watcher) {
+    late CancelableOperation<void> operation;
+    operation =
+        _debounceEvents(watcher.events).forEachCancelable((event) async {
       var extension = p.extension(event.path);
       if (extension != '.sass' && extension != '.scss' && extension != '.css') {
-        continue;
+        return;
       }
 
       switch (event.type) {
         case ChangeType.MODIFY:
           var success = await _handleModify(event.path);
-          if (!success && _options.stopOnError) return;
+          if (!success && _options.stopOnError) operation.cancel();
           break;
 
         case ChangeType.ADD:
           var success = await _handleAdd(event.path);
-          if (!success && _options.stopOnError) return;
+          if (!success && _options.stopOnError) operation.cancel();
           break;
 
         case ChangeType.REMOVE:
           var success = await _handleRemove(event.path);
-          if (!success && _options.stopOnError) return;
+          if (!success && _options.stopOnError) operation.cancel();
           break;
       }
-    }
+    });
+    return operation.valueOrCancellation();
   }
 
   /// Handles a modify event for the stylesheet at [path].

lib/src/utils.dart

@@ -2,8 +2,10 @@
 // MIT-style license that can be found in the LICENSE file or at
 // https://opensource.org/licenses/MIT.
 
+import 'dart:async';
 import 'dart:math' as math;
 
+import 'package:async/async.dart';
 import 'package:charcode/charcode.dart';
 import 'package:collection/collection.dart';
 import 'package:source_span/source_span.dart';
@@ -403,3 +405,29 @@ extension SpanExtensions on FileSpan {
         : file.span(this.start.offset + start, this.start.offset + end + 1);
   }
 }
+
+extension StreamExtension<T> on Stream<T> {
+  /// Like [Stream.forEach], but returns a [CancelableOperation] that cancels
+  /// the underlying stream subscription once it's cancelled.
+  ///
+  /// This also fixed dart-lang/sdk#45645.
+  CancelableOperation<void> forEachCancelable(
+      FutureOr<void> callback(T element)) {
+    var subscription = listen(null);
+    var completer = CancelableCompleter<void>(onCancel: subscription.cancel);
+    subscription.onData((event) async {
+      subscription.pause();
+      try {
+        await callback(event);
+        subscription.resume();
+      } catch (error, stackTrace) {
+        subscription.cancel();
+        completer.completeError(error, stackTrace);
+      }
+    });
+    subscription.onError(completer.completeError);
+    subscription.onDone(completer.complete);
+
+    return completer.operation;
+  }
+}