diff --git a/CHANGELOG.md b/CHANGELOG.md index 96da65a..4d1e718 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,33 @@ +# 4.2.0 + +DDP session resumption, matching +[meteor/meteor#14051](https://github.com/meteor/meteor/pull/14051) (merged +2026-03-06). + +Added: +- After an unexpected disconnect the client keeps its DDP session id and asks + the server to resume it, sending `session` and `receivedCount` in `connect`. + When the server agrees (same session id back) the login and subscriptions + carry on untouched, messages published during the gap arrive in order, and + `onReconnect` callbacks are not run. When it does not, the reconnect + proceeds exactly as before: callbacks and re-subscribe. +- `DdpClient.resumedSession` reports whether the latest `connected` resumed + the previous session; `DdpClient.receivedCount` exposes the message count. +- `disconnect()` sends a DDP `disconnect` message before closing the socket so + the server drops the session at once instead of keeping it for the grace + period. + +Changed: +- In-flight method calls are failed once the reconnect completes (resumed or + not), on an explicit `disconnect()`, or when the retry limit is reached, + rather than the instant the socket drops. They still fail: a request sent + just before the drop may never have reached the server, and that cannot be + told apart from a slow method, so hanging forever is not an option. + +Compatibility: +- Servers without session resumption ignore the extra fields and start a new + session, so behaviour against them is unchanged. + # 4.1.0 Connection lifecycle fixes. The theme is a device that goes to sleep: the diff --git a/README.md b/README.md index b92df4e..6cd47cc 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ Add the package to your `pubspec.yaml`: ```yaml dependencies: - dart_meteor: ^4.1.0 + dart_meteor: ^4.2.0 ``` ## Quick start @@ -259,6 +259,35 @@ attempts (0s, 5s, 10s, … up to `maxRetryInterval`) so an unreachable server does not keep the radio busy. `disconnect()` is final: the client stays offline until you call `reconnect()`. +### Session resumption + +Meteor servers that include [meteor/meteor#14051](https://github.com/meteor/meteor/pull/14051) +keep a session alive for a grace period (15 s by default, +`Meteor.server.options.disconnectGracePeriod`) after an ungraceful disconnect. +The client asks to resume that session on reconnect, sending its DDP session id +and the number of messages it has received so far. If the server still has the +session and nothing was lost in between, the reconnect is seamless: + +- the login is still in place — no resume-token round trip; +- subscriptions are not re-sent, and documents published while the client was + away are delivered in order on the new socket; +- `onConnection` does not fire again on the server, and the connection id is + unchanged. + +If the server cannot resume (grace period expired, a message was lost, the +server restarted, or it predates that change), it starts a new session and the +client falls back to the usual reconnect: re-login and re-subscribe. +`meteor.connection.resumedSession` tells you which happened after each +reconnect. + +Method calls that were in flight fail with `MeteorConnectionError` in both +cases. A request written just before the socket dropped may never have reached +the server, and the client cannot distinguish that from a slow method, so it +reports the uncertainty instead of waiting forever. + +`disconnect()` sends a DDP `disconnect` message first, so the server frees the +session immediately instead of holding it open for the grace period. + The timings are configurable if the defaults do not suit your server: ```dart @@ -327,9 +356,11 @@ same fields you get in a Meteor web client. A call that was still in flight when the connection dropped — because the device slept, or the network went away — throws `MeteorConnectionError` -instead. The two are worth distinguishing: `MeteorError` means the server -considered the request and said no, while `MeteorConnectionError` means you -never heard back and the method may or may not have run. +(whether or not the session is then [resumed](#session-resumption)). The two +errors are worth distinguishing: +`MeteorError` means the server considered the request and said no, while +`MeteorConnectionError` means you never heard back and the method may or may +not have run. ```dart try { @@ -349,6 +380,11 @@ Calls are not resent automatically after a reconnect: a method like See [CHANGELOG.md](CHANGELOG.md) for the full history. The notable breaking changes: +- **4.2.0** — DDP session resumption. Against a server with + [meteor/meteor#14051](https://github.com/meteor/meteor/pull/14051) a brief + network drop no longer re-runs the login or re-sends subscriptions, and + `onReconnect` callbacks are not invoked on a resumed session. Older servers + behave exactly as before. - **4.1.0** — two behaviour changes worth knowing about, both fixes. A method call that is in flight when the connection drops now throws `MeteorConnectionError` instead of hanging forever, so `await meteor.call(…)` diff --git a/example/.gitignore b/example/.gitignore index 00d30e5..ee9d2a0 100644 --- a/example/.gitignore +++ b/example/.gitignore @@ -49,3 +49,5 @@ app.*.map.json # FVM Version Cache .fvm/ + +.wrangler/ diff --git a/lib/src/ddp_client.dart b/lib/src/ddp_client.dart index 92d8fb6..88d13cd 100644 --- a/lib/src/ddp_client.dart +++ b/lib/src/ddp_client.dart @@ -138,6 +138,16 @@ class DdpClient { final Map _onReconnectCallbacks = {}; String? serverId; String? sessionId; + + /// Number of DDP messages received from the server in the current session, + /// excluding `ping`/`pong` (and the pre-session `server_id` frame). Sent with + /// `connect` so a server that supports session resumption (Meteor PR #14051) + /// can verify nothing was lost while we were away. + int _receivedCount = 0; + + /// Whether the most recent `connected` message resumed the previous session + /// rather than starting a new one. + bool _lastConnectResumedSession = false; int _currentMethodId = 0; Timer? _pingPeriodicTimer; Timer? _pongTimeoutTimer; @@ -162,7 +172,8 @@ class DdpClient { Duration? stalenessThreshold, }) : pingInterval = pingInterval ?? const Duration(seconds: pingIntervalSeconds), - pongTimeout = pongTimeout ?? const Duration(seconds: pongTimeoutSeconds), + pongTimeout = + pongTimeout ?? const Duration(seconds: pongTimeoutSeconds), maxRetryInterval = maxRetryInterval ?? const Duration(seconds: 30), stalenessThreshold = stalenessThreshold ?? Duration( @@ -299,9 +310,8 @@ class DdpClient { } if (_connectionStatus.status == DdpConnectionStatusValues.connected) { var last = _lastMessageReceivedAt; - var silentFor = last == null - ? stalenessThreshold - : DateTime.now().difference(last); + var silentFor = + last == null ? stalenessThreshold : DateTime.now().difference(last); if (silentFor >= stalenessThreshold) { printDebug( 'Connection considered stale - nothing received for $silentFor', @@ -317,13 +327,25 @@ class DdpClient { _connect(); } + /// `true` if the last `connected` message from the server resumed the + /// previous DDP session (same session id, no data lost), `false` if it + /// started a fresh one. Only meaningful while connected. + bool get resumedSession => _lastConnectResumedSession; + + /// Messages received from the server in this session, excluding ping/pong. + /// Exposed for tests and diagnostics. + int get receivedCount => _receivedCount; + /// Disconnect the client from the server. The client stays offline until /// [reconnect] is called. void disconnect() { printDebug('Begin of disconnect()'); _isTryToReconnect = false; _cancelScheduledReconnect(); - _teardownConnection('Disconnected by the client'); + // Tell the server this is intentional so it drops the session right away + // instead of holding it open for the resumption grace period. + _sendMsgDisconnect(); + _teardownConnection('Disconnected by the client', keepSession: false); _connectionStatus.retryCount = 0; _connectionStatus.connected = false; _connectionStatus.status = DdpConnectionStatusValues.offline; @@ -333,9 +355,14 @@ class DdpClient { } /// Close the current socket and release everything attached to it, without - /// deciding whether to reconnect. [reason] is reported to any in-flight - /// method calls. - void _teardownConnection(String reason) { + /// deciding whether to reconnect. + /// + /// With [keepSession] the session id and message count survive so the next + /// `connect` can try to resume the session. In-flight method calls are kept + /// until the outcome of that reconnect is known and failed then (see + /// [_onConnected]); without [keepSession] everything is forgotten at once + /// and [reason] is reported to them immediately. + void _teardownConnection(String reason, {required bool keepSession}) { _pingPeriodicTimer?.cancel(); _pingPeriodicTimer = null; _pongTimeoutTimer?.cancel(); @@ -356,9 +383,17 @@ class DdpClient { } serverId = null; - sessionId = null; _lastMessageReceivedAt = null; - _failAllPendingMethodCalls(reason); + if (!keepSession) { + _forgetSession(); + _failAllPendingMethodCalls(reason); + } + } + + void _forgetSession() { + sessionId = null; + _receivedCount = 0; + _lastConnectResumedSession = false; } /// Handle a connection that dropped on its own (socket closed, error, or a @@ -370,7 +405,7 @@ class DdpClient { return; } printDebug('Connection lost: $reason'); - _teardownConnection(reason); + _teardownConnection(reason, keepSession: true); _connectionStatus.connected = false; _connectionStatus.status = DdpConnectionStatusValues.offline; _connectionStatus.reason = reason; @@ -491,6 +526,7 @@ class DdpClient { _connectionStatus.status = DdpConnectionStatusValues.failed; _connectionStatus.reason = 'DDP. Reach max retry attempt'; _emitStatus(); + _failAllPendingMethodCalls('DDP. Reach max retry attempt'); } } @@ -511,7 +547,11 @@ class DdpClient { 'support': ['1', 'pre1', 'pre2'], }; if (sessionId != null) { + // Ask to resume. The server only does so if it still has the session + // and its sent count equals our received count; otherwise it starts + // a new session. Servers without resumption support ignore both. data['session'] = sessionId!; + data['receivedCount'] = _receivedCount; } var msg = json.encode(data); printDebug('Send: $msg'); @@ -519,6 +559,19 @@ class DdpClient { } } + void _sendMsgDisconnect() { + var socket = _socket; + if (socket != null && _connectionStatus.connected) { + var msg = json.encode({'msg': 'disconnect'}); + printDebug('Send: $msg'); + try { + socket.sink.add(msg); + } catch (err) { + printDebug('Failed to send disconnect: $err'); + } + } + } + /// Re-send every live subscription on a freshly established connection. /// Called only once the server has replied `connected` and the reconnect /// callbacks (in practice, the login resume) have finished. @@ -601,17 +654,30 @@ class DdpClient { } } - /// Runs once the server accepted the connection: mark the client connected, - /// give the reconnect callbacks a chance to restore the login, then re-send - /// the subscriptions. + /// Runs once the server accepted the connection. + /// + /// If the server handed back the session id we asked to resume, the session + /// continues where it left off: the login and subscriptions are still live + /// on the server, so nothing is re-sent. Otherwise this is a new session: + /// mark the client connected, give the reconnect callbacks a chance to + /// restore the login, then re-send the subscriptions. + /// + /// In-flight method calls are failed either way. A request written to the + /// socket just before it dropped may never have reached the server, and + /// there is no way to tell that apart from a slow method - so rather than + /// leave the caller hanging forever, report it and let them retry. Future _onConnected(Map dataMap) async { + var newSessionId = dataMap['session']; + var resumed = sessionId != null && newSessionId == sessionId; + _lastConnectResumedSession = resumed; + sessionId = newSessionId; + _connectionStatus.connected = true; _connectionStatus.status = DdpConnectionStatusValues.connected; _connectionStatus.reason = null; _connectionStatus.retryCount = 0; _connectionStatus.retryTime = Duration(seconds: 0); _emitStatus(); - sessionId = dataMap['session']; _pingPeriodicTimer?.cancel(); _pingPeriodicTimer = Timer.periodic(pingInterval, (timer) { @@ -623,6 +689,19 @@ class DdpClient { _sendMsgPing(); }); + _failAllPendingMethodCalls(resumed + ? 'Connection dropped while the call was in flight' + : 'Connection was re-established as a new session'); + + if (resumed) { + printDebug('Resumed DDP session $sessionId'); + return; + } + + // New session: the 'connected' message itself is the first counted + // message. + _receivedCount = 1; + var callbacks = List.from( _onReconnectCallbacks.values, ); @@ -655,6 +734,11 @@ class DdpClient { _pongTimeoutTimer = null; return; } + if (msg != null) { + // Mirrors the server's sentCount: every real DDP message counts, the + // pre-session `server_id` frame (no `msg` field) and ping/pong do not. + _receivedCount++; + } if (_connectionStatus.status == DdpConnectionStatusValues.connecting) { if (dataMap['server_id'] != null) { serverId = dataMap['server_id']; @@ -665,7 +749,7 @@ class DdpClient { unawaited(_onConnected(dataMap)); } else if (msg == 'failed') { serverId = null; - sessionId = null; + _forgetSession(); _connectionStatus.connected = false; _connectionStatus.status = DdpConnectionStatusValues.failed; _connectionStatus.reason = @@ -790,7 +874,7 @@ class DdpClient { if (_isTryToReconnect) { _handleConnectionLost('The websocket was closed by the other side'); } else { - _teardownConnection('The websocket was closed'); + _teardownConnection('The websocket was closed', keepSession: false); } } @@ -798,7 +882,7 @@ class DdpClient { if (_isTryToReconnect) { _handleConnectionLost('Websocket error: $error'); } else { - _teardownConnection('Websocket error: $error'); + _teardownConnection('Websocket error: $error', keepSession: false); } } } diff --git a/pubspec.yaml b/pubspec.yaml index 9de17d0..31ea90b 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -1,6 +1,6 @@ name: dart_meteor description: This library make connection between meteor backend and flutter app easily. Design to work seamlessly with StreamBuilder and FutureBuilder. -version: 4.1.0 +version: 4.2.0 homepage: https://github.com/tanutapi/dart_meteor environment: diff --git a/test/ddp_mock_server_test.dart b/test/ddp_mock_server_test.dart index afe5554..c7527ec 100644 --- a/test/ddp_mock_server_test.dart +++ b/test/ddp_mock_server_test.dart @@ -43,7 +43,7 @@ void main() { test('completes the version 1 handshake and exposes ids', () async { expect(meteor.connection.serverId, '0'); - expect(meteor.connection.sessionId, 'mock-session-id'); + expect(meteor.connection.sessionId, 'mock-session-1'); }); test('method call returns the result', () async { diff --git a/test/lifecycle_test.dart b/test/lifecycle_test.dart index 802a13e..93769f2 100644 --- a/test/lifecycle_test.dart +++ b/test/lifecycle_test.dart @@ -68,7 +68,8 @@ void main() { expect(await meteor.call('methodThatReturnNumber'), 42); }, timeout: Timeout(Duration(seconds: 30))); - test('an in-flight method call fails instead of hanging when the socket drops', + test( + 'an in-flight method call fails instead of hanging when the socket drops', () async { server.silentMethods.add('neverReturns'); var future = meteor.call('neverReturns'); @@ -102,8 +103,8 @@ void main() { expect(server.connectionCount, greaterThan(socketsBefore)); var msgs = server.messagesOnLatestSocket; - var loginIndex = msgs.indexWhere( - (m) => m['msg'] == 'method' && m['method'] == 'login'); + var loginIndex = msgs + .indexWhere((m) => m['msg'] == 'method' && m['method'] == 'login'); var subIndex = msgs.indexWhere((m) => m['msg'] == 'sub'); expect(loginIndex, isNonNegative, reason: 'the resume login must be sent on the new socket'); @@ -154,6 +155,207 @@ void main() { }, timeout: Timeout(Duration(seconds: 30))); }); + group('session resumption (meteor/meteor#14051)', () { + late MockDdpServer server; + late MeteorClient meteor; + + setUp(() async { + server = MockDdpServer()..supportsResumption = true; + server.itemsCollection['doc1'] = {'title': 'First'}; + await server.start(); + meteor = _connectClient(server.port); + await _waitForConnected(meteor); + }); + + tearDown(() async { + meteor.disconnect(); + await server.stop(); + }); + + test('receivedCount counts every message except ping/pong', () async { + // Let a few client pings and server pongs go by. + await Future.delayed(Duration(seconds: 1)); + expect(meteor.connection.receivedCount, 1, + reason: 'only the connected message has been counted so far'); + expect(server.sessions.single.sentCount, 1); + + meteor.subscribe('items'); + await Future.delayed(Duration(milliseconds: 300)); + // added + ready + expect(meteor.connection.receivedCount, 3); + expect(meteor.connection.receivedCount, server.sessions.single.sentCount, + reason: 'client and server counts must stay in lockstep'); + }, timeout: Timeout(Duration(seconds: 30))); + + test('an unexpected drop resumes the session without re-login or re-sub', + () async { + await meteor.loginWithPassword('user1', 'password1'); + meteor.subscribe('items'); + await Future.delayed(Duration(milliseconds: 200)); + var sessionBefore = meteor.connection.sessionId; + var socketsBefore = server.connectionCount; + + server.closeAllSockets(); + await _waitForDisconnected(meteor); + await _waitForConnected(meteor); + await Future.delayed(Duration(milliseconds: 300)); + + expect(server.connectionCount, greaterThan(socketsBefore)); + expect(meteor.connection.sessionId, sessionBefore, + reason: 'the server must hand back the same session id'); + expect(meteor.connection.resumedSession, isTrue); + expect(server.sessionCount, 1, + reason: 'no new server session (onConnection) on resume'); + var msgs = server.messagesOnLatestSocket; + expect(msgs.first['msg'], 'connect'); + expect(msgs.first['session'], sessionBefore); + expect(msgs.first['receivedCount'], isA()); + expect(msgs.where((m) => m['msg'] == 'method'), isEmpty, + reason: 'the login must not be re-sent on a resumed session'); + expect(msgs.where((m) => m['msg'] == 'sub'), isEmpty, + reason: 'subscriptions must not be re-sent on a resumed session'); + expect(await meteor.userId().first, 'user1-id'); + // The session is still fully usable afterwards. + expect(await meteor.call('methodThatReturnNumber'), 42); + }, timeout: Timeout(Duration(seconds: 30))); + + test('messages published during the gap are delivered after the resume', + () async { + meteor.subscribe('items'); + await Future.delayed(Duration(milliseconds: 200)); + var items = >[]; + var sub = meteor.collection('items').listen(items.add); + + server.closeAllSockets(); + await _waitForDisconnected(meteor); + server.broadcast({ + 'msg': 'added', + 'collection': 'items', + 'id': 'doc2', + 'fields': {'title': 'Added while away'}, + }); + await _waitForConnected(meteor); + await Future.delayed(Duration(milliseconds: 300)); + await sub.cancel(); + + expect(meteor.connection.resumedSession, isTrue); + expect(items.last['doc2']?['title'], 'Added while away'); + expect(meteor.connection.receivedCount, server.sessions.single.sentCount); + }, timeout: Timeout(Duration(seconds: 30))); + + test('an in-flight method call fails on resume rather than hanging', + () async { + // The request may have been lost with the socket; the client cannot + // tell, so it must not leave the caller waiting forever. + server.silentMethods.add('neverReturns'); + var failed = expectLater( + meteor.call('neverReturns').timeout(Duration(seconds: 5)), + throwsA(isA())); + await Future.delayed(Duration(milliseconds: 100)); + server.closeAllSockets(); + await _waitForDisconnected(meteor); + await _waitForConnected(meteor); + + expect(meteor.connection.resumedSession, isTrue); + await failed; + }, timeout: Timeout(Duration(seconds: 30))); + + test('a message count mismatch starts a new session and re-subscribes', + () async { + await meteor.loginWithPassword('user1', 'password1'); + meteor.subscribe('items'); + await Future.delayed(Duration(milliseconds: 200)); + var sessionBefore = meteor.connection.sessionId; + server.silentMethods.add('neverReturns'); + var pending = expectLater( + meteor.call('neverReturns'), throwsA(isA())); + + server.closeAllSockets(); + await _waitForDisconnected(meteor); + // Pretend the server sent something the client never got. + server.sessions.single.sentCount++; + await _waitForConnected(meteor); + await Future.delayed(Duration(milliseconds: 300)); + + expect(meteor.connection.sessionId, isNot(sessionBefore)); + expect(meteor.connection.resumedSession, isFalse); + expect(server.sessionCount, 2); + expect(meteor.connection.receivedCount, server.sessions.single.sentCount); + var msgs = server.messagesOnLatestSocket; + expect(msgs.any((m) => m['msg'] == 'method' && m['method'] == 'login'), + isTrue, + reason: 'a new session must re-run the login'); + expect(msgs.any((m) => m['msg'] == 'sub'), isTrue, + reason: 'a new session must re-send subscriptions'); + await pending; + }, timeout: Timeout(Duration(seconds: 30))); + + test('an explicit disconnect tells the server and is never resumed', + () async { + var sessionBefore = meteor.connection.sessionId; + meteor.disconnect(); + await Future.delayed(Duration(milliseconds: 200)); + + expect(server.messagesOnLatestSocket.last['msg'], 'disconnect'); + expect(server.sessions, isEmpty, + reason: 'a graceful disconnect must free the session immediately'); + + meteor.reconnect(); + await _waitForConnected(meteor); + expect(meteor.connection.sessionId, isNot(sessionBefore)); + expect(meteor.connection.resumedSession, isFalse); + expect( + server.messagesOnLatestSocket.first.containsKey('session'), isFalse, + reason: 'after an explicit disconnect the client must not ask to ' + 'resume'); + }, timeout: Timeout(Duration(seconds: 30))); + + test('a server-initiated close is not resumed', () async { + var sessionBefore = meteor.connection.sessionId; + server.closeAllSessions(); + await _waitForDisconnected(meteor); + await _waitForConnected(meteor); + + expect(meteor.connection.sessionId, isNot(sessionBefore)); + expect(meteor.connection.resumedSession, isFalse); + expect(server.sessionCount, 2); + }, timeout: Timeout(Duration(seconds: 30))); + + test('a session whose grace period expired is not resumed', () async { + server.disconnectGracePeriod = Duration.zero; + var sessionBefore = meteor.connection.sessionId; + server.closeAllSockets(); + await _waitForDisconnected(meteor); + await _waitForConnected(meteor); + + expect(meteor.connection.sessionId, isNot(sessionBefore)); + expect(meteor.connection.resumedSession, isFalse); + expect(await meteor.call('methodThatReturnNumber'), 42); + }, timeout: Timeout(Duration(seconds: 30))); + + test('a server without resumption support still gets a working reconnect', + () async { + server.supportsResumption = false; + await meteor.loginWithPassword('user1', 'password1'); + meteor.subscribe('items'); + await Future.delayed(Duration(milliseconds: 200)); + var sessionBefore = meteor.connection.sessionId; + + server.closeAllSockets(); + await _waitForDisconnected(meteor); + await _waitForConnected(meteor); + await Future.delayed(Duration(milliseconds: 300)); + + // The client asked to resume, the server ignored it and started fresh. + expect(server.messagesOnLatestSocket.first['session'], sessionBefore); + expect(meteor.connection.sessionId, isNot(sessionBefore)); + expect(meteor.connection.resumedSession, isFalse); + expect( + server.messagesOnLatestSocket.any((m) => m['msg'] == 'sub'), isTrue); + expect(await meteor.userId().first, 'user1-id'); + }, timeout: Timeout(Duration(seconds: 30))); + }); + group('reconnect policy', () { test('a user-initiated disconnect is not undone by a pending retry', () async { @@ -180,8 +382,7 @@ void main() { // Accepts TCP but refuses the websocket upgrade, so every attempt fails // the same way and is countable. var attempts = 0; - var httpServer = - await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + var httpServer = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); httpServer.listen((req) async { attempts++; req.response.statusCode = HttpStatus.notFound; diff --git a/test/mock_ddp_server.dart b/test/mock_ddp_server.dart index cfa27da..4f0dab9 100644 --- a/test/mock_ddp_server.dart +++ b/test/mock_ddp_server.dart @@ -4,16 +4,68 @@ /// needs a real Meteor server or docker. library; +import 'dart:async'; import 'dart:convert'; import 'dart:io'; import 'package:crypto/crypto.dart'; +/// Server-side state of one DDP session, modelled on `Session` in Meteor's +/// `ddp-server/livedata_server.js` after session resumption landed +/// (meteor/meteor#14051). +class MockDdpSession { + MockDdpSession(this.id); + + final String id; + WebSocket? socket; + + /// Messages sent on this session excluding ping/pong, compared against the + /// client's `receivedCount` on reconnect. + int sentCount = 0; + + /// Non-null while the session is disconnected and waiting to be resumed. + List>? messageQueue; + Timer? removeTimer; + + /// Set by a client `disconnect` message; such a session is never resumed. + bool expectingDisconnect = false; + + /// Ids of the subscriptions the client has open on this session. + final Set subscriptionIds = {}; +} + class MockDdpServer { HttpServer? _httpServer; final List _sockets = []; + final Map _sessionBySocket = {}; + final Map _sessions = {}; + int _sessionCounter = 0; int? _boundPort; + /// When true the server behaves like Meteor with meteor/meteor#14051: an + /// ungracefully dropped session is kept for [disconnectGracePeriod] and + /// resumed if the client reconnects with the same session id and a matching + /// message count. When false (default) every connect starts a new session, + /// like Meteor releases before that change. + bool supportsResumption = false; + + /// How long a dropped session is kept around for resumption. + Duration disconnectGracePeriod = const Duration(seconds: 15); + + /// Messages queued for a dropped session before it is given up on. + int maxMessageQueueLength = 100; + + /// Session ids handed out by `connected`, in order. A resumed session + /// repeats the previous id. + final List sessionIdsSent = []; + + /// How many sessions were created (resumptions do not count), i.e. how many + /// times `onConnection` would have fired on a real server. + int sessionCount = 0; + + /// Sessions currently known to the server, live or awaiting resumption. + Iterable get sessions => _sessions.values; + /// The port the server is (or last was) bound to. Stable across a /// [stop]/[start] cycle so tests can simulate a server outage. int get port => _httpServer?.port ?? _boundPort!; @@ -40,6 +92,10 @@ class MockDdpServer { /// left with an in-flight call. final Set silentMethods = {}; + /// Methods named here are answered only after the given delay, so a result + /// can land while the client is disconnected. + final Map delayedMethods = {}; + /// How many websocket connections have been accepted over this server's /// lifetime, including across restarts. int connectionCount = 0; @@ -67,7 +123,7 @@ class MockDdpServer { // Meteor sends server_id as the very first message on the socket. socket.add(json.encode({'server_id': '0'})); socket.listen((data) => _onMessage(socket, inbox, data), - onDone: () => _sockets.remove(socket)); + onDone: () => _onSocketClosed(socket)); } else { req.response.statusCode = HttpStatus.notFound; await req.response.close(); @@ -80,18 +136,136 @@ class MockDdpServer { await socket.close(); } _sockets.clear(); + for (var session in _sessions.values) { + session.removeTimer?.cancel(); + } + _sessions.clear(); + _sessionBySocket.clear(); await _httpServer?.close(force: true); _httpServer = null; } + /// Drop every socket without warning - what the client sees on a network + /// loss. Sessions stay resumable if [supportsResumption] is on. void closeAllSockets() { for (var socket in List.from(_sockets)) { + // Detach synchronously: a real server notices its own close at once, + // whereas the websocket's onDone only fires after the close handshake, + // by which time a fast client may already be reconnecting. + _onSocketClosed(socket); socket.close(); } _sockets.clear(); } + /// Server-initiated close of every session (`connection.close()` on the + /// server). Never resumable, regardless of [supportsResumption]. + void closeAllSessions() { + for (var session in List.from(_sessions.values)) { + session.expectingDisconnect = true; + _destroySession(session); + } + closeAllSockets(); + } + + void _onSocketClosed(WebSocket socket) { + _sockets.remove(socket); + var session = _sessionBySocket.remove(socket); + if (session == null || session.socket != socket) { + return; + } + session.socket = null; + if (!supportsResumption || session.expectingDisconnect) { + _destroySession(session); + return; + } + // Ungraceful disconnect: queue outgoing messages and wait for a resume. + session.messageQueue = []; + session.removeTimer?.cancel(); + session.removeTimer = + Timer(disconnectGracePeriod, () => _destroySession(session)); + } + + void _destroySession(MockDdpSession session) { + session.removeTimer?.cancel(); + session.removeTimer = null; + session.messageQueue = null; + _sessions.remove(session.id); + } + + void _handleConnect(WebSocket socket, Map msg) { + if (msg['version'] != '1' || !(msg['support'] as List).contains('1')) { + _sendRaw(socket, {'msg': 'failed', 'version': '1'}); + return; + } + var existing = _sessions[msg['session']]; + var resumable = supportsResumption && + existing != null && + existing.socket == null && + existing.removeTimer != null && + !existing.expectingDisconnect && + existing.sentCount == msg['receivedCount']; + if (resumable) { + existing.removeTimer?.cancel(); + existing.removeTimer = null; + var queue = existing.messageQueue ?? const []; + existing.messageQueue = null; + existing.socket = socket; + _sessionBySocket[socket] = existing; + sessionIdsSent.add(existing.id); + _sendOn(existing, {'msg': 'connected', 'session': existing.id}); + for (var queued in queue) { + _sendOn(existing, queued); + } + return; + } + if (existing != null) { + // Out of date (or not resumable) - drop the old session immediately. + _destroySession(existing); + } + var session = MockDdpSession('mock-session-${++_sessionCounter}'); + session.socket = socket; + _sessions[session.id] = session; + _sessionBySocket[socket] = session; + sessionCount++; + sessionIdsSent.add(session.id); + _sendOn(session, {'msg': 'connected', 'session': session.id}); + } + + /// Send on the session a socket belongs to, so the message is counted and, + /// while the session is disconnected, queued. void _send(WebSocket socket, Map msg) { + var session = _sessionBySocket[socket]; + if (session == null) { + _sendRaw(socket, msg); + return; + } + _sendOn(session, msg); + } + + void _sendOn(MockDdpSession session, Map msg) { + var counted = msg['msg'] != 'ping' && msg['msg'] != 'pong'; + var queue = session.messageQueue; + if (queue != null) { + if (counted) { + queue.add(msg); + if (queue.length > maxMessageQueueLength) { + _destroySession(session); + } + } + return; + } + var socket = session.socket; + if (socket == null) { + return; + } + if (counted) { + session.sentCount++; + } + _sendRaw(socket, msg); + } + + void _sendRaw(WebSocket socket, Map msg) { // A message can be in flight when the test closes the socket underneath // us; a real server would just drop it. if (socket.readyState != WebSocket.open) { @@ -110,15 +284,17 @@ class MockDdpServer { inbox.add(msg); switch (msg['msg']) { case 'connect': - if ((msg['version'] == '1') && (msg['support'] as List).contains('1')) { - _send(socket, {'msg': 'connected', 'session': 'mock-session-id'}); - } else { - _send(socket, {'msg': 'failed', 'version': '1'}); - } + _handleConnect(socket, msg); + break; + case 'disconnect': + // Graceful disconnect: the session is torn down as soon as the + // socket goes, never resumed. + _sessionBySocket[socket]?.expectingDisconnect = true; break; case 'ping': if (respondToPings) { - _send(socket, {'msg': 'pong', if (msg['id'] != null) 'id': msg['id']}); + _send( + socket, {'msg': 'pong', if (msg['id'] != null) 'id': msg['id']}); } break; case 'pong': @@ -127,9 +303,11 @@ class MockDdpServer { _handleMethod(socket, msg); break; case 'sub': + _sessionBySocket[socket]?.subscriptionIds.add(msg['id']); _handleSub(socket, msg); break; case 'unsub': + _sessionBySocket[socket]?.subscriptionIds.remove(msg['id']); _send(socket, {'msg': 'nosub', 'id': msg['id']}); break; } @@ -142,6 +320,20 @@ class MockDdpServer { if (silentMethods.contains(msg['method'])) { return; } + var delay = delayedMethods[msg['method']]; + if (delay != null) { + var session = _sessionBySocket[socket]; + Timer(delay, () { + if (session != null) { + _sendOn(session, {'msg': 'result', 'id': id, 'result': params}); + _sendOn(session, { + 'msg': 'updated', + 'methods': [id] + }); + } + }); + return; + } switch (msg['method']) { case 'login': var loginData = params.isNotEmpty @@ -271,10 +463,12 @@ class MockDdpServer { } } - /// Push a change on the `items` collection to every connected client. + /// Push a message to every session, live or waiting to be resumed (for + /// the latter it is queued, exactly like an observe callback firing during + /// the grace period). void broadcast(Map msg) { - for (var socket in _sockets) { - _send(socket, msg); + for (var session in List.from(_sessions.values)) { + _sendOn(session, msg); } } }