fix(sql_connect): e2e listen test refactor

packages/firebase_data_connect/firebase_data_connect/example/integration_test/listen_e2e.dart

@@ -87,76 +87,90 @@ void runListenTests() {
         expect(initialValue.data.movies.length, 0,
             reason: 'Initial movie list should be empty');
 
-        final Completer<void> isReady = Completer<void>();
-        final Completer<bool> hasBeenListened = Completer<bool>();
-        int count = 0;
+        final listener1Ready = Completer<void>();
+        final listener2Ready = Completer<void>();
+        final listener1ReceivedUpdate = Completer<void>();
+        final listener2ReceivedUpdate = Completer<void>();
+
+        int count1 = 0;
+        int count2 = 0;
 
         final listener1 = MoviesConnector.instance
             .listMovies()
             .ref()
             .subscribe()
             .listen((value) {
-          final movies = value.data.movies;
-
-          if (count == 0) {
-            expect(movies.length, 0,
-                reason: 'First emission should contain an empty list');
-            isReady.complete();
-          } else {
-            expect(movies.length, 1,
-                reason: 'Second emission should contain one movie');
-            expect(movies[0].title, 'The Matrix',
-                reason: 'The movie should be The Matrix');
-            hasBeenListened.complete(true);
+          count1++;
+          if (count1 == 1 && !listener1Ready.isCompleted) {
+            listener1Ready.complete();
+          } else if (count1 == 2 && !listener1ReceivedUpdate.isCompleted) {
+            listener1ReceivedUpdate.complete();
           }
-          count++;
         });
-        int listener2Count = 0;
+
         final listener2 = MoviesConnector.instance
             .listMovies()
             .ref()
             .subscribe()
             .listen((value) {
-          listener2Count++;
+          count2++;
+          if (count2 == 1 && !listener2Ready.isCompleted) {
+            listener2Ready.complete();
+          } else if (count2 == 3 && !listener2ReceivedUpdate.isCompleted) {
+            listener2ReceivedUpdate.complete();
+          }
         });
 
-        // Wait for the listener to be ready
-        await isReady.future;
-
-        // Create the movie
-        await MoviesConnector.instance
-            .createMovie(
-              genre: 'Action',
-              title: 'The Matrix',
-              releaseYear: 1999,
-            )
-            .rating(4.5)
-            .ref()
-            .execute();
-
-        await MoviesConnector.instance.listMovies().ref().execute();
-
-        // Wait for the listener to receive the movie update
-        final bool hasListenerReceived = await hasBeenListened.future;
-
-        // Cancel the listener and wait for it to finish
-        await listener1.cancel();
-        expect(hasListenerReceived, isTrue,
-            reason: 'The stream should have emitted new data');
-        // Create the movie
-        await MoviesConnector.instance
-            .createMovie(
-              genre: 'Adventure',
-              title: 'Raiders of the Lost Arc',
-              releaseYear: 1999,
-            )
-            .rating(4.5)
-            .ref()
-            .execute();
-        await Future.delayed(const Duration(seconds: 5));
-        expect(count, equals(2));
-        expect(listener2Count, equals(3));
-        await listener2.cancel();
+        try {
+          // Wait for both listeners to be ready with initial emission
+          await Future.wait([
+            listener1Ready.future,
+            listener2Ready.future,
+          ]).timeout(_listenTimeout);
+
+          // Create first movie
+          await MoviesConnector.instance
+              .createMovie(
+                genre: 'Action',
+                title: 'The Matrix',
+                releaseYear: 1999,
+              )
+              .rating(4.5)
+              .ref()
+              .execute();
+
+          await MoviesConnector.instance.listMovies().ref().execute();
+
+          // Wait for listener1 to receive the update
+          await listener1ReceivedUpdate.future.timeout(_listenTimeout);
+
+          // Cancel listener1
+          await listener1.cancel();
+
+          // Create second movie
+          await MoviesConnector.instance
+              .createMovie(
+                genre: 'Adventure',
+                title: 'Raiders of the Lost Arc',
+                releaseYear: 1999,
+              )
+              .rating(4.5)
+              .ref()
+              .execute();
+
+          await MoviesConnector.instance.listMovies().ref().execute();
+
+          // Wait deterministically for listener2's 3rd emission
+          await listener2ReceivedUpdate.future.timeout(_listenTimeout);
+
+          expect(count1, equals(2),
+              reason: 'Canceled listener should not receive further updates');
+          expect(count2, equals(3),
+              reason: 'Active listener should receive all updates');
+        } finally {
+          await listener1.cancel();
+          await listener2.cancel();
+        }
       });
     },
   );