DdsExtension.onEventWithHistory implementation seems incorrect

[mraleph]

The implementation does the following intermixing of async and sync code:

      final history = (await getStreamHistory(stream)).history;
      Event? firstStreamEvent;
      unawaited(streamEvents.peek.then((e) {
        firstStreamEvent = e;
      }));
      for (final event in history) {
        if (firstStreamEvent != null &&
            event.timestamp! > firstStreamEvent!.timestamp!) {
          break;
        }

Future.then is never executed synchronously (at least not for builtin implementation of Future) - which means firstStreamEvent is always null and the code does not do what it is expected to do.

Also the way code is written it leaves any errors thrown by peek unhandled instead of properly forwarding them into Stream returned by onEventWithHistory.

This was revealed when I tried adding code to properly close event streams when underlying VmService connection goes away. I have now undone this change because onEventWithHistory needs to be fixed first.

diff --git a/pkg/vm_service/lib/src/vm_service.dart b/pkg/vm_service/lib/src/vm_service.dart
index e14557b3bf3..a32c9f4915f 100644
--- a/pkg/vm_service/lib/src/vm_service.dart
+++ b/pkg/vm_service/lib/src/vm_service.dart
@@ -1890,6 +1890,9 @@ class VmService {
         'Service connection disposed',
       ));
     });
+    for (var controller in _eventControllers.values) {
+      await controller.close();
+    }
     _outstandingRequests.clear();
     final handler = _disposeHandler;
     if (handler != null) {

[bkonyi]

@derekxu16 can you take a look?