diff --git a/AGENTS.md b/AGENTS.md index 5ccba8d..d7b7dde 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -66,6 +66,7 @@ Run analyze, format and tests before every commit. | `lib/src/livestream/` | WHIP/WHEP over plain HTTP, independent of the room connection | | `lib/src/models/`, `wire.dart`, `logger.dart`, `version.dart` | public immutable types and shared helpers | | `lib/src/fishjam_client.dart`, `lib/src/livestream_api.dart`, `lib/fishjam_client.dart` | the public facade; wiring and type conversion only | +| `lib/src/connectivity_network_source.dart` | the `NetworkSource` (`signaling/network_source.dart`) on `connectivity_plus`; the only file that imports it | | `test/fakes/`, `test/support/`, `test/fixtures/` | fakes, fixture replayer, golden sequences | ## Testing tools diff --git a/README.md b/README.md index db60299..b43cffd 100644 --- a/README.md +++ b/README.md @@ -101,11 +101,15 @@ await client.dispose(); A lost connection is re-established automatically (`ReconnectConfig`): +- A change of network (Wi-Fi to mobile, a different Wi-Fi, going offline) starts reconnecting at once instead + of waiting for the old connection to time out; while the device is offline, attempts wait for a network + (at most `ReconnectConfig.offlineTimeout`, 60 s by default). Gaining a network (mobile next to Wi-Fi) is + not a change. - From `ReconnectionStarted` until `Reconnected` the status is `reconnecting` and calls that need the session (track operations, metadata updates, `createDataChannels()`) fail at once with `StateError`. - Before the new `Joined`, the remote tracks and peers you knew are reported with `TrackRemoved` and `PeerLeft` / `ComponentRemoved`; the new `Joined` lists the room as it is now. -- Local tracks are re-published only with `ReconnectConfig(addTracksOnReconnect: true)`, under new track ids +- Local tracks are re-published (`ReconnectConfig.addTracksOnReconnect`, on by default), under new track ids (the old ids get `LocalTrackRemoved`). Create the data channels again after `Reconnected`. `replaceTrack(id, null)` mutes by sending nothing; the device keeps capturing until you stop the track. diff --git a/SPEC.md b/SPEC.md index 9c8702f..6fbe8c3 100644 --- a/SPEC.md +++ b/SPEC.md @@ -38,6 +38,8 @@ Conventions: - Text frames from the server are ignored (logged at debug level). - The server pings every 30 s. The client additionally sends its own pings (interval 15 s) and treats a missing pong within one interval as a connection loss (retryable, [§6](#6-reconnection)). +- Opening the socket (connection, TLS and upgrade) times out after 10 s; a timeout fails like a socket that + cannot be opened. Closing a socket that is still opening aborts the open, which fails the same way. ### 2.2 Messages @@ -107,7 +109,12 @@ idle ──connect()──▶ connecting ──authenticated──▶ authentica - join close reason → `JoinException(reason)` (and `JoinFailed` is emitted), - any other close or socket failure before `Joined` → `ConnectionException(code, reason)`. Reconnection is **not** attempted for failures before the first `Joined`. -6. Calling `connect()` while not `idle`/`disconnected` throws `StateError`. +6. Calling `connect()` while not `idle`/`disconnected` throws `StateError`. The status is `disconnected` + from the moment a session is given up, so `connect()` may be called from a listener of an event emitted + then (e.g. `ReconnectionFailed`) or without awaiting `leave()`. Such a call opens its socket only once the + given-up session is torn down and its `Disconnected` emitted, so nothing of the old session (events, the + audio mode of [§20b](#20b-audio-output-routing)) lands in the new one. That teardown settles only the + `connect()` call that started it. ### 3.3 `leave()` @@ -168,19 +175,20 @@ All client → server messages go through a single ordered queue in front of the ## 6. Reconnection Configuration (`ReconnectConfig`): `maxAttempts` (default 3), `initialDelay` (default 500 ms), `delay` -(default 500 ms), `addTracksOnReconnect` (default false). `ReconnectConfig.disabled` turns it off. Reconnect is -enabled with defaults unless configured otherwise. +(default 500 ms), `addTracksOnReconnect` (default true, D27), `offlineTimeout` (default 60 s, [§6.1](#61-network-changes)). +`ReconnectConfig.disabled` turns it off. Reconnect is enabled with defaults unless configured otherwise. States: `idle`, `reconnecting`, `failed`. | From | Trigger | Action | To | |---|---|---|---| -| idle (joined) | retryable loss: retryable close, socket error, pong timeout, peer connection `failed`, ICE `failed` | snapshot local peer metadata (`peer` part) and local tracks; emit `ReconnectionStarted`; schedule attempt 1 | reconnecting | -| reconnecting | attempt timer fires | tear down old session silently (no `Disconnected`); start a new session with the snapshot metadata | reconnecting | +| idle (joined) | retryable loss: retryable close, socket error, pong timeout, peer connection `failed`, ICE `failed`, ICE `disconnected` for 3 s ([§8.4](#84-peer-connection)), network change ([§6.1](#61-network-changes)) | snapshot local peer metadata (`peer` part) and local tracks; emit `ReconnectionStarted`; schedule attempt 1 | reconnecting | +| reconnecting | attempt timer fires while online | tear down old session silently (no `Disconnected`) without waiting for its socket to finish closing; start a new session with the snapshot metadata | reconnecting | +| reconnecting | the network is lost or comes back or changes | see [§6.1](#61-network-changes) | reconnecting | | reconnecting | retryable loss in the new session before `connected` (including a new session that cannot be started at all) | schedule next attempt if attempts remain | reconnecting | | reconnecting | retryable loss, no attempts left | clear queue; emit `ReconnectionFailed`, then `Disconnected(DisconnectReason.reconnectionFailed)` | failed | | reconnecting | terminal close (auth/join/session ended/protocol) | cancel timer; clear queue; emit the matching error event and `Disconnected` | failed | -| reconnecting | `connected` | emit `TrackRemoved` for every remote track and then `PeerLeft` / `ComponentRemoved` for every remote endpoint the app was told about by the lost session; emit `Joined` (its snapshot is the source of truth); emit `LocalTrackRemoved(oldTrackId)` for every snapshotted track (once per reconnection, also without `addTracksOnReconnect`); if `addTracksOnReconnect`, re-add every snapshotted track whose media track is still live (skip, do not abort on, ended or muted tracks); when all re-adds settled (success or failure): flush queue, emit `Reconnected`, reset attempt counter | idle | +| reconnecting | `connected` | emit `TrackRemoved` for every remote track and then `PeerLeft` / `ComponentRemoved` for every remote endpoint the app was told about by the lost session; emit `Joined` (its snapshot is the source of truth); emit `LocalTrackRemoved` for every local track id the app was told about by the lost sessions (also without `addTracksOnReconnect`); if `addTracksOnReconnect`, re-add every snapshotted track whose media track is still live (skip, do not abort on, ended or muted tracks), all at once and in snapshot order, so they ride one negotiation cycle ([§8.3](#83-command-queue)); when all re-adds settled (success or failure): flush queue, emit `Reconnected`, reset attempt counter | idle | | any | `leave()` | cancel timer; clear queue | idle | - Delay of attempt `n` (1-based) is `initialDelay + (n - 1) * delay`. @@ -191,13 +199,59 @@ States: `idle`, `reconnecting`, `failed`. of its events are forwarded and its close or failure is ignored. App operations that need the session (track operations, metadata updates, `setTargetTrackEncoding`, `createDataChannels()`) fail at once with a `StateError` naming the reconnecting state; nothing is queued for later. -- Every local track id of the lost session ends with `LocalTrackRemoved`: the tracks went away with the - session. A re-added track gets a **new** track id; `LocalTrackAdded` is emitted for it after the removals. - If an attempt that already re-added tracks is lost before `Reconnected`, those new ids are reported - removed when the next attempt starts. +- An attempt replaced after its `Joined` without a loss of its own (it joined while offline, or a new network + appeared while its tracks were restored, [§6.1](#61-network-changes)) counts as lost too: the remote + endpoints it announced are reported gone before the next `Joined`. +- Every local track id of a lost session ends with `LocalTrackRemoved` right after the next `Joined` (or with + the `Disconnected` that ends a failed reconnection): the tracks went away with the session. This includes + the new ids of tracks an attempt re-added before it was lost. A re-added track gets a **new** track id; + `LocalTrackAdded` is emitted for it after the removals. +- Only tracks the app got `LocalTrackAdded` for (and no `LocalTrackRemoved` yet) are snapshotted, reported + removed and re-added. A track whose `addTrack` was still negotiating when the session was lost is not: its + `addTrack` fails, so the app never learned its id and does not expect it published. Neither is a track an + abandoned attempt negotiated without its `LocalTrackAdded` reaching the app. - A **muted** track (no media track) is not re-added: there is nothing to publish. The app adds the track again when it unmutes. +### 6.1 Network changes + +The client watches the kinds of network the device uses (Wi-Fi, mobile, ethernet, VPN, other), as the OS +reports them. A socket and a peer connection are bound to the network they were opened on, so losing it means +the session will not survive; reacting at once avoids waiting for a pong timeout or an ICE failure. + +- With reconnection enabled, watching starts at `connect()` and stops when the client is disconnected; with + reconnection disabled the network is not watched. The first value read is the baseline. **Online** means at + least one network is reported; **offline** means none. +- Every report is acted on when it arrives, compared with the previous report. The brief "offline" some + platforms report while switching networks is not waited out: no attempt starts while offline, and the network + reported after it starts one at once. +- Kinds of change: + - **lost**: now offline; + - **available**: online after being offline; + - **changed**: a network went away while another remains (Wi-Fi → mobile, a VPN going down). + - Only gaining a network (mobile added next to Wi-Fi, a VPN coming up) and a repeat of the same set are no + change: the connection keeps its path. +- A switch between two Wi-Fi networks looks like the same set: some platforms report it as offline and then + online (`lost`, then `available`), others not at all, and then the socket or ICE reports the loss. + +| Status | Change | Action | +|---|---|---| +| joined, reconnection enabled | changed | retryable loss `network changed` (§6), without waiting for the socket or ICE to fail | +| joined, reconnection enabled | lost | retryable loss `network lost` (§6) | +| before the first `Joined` of a `connect()` | any | no loss (the attempt fails or succeeds on its own); whether the device is online is remembered | +| reconnecting | lost | a scheduled attempt is cancelled; no attempt starts while offline, and a loss while offline waits for the network instead of using up an attempt | +| reconnecting | available or changed | the attempt budget starts over (`maxAttempts` on the new network). A scheduled attempt is cancelled, an attempt in progress is abandoned (its session torn down silently) and a new attempt starts at once | +| reconnecting | offline for `offlineTimeout` | as "no attempts left" (§6): `ReconnectionFailed`, then `Disconnected(reconnectionFailed)` | + +- A session of a reconnection that reaches `connected` while the device is offline, or goes offline before its + tracks are restored, does not complete it: no `Reconnected`; the reconnection waits for the network and starts + a new attempt then. +- One reconnection makes at most `3 × maxAttempts` attempts in total, so a network that keeps changing cannot + keep it going forever; when that cap is reached the reconnection fails as when no attempts are left. +- When the OS cannot be asked, the client assumes it is always online and never reports a change. This is only + logged: the client asks for the networks in use before it subscribes to reports, so nothing reaches the app's + error handler. + ## 7. Media event processing - Server media events are processed **strictly one at a time, in arrival order**. Handling of event N+1 starts @@ -284,17 +338,23 @@ Rules: candidate (end of gathering) is not sent. Remote candidates are added with `addIceCandidate`; failures are logged. - Connection state `failed` or ICE connection state `failed` emits `ConnectionError(message)` and is a - retryable loss ([§6](#6-reconnection)). + retryable loss ([§6](#6-reconnection)). So does ICE connection state `disconnected` that has not turned + `connected` or `completed` within 3 s: the OS takes much longer to report `failed`, while a short + `disconnected` recovers on its own. - A new session always creates a new peer connection; the old one is closed. ## 9. Receive transceiver sync On every `offerData{tracksTypes{audio, video}}`, the client makes the number of **live receive transceivers** -per kind equal the server's numbers. A transceiver is a live receive transceiver when its direction is -`recvonly` and its current direction is not `stopped`; its kind is its receiver track kind. +per kind equal the server's numbers. The live receive transceivers are the `recvonly` transceivers this +session added and has not stopped (the client always makes the offer, so it created every transceiver), +identified by their receiver track id. The SDK does not ask transceivers for their direction: on Android +`flutter_webrtc` disposes the handles of transceivers without a mid whenever it looks up a negotiated one, and +on iOS transceivers without a mid share one id. 1. Count live receive transceivers per kind. -2. If there are more than requested, stop the excess, choosing transceivers with no `mid` first. +2. If there are more than requested, stop the excess, choosing transceivers with no `mid` first. All + transceivers without a mid are stopped before any with a mid, from one `getTransceivers` snapshot. 3. If there are fewer, add `recvonly` transceivers: first all missing **video**, then all missing **audio**. 4. Negative deltas never throw. @@ -659,6 +719,10 @@ through `dart:developer`'s `log` only when `debug` is true. The SDK never calls | D21 | local track ids of the lost session silently disappear on reconnect | `LocalTrackRemoved` for every old id; restored tracks get `LocalTrackAdded` with new ids (§6) | apps can drop stale ids | | D22 | remote peers and tracks of the lost session are kept and re-announced with `peerJoined` / `trackAdded` after reconnecting | `TrackRemoved` and `PeerLeft` / `ComponentRemoved` for all of them before the new `Joined`, whose snapshot is the source of truth (§6) | no stale or duplicate remote state | | D23 | (React Native SDK) audio output routing goes through Android Telecom while a Telecom call is active | always `AudioManager` (§20b) | this SDK has no Telecom integration yet | +| D24 | a network switch is noticed only when the socket or ICE fails; attempts are used up while the device is offline | losing a network is a retryable loss; attempts wait while offline (at most `offlineTimeout`) and restart with a fresh budget on a new network (§6.1) | fast recovery on mobile; a tunnel does not exhaust the attempts | +| D25 | ICE `disconnected` only renegotiates, however long it lasts | `disconnected` for 3 s is a retryable loss (§8.4) | do not wait for a late `failed` | +| D26 | opening the socket has no timeout | 10 s (§2.1) | an attempt on a dead network does not hang | +| D27 | `addTracksOnReconnect` defaults to false: after a reconnection the user publishes nothing until the app adds its tracks again | defaults to true (§6) | a network switch on mobile should not silently turn off the camera and microphone | ## 23. Golden fixtures diff --git a/example/.env.example b/example/.env.example new file mode 100644 index 0000000..f76e3a2 --- /dev/null +++ b/example/.env.example @@ -0,0 +1,10 @@ +# Copy to .env.local, then: flutter run --dart-define-from-file=.env.local + +# https://fishjam.io/app +FISHJAM_ID= + +# https://fishjam.io/app/sandbox (prototyping only) +SANDBOX_API_URL= + +# Or a peer token from your backend (server SDK) +# PEER_TOKEN= diff --git a/example/README.md b/example/README.md index 8d3e922..9cbbeed 100644 --- a/example/README.md +++ b/example/README.md @@ -1,7 +1,7 @@ # fishjam_example A minimal Fishjam room built with `fishjam_client`: it publishes the camera and microphone, renders the -other peers' video, has mute toggles, a chat over the reliable data channel, and a leave button. +other peers' video, and has mute toggles and a leave button. ## Run diff --git a/example/lib/join_screen.dart b/example/lib/join_screen.dart index e11f4e6..2f2acf5 100644 --- a/example/lib/join_screen.dart +++ b/example/lib/join_screen.dart @@ -19,7 +19,7 @@ class _JoinScreenState extends State { final _form = GlobalKey(); final _name = TextEditingController(text: 'flutter'); // Optional prefill: flutter run --dart-define-from-file=.env.local - // (.env.local holds FISHJAM_ID, PEER_TOKEN, SANDBOX_API_URL). + // (copy example/.env.example to .env.local). final _fishjamUrl = TextEditingController( text: const String.fromEnvironment('FISHJAM_ID'), ); diff --git a/example/lib/room_screen.dart b/example/lib/room_screen.dart index ac1b12a..aa1adb7 100644 --- a/example/lib/room_screen.dart +++ b/example/lib/room_screen.dart @@ -1,7 +1,5 @@ import 'dart:async'; -import 'dart:convert'; import 'dart:io'; -import 'dart:typed_data'; import 'package:fishjam_client/fishjam_client.dart'; import 'package:flutter/material.dart'; @@ -10,7 +8,7 @@ import 'package:flutter_webrtc/flutter_webrtc.dart'; import 'sandbox.dart'; /// A minimal room: publishes camera and microphone, renders remote video, -/// mute toggles, a data channel chat and leave. +/// mute toggles and leave. class RoomScreen extends StatefulWidget { const RoomScreen({ super.key, @@ -32,22 +30,12 @@ class _RemoteVideo { final String label; } -class _ChatMessage { - const _ChatMessage(this.from, this.text); - - final String from; - final String text; -} - class _RoomScreenState extends State { - // Tracks are published again after a reconnection (with new ids). - final _client = FishjamClient( - reconnect: const ReconnectConfig(addTracksOnReconnect: true), - ); + // Tracks are published again after a reconnection, with new ids + // (`ReconnectConfig.addTracksOnReconnect`, on by default). + final _client = FishjamClient(); final _localRenderer = RTCVideoRenderer(); final _remote = {}; - final _messages = <_ChatMessage>[]; - final _chatInput = TextEditingController(); final _subscriptions = >[]; MediaStream? _localStream; @@ -172,9 +160,7 @@ class _RoomScreenState extends State { Future _start() async { await _localRenderer.initialize(); - _subscriptions - ..add(_client.events.listen(_onEvent)) - ..add(_client.dataStream(reliable: true).listen(_onChat)); + _subscriptions.add(_client.events.listen(_onEvent)); try { // Asks for camera and microphone permission. final stream = await navigator.mediaDevices.getUserMedia({ @@ -194,7 +180,6 @@ class _RoomScreenState extends State { ); _setStatus('Connected'); await Future.wait([_publishCamera(), _publishMicrophone()]); - await _client.createDataChannels(); } on Object catch (e) { _setStatus('Error: $e'); } @@ -248,8 +233,6 @@ class _RoomScreenState extends State { _setStatus('Reconnecting…'); case Reconnected(): _setStatus('Connected'); - // Data channels belong to the session: create them again. - unawaited(_client.createDataChannels().catchError((Object _) {})); case Disconnected(:final reason) when reason != DisconnectReason.left: _setStatus('Disconnected: ${reason.name}'); default: @@ -284,28 +267,6 @@ class _RoomScreenState extends State { await previous?.renderer.dispose(); } - void _onChat(Uint8List data) { - try { - final json = jsonDecode(utf8.decode(data)) as Map; - setState( - () => _messages.add(_ChatMessage('${json['from']}', '${json['text']}')), - ); - } on Object { - // Not a chat message. - } - } - - void _sendChat() { - final text = _chatInput.text.trim(); - if (text.isEmpty || !_client.dataChannelsReady) return; - _client.publishData( - utf8.encode(jsonEncode({'from': widget.displayName, 'text': text})), - reliable: true, - ); - _chatInput.clear(); - setState(() => _messages.add(_ChatMessage(widget.displayName, text))); - } - /// Muting replaces the sent media with nothing; unmuting puts it back. /// `replaceTrack(id, null)` only stops sending: the camera keeps capturing /// (and the local preview keeps running). Set `track.enabled = false` or @@ -384,7 +345,6 @@ class _RoomScreenState extends State { } unawaited(stream.dispose()); } - _chatInput.dispose(); super.dispose(); } @@ -414,7 +374,6 @@ class _RoomScreenState extends State { children: tiles, ), ), - SizedBox(height: 180, child: _chat()), _controls(), ], ), @@ -422,45 +381,6 @@ class _RoomScreenState extends State { ); } - Widget _chat() => Column( - children: [ - Expanded( - child: ListView( - padding: const EdgeInsets.symmetric(horizontal: 12), - children: [ - for (final m in _messages) - Text.rich( - TextSpan( - children: [ - TextSpan( - text: '${m.from}: ', - style: const TextStyle(fontWeight: FontWeight.bold), - ), - TextSpan(text: m.text), - ], - ), - ), - ], - ), - ), - Padding( - padding: const EdgeInsets.symmetric(horizontal: 12), - child: Row( - children: [ - Expanded( - child: TextField( - controller: _chatInput, - decoration: const InputDecoration(hintText: 'Message'), - onSubmitted: (_) => _sendChat(), - ), - ), - IconButton(onPressed: _sendChat, icon: const Icon(Icons.send)), - ], - ), - ), - ], - ); - Widget _controls() => Padding( padding: const EdgeInsets.all(8), child: Row( diff --git a/example/pubspec.lock b/example/pubspec.lock index bdc7460..7fd5081 100644 --- a/example/pubspec.lock +++ b/example/pubspec.lock @@ -57,6 +57,22 @@ packages: url: "https://pub.dev" source: hosted version: "1.19.1" + connectivity_plus: + dependency: transitive + description: + name: connectivity_plus + sha256: d7749fd36d8b404a4935612b94eca7c3ce3705377ab678e24ff2a4301436f741 + url: "https://pub.dev" + source: hosted + version: "7.3.2" + connectivity_plus_platform_interface: + dependency: transitive + description: + name: connectivity_plus_platform_interface + sha256: "3c09627c536d22fd24691a905cdd8b14520de69da52c7a97499c8be5284a32ed" + url: "https://pub.dev" + source: hosted + version: "2.1.0" crypto: dependency: transitive description: @@ -73,6 +89,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.8.2" + dbus: + dependency: transitive + description: + name: dbus + sha256: f7db8504e715b835f5ebe5fe74604ccd961d504c73389a5d8e81bafc2e2f3612 + url: "https://pub.dev" + source: hosted + version: "0.8.0" fake_async: dependency: transitive description: @@ -135,6 +159,11 @@ packages: description: flutter source: sdk version: "0.0.0" + flutter_web_plugins: + dependency: transitive + description: flutter + source: sdk + version: "0.0.0" flutter_webrtc: dependency: "direct main" description: @@ -273,6 +302,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.18.3" + nm: + dependency: transitive + description: + name: nm + sha256: "2ae480aa3379716b8bb58dfebb168e2919394c92b629efaaa0baaa5d140786e6" + url: "https://pub.dev" + source: hosted + version: "0.6.0" objective_c: dependency: transitive description: @@ -345,6 +382,14 @@ packages: url: "https://pub.dev" source: hosted version: "2.3.0" + petitparser: + dependency: transitive + description: + name: petitparser + sha256: "91bd59303e9f769f108f8df05e371341b15d59e995e6806aefab827b58336675" + url: "https://pub.dev" + source: hosted + version: "7.0.2" platform: dependency: transitive description: @@ -526,6 +571,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.1.0" + xml: + dependency: transitive + description: + name: xml + sha256: "67f0aff7be013d107995e9b75bf4e7f2c3ef2dfdb2c8e68024bba0a7fd5756a4" + url: "https://pub.dev" + source: hosted + version: "7.0.1" yaml: dependency: transitive description: diff --git a/lib/src/connectivity_network_source.dart b/lib/src/connectivity_network_source.dart new file mode 100644 index 0000000..08f351f --- /dev/null +++ b/lib/src/connectivity_network_source.dart @@ -0,0 +1,18 @@ +import 'package:connectivity_plus/connectivity_plus.dart'; +import 'package:fishjam_client/src/signaling/network_source.dart'; + +/// [NetworkSource] backed by `connectivity_plus`. +final class ConnectivityNetworkSource implements NetworkSource { + @override + Future> getNetworkTypes() async => + _kinds(await Connectivity().checkConnectivity()); + + @override + Stream> get networkTypesChanged => + Connectivity().onConnectivityChanged.map(_kinds); + + static Set _kinds(List results) => { + for (final result in results) + if (result != ConnectivityResult.none) result.name, + }; +} diff --git a/lib/src/engine/receive_transceivers.dart b/lib/src/engine/receive_transceivers.dart index 806340c..d9f6598 100644 --- a/lib/src/engine/receive_transceivers.dart +++ b/lib/src/engine/receive_transceivers.dart @@ -1,43 +1,104 @@ +import 'package:fishjam_client/src/logger.dart'; import 'package:fishjam_client/src/proto/fishjam/media_events/server/server.pb.dart' as server; import 'package:fishjam_client/src/rtc/rtc_adapter.dart'; -/// Makes the live receive transceivers per kind match the numbers the -/// server asked for in `offerData`, before the offer is created. -Future syncReceiveTransceivers( - RtcPeerConnection pc, - server.MediaEvent_OfferData_TrackTypes types, -) async { - final live = >{ +/// The receive-only transceivers of one peer connection (SPEC §9). +/// +/// This client always makes the offer, so every transceiver in the peer +/// connection was created here: the receive ones are recorded by their +/// receiver's track id, which stays the same in every snapshot, instead of +/// asking each transceiver for its direction. Per-transceiver native calls +/// are unreliable in the WebRTC plugin: on Android a call on a negotiated +/// transceiver disposes the handles of the ones without a mid, and on iOS all +/// transceivers without a mid share one id. +class ReceiveTransceivers { + ReceiveTransceivers({required this._logger}); + + final FishjamLogger _logger; + + /// Missing transceivers are added video first (SPEC §9). + static const List _kinds = [MediaKind.video, MediaKind.audio]; + + /// Receiver track ids of our receive-only transceivers, per kind. A + /// transceiver is identified by its receiver's track id (see the class doc). + final _receiverTrackIds = >{ MediaKind.video: [], MediaKind.audio: [], }; - for (final t in await pc.getTransceivers()) { - if (await t.getDirection() == TransceiverDirection.recvonly && - await t.getCurrentDirection() != TransceiverDirection.stopped) { - live[t.receiver.kind]!.add(t); + + /// Makes the live receive transceivers per kind match the numbers the + /// server asked for in `offerData`, before the offer is created. + Future sync( + RtcPeerConnection pc, + server.MediaEvent_OfferData_TrackTypes types, + ) async { + final wantedCount = { + MediaKind.video: types.video, + MediaKind.audio: types.audio, + }; + // Skip the native `getTransceivers()` call when nothing needs stopping. + if (_receiverTrackIds.entries.any( + (e) => e.value.length > wantedCount[e.key]!, + )) { + await _stopExcess(pc, wantedCount); } - } - final wanted = {MediaKind.video: types.video, MediaKind.audio: types.audio}; - for (final kind in [MediaKind.video, MediaKind.audio]) { - final have = live[kind]!; - final excess = have.length - wanted[kind]!; - if (excess <= 0) continue; - // Transceivers without a mid were never negotiated: stop them first. - final candidates = [ - ...have.where((t) => t.mid == null), - ...have.where((t) => t.mid != null), - ]; - for (final t in candidates.take(excess)) { - await t.stop(); + for (final kind in _kinds) { + for ( + var i = _receiverTrackIds[kind]!.length; + i < wantedCount[kind]!; + i++ + ) { + final transceiver = await pc.addTransceiverOfKind( + kind, + const TransceiverInit(direction: TransceiverDirection.recvonly), + ); + final id = transceiver.receiver.track?.id; + if (id == null) { + // Unrecorded, it is added again on the next offerData. + _logger.warn( + 'A ${kind.name} receive transceiver has no receiver ' + 'track; it is not counted', + ); + } else { + _receiverTrackIds[kind]!.add(id); + } + } } } - for (final kind in [MediaKind.video, MediaKind.audio]) { - for (var i = live[kind]!.length; i < wanted[kind]!; i++) { - await pc.addTransceiverOfKind( - kind, - const TransceiverInit(direction: TransceiverDirection.recvonly), - ); + + /// Stops the excess, all from one snapshot: first the transceivers + /// without a mid (never negotiated) of every kind, then negotiated ones, + /// because stopping a negotiated one invalidates the handles of those + /// without a mid. + Future _stopExcess( + RtcPeerConnection pc, + Map wantedCount, + ) async { + final transceiverByTrackId = { + for (final t in await pc.getTransceivers()) ?t.receiver.track?.id: t, + }; + for (final ids in _receiverTrackIds.values) { + // Forget ids whose transceiver no longer exists (e.g. the server + // rejected its media section), so the lookups below cannot fail. + ids.removeWhere((id) => !transceiverByTrackId.containsKey(id)); + } + for (final stopNegotiated in [false, true]) { + for (final kind in _kinds) { + final ids = _receiverTrackIds[kind]!; + final excess = ids.length - wantedCount[kind]!; + if (excess <= 0) continue; + final toStop = ids + .where( + (id) => (transceiverByTrackId[id]!.mid != null) == stopNegotiated, + ) + .take(excess) + .toList(); + for (final id in toStop) { + await transceiverByTrackId[id]!.stop(); + ids.remove(id); + } + } } } } diff --git a/lib/src/engine/rtc_engine.dart b/lib/src/engine/rtc_engine.dart index 6a67b00..8342c3e 100644 --- a/lib/src/engine/rtc_engine.dart +++ b/lib/src/engine/rtc_engine.dart @@ -45,6 +45,7 @@ class RtcEngine { required this.localTracks, required this.remote, this.nativeCallTimeout = const Duration(seconds: 10), + this.iceDisconnectedTimeout = const Duration(seconds: 3), this.onServerError, this.onLocalPeerRemoved, this.onConnectionFailed, @@ -66,6 +67,9 @@ class RtcEngine { /// `setLocalDescription`, `setRemoteDescription`). final Duration nativeCallTimeout; + /// How long ICE may stay `disconnected` before it counts as a failure. + final Duration iceDisconnectedTimeout; + final void Function(String message)? onServerError; final void Function()? onLocalPeerRemoved; @@ -85,6 +89,8 @@ class RtcEngine { String? _localPeerId; List _iceServers = const []; RtcPeerConnection? _pc; + + late final _receiveTransceivers = ReceiveTransceivers(logger: _logger); final _subscriptions = >[]; bool _hasRemoteDescription = false; final _pendingCandidates = []; @@ -96,6 +102,7 @@ class RtcEngine { final _ownedStreams = {}; bool _failed = false; bool _closed = false; + Timer? _iceDisconnectedTimer; void connect(Object? peerMetadata) { localTracks.peerMetadata = peerMetadata; @@ -316,6 +323,7 @@ class RtcEngine { Future close() async { if (_closed) return; _closed = true; + _iceDisconnectedTimer?.cancel(); _commands.failAll(); _negotiation.reset(); if (!_joined.isCompleted) _joined.completeError(StateError('disconnected')); @@ -476,7 +484,7 @@ class RtcEngine { } try { final pc = await _ensurePeerConnection(); - await syncReceiveTransceivers(pc, offerData.tracksTypes); + await _receiveTransceivers.sync(pc, offerData.tracksTypes); await _commands.drainBatch(); final offer = await pc.createOffer().timeout(nativeCallTimeout); // Native stacks emit candidates during setLocalDescription, before @@ -639,6 +647,18 @@ class RtcEngine { _perform(_negotiation.handle(NegotiationInput.iceDisconnected)); }), ); + // The OS reports `failed` only after tens of seconds; a network + // switch should not keep the call frozen that long. + _iceDisconnectedTimer ??= Timer( + iceDisconnectedTimeout, + () => _connectionFailed( + 'ICE connection disconnected for ' + '${iceDisconnectedTimeout.inMilliseconds} ms', + ), + ); + case IceConnectionState.connected || IceConnectionState.completed: + _iceDisconnectedTimer?.cancel(); + _iceDisconnectedTimer = null; case IceConnectionState.failed: _connectionFailed('ICE connection failed'); case _: diff --git a/lib/src/fishjam_client.dart b/lib/src/fishjam_client.dart index 38b0731..0c1e443 100644 --- a/lib/src/fishjam_client.dart +++ b/lib/src/fishjam_client.dart @@ -2,6 +2,7 @@ import 'dart:async'; import 'dart:typed_data'; import 'package:fishjam_client/src/audio_output/audio_output_manager.dart'; +import 'package:fishjam_client/src/connectivity_network_source.dart'; import 'package:fishjam_client/src/data/data_channels.dart'; import 'package:fishjam_client/src/engine/rtc_engine.dart'; import 'package:fishjam_client/src/logger.dart'; @@ -16,11 +17,12 @@ import 'package:fishjam_client/src/rtc/flutter_webrtc_adapter.dart'; import 'package:fishjam_client/src/rtc/rtc_adapter.dart'; import 'package:fishjam_client/src/signaling/auth.dart'; import 'package:fishjam_client/src/signaling/message_queue.dart'; +import 'package:fishjam_client/src/signaling/network_change_detector.dart'; +import 'package:fishjam_client/src/signaling/network_source.dart'; import 'package:fishjam_client/src/signaling/reconnect.dart'; import 'package:fishjam_client/src/signaling/signaling_session.dart'; import 'package:fishjam_client/src/signaling/transport.dart'; import 'package:fishjam_client/src/stats/stats_reporter.dart'; -import 'package:fishjam_client/src/tracks/local_track.dart'; import 'package:fishjam_client/src/tracks/local_track_registry.dart'; import 'package:fishjam_client/src/tracks/remote_track.dart'; import 'package:fishjam_client/src/version.dart'; @@ -55,22 +57,32 @@ class FishjamClient { }) : this.internal( debug: debug, reconnect: reconnect, + networkSource: ConnectivityNetworkSource(), audioOutput: AudioOutputManager.instance, ); - /// A client on a custom WebRTC adapter and socket transport (tests only; - /// they default to `flutter_webrtc` and `dart:io` WebSockets). Without - /// `audioOutput` the audio mode is never changed. + /// A client on a custom WebRTC adapter, socket transport and network + /// source (tests only; they default to `flutter_webrtc`, `dart:io` + /// WebSockets and no network watching). Without `audioOutput` the audio + /// mode is never changed. @internal FishjamClient.internal({ this.debug = false, this.reconnect = const ReconnectConfig(), RtcAdapter? rtcAdapter, this._transportFactory, + NetworkSource? networkSource, this._audioOutput, }) : _adapter = rtcAdapter ?? const FlutterWebrtcAdapter(), _logger = FishjamLogger(enabled: debug) { _reconnect = ReconnectManager(config: reconnect, host: _Host(this)); + if (networkSource != null && reconnect.isEnabled) { + _networkChangeDetector = NetworkChangeDetector( + source: networkSource, + logger: _logger, + onChanged: _onNetworkChanged, + ); + } _dataChannels = DataChannelManager( logger: _logger, negotiate: (apply) => _requireAvailable().engine.negotiate(apply), @@ -105,6 +117,7 @@ class FishjamClient { late final ReconnectManager _reconnect; late final DataChannelManager _dataChannels; late final StatsReporter _stats; + NetworkChangeDetector? _networkChangeDetector; ConnectionStatus _status = ConnectionStatus.idle; @@ -118,6 +131,10 @@ class FishjamClient { /// Settles the pending `connect()` call exactly once. Completer? _connecting; + + /// Completes after the last ended session finished its teardown and + /// emitted `Disconnected`. A new `connect()` waits for it. Never fails. + Future _previousSessionEnded = Future.value(); String _url = ''; String _token = ''; bool _disposed = false; @@ -126,6 +143,11 @@ class FishjamClient { /// by a session that was lost; reported gone before the next `Joined`. ({List peers, List components})? _staleRemote; + /// Ids of local tracks the app received `LocalTrackAdded` for, and not yet + /// `LocalTrackRemoved`. On reconnection only these are reported removed and + /// restored, because the app does not know about the others. + final _announcedLocalTracks = {}; + /// All client events (broadcast, not replayed). Stream get events => _events.stream; @@ -152,15 +174,28 @@ class FishjamClient { _token = token; _reconnect.reset(); _staleRemote = null; + _announcedLocalTracks.clear(); _status = ConnectionStatus.connecting; + unawaited(_networkChangeDetector?.start()); final connecting = _connecting = Completer(); - final session = _newSession(); + unawaited(_openAfterPreviousSessionEnded(_newSession(), peerMetadata)); + return await connecting.future; + } + + /// Waits for the previous session to finish its teardown, then opens + /// [session]. This keeps the event order when the app calls `connect()` from + /// a `ReconnectionFailed` listener: the old `Disconnected` comes before the + /// new session starts. + Future _openAfterPreviousSessionEnded( + _Session session, + Object? peerMetadata, + ) async { try { - await _open(session, peerMetadata); + await _previousSessionEnded; + if (_isCurrent(session)) await _open(session, peerMetadata); } on Object { // Reported through `closed`, which settles connect(). } - return await connecting.future; } /// Leaves the room gracefully. A no-op when not connected. @@ -364,6 +399,11 @@ class FishjamClient { // While reconnecting, the status turns `joined` with `Reconnected`. if (!_reconnect.isReconnecting) _status = ConnectionStatus.joined; _emit(joined); + // Every local track the app was told about belongs to a lost session (the + // set starts empty at connect()) and went away with it. + for (final trackId in _announcedLocalTracks.toList()) { + _emit(LocalTrackRemoved(trackId)); + } _stats.start(); _setCommunicationMode(on: true); final connecting = _connecting; @@ -389,7 +429,7 @@ class FishjamClient { connectError: connectExceptionFor(reason), ); } else if (reason is RetryableClose) { - await _onRetryableLoss(reason); + _reconnectOrEnd(reason); } else { _reconnect.onTerminalClose(reason); await _finish(disconnectedFor(reason), error: errorEventFor(reason)); @@ -400,9 +440,7 @@ class FishjamClient { /// (the engine already emitted `ConnectionError`): a retryable loss. void _onConnectionFailed(_Session session) { if (!_isCurrent(session)) return; - unawaited( - _onRetryableLoss(const RetryableClose(reason: 'peer connection')), - ); + _reconnectOrEnd(const RetryableClose(reason: 'peer connection')); } void _onServerError(_Session session) => @@ -428,27 +466,60 @@ class FishjamClient { }); } - Future _onRetryableLoss(RetryableClose reason) async { + /// While joined, losing or switching the network ends the session: the + /// socket and peer connection are bound to the old network. + void _onNetworkChanged(NetworkChange change) { + _logger.debug('Network ${change.name}'); + // Report the loss first so a reconnection exists. The calls below then + // act on it: online starts an attempt now, offline makes it wait. + if (_status == ConnectionStatus.joined && + change != NetworkChange.available) { + _reconnectOrEnd(RetryableClose(reason: 'network ${change.name}')); + } + if (change == NetworkChange.lost) { + _reconnect.onNetworkLost(); + } else { + _reconnect.onNetworkAvailable(); + } + } + + /// Starts a reconnection or ends the session. Synchronous on purpose: + /// [_onNetworkChanged] calls the reconnect manager right after and needs the + /// reconnection to exist. Only ending the session (`_finish`) continues in + /// the background. + void _reconnectOrEnd(RetryableClose reason) { final session = _session!; final snapshot = ReconnectSnapshot( peerMetadata: session.localTracks.peerMetadata, - tracks: [for (final t in session.localTracks.tracks) t.toContext()], + tracks: [ + for (final t in session.localTracks.tracks) + if (_announcedLocalTracks.contains(t.trackId)) t.toContext(), + ], ); if (_reconnect.onRetryableLoss(reason, snapshot)) { - // From here on the lost session reports nothing. - session.isLost = true; - if (session.isJoined) { - _staleRemote = ( - peers: session.remote.peers, - components: session.remote.components, - ); - } + _markSessionLost(session); _status = ConnectionStatus.reconnecting; _stats.stop(); } else if (_reconnect.state == ReconnectState.failed) { - await _finish(const Disconnected(DisconnectReason.reconnectionFailed)); + unawaited( + _finish(const Disconnected(DisconnectReason.reconnectionFailed)), + ); } else { - await _finish(disconnectedFor(reason)); + unawaited(_finish(disconnectedFor(reason))); + } + } + + /// Ignores all later events from [session]. If it had joined, saves its + /// remote peers and components so they are reported as gone before the + /// next `Joined` (SPEC §22, D22). + void _markSessionLost(_Session session) { + if (session.isLost) return; + session.isLost = true; + if (session.isJoined) { + _staleRemote = ( + peers: session.remote.peers, + components: session.remote.components, + ); } } @@ -471,11 +542,23 @@ class FishjamClient { _session = null; _staleRemote = null; _status = ConnectionStatus.disconnected; - await _teardown(session, sendDisconnect: sendDisconnect); - _setCommunicationMode(on: false); - if (error != null) _emit(error); - _emit(disconnected); + // Take the pending `connect()` completer before the teardown. The app may + // call `connect()` again during the teardown (e.g. from a + // `ReconnectionFailed` listener); that new call must not be failed by + // this one. final connecting = _connecting; + _connecting = null; + _networkChangeDetector?.stop(); + final ended = Completer(); + _previousSessionEnded = ended.future; + try { + await _teardown(session, sendDisconnect: sendDisconnect); + _setCommunicationMode(on: false); + if (error != null) _emit(error); + _emit(disconnected); + } finally { + ended.complete(); + } if (connecting != null && !connecting.isCompleted) { connecting.completeError( connectError ?? @@ -487,16 +570,28 @@ class FishjamClient { } /// In this order: data channels, `disconnect` (when leaving), peer - /// connection and local streams, socket. + /// connection and local streams, socket. Without [waitForSocket] the + /// socket close is started but not awaited. Future _teardown( _Session session, { required bool sendDisconnect, + bool waitForSocket = true, }) async { _stats.stop(); await _dataChannels.close(); if (sendDisconnect) session.engine.sendDisconnect(); await session.engine.close(); - await session.signaling.close(); + final socketClosing = session.signaling.close(); + if (waitForSocket) { + await socketClosing; + } else { + // The socket has often failed already, and then so does its close. + unawaited( + socketClosing.catchError( + (Object e) => _logger.debug('Closing the old socket failed: $e'), + ), + ); + } if (identical(_socket, session.signaling)) _socket = null; } @@ -555,6 +650,13 @@ class FishjamClient { } void _emit(FishjamEvent event) { + switch (event) { + case LocalTrackAdded(:final track): + _announcedLocalTracks.add(track.trackId); + case LocalTrackRemoved(:final trackId): + _announcedLocalTracks.remove(trackId); + case _: + } if (!_events.isClosed) _events.add(event); } @@ -593,9 +695,6 @@ class _Session { /// Whether `Joined` was emitted for this session. bool isJoined = false; - /// Whether the session was started by a reconnection attempt. - bool isReconnectAttempt = false; - /// Whether a reconnection replaces this session; it reports nothing more. bool isLost = false; @@ -612,20 +711,16 @@ class _Host implements ReconnectHost { @override Future startSession(Object? peerMetadata) async { final old = _client._session; - // The new session becomes current first, so the old one is torn down - // silently (no `Disconnected`) and its late reports are ignored. - final session = _client._newSession()..isReconnectAttempt = true; + // Make the new session current first, so tearing down the old one emits + // no `Disconnected` and its late events are ignored. + final session = _client._newSession(); if (old != null) { - // Tracks a previous attempt restored were announced with - // `LocalTrackAdded`; they go away with that session. - if (old.isReconnectAttempt) { - for (final track in old.localTracks.tracks) { - if (track.negotiationStatus == TrackNegotiationStatus.negotiated) { - _client._emit(LocalTrackRemoved(track.trackId)); - } - } - } - await _client._teardown(old, sendDisconnect: false); + // The old session may not be marked lost yet (an attempt replaced + // after its `Joined`). + _client._markSessionLost(old); + // Do not wait for the old socket to close: on a dead network the close + // handshake can take seconds. + await _client._teardown(old, sendDisconnect: false, waitForSocket: false); } if (!identical(_client._session, session)) return; try { diff --git a/lib/src/models/config.dart b/lib/src/models/config.dart index 6d10a5a..dc8090a 100644 --- a/lib/src/models/config.dart +++ b/lib/src/models/config.dart @@ -8,7 +8,8 @@ final class ReconnectConfig { this.maxAttempts = 3, this.initialDelay = const Duration(milliseconds: 500), this.delay = const Duration(milliseconds: 500), - this.addTracksOnReconnect = false, + this.addTracksOnReconnect = true, + this.offlineTimeout = const Duration(seconds: 60), }); /// Reconnection turned off. @@ -27,9 +28,14 @@ final class ReconnectConfig { /// Additional delay added for every following attempt. final Duration delay; - /// Whether local tracks are published again after reconnecting. + /// Whether local tracks are published again (with new ids) after + /// reconnecting. final bool addTracksOnReconnect; + /// How long a reconnection waits for the device to get a network before + /// it fails with `ReconnectionFailed`. + final Duration offlineTimeout; + bool get isEnabled => maxAttempts > 0; /// Delay before attempt [attempt] (1-based): @@ -43,11 +49,17 @@ final class ReconnectConfig { other.maxAttempts == maxAttempts && other.initialDelay == initialDelay && other.delay == delay && - other.addTracksOnReconnect == addTracksOnReconnect; + other.addTracksOnReconnect == addTracksOnReconnect && + other.offlineTimeout == offlineTimeout; @override - int get hashCode => - Object.hash(maxAttempts, initialDelay, delay, addTracksOnReconnect); + int get hashCode => Object.hash( + maxAttempts, + initialDelay, + delay, + addTracksOnReconnect, + offlineTimeout, + ); } /// Connection status of the client. diff --git a/lib/src/rtc/flutter_webrtc_adapter_mappers.dart b/lib/src/rtc/flutter_webrtc_adapter_mappers.dart index 0e98282..0a1c878 100644 --- a/lib/src/rtc/flutter_webrtc_adapter_mappers.dart +++ b/lib/src/rtc/flutter_webrtc_adapter_mappers.dart @@ -23,15 +23,6 @@ webrtc.RTCSessionDescription _toNativeDescription( SessionDescription description, ) => webrtc.RTCSessionDescription(description.sdp, description.type.name); -TransceiverDirection _fromNativeDirection(webrtc.TransceiverDirection d) => - switch (d) { - webrtc.TransceiverDirection.SendRecv => TransceiverDirection.sendrecv, - webrtc.TransceiverDirection.SendOnly => TransceiverDirection.sendonly, - webrtc.TransceiverDirection.RecvOnly => TransceiverDirection.recvonly, - webrtc.TransceiverDirection.Inactive => TransceiverDirection.inactive, - webrtc.TransceiverDirection.Stopped => TransceiverDirection.stopped, - }; - webrtc.TransceiverDirection _toNativeDirection(TransceiverDirection d) => switch (d) { TransceiverDirection.sendrecv => webrtc.TransceiverDirection.SendRecv, diff --git a/lib/src/rtc/flutter_webrtc_adapter_peer_connection.dart b/lib/src/rtc/flutter_webrtc_adapter_peer_connection.dart index d915768..7fd42df 100644 --- a/lib/src/rtc/flutter_webrtc_adapter_peer_connection.dart +++ b/lib/src/rtc/flutter_webrtc_adapter_peer_connection.dart @@ -200,16 +200,6 @@ final class _Transceiver implements RtcRtpTransceiver { @override String? get mid => _midOrNull(_transceiver.mid); - @override - Future getDirection() async => - _fromNativeDirection(await _transceiver.getDirection()); - - @override - Future getCurrentDirection() async { - final direction = await _transceiver.getCurrentDirection(); - return direction == null ? null : _fromNativeDirection(direction); - } - @override RtcRtpSender get sender => _Sender(_transceiver.sender); diff --git a/lib/src/rtc/rtc_adapter.dart b/lib/src/rtc/rtc_adapter.dart index 557dc1e..29df4bb 100644 --- a/lib/src/rtc/rtc_adapter.dart +++ b/lib/src/rtc/rtc_adapter.dart @@ -99,11 +99,6 @@ abstract interface class RtcRtpTransceiver { /// Media id; `null` until a local description was applied. String? get mid; - Future getDirection(); - - /// Negotiated direction; `null` before the first negotiation. - Future getCurrentDirection(); - RtcRtpSender get sender; RtcRtpReceiver get receiver; diff --git a/lib/src/signaling/network_change_detector.dart b/lib/src/signaling/network_change_detector.dart new file mode 100644 index 0000000..6c73ff5 --- /dev/null +++ b/lib/src/signaling/network_change_detector.dart @@ -0,0 +1,84 @@ +import 'dart:async'; + +import 'package:fishjam_client/src/logger.dart'; +import 'package:fishjam_client/src/signaling/network_source.dart'; + +enum NetworkChange { + /// The device went offline. + lost, + + /// The device is online again after being offline. + available, + + /// A network went away while another remains. + changed, +} + +/// Compares each OS report with the previous one and calls [_onChanged] with +/// the kind of change. If the OS cannot be asked, logs a warning and reports +/// nothing (SPEC §6.1). +class NetworkChangeDetector { + NetworkChangeDetector({ + required this._source, + required this._logger, + required this._onChanged, + }); + + final NetworkSource _source; + final FishjamLogger _logger; + final void Function(NetworkChange change) _onChanged; + + StreamSubscription>? _subscription; + int _generation = 0; + + /// `null` until [start] reads the first value. + Set? _previousNetworkStates; + + /// Starts watching with a new baseline (stops a previous watch first). + Future start() async { + stop(); + final generation = _generation; + try { + // Read first: a platform that cannot be asked reports a failed + // subscription to the framework's error handler, not to the stream. + final networks = await _source.getNetworkTypes(); + if (generation != _generation) return; + _previousNetworkStates = networks; + _subscription = _source.networkTypesChanged.listen( + _onNetworksReported, + onError: (Object e) { + _logger.warn('Network reports failed; stopped watching them', e); + stop(); + }, + ); + } on Object catch (e) { + if (generation == _generation) { + _logger.warn('Cannot read the network state; not watching it', e); + } + } + } + + void stop() { + _generation++; + unawaited(_subscription?.cancel()); + _subscription = null; + _previousNetworkStates = null; + } + + void _onNetworksReported(Set networks) { + final change = _classifyNetworkChange(_previousNetworkStates!, networks); + _previousNetworkStates = networks; + if (change != null) _onChanged(change); + } + + static NetworkChange? _classifyNetworkChange( + Set previous, + Set current, + ) { + if (current.isEmpty) return previous.isEmpty ? null : NetworkChange.lost; + if (previous.isEmpty) return NetworkChange.available; + // Only losing a network can break the connection's path; gaining one + // (mobile data next to Wi-Fi, a VPN coming up) leaves it in place. + return current.containsAll(previous) ? null : NetworkChange.changed; + } +} diff --git a/lib/src/signaling/network_source.dart b/lib/src/signaling/network_source.dart new file mode 100644 index 0000000..242dcf5 --- /dev/null +++ b/lib/src/signaling/network_source.dart @@ -0,0 +1,9 @@ +/// Reports which kinds of network the device uses, as the OS sees them. +abstract interface class NetworkSource { + /// The kinds of network in use (e.g. `wifi`, `mobile`); empty when + /// offline. Throws when the OS cannot be asked. + Future> getNetworkTypes(); + + /// Every report the OS makes; a report may repeat the previous one. + Stream> get networkTypesChanged; +} diff --git a/lib/src/signaling/reconnect.dart b/lib/src/signaling/reconnect.dart index 8ff18f0..0050bef 100644 --- a/lib/src/signaling/reconnect.dart +++ b/lib/src/signaling/reconnect.dart @@ -35,9 +35,11 @@ abstract interface class ReconnectHost { Future startSession(Object? peerMetadata); /// Tears the session down and emits `Disconnected(reason)`. Called only - /// when the manager gives up on its own: [startSession] threw on the last - /// attempt (`ReconnectionFailed` was emitted just before). Later reports - /// about the abandoned session must be ignored. + /// where no caller waits for a result (`ReconnectionFailed` was emitted + /// just before): [startSession] threw on the last attempt, the device was + /// offline for `offlineTimeout`, or the network changed after the total + /// attempt cap was reached. Later reports about the abandoned session must + /// be ignored. void endSession(DisconnectReason reason); /// Publishes [track]'s media again; the new track gets a new id. @@ -59,16 +61,19 @@ abstract interface class ReconnectHost { /// - The caller reports every loss of the current session — including a new /// session that fails to open — through [onRetryableLoss] or /// [onTerminalClose], and every `connected` through [onSessionJoined]. -/// - The manager emits only `ReconnectionStarted`, `ReconnectionFailed`, -/// `Reconnected` and, on the first `connected` of a reconnection, -/// `LocalTrackRemoved` for every snapshotted track (restored tracks get -/// new ids). It never emits `Disconnected`: when +/// - The caller reports every network change through [onNetworkLost] or +/// [onNetworkAvailable]. While joined it first reports a lost or changed +/// network as a retryable loss, so the change then acts on the +/// reconnection that loss started. +/// - The manager emits only `ReconnectionStarted`, `ReconnectionFailed` and +/// `Reconnected`. It never emits `Disconnected`: when /// [onRetryableLoss] returns `false` the caller ends the session with /// `Disconnected(reconnectionFailed)` if [state] is /// [ReconnectState.failed], else `Disconnected(connectionLost)`; after a /// terminal close the caller emits the error event and `Disconnected`. -/// The one exception is [ReconnectHost.startSession] throwing on the last -/// attempt, where the manager calls [ReconnectHost.endSession]. +/// Where no caller waits for a result (timers, network events), the +/// manager ends the session itself through [ReconnectHost.endSession] (the +/// cases are listed there). /// - With `addTracksOnReconnect`, only tracks with a live media track are /// re-added. Muted tracks (no media track) and ended tracks are /// not; a failed re-add does not stop the others. @@ -80,14 +85,28 @@ class ReconnectManager { ReconnectState _state = ReconnectState.idle; ReconnectSnapshot? _snapshot; - Timer? _timer; + Timer? _nextAttemptTimer; - /// Number of attempts started in the current reconnection. - int _attempt = 0; + /// Number of attempts started on the current network in the current + /// reconnection (the budget of `maxAttempts`). + int _attemptsOnThisNetwork = 0; - /// Whether `LocalTrackRemoved` was emitted for the snapshot's tracks in - /// the current reconnection (once, on the first `connected`). - bool _removalsReported = false; + /// Number of attempts started in the current reconnection on any network: + /// a network that keeps changing restarts the budget, so this caps it. + int _totalAttempts = 0; + + /// Whether the latest attempt is still open: started, and neither lost nor + /// left waiting for the network (it stays set while a joined attempt + /// restores its tracks). + bool _attemptInProgress = false; + + /// While offline no attempt starts, and a loss waits for the network + /// instead of using up the budget. + bool _online = true; + + /// Runs while reconnecting offline; fires [_failAndEndSession] at + /// `offlineTimeout`. + Timer? _offlineTimer; /// Bumped whenever the current attempt is superseded, so a pending track /// restoration of an outdated session does not complete the reconnection. @@ -109,22 +128,27 @@ class ReconnectManager { if (!_config.isEnabled) return false; _state = ReconnectState.reconnecting; _snapshot = snapshot; - _attempt = 0; - _removalsReported = false; _host ..holdQueue() ..emit(const ReconnectionStarted()); - _scheduleNextAttempt(); + if (_online) { + _scheduleNextAttempt(); + } else { + _startOfflineTimeout(); + } return true; case ReconnectState.reconnecting: - if (_timer != null) return true; + if (_nextAttemptTimer != null) return true; _generation++; - if (_attempt < _config.maxAttempts) { + _attemptInProgress = false; + // Offline: wait for the network; do not count an attempt. + if (!_online) return true; + if (_attemptsOnThisNetwork < _config.maxAttempts && + _totalAttempts < _maxTotalAttempts) { _scheduleNextAttempt(); return true; } - _finish(ReconnectState.failed); - _host.emit(const ReconnectionFailed()); + _markFailed(); return false; } } @@ -136,64 +160,126 @@ class ReconnectManager { if (_state == ReconnectState.reconnecting) { _finish(ReconnectState.failed); } else { - _cancelTimer(); + _cancelNextAttemptTimer(); } } /// A session reached `connected`. In [ReconnectState.reconnecting]: - /// reports every snapshotted track removed (once per reconnection), /// restores tracks, releases the queue and emits `Reconnected`. /// - /// Only live tracks are restored; ended or muted ones are skipped and a - /// failed restore does not stop the others. If the session is lost - /// while restoring, `Reconnected` is not emitted for it. + /// Only live tracks are restored, all at once; ended or muted ones are + /// skipped and a failed restore does not stop the others. If the session + /// is lost or the device goes offline before it is restored, `Reconnected` + /// is not emitted for it. Future onSessionJoined() async { if (_state != ReconnectState.reconnecting) return; - final generation = _generation; - final tracks = _snapshot!.tracks; - if (!_removalsReported) { - _removalsReported = true; - for (final track in tracks) { - _host.emit(LocalTrackRemoved(track.trackId)); - } + if (_online && _config.addTracksOnReconnect) { + final generation = _generation; + // Start all restores at once so they usually share one renegotiation. + // A failed restore is ignored, so the others still run; that track is + // not published again and the app gets no `LocalTrackAdded` for it. + await Future.wait([ + for (final track in _snapshot!.tracks) + if (track.rtcTrack?.isLive ?? false) + _host.restoreTrack(track).then((_) {}, onError: (_) {}), + ]); + if (generation != _generation) return; } - if (_config.addTracksOnReconnect) { - for (final track in tracks) { - if (!(track.rtcTrack?.isLive ?? false)) continue; - try { - await _host.restoreTrack(track); - } on Object { - // The host reports the failure; the other tracks still go. - } - if (generation != _generation) return; - } + if (!_online) { + // The device went offline before or during the restore. This session + // will not survive, so do not emit `Reconnected`; wait for the network + // instead (the offline timeout is already running). + _attemptInProgress = false; + return; } _finish(ReconnectState.idle, release: true); _host.emit(const Reconnected()); } + /// The device has no network. While reconnecting, a scheduled attempt is + /// cancelled and the reconnection waits, at most `offlineTimeout`, for + /// [onNetworkAvailable]. Outside a reconnection it is only remembered. + void onNetworkLost() { + _online = false; + if (_state != ReconnectState.reconnecting) return; + _cancelNextAttemptTimer(); + _startOfflineTimeout(); + } + + /// The device is online again or switched networks. While reconnecting: + /// reset the attempt budget and start a new attempt now. + void onNetworkAvailable() { + _online = true; + _cancelOfflineTimer(); + if (_state != ReconnectState.reconnecting) return; + if (_totalAttempts >= _maxTotalAttempts) { + // If an attempt is still running, its result decides; otherwise fail + // now. + if (!_attemptInProgress) _failAndEndSession(); + return; + } + _cancelNextAttemptTimer(); + _startAttempt(1); + } + /// Cancels everything and returns to [ReconnectState.idle] (on `leave()` - /// and on a new `connect()`); clears the queue. - void reset() => _finish(ReconnectState.idle); + /// and on a new `connect()`); clears the queue and forgets that the device + /// was offline. + void reset() { + _finish(ReconnectState.idle); + _online = true; + } + + int get _maxTotalAttempts => 3 * _config.maxAttempts; + + /// Calls [_failAndEndSession] after `offlineTimeout` unless + /// [onNetworkAvailable] comes first. + void _startOfflineTimeout() { + _offlineTimer ??= Timer(_config.offlineTimeout, () { + _offlineTimer = null; + _failAndEndSession(); + }); + } + + /// Emits `ReconnectionFailed`. The caller ends the session (used when + /// [onRetryableLoss] returns `false`). + void _markFailed() { + _finish(ReconnectState.failed); + _host.emit(const ReconnectionFailed()); + } + + /// Emits `ReconnectionFailed` and ends the session. Used where no caller is + /// waiting for a result (timers, network events). + void _failAndEndSession() { + _markFailed(); + _host.endSession(DisconnectReason.reconnectionFailed); + } void _scheduleNextAttempt() { - final attempt = _attempt + 1; - _timer = Timer(_config.delayBeforeAttempt(attempt), () { - _timer = null; - _attempt = attempt; - final generation = ++_generation; - unawaited( - _host - .startSession(_snapshot!.peerMetadata) - .then( - (_) {}, - onError: (Object error) => - _onStartFailed(generation, RetryableClose(error: error)), - ), - ); + final attempt = _attemptsOnThisNetwork + 1; + _nextAttemptTimer = Timer(_config.delayBeforeAttempt(attempt), () { + _nextAttemptTimer = null; + _startAttempt(attempt); }); } + /// Starts [attempt], superseding any attempt still in progress. + void _startAttempt(int attempt) { + _attemptsOnThisNetwork = attempt; + _totalAttempts++; + _attemptInProgress = true; + final generation = ++_generation; + unawaited( + _host + .startSession(_snapshot!.peerMetadata) + .then( + (_) {}, + onError: (Object error) => + _onStartFailed(generation, RetryableClose(error: error)), + ), + ); + } + /// [ReconnectHost.startSession] threw. Unless the attempt was already /// superseded (its loss reported, a terminal close, `reset`), this is the /// attempt's retryable loss. @@ -207,11 +293,14 @@ class ReconnectManager { /// Leaves the current reconnection: cancels the timer, invalidates pending /// restores, and releases (success) or clears (otherwise) the queue. void _finish(ReconnectState state, {bool release = false}) { - _cancelTimer(); + _cancelNextAttemptTimer(); + _cancelOfflineTimer(); _generation++; _state = state; _snapshot = null; - _attempt = 0; + _attemptsOnThisNetwork = 0; + _totalAttempts = 0; + _attemptInProgress = false; if (release) { _host.releaseQueue(); } else { @@ -219,8 +308,13 @@ class ReconnectManager { } } - void _cancelTimer() { - _timer?.cancel(); - _timer = null; + void _cancelNextAttemptTimer() { + _nextAttemptTimer?.cancel(); + _nextAttemptTimer = null; + } + + void _cancelOfflineTimer() { + _offlineTimer?.cancel(); + _offlineTimer = null; } } diff --git a/lib/src/signaling/transport.dart b/lib/src/signaling/transport.dart index 7ef59ae..60d1803 100644 --- a/lib/src/signaling/transport.dart +++ b/lib/src/signaling/transport.dart @@ -78,15 +78,24 @@ final class WebSocketTransport implements SignalingTransport { /// Ignored text frames are logged at debug level with [logger]. WebSocketTransport({ this.pingInterval = const Duration(seconds: 15), + this.connectTimeout = const Duration(seconds: 10), FishjamLogger? logger, }) : _logger = logger ?? FishjamLogger.disabled(); /// Interval between client pings, and the pong timeout. final Duration pingInterval; + /// Limit for opening the socket: connection, TLS and upgrade together. + final Duration connectTimeout; + final FishjamLogger _logger; WebSocket? _socket; + + /// A dedicated `HttpClient` for this socket. Force-closing it is the only + /// way to abort a connect that is still in progress (on timeout or + /// [close]). + HttpClient? _httpClient; final _incoming = StreamController(); final _closed = Completer(); @@ -101,13 +110,20 @@ final class WebSocketTransport implements SignalingTransport { @override Future connect(Uri url) async { - if (_socket != null) throw StateError('A transport can be used once'); + if (_httpClient != null) throw StateError('A transport can be used once'); + final httpClient = _httpClient = HttpClient(); try { - _socket = await WebSocket.connect(url.toString()) - ..pingInterval = pingInterval; + _socket = + await WebSocket.connect( + url.toString(), + customClient: httpClient, + ).timeout(connectTimeout) + ..pingInterval = pingInterval; } on Object catch (e) { _finish(TransportClosed(error: e)); throw TransportException('Cannot connect to $url', e); + } finally { + httpClient.close(force: true); } _socket!.listen( (data) { @@ -136,6 +152,7 @@ final class WebSocketTransport implements SignalingTransport { Future close([int code = 1000, String? reason]) async { final socket = _socket; if (socket == null) { + _httpClient?.close(force: true); _finish(TransportClosed(code: code, reason: reason)); return; } diff --git a/pubspec.lock b/pubspec.lock index 8c1b1ef..3d2f08e 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -73,6 +73,22 @@ packages: url: "https://pub.dev" source: hosted version: "1.19.1" + connectivity_plus: + dependency: "direct main" + description: + name: connectivity_plus + sha256: d7749fd36d8b404a4935612b94eca7c3ce3705377ab678e24ff2a4301436f741 + url: "https://pub.dev" + source: hosted + version: "7.3.2" + connectivity_plus_platform_interface: + dependency: "direct dev" + description: + name: connectivity_plus_platform_interface + sha256: "3c09627c536d22fd24691a905cdd8b14520de69da52c7a97499c8be5284a32ed" + url: "https://pub.dev" + source: hosted + version: "2.1.0" convert: dependency: transitive description: @@ -105,6 +121,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.8.2" + dbus: + dependency: transitive + description: + name: dbus + sha256: f7db8504e715b835f5ebe5fe74604ccd961d504c73389a5d8e81bafc2e2f3612 + url: "https://pub.dev" + source: hosted + version: "0.8.0" fake_async: dependency: "direct dev" description: @@ -147,6 +171,11 @@ packages: description: flutter source: sdk version: "0.0.0" + flutter_web_plugins: + dependency: transitive + description: flutter + source: sdk + version: "0.0.0" flutter_webrtc: dependency: "direct main" description: @@ -275,6 +304,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.19.0" + nm: + dependency: transitive + description: + name: nm + sha256: "2ae480aa3379716b8bb58dfebb168e2919394c92b629efaaa0baaa5d140786e6" + url: "https://pub.dev" + source: hosted + version: "0.6.0" objective_c: dependency: transitive description: @@ -347,6 +384,14 @@ packages: url: "https://pub.dev" source: hosted version: "2.3.0" + petitparser: + dependency: transitive + description: + name: petitparser + sha256: "2f1564aadb14215e478563b5406cb80729d022158c01dc50c96013c2f7eb4790" + url: "https://pub.dev" + source: hosted + version: "7.1.0" platform: dependency: transitive description: @@ -528,6 +573,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.1.0" + xml: + dependency: transitive + description: + name: xml + sha256: "0fe8cc98946def31cbcbc8734aa4067031e6f65ea3a78a508060bfed2200bfeb" + url: "https://pub.dev" + source: hosted + version: "7.1.0" yaml: dependency: transitive description: diff --git a/pubspec.yaml b/pubspec.yaml index 1321b99..b6b4e8f 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -13,6 +13,7 @@ environment: dependencies: collection: ^1.19.0 + connectivity_plus: ^7.3.2 flutter: sdk: flutter flutter_webrtc: ^1.6.2 @@ -22,6 +23,7 @@ dependencies: uuid: ^4.6.0 dev_dependencies: + connectivity_plus_platform_interface: ^2.1.0 fake_async: ^1.3.3 flutter_test: sdk: flutter diff --git a/test/client/audio_routing_test.dart b/test/client/audio_routing_test.dart index 1dfc8c3..b2a58a3 100644 --- a/test/client/audio_routing_test.dart +++ b/test/client/audio_routing_test.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:fishjam_client/fishjam_client.dart'; import 'package:fishjam_client/src/proto/fishjam/media_events/server/server.pb.dart' as server; @@ -62,6 +64,31 @@ void main() { expect(calls, ['enterCommunicationMode', 'exitCommunicationMode']); }); + test('a connect() made while the previous session is torn down opens ' + 'after it, so the mode is exited and entered again', () async { + await h.join(); + final gate = h.transport.closeGate = Completer(); + final leaving = h.client.leave(); + await settle(); + final connecting = h.client.connect(url: 'ws://f', token: 't'); + await settle(); + expect(h.transport.isOpen, isFalse); + + gate.complete(); + await leaving; + await settle(); + await h.rejoin(); + await connecting; + + expect(calls, [ + 'enterCommunicationMode', + 'exitCommunicationMode', + 'enterCommunicationMode', + ]); + expect(h.events.whereType(), hasLength(1)); + expect(h.events.last, isA()); + }); + test('never entered when connect() fails before Joined', () async { final connecting = h.client.connect(url: 'ws://f', token: 't'); final failed = expectLater( diff --git a/test/client/client_fixtures_test.dart b/test/client/client_fixtures_test.dart index 9cc6c55..7fd78be 100644 --- a/test/client/client_fixtures_test.dart +++ b/test/client/client_fixtures_test.dart @@ -111,16 +111,21 @@ Object? _sync(void Function() call) { return null; } +/// The fixture's `reconnect` settings; missing ones keep the SDK defaults. ReconnectConfig _reconnectConfig(Map config) { + const defaults = ReconnectConfig(); final json = config['reconnect'] as Map?; - if (json == null) return const ReconnectConfig(); + if (json == null) return defaults; + Duration? ms(String key) => switch (json[key]) { + final int value => Duration(milliseconds: value), + _ => null, + }; return ReconnectConfig( - maxAttempts: (json['maxAttempts'] as int?) ?? 3, - initialDelay: Duration( - milliseconds: (json['initialDelayMs'] as int?) ?? 500, - ), - delay: Duration(milliseconds: (json['delayMs'] as int?) ?? 500), - addTracksOnReconnect: (json['addTracksOnReconnect'] as bool?) ?? false, + maxAttempts: json['maxAttempts'] as int? ?? defaults.maxAttempts, + initialDelay: ms('initialDelayMs') ?? defaults.initialDelay, + delay: ms('delayMs') ?? defaults.delay, + addTracksOnReconnect: + json['addTracksOnReconnect'] as bool? ?? defaults.addTracksOnReconnect, ); } diff --git a/test/client/client_harness.dart b/test/client/client_harness.dart index 902b79c..7528acc 100644 --- a/test/client/client_harness.dart +++ b/test/client/client_harness.dart @@ -11,6 +11,7 @@ import 'package:fishjam_client/src/proto/fishjam/media_events/server/server.pb.d as server; import 'package:fishjam_client/src/proto/fishjam/peer_notifications.pb.dart'; +import '../fakes/fake_network.dart'; import '../fakes/fake_rtc.dart'; import '../fakes/fake_transport.dart'; import '../public/native_fakes.dart'; @@ -30,6 +31,7 @@ class ClientHarness { reconnect: reconnect, rtcAdapter: rtc, transportFactory: transports.call, + networkSource: network, audioOutput: audioOutput, ); _subscription = client.events.listen(events.add); @@ -38,6 +40,9 @@ class ClientHarness { /// Transports created by the client. final transports = FakeTransportFactory(); + /// The device's networks, as the OS reports them. + final network = FakeNetworkSource(); + /// The WebRTC fake. final FakeRtcAdapter rtc; @@ -87,9 +92,19 @@ class ClientHarness { peerMetadata: peerMetadata, ); await settle(); + await rejoin(roomType: roomType, endpoints: endpoints); + await connected; + } + + /// The server lets the current transport in: authenticates it with + /// [roomType] and sends `connected` with [endpoints] besides the local + /// peer (a first join or a reconnection attempt). + Future rejoin({ + PeerMessage_RoomType roomType = PeerMessage_RoomType.ROOM_TYPE_CONFERENCE, + Map endpoints = const {}, + }) async { await authenticate(roomType: roomType); await connectedEvent(endpoints: endpoints); - await connected; } /// The server accepts the `authRequest` on the current transport. @@ -126,21 +141,29 @@ class ClientHarness { await settle(); } - /// The server runs one negotiation cycle: `offerData`, then an answer - /// that maps the mids of the client's offer. + /// The server runs one negotiation cycle: [offerData], then [answer]. Future negotiate({int audio = 0, int video = 0}) async { - await receive( - server.MediaEvent( - offerData: server.MediaEvent_OfferData( - tracksTypes: server.MediaEvent_OfferData_TrackTypes( - audio: audio, - video: video, - ), + await offerData(audio: audio, video: video); + await answer(); + } + + /// The server sends `offerData` asking for [audio] and [video] receive + /// transceivers. + Future offerData({int audio = 0, int video = 0}) => receive( + server.MediaEvent( + offerData: server.MediaEvent_OfferData( + tracksTypes: server.MediaEvent_OfferData_TrackTypes( + audio: audio, + video: video, ), ), - ); + ), + ); + + /// The server answers the client's last offer, mapping its mids. + Future answer() { final offer = sentOf(peer.MediaEvent_Content.sdpOffer).last.sdpOffer; - await receive( + return receive( server.MediaEvent( sdpAnswer: server.MediaEvent_SdpAnswer( sdp: 'v=0\r\ns=fake-answer\r\n', diff --git a/test/client/connectivity_missing_plugin_test.dart b/test/client/connectivity_missing_plugin_test.dart new file mode 100644 index 0000000..4da938a --- /dev/null +++ b/test/client/connectivity_missing_plugin_test.dart @@ -0,0 +1,29 @@ +import 'package:fishjam_client/src/connectivity_network_source.dart'; +import 'package:fishjam_client/src/logger.dart'; +import 'package:fishjam_client/src/signaling/network_change_detector.dart'; +import 'package:flutter_test/flutter_test.dart'; + +// No platform implementation is registered here, as in an app's widget test. +void main() { + testWidgets('without the plugin, watching only logs a warning; nothing is ' + 'reported to the framework', (tester) async { + final logs = []; + final detector = NetworkChangeDetector( + source: ConnectivityNetworkSource(), + logger: FishjamLogger( + enabled: true, + sink: (level, message, {error, stackTrace}) => + logs.add('${level.name}: $message'), + ), + onChanged: (_) {}, + ); + + // Platform channels answer in real time. + await tester.runAsync(() async { + await detector.start(); + detector.stop(); + }); + + expect(logs, [startsWith('warn')]); + }); +} diff --git a/test/client/connectivity_network_source_test.dart b/test/client/connectivity_network_source_test.dart new file mode 100644 index 0000000..afb42e9 --- /dev/null +++ b/test/client/connectivity_network_source_test.dart @@ -0,0 +1,55 @@ +import 'dart:async'; + +import 'package:connectivity_plus_platform_interface/connectivity_plus_platform_interface.dart'; +import 'package:fishjam_client/src/connectivity_network_source.dart'; +import 'package:flutter_test/flutter_test.dart'; + +class _FakePlatform extends ConnectivityPlatform { + List current = [ConnectivityResult.wifi]; + final reports = StreamController>.broadcast(); + + @override + Future> checkConnectivity() async => current; + + @override + Stream> get onConnectivityChanged => reports.stream; +} + +void main() { + late _FakePlatform platform; + + setUp(() => ConnectivityPlatform.instance = platform = _FakePlatform()); + + test('reports the kinds of network in use', () async { + platform.current = [ConnectivityResult.mobile, ConnectivityResult.vpn]; + + expect(await ConnectivityNetworkSource().getNetworkTypes(), { + 'mobile', + 'vpn', + }); + }); + + test('none is reported as no network', () async { + final reports = >[]; + final subscription = ConnectivityNetworkSource().networkTypesChanged.listen( + reports.add, + ); + addTearDown(subscription.cancel); + + platform.reports + ..add([ConnectivityResult.none]) + ..add([ + ConnectivityResult.wifi, + ConnectivityResult.ethernet, + ConnectivityResult.bluetooth, + ConnectivityResult.satellite, + ConnectivityResult.other, + ]); + await Future.delayed(Duration.zero); + + expect(reports, [ + {}, + {'wifi', 'ethernet', 'bluetooth', 'satellite', 'other'}, + ]); + }); +} diff --git a/test/client/fishjam_client_test.dart b/test/client/fishjam_client_test.dart index 942e898..7031906 100644 --- a/test/client/fishjam_client_test.dart +++ b/test/client/fishjam_client_test.dart @@ -21,13 +21,13 @@ const _fast = ReconnectConfig( maxAttempts: 2, initialDelay: Duration(milliseconds: 20), delay: Duration(milliseconds: 20), + addTracksOnReconnect: false, ); const _fastRestore = ReconnectConfig( maxAttempts: 2, initialDelay: Duration(milliseconds: 20), delay: Duration(milliseconds: 20), - addTracksOnReconnect: true, ); Future _wait(int ms) async { @@ -162,6 +162,54 @@ void main() { expect(h.eventsOf().single.reason, DisconnectReason.left); }); + test('connect() during the teardown of the previous session opens its ' + 'socket once that session has ended', () async { + h = ClientHarness(); + await h.join(); + final gate = h.transport.closeGate = Completer(); + final leaving = h.client.leave(); + await settle(); + final connecting = h.client.connect( + url: 'ws://fishjam.test', + token: 'peer-token', + ); + await settle(); + expect(h.transport.isOpen, isFalse); + h.events.clear(); + + gate.complete(); + await leaving; + await settle(); + await h.rejoin(); + await connecting; + expect(h.events.map((e) => e.runtimeType), [ + Disconnected, + AuthSucceeded, + Joined, + ]); + }); + + test('leave() while connect() waits for the previous teardown settles ' + 'it; its socket never opens', () async { + h = ClientHarness(); + await h.join(); + final gate = h.transport.closeGate = Completer(); + final leaving = h.client.leave(); + await settle(); + final connecting = expectLater( + h.client.connect(url: 'ws://fishjam.test', token: 'peer-token'), + throwsA(isA()), + ); + await settle(); + await h.client.leave(); + + gate.complete(); + await leaving; + await connecting; + expect(h.transport.connectedUrl, isNull); + expect(h.client.status, ConnectionStatus.disconnected); + }); + test('audio-only room: AuthSucceeded reports it and video tracks are ' 'rejected synchronously (SPEC §20)', () async { h = ClientHarness(); @@ -385,6 +433,15 @@ void main() { }); group('reconnection (SPEC §6)', () { + /// Loses the session; the next attempt joins and restores the tracks. + Future reconnectAndRestore() async { + h.transport.serverClose(code: 1011, reason: 'boom'); + await _wait(40); + await h.rejoin(); + await h.negotiate(); + await settle(); + } + test('a peer connection failure replaces the session', () async { h = ClientHarness(reconnect: _fast); await h.join(); @@ -400,8 +457,7 @@ void main() { expect(h.transports.created, hasLength(2)); expect(firstTransport.closedByClient, isTrue); expect(firstPc.isClosed, isTrue); - await h.authenticate(); - await h.connectedEvent(); + await h.rejoin(); await settle(); expect(h.eventsOf(), hasLength(1)); expect(h.client.status, ConnectionStatus.joined); @@ -415,23 +471,9 @@ void main() { final adding = h.client.addTrack(h.track('audio')); final result = expectLater(adding, throwsA(anything)); await settle(); - await h.receive( - server.MediaEvent( - offerData: server.MediaEvent_OfferData( - tracksTypes: server.MediaEvent_OfferData_TrackTypes(), - ), - ), - ); - final offer = h.sentOf(_C.sdpOffer).last.sdpOffer; + await h.offerData(); h.pc.failNext('setRemoteDescription', StateError('bad answer')); - await h.receive( - server.MediaEvent( - sdpAnswer: server.MediaEvent_SdpAnswer( - sdp: 'v=0', - midToTrackId: offer.midToTrackId.entries, - ), - ), - ); + await h.answer(); await result; await settle(); expect(h.eventsOf(), hasLength(1)); @@ -453,8 +495,7 @@ void main() { h.transport.serverClose(code: 1011, reason: 'boom'); await _wait(40); - await h.authenticate(); - await h.connectedEvent(); + await h.rejoin(); // The restored camera is being negotiated. expect(h.sentOf(_C.renegotiateTracks), isNotEmpty); await h.negotiate(); @@ -476,8 +517,48 @@ void main() { expect(h.client.localPeer!.tracks.keys, [restored.trackId]); }); - test('tracks restored by an attempt that is then lost are reported ' - 'removed too', () async { + test( + 'a later reconnection reports each restored track removed once', + () async { + h = ClientHarness(reconnect: _fastRestore); + await h.join(); + final cam = await h.publish('video'); + h.events.clear(); + + await reconnectAndRestore(); + final cam2 = h.eventsOf().single.track.trackId; + await reconnectAndRestore(); + + expect(h.eventsOf().map((e) => e.trackId), [ + cam, + cam2, + ]); + expect(h.eventsOf(), hasLength(2)); + }, + ); + + test( + 'live tracks are restored together, in one negotiation cycle', + () async { + h = ClientHarness(reconnect: _fastRestore); + await h.join(); + await h.publish('video'); + await h.publish('audio'); + final renegotiations = h.sentOf(_C.renegotiateTracks).length; + h.events.clear(); + + await reconnectAndRestore(); + + expect(h.sentOf(_C.renegotiateTracks), hasLength(renegotiations + 1)); + final offer = h.sentOf(_C.sdpOffer).last.sdpOffer; + expect(offer.trackIdToMetadataJson, hasLength(2)); + expect(h.eventsOf(), hasLength(2)); + expect(h.eventsOf(), hasLength(1)); + }, + ); + + test('an attempt lost while its tracks are being restored adds none of ' + 'them; the next attempt restores them all', () async { h = ClientHarness(reconnect: _fastRestore); await h.join(); final cam = await h.publish('video'); @@ -486,26 +567,17 @@ void main() { h.transport.serverClose(code: 1011, reason: 'boom'); await _wait(40); - await h.authenticate(); - await h.connectedEvent(); - await h.negotiate(); // the camera is restored - final cam2 = h.eventsOf().single.track.trackId; - h.transport.serverClose(code: 1011, reason: 'boom'); // mic pending + await h.rejoin(); + h.transport.serverClose(code: 1011, reason: 'boom'); // restore pending await _wait(60); - await h.authenticate(); - await h.connectedEvent(); - await h.negotiate(); + await h.rejoin(); await h.negotiate(); await settle(); - expect(h.eventsOf().map((e) => e.trackId), [ - cam, - mic, - cam2, - ]); + expect(h.eventsOf().map((e) => e.trackId), [cam, mic]); final added = h.eventsOf().map((e) => e.track.trackId); - expect(added, hasLength(3)); - expect(h.client.localPeer!.tracks.keys, added.skip(1)); + expect(added, hasLength(2)); + expect(h.client.localPeer!.tracks.keys, added); expect(h.eventsOf(), hasLength(1)); }); @@ -518,8 +590,7 @@ void main() { h.events.clear(); h.transport.loseConnection(); await _wait(40); - await h.authenticate(); - await h.connectedEvent(); + await h.rejoin(); await settle(); expect(h.eventsOf().single.trackId, cam); expect(h.eventsOf(), isEmpty); @@ -587,8 +658,7 @@ void main() { client.events.listen(events.add); final connecting = client.connect(url: 'ws://f', token: 't'); await settle(); - await h.authenticate(); - await h.connectedEvent(); + await h.rejoin(); await connecting; failing = true; h.transport.serverClose(code: 1011, reason: 'boom'); @@ -644,8 +714,7 @@ void main() { expect(h.eventsOf(), isEmpty); await _wait(40); - await h.authenticate(); - await h.connectedEvent(); + await h.rejoin(); // Joined, but the track is still being restored. expect(h.eventsOf(), hasLength(2)); expect(h.client.status, ConnectionStatus.reconnecting); @@ -682,8 +751,7 @@ void main() { h.events.clear(); h.transport.serverClose(code: 1011, reason: 'boom'); await _wait(40); - await h.authenticate(); - await h.connectedEvent(endpoints: {'bob': endpoints['bob']!}); + await h.rejoin(endpoints: {'bob': endpoints['bob']!}); await settle(); final described = [ @@ -710,6 +778,329 @@ void main() { }); }); + group('network changes (SPEC §6.1)', () { + const slow = ReconnectConfig( + initialDelay: Duration(seconds: 5), + delay: Duration(seconds: 5), + ); + const offlineFor50ms = ReconnectConfig( + offlineTimeout: Duration(milliseconds: 50), + ); + + test('a change while joined replaces the session at once', () async { + h = ClientHarness(reconnect: slow); + await h.join(); + final first = h.transport; + h.network.report({'mobile'}); + await settle(); + expect(h.eventsOf(), hasLength(1)); + expect(h.transports.created, hasLength(2)); + expect(first.closedByClient, isTrue); + await h.rejoin(); + expect(h.eventsOf(), hasLength(1)); + expect(h.eventsOf(), isEmpty); + }); + + test('the new session does not wait for the old socket to close', () async { + h = ClientHarness(reconnect: slow); + await h.join(); + h.transport.closeGate = Completer(); + h.network.report({'mobile'}); + await settle(); + expect(h.transports.created, hasLength(2)); + expect(h.transport.sentMessages.single.hasAuthRequest(), isTrue); + }); + + test('an old socket whose close fails is not an unhandled error', () async { + h = ClientHarness(reconnect: slow); + await h.join(); + final gate = h.transport.closeGate = Completer(); + h.network.report({'mobile'}); + await settle(); + gate.completeError(StateError('socket failed')); + await _wait(10); + expect(h.transports.created, hasLength(2)); + }); + + test('offline: the attempt waits for the network, then starts at ' + 'once', () async { + h = ClientHarness(reconnect: _fast); + await h.join(); + h.network.report({}); + await _wait(100); + expect(h.eventsOf(), hasLength(1)); + expect(h.transports.created, hasLength(1)); + h.network.report({'mobile'}); + await settle(); + expect(h.transports.created, hasLength(2)); + }); + + test('gaining a network next to the current one changes nothing', () async { + h = ClientHarness(reconnect: _fast); + await h.join(); + h.network.report({'wifi', 'mobile'}); + await _wait(40); + expect(h.eventsOf(), isEmpty); + expect(h.transports.created, hasLength(1)); + }); + + test('a brief offline while switching networks is one change', () async { + h = ClientHarness(reconnect: slow); + await h.join(); + h.network.report({}); + await _wait(10); + h.network.report({'mobile'}); + await _wait(100); + expect(h.eventsOf(), hasLength(1)); + expect(h.transports.created, hasLength(2)); + }); + + test('offline for offlineTimeout ends the reconnection', () async { + h = ClientHarness(reconnect: offlineFor50ms); + await h.join(); + h.network.report({}); + await _wait(100); + expect(h.eventsOf(), hasLength(1)); + expect( + h.eventsOf().single.reason, + DisconnectReason.reconnectionFailed, + ); + expect(h.client.status, ConnectionStatus.disconnected); + }); + + group('an attempt replaced after its Joined has its remote endpoints ' + 'reported gone before the next Joined (D22)', () { + final bob = server.MediaEvent_Endpoint( + endpointType: 'exwebrtc', + metadataJson: '{"peer":{"name":"bob"},"server":{}}', + trackIdToTrack: { + 'bob:cam': server.MediaEvent_Track( + metadataJson: '{"type":"camera","paused":false}', + ), + }.entries, + ); + + List remoteEvents() => [ + for (final e in h.events) + switch (e) { + TrackRemoved(:final track) => 'TrackRemoved ${track.trackId}', + PeerLeft(:final peer) => 'PeerLeft ${peer.id}', + Joined(:final peers) => 'Joined ${[for (final p in peers) p.id]}', + _ => null, + }, + ].nonNulls.toList(); + + test('one that joined while offline', () async { + h = ClientHarness(reconnect: slow); + await h.join(); + h.events.clear(); + h.network.report({'mobile'}); + await _wait(10); + h.network.report({}); + await _wait(10); + await h.rejoin(endpoints: {'bob': bob}); + h.network.report({'wifi'}); + await _wait(10); + expect(h.transports.created, hasLength(3)); + await h.rejoin(); + await settle(); + + expect(remoteEvents(), [ + 'Joined [bob]', + 'TrackRemoved bob:cam', + 'PeerLeft bob', + 'Joined []', + ]); + expect(h.eventsOf(), hasLength(1)); + }); + + test('one still restoring its tracks', () async { + h = ClientHarness(reconnect: _fastRestore); + await h.join(); + await h.publish('audio'); + h.events.clear(); + h.network.report({'mobile'}); + await _wait(10); + // No negotiation follows, so the restore stays pending. + await h.rejoin(endpoints: {'bob': bob}); + h.network.report({'wifi'}); + await _wait(10); + expect(h.transports.created, hasLength(3)); + await h.rejoin(); + await settle(); + + expect(remoteEvents(), [ + 'Joined [bob]', + 'TrackRemoved bob:cam', + 'PeerLeft bob', + 'Joined []', + ]); + }); + }); + + test('an attempt that joins while offline reports the old local tracks ' + 'removed right after its Joined', () async { + h = ClientHarness(reconnect: slow); + await h.join(); + final cam = await h.publish('video'); + h.events.clear(); + List sequence() => [ + for (final e in h.events) + switch (e) { + Joined() => 'Joined', + LocalTrackAdded(:final track) => 'added ${track.trackId}', + LocalTrackRemoved(:final trackId) => 'removed $trackId', + Reconnected() => 'Reconnected', + _ => null, + }, + ].nonNulls.toList(); + + h.network.report({'mobile'}); + await _wait(10); + h.network.report({}); + await _wait(10); + await h.rejoin(); + expect(sequence(), ['Joined', 'removed $cam']); + + h.network.report({'wifi'}); + await _wait(10); + await h.rejoin(); + await h.negotiate(); + await settle(); + final cam2 = h.eventsOf().single.track.trackId; + expect(sequence(), [ + 'Joined', + 'removed $cam', + 'Joined', + 'added $cam2', + 'Reconnected', + ]); + }); + + group('only local tracks the app was told about are reported removed and ' + 'restored', () { + List localEvents() => [ + for (final e in h.events) + switch (e) { + LocalTrackAdded(:final track) => 'added ${track.trackId}', + LocalTrackRemoved(:final trackId) => 'removed $trackId', + Reconnected() => 'Reconnected', + _ => null, + }, + ].nonNulls.toList(); + + test('a track still being added when the network changes', () async { + h = ClientHarness(reconnect: _fastRestore); + await h.join(); + final adding = h.client.addTrack(h.track('audio')); + await settle(); + h.network.report({'mobile'}); + await expectLater(adding, throwsStateError); + await _wait(10); + expect(h.transports.created, hasLength(2)); + await h.rejoin(); + await settle(); + + expect(localEvents(), ['Reconnected']); + }); + + test('a track restored by an attempt that failed before announcing ' + 'it', () async { + h = ClientHarness(reconnect: _fastRestore); + await h.join(); + final first = await h.publish('audio'); + h.events.clear(); + h.transport.serverClose(code: 1011, reason: 'boom'); + await _wait(40); + await h.rejoin(); + await h.offerData(); + h.pc.setConnectionState(PeerConnectionState.failed); + await settle(); + // The answer reaches the attempt after it was given up; the next + // attempt starts 40 ms after the loss. + await h.answer(); + await _wait(40); + await h.rejoin(); + await h.negotiate(); + await settle(); + + final restored = h.eventsOf().single.track.trackId; + expect(localEvents(), [ + 'removed $first', + 'added $restored', + 'Reconnected', + ]); + }); + }); + + test('connect() from a ReconnectionFailed listener is not failed by the ' + 'teardown of the given-up session', () async { + h = ClientHarness(reconnect: offlineFor50ms); + await h.join(); + Object? retryError; + var retried = false; + final retry = h.client.events.listen((e) { + if (e is! ReconnectionFailed) return; + retried = true; + unawaited( + h.client + .connect(url: 'ws://fishjam.test', token: 'peer-token') + .then((_) {}, onError: (Object e) => retryError = e), + ); + }); + addTearDown(retry.cancel); + h.network.report({}); + await _wait(100); + expect(retried, isTrue); + expect(retryError, isNull); + await h.rejoin(); + await settle(); + + expect(retryError, isNull); + expect(h.client.status, ConnectionStatus.joined); + }); + + test('a report equal to the start value changes nothing', () async { + h = ClientHarness(reconnect: _fast); + await h.join(); + h.network.report({'wifi'}); + await _wait(40); + expect(h.eventsOf(), isEmpty); + }); + + test('a change before Joined is ignored', () async { + h = ClientHarness(reconnect: _fast); + final connected = h.client.connect(url: 'ws://fishjam.test', token: 't'); + await settle(); + h.network.report({'mobile'}); + await settle(); + await h.rejoin(); + await connected; + await _wait(40); + expect(h.eventsOf(), isEmpty); + expect(h.transports.created, hasLength(1)); + }); + + test('with reconnection disabled, the network is not watched', () async { + h = ClientHarness(reconnect: ReconnectConfig.disabled); + await h.join(); + expect(h.network.isWatched, isFalse); + h.network.report({'mobile'}); + await _wait(40); + expect(h.client.status, ConnectionStatus.joined); + expect(h.eventsOf(), isEmpty); + }); + + test('watching starts with connect() and stops on leave', () async { + h = ClientHarness(); + expect(h.network.isWatched, isFalse); + await h.join(); + expect(h.network.isWatched, isTrue); + await h.client.leave(); + expect(h.network.isWatched, isFalse); + }); + }); + group('statistics (SPEC §17)', () { test('reports every 10 s while joined and stops on teardown', () { fakeAsync((async) { diff --git a/test/engine/engine_harness.dart b/test/engine/engine_harness.dart index acc89bb..03b7ad7 100644 --- a/test/engine/engine_harness.dart +++ b/test/engine/engine_harness.dart @@ -31,6 +31,7 @@ class EngineHarness { FakeRtcAdapter? rtc, RemoteStoreFactory remoteStore = RemoteStore.new, Duration nativeCallTimeout = const Duration(seconds: 10), + Duration iceDisconnectedTimeout = const Duration(seconds: 3), }) : logger = logger ?? FishjamLogger.disabled(), rtc = rtc ?? FakeRtcAdapter() { localTracks = LocalTrackRegistry( @@ -46,6 +47,7 @@ class EngineHarness { engine = RtcEngine( adapter: this.rtc, nativeCallTimeout: nativeCallTimeout, + iceDisconnectedTimeout: iceDisconnectedTimeout, logger: this.logger, sendMediaEvent: sent.add, emit: events.add, diff --git a/test/engine/receive_transceivers_test.dart b/test/engine/receive_transceivers_test.dart new file mode 100644 index 0000000..5c8f482 --- /dev/null +++ b/test/engine/receive_transceivers_test.dart @@ -0,0 +1,22 @@ +import 'package:fishjam_client/src/engine/receive_transceivers.dart'; +import 'package:fishjam_client/src/proto/fishjam/media_events/server/server.pb.dart' + as server; +import 'package:fishjam_client/src/rtc/rtc_types.dart'; +import 'package:flutter_test/flutter_test.dart'; + +import '../fakes/fake_rtc.dart'; +import '../tracks/log_capture.dart'; + +void main() { + test('a receive transceiver added without a receiver track is reported, ' + 'since it cannot be recorded', () async { + final logs = LogCapture(); + final receive = ReceiveTransceivers(logger: logs.logger); + final pc = FakePeerConnection(const RtcConfiguration()) + ..receiversWithoutTrack = true; + + await receive.sync(pc, server.MediaEvent_OfferData_TrackTypes(video: 1)); + + expect(logs.warnings, [contains('no receiver track')]); + }); +} diff --git a/test/engine/rtc_engine_test.dart b/test/engine/rtc_engine_test.dart index 950f3e8..20eb684 100644 --- a/test/engine/rtc_engine_test.dart +++ b/test/engine/rtc_engine_test.dart @@ -420,6 +420,55 @@ void main() { expect(logs, contains(startsWith('warn'))); }); + group('ICE disconnected that lasts (SPEC §8.4)', () { + const timeout = Duration(milliseconds: 20); + + Future disconnected() async { + final h = EngineHarness(iceDisconnectedTimeout: timeout); + await h.join(); + await h.offerData(); + await h.answer(); + h.pc.setIceConnectionState(IceConnectionState.disconnected); + await settle(); + return h; + } + + Future outlast() async { + await Future.delayed(timeout * 2); + await settle(); + } + + test( + 'is one ConnectionError, reported as a connection failure', + () async { + final h = await disconnected(); + await outlast(); + expect(h.events.whereType(), hasLength(1)); + expect(h.connectionFailures, hasLength(1)); + }, + ); + + for (final recovered in [ + IceConnectionState.connected, + IceConnectionState.completed, + ]) { + test('is nothing when ICE turns ${recovered.name} in time', () async { + final h = await disconnected(); + h.pc.setIceConnectionState(recovered); + await outlast(); + expect(h.events.whereType(), isEmpty); + expect(h.connectionFailures, isEmpty); + }); + } + + test('is nothing once the engine was closed', () async { + final h = await disconnected(); + await h.engine.close(); + await outlast(); + expect(h.events.whereType(), isEmpty); + }); + }); + test('ICE or peer connection failure emits one ConnectionError and ' 'reports it', () async { final (h, _) = await _joined(); @@ -487,6 +536,29 @@ void main() { expect(await liveRecv(h), isEmpty); }); + test( + 'orphans of every kind are stopped before any negotiated one', + () async { + final (h, _) = await _joined(); + await h.offerData(video: 1); + await h.answer(); + h.pc.failNext('setLocalDescription', StateError('failed')); + await h.offerData(video: 1, audio: 1); + final states = h.pc.transceiverStates; + expect(states.map((t) => (t.kind, t.mid)), [ + (MediaKind.video, '0'), + (MediaKind.audio, null), + ]); + + await h.offerData(); + expect(h.pc.callsOf('stopTransceiver').map((c) => c.args), [ + {'mid': null, 'kind': 'audio'}, + {'mid': '0', 'kind': 'video'}, + ]); + expect(states.every((t) => t.isStopped), isTrue); + }, + ); + test('stopped transceivers are not counted', () async { final (h, _) = await _joined(); await h.offerData(audio: 1); diff --git a/test/fakes/fake_network.dart b/test/fakes/fake_network.dart new file mode 100644 index 0000000..9063cfc --- /dev/null +++ b/test/fakes/fake_network.dart @@ -0,0 +1,54 @@ +/// Scriptable [NetworkSource] for unit tests. +library; + +import 'dart:async'; + +import 'package:fishjam_client/src/signaling/network_source.dart'; + +/// A fake OS network report: the test sets the current networks and pushes +/// reports. +class FakeNetworkSource implements NetworkSource { + /// Starts with [networks] in use. + FakeNetworkSource({Set networks = const {'wifi'}}) + : _current = networks; + + Set _current; + final _reports = StreamController>.broadcast(); + + /// When set, [getNetworkTypes] waits for it. + Completer? getNetworkTypesGate; + + /// When set, [getNetworkTypes] throws it. + Error? getNetworkTypesError; + + /// When set, reading [networkTypesChanged] throws it. + Error? networkTypesChangedError; + + /// Whether anyone listens to [networkTypesChanged]. + bool get isWatched => _reports.hasListener; + + @override + Future> getNetworkTypes() async { + final gate = getNetworkTypesGate; + if (gate != null) await gate.future; + final error = getNetworkTypesError; + if (error != null) throw error; + return _current; + } + + @override + Stream> get networkTypesChanged { + final error = networkTypesChangedError; + if (error != null) throw error; + return _reports.stream; + } + + /// The OS reports [networks] (also when they did not change). + void report(Set networks) { + _current = networks; + _reports.add(networks); + } + + /// The OS report stream fails. + void fail(Object error) => _reports.addError(error); +} diff --git a/test/fakes/fake_rtc.dart b/test/fakes/fake_rtc.dart index fa087ae..833a096 100644 --- a/test/fakes/fake_rtc.dart +++ b/test/fakes/fake_rtc.dart @@ -151,22 +151,33 @@ class FakeTransceiverState { /// A snapshot view of a transceiver: [mid] is the value at the time the view /// was obtained, like `flutter_webrtc`'s objects. +/// +/// Models how `flutter_webrtc` reaches native transceivers on Android: a view +/// without a mid holds a native handle that the next native +/// `getTransceivers()` disposes, and [stop] on a view with a mid makes that +/// call itself (it looks the transceiver up by mid). [stop] on a disposed +/// handle throws. class FakeRtpTransceiver implements RtcRtpTransceiver { - FakeRtpTransceiver._(this.state, this._pc) : mid = state.mid; + FakeRtpTransceiver._(this.state, this._pc) + : mid = state.mid, + _handleGeneration = _pc._handleGeneration; /// The live state behind this view. final FakeTransceiverState state; final FakePeerConnection _pc; + final int _handleGeneration; @override final String? mid; - @override - Future getDirection() async => state.direction; - - @override - Future getCurrentDirection() async => - state.currentDirection; + /// Resolves the native transceiver, as `flutter_webrtc` does. + void _resolve() { + if (mid != null) { + _pc._handleGeneration++; + } else if (_handleGeneration != _pc._handleGeneration) { + throw StateError('RtpTransceiver has been disposed.'); + } + } @override RtcRtpSender get sender => state.sender; @@ -176,6 +187,7 @@ class FakeRtpTransceiver implements RtcRtpTransceiver { @override Future stop() async { + _resolve(); _pc._record('stopTransceiver', {'mid': state.mid, 'kind': state.kind.name}); state ..direction = TransceiverDirection.stopped @@ -242,15 +254,17 @@ class FakeRtpSender implements RtcRtpSender { Future> getStats() async => stats; } -/// A fake receiver; it always has a remote track of its kind. +/// A fake receiver; it has a remote track of its kind unless created while +/// [FakePeerConnection.receiversWithoutTrack] was set. class FakeRtpReceiver implements RtcRtpReceiver { - FakeRtpReceiver._(this.kind) : track = FakeMediaTrack(kind: kind); + FakeRtpReceiver._(this.kind, {required bool withTrack}) + : track = withTrack ? FakeMediaTrack(kind: kind) : null; @override final MediaKind kind; @override - final RtcMediaTrack track; + final RtcMediaTrack? track; } /// A fake data channel. @@ -338,7 +352,14 @@ class FakePeerConnection implements RtcPeerConnection { /// Whether [close] was called. bool isClosed = false; + /// Transceivers added while set get a receiver without a track. + bool receiversWithoutTrack = false; + final List _transceivers = []; + + /// Bumped by every native `getTransceivers()`; views without a mid from an + /// earlier generation are disposed (see [FakeRtpTransceiver]). + int _handleGeneration = 0; final Map _failures = {}; int _offers = 0; int _mids = 0; @@ -477,16 +498,17 @@ class FakePeerConnection implements RtcPeerConnection { track, init.sendEncodings.isEmpty ? const [RtpEncoding()] : init.sendEncodings, ), - receiver: FakeRtpReceiver._(kind), + receiver: FakeRtpReceiver._(kind, withTrack: !receiversWithoutTrack), ); _transceivers.add(state); return FakeRtpTransceiver._(state, this); } @override - Future> getTransceivers() async => [ - for (final t in _transceivers) FakeRtpTransceiver._(t, this), - ]; + Future> getTransceivers() async { + _handleGeneration++; + return [for (final t in _transceivers) FakeRtpTransceiver._(t, this)]; + } @override Future> getSenders() async => [ diff --git a/test/fakes/fake_rtc_test.dart b/test/fakes/fake_rtc_test.dart index 1722daf..f0db36b 100644 --- a/test/fakes/fake_rtc_test.dart +++ b/test/fakes/fake_rtc_test.dart @@ -13,6 +13,25 @@ void main() { }); group('transceivers', () { + test('model flutter_webrtc on Android: a lookup by mid disposes the ' + 'handles of transceivers without a mid', () async { + await pc.addTransceiverOfKind( + MediaKind.audio, + const TransceiverInit(direction: TransceiverDirection.recvonly), + ); + await pc.setLocalDescription(await pc.createOffer()); + await pc.addTransceiverOfKind( + MediaKind.video, + const TransceiverInit(direction: TransceiverDirection.recvonly), + ); + final [negotiated, fresh] = await pc.getTransceivers(); + expect(negotiated.mid, isNotNull); + expect(fresh.mid, isNull); + + await negotiated.stop(); + await expectLater(fresh.stop(), throwsStateError); + }); + test( 'mids are assigned by setLocalDescription and views are snapshots', () async { @@ -57,8 +76,10 @@ void main() { const SessionDescription(type: SdpType.answer, sdp: 'answer'), ); - final t = (await pc.getTransceivers()).single; - expect(await t.getCurrentDirection(), TransceiverDirection.recvonly); + expect( + pc.transceiverStates.single.currentDirection, + TransceiverDirection.recvonly, + ); expect(pc.signalingState, SignalingState.stable); expect(pc.remoteDescription?.sdp, 'answer'); await Future.delayed(Duration.zero); @@ -74,9 +95,9 @@ void main() { await t.stop(); - final fresh = (await pc.getTransceivers()).single; - expect(await fresh.getDirection(), TransceiverDirection.stopped); - expect(await fresh.getCurrentDirection(), TransceiverDirection.stopped); + final state = pc.transceiverStates.single; + expect(state.isStopped, isTrue); + expect(state.currentDirection, TransceiverDirection.stopped); expect(pc.calls.last.method, 'stopTransceiver'); }); @@ -92,7 +113,10 @@ void main() { final fresh = (await pc.getTransceivers()).single; expect(fresh.sender.track, isNull); - expect(await fresh.getDirection(), TransceiverDirection.inactive); + expect( + pc.transceiverStates.single.direction, + TransceiverDirection.inactive, + ); expect((await pc.getSenders()).single.track, isNull); }, ); diff --git a/test/fakes/fake_transport.dart b/test/fakes/fake_transport.dart index bc93a72..69d46b2 100644 --- a/test/fakes/fake_transport.dart +++ b/test/fakes/fake_transport.dart @@ -61,6 +61,10 @@ class FakeTransport implements SignalingTransport { /// Whether the client called [close]. bool closedByClient = false; + /// When set, [close] waits for it before closing, like a socket whose + /// close handshake cannot reach the server. + Completer? closeGate; + /// Code passed to [close]. int? clientCloseCode; @@ -108,6 +112,8 @@ class FakeTransport implements SignalingTransport { closedByClient = true; clientCloseCode = code; clientCloseReason = reason; + final gate = closeGate; + if (gate != null) await gate.future; _finish(TransportClosed(code: code, reason: reason)); } diff --git a/test/models/models_test.dart b/test/models/models_test.dart index 95c0940..bfeab93 100644 --- a/test/models/models_test.dart +++ b/test/models/models_test.dart @@ -171,14 +171,34 @@ void main() { }); group('ReconnectConfig (SPEC §6)', () { - test('defaults enable three attempts', () { + test('defaults enable three attempts that publish the tracks again', () { const config = ReconnectConfig(); expect(config.isEnabled, isTrue); expect(config.maxAttempts, 3); - expect(config.addTracksOnReconnect, isFalse); + expect(config.addTracksOnReconnect, isTrue); expect(ReconnectConfig.disabled.isEnabled, isFalse); }); + test( + 'waits at most 60 s for the network; the limit is part of equality', + () { + expect( + const ReconnectConfig().offlineTimeout, + const Duration(seconds: 60), + ); + const shorter = ReconnectConfig(offlineTimeout: Duration(seconds: 5)); + expect(shorter, isNot(const ReconnectConfig())); + expect( + shorter, + const ReconnectConfig(offlineTimeout: Duration(seconds: 5)), + ); + expect( + shorter.hashCode, + const ReconnectConfig(offlineTimeout: Duration(seconds: 5)).hashCode, + ); + }, + ); + test('delay of attempt n is initialDelay + (n - 1) * delay', () { const config = ReconnectConfig( initialDelay: Duration(milliseconds: 100), diff --git a/test/signaling/network_change_detector_test.dart b/test/signaling/network_change_detector_test.dart new file mode 100644 index 0000000..b27304b --- /dev/null +++ b/test/signaling/network_change_detector_test.dart @@ -0,0 +1,204 @@ +import 'dart:async'; + +import 'package:fake_async/fake_async.dart'; +import 'package:fishjam_client/src/logger.dart'; +import 'package:fishjam_client/src/signaling/network_change_detector.dart'; +import 'package:flutter_test/flutter_test.dart'; + +import '../fakes/fake_network.dart'; + +void main() { + late FakeNetworkSource source; + late List changes; + late List logs; + late NetworkChangeDetector detector; + + setUp(() { + source = FakeNetworkSource(); + changes = []; + logs = []; + detector = NetworkChangeDetector( + source: source, + logger: FishjamLogger( + enabled: true, + sink: (level, message, {error, stackTrace}) => + logs.add('${level.name}: $message'), + ), + onChanged: changes.add, + ); + }); + + /// Starts the detector inside [body]'s fake time. + void run(void Function(FakeAsync async) body) => fakeAsync((async) { + unawaited(detector.start()); + async.flushMicrotasks(); + body(async); + }); + + test('the networks in use at start are the baseline, not a change', () { + run((async) { + source.report({'wifi'}); + async.flushMicrotasks(); + + expect(changes, isEmpty); + }); + }); + + test('a report is acted on at once', () { + run((async) { + source.report({'mobile'}); + async.flushMicrotasks(); + + expect(changes, [NetworkChange.changed]); + }); + }); + + test('offline is lost; online after offline is available, even when it ' + 'comes right after', () { + run((async) { + source + ..report({}) + ..report({'wifi'}); + async.flushMicrotasks(); + + expect(changes, [NetworkChange.lost, NetworkChange.available]); + }); + }); + + test('a repeated report is no change', () { + run((async) { + source + ..report({}) + ..report({}) + ..report({'wifi'}) + ..report({'wifi'}); + async.flushMicrotasks(); + + expect(changes, [NetworkChange.lost, NetworkChange.available]); + }); + }); + + test('only gaining a network is no change; losing one later is', () { + run((async) { + source.report({'wifi', 'mobile'}); + async.flushMicrotasks(); + expect(changes, isEmpty); + + source.report({'mobile'}); + async.flushMicrotasks(); + expect(changes, [NetworkChange.changed]); + }); + }); + + test('reports are listened to only once the start value was read, so an ' + 'OS that cannot be asked is never subscribed to', () { + fakeAsync((async) { + final gate = source.getNetworkTypesGate = Completer(); + unawaited(detector.start()); + async.flushMicrotasks(); + expect(source.isWatched, isFalse); + + gate.complete(); + async.flushMicrotasks(); + expect(source.isWatched, isTrue); + source.report({'wifi'}); + async.flushMicrotasks(); + expect(changes, isEmpty); + }); + }); + + test('of two starts in a row only the newer one listens, so stop ends ' + 'watching', () { + fakeAsync((async) { + unawaited(detector.start()); + unawaited(detector.start()); + async.flushMicrotasks(); + detector.stop(); + async.flushMicrotasks(); + + expect(source.isWatched, isFalse); + }); + }); + + test('stop while the start value is read: nothing is listened to', () { + fakeAsync((async) { + final gate = source.getNetworkTypesGate = Completer(); + unawaited(detector.start()); + async.flushMicrotasks(); + detector.stop(); + gate.complete(); + async.flushMicrotasks(); + + expect(source.isWatched, isFalse); + }); + }); + + test( + 'a superseded start whose read fails leaves the newer watch running', + () { + fakeAsync((async) { + final gate = source.getNetworkTypesGate = Completer(); + unawaited(detector.start()); + async.flushMicrotasks(); + source.getNetworkTypesGate = null; + unawaited(detector.start()); + async.flushMicrotasks(); + expect(source.isWatched, isTrue); + + source.getNetworkTypesError = StateError('no answer'); + gate.complete(); + async.flushMicrotasks(); + + expect(source.isWatched, isTrue); + expect(logs, isEmpty); + }); + }, + ); + + test('when the OS cannot be asked, it warns, reports nothing and does not ' + 'listen', () { + source.getNetworkTypesError = StateError('no plugin'); + run((async) { + expect(source.isWatched, isFalse); + source.report({'mobile'}); + async.flushMicrotasks(); + + expect(changes, isEmpty); + expect(logs, [startsWith('warn')]); + }); + }); + + test('when the report stream cannot be opened, it warns and reports ' + 'nothing', () { + source.networkTypesChangedError = StateError('no binding'); + run((async) { + expect(logs, [startsWith('warn')]); + expect(changes, isEmpty); + }); + }); + + test('a failing report stream warns and stops watching', () { + run((async) { + source.fail(StateError('channel gone')); + async.flushMicrotasks(); + + expect(logs, [startsWith('warn')]); + expect(source.isWatched, isFalse); + }); + }); + + test('stop ends watching; start after stop takes a new baseline', () { + run((async) { + detector.stop(); + async.flushMicrotasks(); + expect(source.isWatched, isFalse); + + source.report({'mobile'}); + unawaited(detector.start()); + async.flushMicrotasks(); + source.report({'mobile'}); + async.flushMicrotasks(); + expect(changes, isEmpty); + }); + }); +} diff --git a/test/signaling/reconnect_test.dart b/test/signaling/reconnect_test.dart index fe9234d..f282fd1 100644 --- a/test/signaling/reconnect_test.dart +++ b/test/signaling/reconnect_test.dart @@ -89,8 +89,15 @@ LocalTrackContext _track(String id, {RtcMediaTrack? media}) => const _loss = RetryableClose(code: 1011, reason: 'Internal server error'); -// maxAttempts 3, initialDelay 500 ms (the defaults), delay 300 ms. -const _config = ReconnectConfig(delay: Duration(milliseconds: 300)); +// maxAttempts 3, initialDelay 500 ms (the defaults), delay 300 ms, tracks +// not restored. +const _config = ReconnectConfig( + delay: Duration(milliseconds: 300), + addTracksOnReconnect: false, +); + +// The SDK defaults: tracks are restored. +const _restoring = ReconnectConfig(); void main() { late _Host host; @@ -248,25 +255,19 @@ void main() { async.flushMicrotasks(); expect(host.restored, isEmpty); - // Every old track id is reported removed, before Reconnected. - expect( - host.events.whereType().map((e) => e.trackId), - ['t1'], - ); - expect(host.calls.sublist(host.calls.length - 3), [ - 'emit:LocalTrackRemoved', + expect(host.calls.sublist(host.calls.length - 2), [ 'releaseQueue', 'emit:Reconnected', ]); }); }); - test('restores live tracks in order, skips ended and muted ones, ' - 'survives failures, then releases the queue and emits ' + test('restores live tracks together and in order, skips ended and muted ' + 'ones, survives failures, then releases the queue and emits ' 'Reconnected', () { fakeAsync((async) { final ended = FakeMediaTrack.video()..end(); - final m = manager(const ReconnectConfig(addTracksOnReconnect: true)) + final m = manager(_restoring) ..onRetryableLoss( _loss, snapshot( @@ -288,8 +289,13 @@ void main() { unawaited(m.onSessionJoined().then((_) => done = true)); async.flushMicrotasks(); - // The queue stays held until every restore settled. - expect(host.calls.last, 'restoreTrack:cam'); + // Every restore starts before the first one completes, so they ride + // one negotiation cycle; the queue stays held until all settled. + expect(host.calls.where((c) => c.startsWith('restoreTrack')), [ + 'restoreTrack:cam', + 'restoreTrack:broken', + 'restoreTrack:mic', + ]); expect(host.count('releaseQueue'), 0); expect(m.isReconnecting, isTrue); expect(done, isFalse); @@ -303,20 +309,7 @@ void main() { 'restoreTrack:broken', 'restoreTrack:mic', ]); - expect(host.restored, ['cam', 'mic']); - // Every old track id is reported removed, in order, before the - // first restore (restored tracks get new ids). - expect( - host.events.whereType().map((e) => e.trackId), - ['cam', 'ended', 'muted', 'broken', 'mic'], - ); - final firstRestore = host.calls.indexOf('restoreTrack:cam'); - expect( - host.calls - .sublist(0, firstRestore) - .where((c) => c == 'emit:LocalTrackRemoved'), - hasLength(5), - ); + expect(host.restored, unorderedEquals(['cam', 'mic'])); expect(host.calls.sublist(host.calls.length - 2), [ 'releaseQueue', 'emit:Reconnected', @@ -328,7 +321,7 @@ void main() { test('a loss during restoration cancels Reconnected and schedules the ' 'next attempt', () { fakeAsync((async) { - final m = manager(const ReconnectConfig(addTracksOnReconnect: true)) + final m = manager(_restoring) ..onRetryableLoss( _loss, snapshot( @@ -347,7 +340,7 @@ void main() { host.pendingRestores['cam']!.complete(); async.flushMicrotasks(); - expect(host.count('restoreTrack:mic'), 0); + expect(host.count('restoreTrack:mic'), 1); expect(host.count('emit:Reconnected'), 0); expect(host.count('releaseQueue'), 0); expect(m.isReconnecting, isTrue); @@ -355,14 +348,9 @@ void main() { async.elapse(const Duration(milliseconds: 1000)); expect(host.count('startSession'), 2); - // The next attempt restores the snapshot again, but the old ids - // were already reported removed. + // The next attempt restores the snapshot again. unawaited(m.onSessionJoined()); async.flushMicrotasks(); - expect( - host.events.whereType().map((e) => e.trackId), - ['cam', 'mic'], - ); expect(host.count('restoreTrack:cam'), 2); expect(host.count('emit:Reconnected'), 1); }); @@ -392,7 +380,7 @@ void main() { test('a terminal close during restoration prevents Reconnected', () { fakeAsync((async) { - final m = manager(const ReconnectConfig(addTracksOnReconnect: true)) + final m = manager(_restoring) ..onRetryableLoss( _loss, snapshot(tracks: [_track('cam', media: FakeMediaTrack.video())]), @@ -535,4 +523,237 @@ void main() { expect(m.onRetryableLoss(_loss, snapshot()), isTrue); }); }); + + group('network changes (SPEC §6.1)', () { + const offline10s = ReconnectConfig(offlineTimeout: Duration(seconds: 10)); + + test('losing the network cancels the scheduled attempt; while offline ' + 'nothing starts and a loss waits instead of using an attempt', () { + fakeAsync((async) { + final m = manager() + ..onRetryableLoss(_loss, snapshot()) + ..onNetworkLost(); + async.elapse(const Duration(seconds: 30)); + expect(host.count('startSession'), 0); + expect(m.onRetryableLoss(_loss, snapshot()), isTrue); + async.elapse(const Duration(seconds: 20)); + expect(host.count('startSession'), 0); + expect(m.isReconnecting, isTrue); + }); + }); + + test('a loss while already offline waits for the network', () { + fakeAsync((async) { + final m = manager()..onNetworkLost(); + expect(m.onRetryableLoss(_loss, snapshot()), isTrue); + async.elapse(const Duration(seconds: 30)); + expect(host.count('startSession'), 0); + expect(host.count('emit:ReconnectionStarted'), 1); + }); + }); + + test('the network back starts an attempt at once with a fresh budget', () { + fakeAsync((async) { + final m = manager() + ..onRetryableLoss(_loss, snapshot()) + ..onNetworkLost() + ..onNetworkAvailable(); + async.flushMicrotasks(); + expect(host.count('startSession'), 1); + + expect(m.onRetryableLoss(_loss, snapshot()), isTrue); + async.elapse(const Duration(milliseconds: 800)); + expect(m.onRetryableLoss(_loss, snapshot()), isTrue); + async.elapse(const Duration(milliseconds: 1100)); + expect(host.count('startSession'), 3); + expect(m.onRetryableLoss(_loss, snapshot()), isFalse); + }); + }); + + test('a new network abandons the scheduled attempt and starts one at ' + 'once, with a fresh budget', () { + fakeAsync((async) { + final m = manager()..onRetryableLoss(_loss, snapshot()); + async.elapse(const Duration(milliseconds: 500)); + m.onRetryableLoss(_loss, snapshot()); + async.elapse(const Duration(milliseconds: 800)); + m.onRetryableLoss(_loss, snapshot()); + expect(host.count('startSession'), 2); + + m.onNetworkAvailable(); + async.flushMicrotasks(); + expect(host.count('startSession'), 3); + async.elapse(const Duration(seconds: 5)); + expect(host.count('startSession'), 3); + + expect(m.onRetryableLoss(_loss, snapshot()), isTrue); + async.elapse(const Duration(milliseconds: 800)); + expect(m.onRetryableLoss(_loss, snapshot()), isTrue); + async.elapse(const Duration(milliseconds: 1100)); + expect(host.count('startSession'), 5); + expect(m.onRetryableLoss(_loss, snapshot()), isFalse); + }); + }); + + test('a new network abandons the attempt in progress; its late start ' + 'error is ignored', () { + fakeAsync((async) { + final gate = host.startGate = Completer(); + host.failingStarts = 1; + final m = manager()..onRetryableLoss(_loss, snapshot()); + async.elapse(const Duration(milliseconds: 500)); + expect(host.count('startSession'), 1); + + m.onNetworkAvailable(); + async.flushMicrotasks(); + expect(host.count('startSession'), 2); + + gate.complete(); + async.elapse(const Duration(seconds: 5)); + expect(host.count('startSession'), 2); + expect(m.isReconnecting, isTrue); + }); + }); + + test('a network change during track restoration starts a new attempt ' + 'and cancels Reconnected', () { + fakeAsync((async) { + final pending = host.pendingRestores['t1'] = Completer(); + final m = manager(_restoring) + ..onRetryableLoss( + _loss, + snapshot(tracks: [_track('t1', media: FakeMediaTrack.video())]), + ); + async.elapse(const Duration(milliseconds: 500)); + unawaited(m.onSessionJoined()); + async.flushMicrotasks(); + + m.onNetworkAvailable(); + async.flushMicrotasks(); + expect(host.count('startSession'), 2); + + pending.complete(); + async.flushMicrotasks(); + expect(host.events.whereType(), isEmpty); + }); + }); + + test('a session that joins while offline does not complete the ' + 'reconnection; the network back starts a new attempt', () { + fakeAsync((async) { + final m = manager()..onRetryableLoss(_loss, snapshot()); + async.elapse(const Duration(milliseconds: 500)); + m.onNetworkLost(); + unawaited(m.onSessionJoined()); + async.flushMicrotasks(); + expect(host.events.whereType(), isEmpty); + expect(m.isReconnecting, isTrue); + + m.onNetworkAvailable(); + async.flushMicrotasks(); + expect(host.count('startSession'), 2); + }); + }); + + test('a session that goes offline while its tracks are restored does not ' + 'complete the reconnection; the network back starts a new attempt', () { + fakeAsync((async) { + final pending = host.pendingRestores['t1'] = Completer(); + final m = manager(_restoring) + ..onRetryableLoss( + _loss, + snapshot(tracks: [_track('t1', media: FakeMediaTrack.video())]), + ); + async.elapse(const Duration(milliseconds: 500)); + unawaited(m.onSessionJoined()); + async.flushMicrotasks(); + + m.onNetworkLost(); + pending.complete(); + async.flushMicrotasks(); + expect(host.events.whereType(), isEmpty); + expect(host.count('releaseQueue'), 0); + expect(m.isReconnecting, isTrue); + + m.onNetworkAvailable(); + async.flushMicrotasks(); + expect(host.count('startSession'), 2); + }); + }); + + test('offline for offlineTimeout fails the reconnection', () { + fakeAsync((async) { + final m = manager(offline10s) + ..onRetryableLoss(_loss, snapshot()) + ..onNetworkLost(); + async.elapse(const Duration(milliseconds: 9999)); + expect(m.isReconnecting, isTrue); + + async.elapse(const Duration(milliseconds: 1)); + expect(m.state, ReconnectState.failed); + expect(host.calls.sublist(host.calls.length - 3), [ + 'clearQueue', + 'emit:ReconnectionFailed', + 'endSession:${DisconnectReason.reconnectionFailed}', + ]); + }); + }); + + test('the network back stops the offline timeout', () { + fakeAsync((async) { + final m = manager(offline10s) + ..onRetryableLoss(_loss, snapshot()) + ..onNetworkLost(); + async.elapse(const Duration(seconds: 5)); + m.onNetworkAvailable(); + async.elapse(const Duration(seconds: 30)); + expect(host.events.whereType(), isEmpty); + }); + }); + + test('one reconnection makes at most 3 × maxAttempts attempts', () { + fakeAsync((async) { + final m = manager(const ReconnectConfig(maxAttempts: 2)) + ..onRetryableLoss(_loss, snapshot()); + // A network that keeps changing restarts the budget every time. + for (var i = 0; i < 6; i++) { + async.elapse(const Duration(seconds: 1)); + m.onNetworkAvailable(); + async.flushMicrotasks(); + m.onRetryableLoss(_loss, snapshot()); + } + expect(host.count('startSession'), 6); + expect(m.state, ReconnectState.failed); + expect(host.events.whereType(), hasLength(1)); + }); + }); + + test('changes outside a reconnection start nothing', () { + fakeAsync((async) { + final m = manager() + ..onNetworkLost() + ..onNetworkAvailable(); + async.elapse(const Duration(seconds: 5)); + expect(host.calls, isEmpty); + + m + ..onRetryableLoss(_loss, snapshot()) + ..reset() + ..onNetworkAvailable(); + async.elapse(const Duration(seconds: 5)); + expect(host.count('startSession'), 0); + }); + }); + + test('reset forgets that the device was offline', () { + fakeAsync((async) { + manager() + ..onNetworkLost() + ..reset() + ..onRetryableLoss(_loss, snapshot()); + async.elapse(const Duration(milliseconds: 500)); + expect(host.count('startSession'), 1); + }); + }); + }); } diff --git a/test/signaling/websocket_transport_test.dart b/test/signaling/websocket_transport_test.dart index eca201f..e7114c2 100644 --- a/test/signaling/websocket_transport_test.dart +++ b/test/signaling/websocket_transport_test.dart @@ -17,6 +17,21 @@ Future<(HttpServer, Uri)> _serve( return (server, Uri.parse('ws://127.0.0.1:${server.port}/socket')); } +/// A server that accepts the TCP connection and never answers the upgrade, +/// like a peer on a network that went away mid-handshake. +Future _silentServer() async { + final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); + final accepted = []; + server.listen(accepted.add); + addTearDown(() async { + for (final socket in accepted) { + socket.destroy(); + } + await server.close(); + }); + return Uri.parse('ws://127.0.0.1:${server.port}/socket'); +} + void main() { test('exchanges binary frames and drops text frames', () async { final serverSide = Completer(); @@ -105,6 +120,46 @@ void main() { expect((await transport.onClose).error, isNotNull); }); + test('pings every 15 s and gives up opening a socket after 10 s', () { + final transport = WebSocketTransport(); + + expect(transport.pingInterval, const Duration(seconds: 15)); + expect(transport.connectTimeout, const Duration(seconds: 10)); + }); + + test( + 'a server that never answers the upgrade times the connect out', + () async { + final url = await _silentServer(); + final transport = WebSocketTransport( + connectTimeout: const Duration(milliseconds: 200), + ); + + await expectLater( + transport.connect(url), + throwsA(isA()), + ); + expect((await transport.onClose).error, isA()); + expect(transport.isOpen, isFalse); + }, + ); + + test('close aborts a connect in progress', () async { + final url = await _silentServer(); + final transport = WebSocketTransport(); + final connecting = expectLater( + transport.connect(url), + throwsA(isA()), + ); + await Future.delayed(const Duration(milliseconds: 50)); + + await transport.close(); + + // Without the abort it would wait for the 10 s connect timeout. + await connecting.timeout(const Duration(seconds: 2)); + expect(transport.isOpen, isFalse); + }); + test('a transport is single use', () async { final (server, url) = await _serve((socket) => socket.listen(null)); addTearDown(server.close); diff --git a/tool/check_coverage.sh b/tool/check_coverage.sh index 1d05e6c..2ca6ae1 100755 --- a/tool/check_coverage.sh +++ b/tool/check_coverage.sh @@ -13,6 +13,7 @@ # Never recorded (no executable code): # lib/fishjam_client.dart exports only # lib/src/version.dart constants only +# lib/src/signaling/network_source.dart interface only set -euo pipefail LCOV="${1:-coverage/lcov.info}" @@ -27,7 +28,8 @@ unrecorded=0 while IFS= read -r file; do case "$file" in lib/src/proto/* | lib/src/rtc/flutter_webrtc_adapter*.dart | \ - lib/fishjam_client.dart | lib/src/version.dart) continue ;; + lib/fishjam_client.dart | lib/src/version.dart | \ + lib/src/signaling/network_source.dart) continue ;; esac if ! grep -qx "SF:$file" "$LCOV"; then echo "FAIL: $file has no coverage record; load it from a test" >&2