Opening a WebSocket is one line of code. Keeping it open on a phone is the hard part. The user walks into a lift, the phone switches from Wi-Fi to mobile data, the app goes to the background, the access token expires. The socket dies, sometimes with an error and sometimes without a word.

So every realtime feature ends up with the same code around the socket: reconnect, but not in a tight loop. Get a fresh token before reconnecting, because the old one may have expired. Notice a connection that is dead but still looks open. Re-subscribe to the topics. Do something sensible with messages sent while offline. And stop for good when the server says "you are not allowed", instead of retrying forever.

resilient_ws packages that code. It has two realtime clients for Dart and Flutter that keep working on bad mobile networks: a WebSocket client and a Server-Sent Events client. Both reconnect on their own, fetch a fresh token for every connection, and stop cleanly when told to. Both work on every platform, including the web.

What it does

ResilientWebSocket

  • States. connecting, open, reconnecting and closed, available as a stream and as a getter.
  • Backoff. Exponential backoff with jitter and a cap: 1 s doubling to 30 s by default. The backoff resets only after a connection has stayed up for stableAfter (10 s by default), so a server that accepts and then drops connections does not get hammered.
  • Fresh tokens. The token provider runs before every connection attempt. There are three ways to send the token: an HTTP header (dart:io only), a query parameter, or a first message.
  • Heartbeat. An application-level ping. If no frame arrives within timeout of a ping, the client reconnects. Any inbound frame counts as proof of life, and pongs are not passed on to messages.
  • Outbox. A bounded buffer for messages sent while disconnected, with a dropOldest or dropNewest overflow policy. Buffered messages are flushed in order after a reconnect.
  • Subscriptions. Topic subscriptions are re-sent after every reconnect. The subscribe and unsubscribe frames come from a pluggable SubscriptionCodec.
  • Typed messages. Messages go through a pluggable MessageCodec<T>, with JsonMessageCodec included.
  • Permanent close codes. Close codes that mean "stop" (1008, 4401 and 4403 by default, configurable) end the client instead of starting a retry loop.
  • Network awareness. An optional Stream<bool> of network availability pauses reconnects while offline and reconnects at once when the network returns.
  • Clean close. close() sends a close frame, cancels every timer, and never triggers a reconnect.

SseClient

The parser follows the WHATWG HTML "Server-sent events" section: the event, data (multi-line), id and retry fields, comments and heartbeats, a leading BOM, and CRLF, CR and LF line endings, including a CRLF split across chunks. An event is dispatched only on a blank line.

The client reconnects automatically, sends Last-Event-ID, and honours the server's retry: field. Consecutive failures double the wait, up to maxRetry. The token provider runs before every request. HTTP 204 stops the client, as the spec says. By default 4xx statuses stop it too, while 5xx, 408 and 429 are retried, and both rules are configurable. An optional idleTimeout reconnects when a connection goes silent, and the http.Client can be injected.

How to use it

Add it from GitHub:

dependencies: resilient_ws: git: https://github.com/AbdeldjalilChougui/resilient_ws

A typed WebSocket client:

import 'package:resilient_ws/resilient_ws.dart'; class Tick { Tick(this.topic, this.value); factory Tick.fromJson(Object? json) { final map = json! as Map<String, Object?>; return Tick(map['topic']! as String, map['value']! as num); } final String topic; final num value; Map<String, Object?> toJson() => {'topic': topic, 'value': value}; } Future<String?> readAccessToken() async => 'token'; // your auth layer Future<void> main() async { final ws = ResilientWebSocket<Tick>( url: Uri.parse('wss://api.example.com/ws'), codec: JsonMessageCodec<Tick>( fromJson: Tick.fromJson, toJson: (t) => t.toJson(), ), tokenProvider: readAccessToken, // called before every connect tokenAuth: const TokenAuth.queryParameter(name: 'token'), heartbeat: const Heartbeat( interval: Duration(seconds: 25), timeout: Duration(seconds: 10), ), outboxCapacity: 50, overflowPolicy: OverflowPolicy.dropOldest, permanentCloseCodes: {1008, 4001, 4003}, ); ws.states.listen((s) => print('state: ${s.name}')); ws.messages.listen((tick) => print('${tick.topic}: ${tick.value}')); ws.errors.listen((e) { if (e is PermanentCloseException) print('server said stop: ${e.code}'); }); ws.subscribe('prices'); // re-sent after every reconnect ws.connect(); ws.send(Tick('hello', 1)); // buffered until the socket is open await Future<void>.delayed(const Duration(seconds: 30)); await ws.close(); // no reconnect follows }

