Improve streaming_sync_tests.

Author: rkistner

packages/powersync/test/streaming_sync_test.dart

@@ -44,12 +44,8 @@ void main() {
         var server = await createServer();
 
         credentialsCallback() async {
-          final endpoint = 'http://${server.address.host}:${server.port}';
           return PowerSyncCredentials(
-              endpoint: endpoint,
-              token: 'token',
-              userId: 'u1',
-              expiresAt: DateTime.now());
+              endpoint: server.endpoint, token: 'token');
         }
 
         final pdb = await setupPowerSync(path: path);
@@ -59,12 +55,12 @@ void main() {
 
         await Future.delayed(Duration(milliseconds: random.nextInt(100)));
         if (random.nextBool()) {
-          server.close(force: true).ignore();
+          server.close();
         }
 
         await pdb.close();
 
-        server.close(force: true).ignore();
+        server.close();
       }
     });
 
@@ -81,18 +77,13 @@ void main() {
       // [PowerSync] WARNING: 2023-06-29 16:10:17.667537: Sync Isolate error
       // [Connection closed while receiving data, #0      IOClient.send.<anonymous closure> (package:http/src/io_client.dart:76:13)
 
-      HttpServer? server;
+      TestServer? server;
 
       credentialsCallback() async {
         if (server == null) {
           throw AssertionError('No active server');
         }
-        final endpoint = 'http://${server.address.host}:${server.port}';
-        return PowerSyncCredentials(
-            endpoint: endpoint,
-            token: 'token',
-            userId: 'u1',
-            expiresAt: DateTime.now());
+        return PowerSyncCredentials(endpoint: server.endpoint, token: 'token');
       }
 
       final pdb = await setupPowerSync(path: path);
@@ -107,7 +98,7 @@ void main() {
         // 2ms: HttpException: HttpServer is not bound to a socket
         // 20ms: Connection closed while receiving data
         await Future.delayed(Duration(milliseconds: 20));
-        server.close(force: true).ignore();
+        server.close();
       }
       await pdb.close();
     });

packages/powersync/test/test_server.dart

@@ -1,18 +1,67 @@
 import 'dart:async';
 import 'dart:convert' as convert;
 import 'dart:io';
+import 'dart:math';
 
 import 'package:http/http.dart' show ByteStream;
 import 'package:shelf/shelf.dart';
 import 'package:shelf/shelf_io.dart' as shelf_io;
 import 'package:shelf_router/shelf_router.dart';
 
-Future<HttpServer> createServer() async {
-  var app = Router();
+class TestServer {
+  late HttpServer server;
+  Router app = Router();
+  int connectionCount = 0;
+  int maxConnectionCount = 0;
+  int tokenExpiresIn;
 
-  app.post('/sync/stream', handleSyncStream);
-  // Open on an arbitrary open port
-  var server = await shelf_io.serve(app.call, 'localhost', 0);
+  TestServer({this.tokenExpiresIn = 65});
+
+  Future<void> init() async {
+    app.post('/sync/stream', handleSyncStream);
+    // Open on an arbitrary open port
+    server = await shelf_io.serve(app.call, 'localhost', 0);
+  }
+
+  String get endpoint {
+    return 'http://${server.address.host}:${server.port}';
+  }
+
+  Future<Response> handleSyncStream(Request request) async {
+    connectionCount += 1;
+    maxConnectionCount = max(connectionCount, maxConnectionCount);
+
+    stream() async* {
+      try {
+        var blob = "*" * 5000;
+        for (var i = 0; i < 50; i++) {
+          yield {"token_expires_in": tokenExpiresIn, "blob": blob};
+          await Future.delayed(Duration(microseconds: 1));
+        }
+      } finally {
+        connectionCount -= 1;
+      }
+    }
+
+    return Response.ok(
+      encodeNdjson(stream()),
+      headers: {
+        'Content-Type': 'application/x-ndjson',
+      },
+      context: {
+        'shelf.io.buffer_output': false,
+      },
+    );
+  }
+
+  void close() {
+    server.close(force: true).ignore();
+  }
+}
+
+Future<TestServer> createServer() async {
+  var server = TestServer();
+  await server.init();
   return server;
 }
 
@@ -22,23 +71,3 @@ ByteStream encodeNdjson(Stream<Object> jsonInput) {
   final byteInput = stringInput.transform(convert.utf8.encoder);
   return ByteStream(byteInput);
 }
-
-Future<Response> handleSyncStream(Request request) async {
-  stream() async* {
-    var blob = "*" * 5000;
-    for (var i = 0; i < 50; i++) {
-      yield {"token_expires_in": 5, "blob": blob};
-      await Future.delayed(Duration(microseconds: 1));
-    }
-  }
-
-  return Response.ok(
-    encodeNdjson(stream()),
-    headers: {
-      'Content-Type': 'application/x-ndjson',
-    },
-    context: {
-      'shelf.io.buffer_output': false,
-    },
-  );
-}