[ddc] Update e2e breakpoint pause detection

Make pulling both events from the stream more explicit and
ensure that different errors appear when each event doesn't
arrive in time.

There are several issues with the previous approach:
- `.skip(1)` actually returns a new stream that will skip the
  first event.
- Calling `.timeout()` on the new stream returns yet another
  stream with the timeout.
- It appeared like each event should have a timeout but the
  ordering means both the skip and first event actually shared
  a single timeout.

Change-Id: I60062634a35424455c03b6e1778cc56bbdeb1716
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/294080
Reviewed-by: Mark Zhou <markzipan@google.com>
Commit-Queue: Nicholas Shahan <nshahan@google.com>

pkg/dev_compiler/test/expression_compiler/expression_compiler_e2e_suite.dart

@@ -7,6 +7,7 @@ import 'dart:convert';
 import 'dart:io' show Directory, File, Platform, FileSystemException;
 import 'dart:math';
 
+import 'package:async/async.dart';
 import 'package:browser_launcher/browser_launcher.dart' as browser;
 import 'package:dev_compiler/src/compiler/module_builder.dart';
 import 'package:dev_compiler/src/compiler/shared_command.dart'
@@ -555,23 +556,23 @@ class TestDriver {
     var location = await _jsLocationFromDartLine(script, dartLine);
 
     var bp = await debugger.setBreakpoint(location);
+    final pauseQueue = StreamQueue(pauseController.stream);
     try {
       // Continue to the next breakpoint, ignoring the first pause event
       // since it corresponds to the preemptive URI breakpoint made prior
       // to page navigation.
       await debugger.resume();
-      final event = await pauseController.stream
-          .skip(1)
-          .timeout(Duration(seconds: 5),
-              onTimeout: (event) => throw Exception(
-                  'Unable to find JS preemptive pause event in $output.'))
-          .first
-          .timeout(Duration(seconds: 5),
-              onTimeout: (() => throw Exception(
-                  'Unable to find JS pause event corresponding to line '
-                  '($dartLine -> $location) in $output.')));
+      await pauseQueue.next.timeout(Duration(seconds: 5),
+          onTimeout: () => throw Exception(
+              'Unable to find JS preemptive pause event in $output.'));
+      final event = await pauseQueue.next.timeout(Duration(seconds: 5),
+          onTimeout: () => throw Exception(
+              'Unable to find JS pause event corresponding to line '
+              '($dartLine -> $location) in $output.'));
+
       return await onPause(event);
     } finally {
+      await pauseQueue.cancel();
       await pauseSub.cancel();
       await pauseController.close();
       await consoleSub.cancel();