The default subscription frames are {"op":"sub","topic":"prices"} and {"op":"unsub","topic":"prices"}. If your server speaks another protocol, pass a SubscriptionCodec.from(subscribe: ..., unsubscribe: ...). If it expects the token in a first message, use TokenAuth.firstMessage; the client then sends it before the subscriptions and before any buffered messages.

Server-Sent Events:

import 'package:resilient_ws/resilient_ws.dart'; Future<void> main() async { final sse = SseClient( url: Uri.parse('https://api.example.com/events'), tokenProvider: () async => 'token', // "Authorization: Bearer token" retry: const Duration(seconds: 3), // until the server sends "retry:" idleTimeout: const Duration(seconds: 60), // server heartbeats every 15 s ); sse.events.listen((e) => print('${e.type} #${e.lastEventId}: ${e.data}')); sse.states.listen((s) => print('state: ${s.name}')); sse.connect(); await Future<void>.delayed(const Duration(minutes: 1)); await sse.close(); }

The parser can also be used by itself, as a stream transformer: SseParser.transformer().

Behaviour worth knowing

The README documents the exact rules, which matters for this kind of code:

  • Connection order. On every connect the client sends the first-message token (if used), then every subscription, then the buffered messages.
  • Backoff formula. The delay for attempt n is min(max, initial * multiplier^n), minus a random share of up to jitter (0.2 by default). Jitter only shortens a delay, so max is a hard cap.
  • Failures. A throwing token provider, a handshake that fails or takes longer than connectTimeout, or a dropped connection are all retried. maxAttempts, off by default, stops the client after that many failed attempts in a row.
  • Stale connections. Every connection attempt has a generation number, so frames, errors and close events from a replaced connection are ignored. reconnect() forces a fresh connection, which is useful when an app returns to the foreground.
  • SSE ids. An id: field counts only once the blank line after it arrives, as the spec says, so a half-received event never moves Last-Event-ID.

Things to know

  • Header auth is dart:io only. Browsers do not let a WebSocket set headers. On the web, TokenAuth.header makes the connection attempt fail with UnsupportedError. Use TokenAuth.queryParameter or TokenAuth.firstMessage instead.
  • SSE on the web. The client uses package:http, not the browser's EventSource, so it can send Authorization and Last-Event-ID headers. Custom headers trigger a CORS preflight, so the server must allow authorization, last-event-id and cache-control. A proxy or dev server that buffers or compresses text/event-stream can still delay events, and that is outside the client's control. A browser test in the repository checks that events arrive one by one in Chrome and that the client reconnects with Last-Event-ID.
  • The heartbeat is application-level. The server must answer your ping frame, or send any traffic within timeout. WebSocket protocol-level pings are not configured.
  • Handshake rejections cannot be told apart. A server that rejects the upgrade with HTTP 401 or 403 looks like any other failed handshake to web_socket_channel, so it is retried with backoff, with a fresh token each time. Use maxAttempts to bound it, or have the server accept the upgrade and close with an auth close code such as 4401.
  • Undispatched SSE data is discarded. An event cut off mid-stream is thrown away, as the spec requires.

Why I built it

The reconnect logic around a socket is the same in every app, and it is easy to get subtly wrong. A retry loop without jitter. A token read once at startup. A heartbeat that fires on a connection that was already replaced. A close() that is followed by one last reconnect.

I wanted those rules written once, with the behaviour documented and the SSE parser tested against the examples of the spec itself. I also wanted WebSocket and SSE side by side, with the same ideas for tokens, states and retries, because the choice between them often depends on the server and not on the app.

Get the code

The source, the behaviour notes and the tests are on GitHub: https://github.com/AbdeldjalilChougui/resilient_ws. It's MIT licensed. Issues and pull requests are welcome.

Originally published at abdeldjalilchougui.biz.