wip: mock server

Author: stevensJourney

packages/powersync/test/connected_test.dart

@@ -0,0 +1,63 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:http/http.dart' as http;
+import 'package:test/test.dart';
+
+import 'server/sync_server/mock_sync_server.dart';
+
+void main() {
+  late TestHttpServerHelper testServer;
+
+  setUp(() async {
+    testServer = TestHttpServerHelper();
+    await testServer.start();
+  });
+
+  tearDown(() async {
+    await testServer.stop();
+  });
+
+  test('should receive events from the sync stream without waiting for close',
+      () async {
+    final client = http.Client();
+    final request =
+        http.Request('POST', testServer.uri.replace(path: '/sync/stream'));
+    request.headers['Content-Type'] = 'application/json';
+
+    // Send the request and get the response stream
+    final responseStream = await client.send(request);
+
+    final expectedEvents = ['event1', 'event2', 'event3'];
+    final receivedEvents = <String>[];
+    final completer = Completer<void>();
+
+    // Listen to the response stream for real-time processing of incoming events
+    final subscription = responseStream.stream
+        .transform(utf8.decoder)
+        .transform(LineSplitter())
+        .listen(
+      (event) {
+        receivedEvents.add(event);
+        if (receivedEvents.length == expectedEvents.length) {
+          completer.complete(); // Complete once all events are received
+        }
+      },
+      onError: (e) => completer.completeError(e),
+    );
+
+    // Programmatically trigger events on the server
+    for (final event in expectedEvents) {
+      testServer.addEvent('$event\n');
+      await Future.delayed(
+          Duration(milliseconds: 100)); // Small delay for each event
+    }
+
+    // Wait for the events to be received
+    await completer.future.timeout(Duration(seconds: 5));
+    await subscription.cancel();
+    client.close();
+
+    expect(receivedEvents.toSet().containsAll(expectedEvents.toSet()), isTrue);
+  });
+}

packages/powersync/test/server/sync_server/mock_sync_server.dart

@@ -0,0 +1,57 @@
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:shelf/shelf.dart';
+import 'package:shelf/shelf_io.dart' as io;
+import 'package:shelf_router/shelf_router.dart';
+
+class TestHttpServerHelper {
+  final StreamController<String> _controller = StreamController<String>();
+  late HttpServer _server;
+  // late Timer _timer;
+
+  Uri get uri => Uri.parse('http://localhost:${_server.port}');
+
+  // Start the HTTP server
+  Future<void> start() async {
+    final router = Router()
+      ..post('/sync/stream', (Request request) async {
+        // Respond immediately with a stream
+        return Response.ok(_controller.stream.transform(utf8.encoder),
+            headers: {
+              'Content-Type': 'text/event-stream',
+              'Cache-Control': 'no-cache',
+              'Connection': 'keep-alive',
+              'Transfer-Encoding': 'identity', // Use chunked transfer encoding
+            },
+            context: {
+              "shelf.io.buffer_output": false
+            });
+      });
+
+    // Sending a newline gets the stream going and resolving on the client side
+    _controller.add("\n");
+
+    // // Add data. The mock stream does not seem to resolve without data
+    // _timer = Timer.periodic(Duration(seconds: 1), (Timer timer) {
+    //   // This code will execute every second
+    //   _controller.add('{ "token_expires_in": 3600}\n');
+    // });
+
+    _server = await io.serve(router, 'localhost', 0);
+    print('Test server running at ${_server.address}:${_server.port}');
+  }
+
+  // Programmatically add data to the stream
+  void addEvent(String data) {
+    _controller.add(data);
+  }
+
+  // Stop the HTTP server
+  Future<void> stop() async {
+    // _timer.cancel();
+    await _controller.close();
+    await _server.close();
+  }
+}