baaba_realtime_client 3.0.1

Real-time client for Flutter: ASP.NET Core SignalR, classic SignalR 2.x and WebSockets. Either-based errors, auto-reconnect, reliable sends, offline queue, test doubles.

baaba_realtime_client

A real-time client for Flutter: ASP.NET Core SignalR (via signalr_netcore), classic ASP.NET SignalR 2.x and raw WebSockets, behind one API that never throws.

• Every async call returns Either<RealtimeFailure, T> — no try/catch around your hub calls
• Auto-reconnect with exponential backoff and full jitter, and a reconnectedStream to catch up on
• Automatic room re-join, dead-connection detection, opt-in offline queueing
• Reliable sends: idempotency key, acknowledgement, retries, per-message status
• Connection quality from measured round trips, for a "weak connection" banner
• One socket shared by every feature that needs it
• JSON or MessagePack on ASP.NET Core; session resumption on SignalR 2.x
• One errorStream for every failure, including ones no call site could see
• A full fake client for testing, with no sockets involved

Install

dependencies:
  baaba_realtime_client: ^3.0.1

Requires equatable 3.x if your app depends on it directly.

Quick start

import 'package:baaba_realtime_client/baaba_realtime_client.dart';

RealtimeClient.init(RealtimeConfig(
  uri: Uri.parse('https://api.example.com/chatHub'),
  auth: RealtimeAuthConfig(getToken: () => secureStorage.read('access_token')),
));

final client = RealtimeClient.instance();

await client.connect();

That is the whole setup. Negotiation, transport choice, the handshake, keep-alive and reconnection are handled for you. https:// and wss:// are both fine.

Need two hubs (chat and calls, say)? Use RealtimeClient.create(config) for independent instances instead of the singleton. Need two features on the same hub? Share one connection — see Sharing one connection.

Classic ASP.NET SignalR (2.x)

A 2.x hub (Microsoft.AspNet.SignalR, /signalr) speaks a different protocol from ASP.NET Core SignalR, so it has its own transport type:

RealtimeConfig(
  uri: Uri.parse('https://api.example.com/signalr'),
  transportType: RealtimeTransportType.signalrClassic,
  classicSignalR: const ClassicSignalROptions(
    hubs: ['chatHub', 'pulseHub'],  // all on one socket
    pingMethod: 'Ping',             // optional: a hub method to call as the heartbeat
  ),
)

• invoke resolves with the hub method's return value, and fails with serverError when it throws.
• Calls go to the first hub; qualify the method for another: client.invoke('pulseHub.Like', …). Incoming events carry their hub in RealtimeMessage.metadata['hub'].
• A dropped connection first resumes its session through /reconnect. The server replays what it sent meanwhile and keeps the connection id and groups, and RealtimeReconnection.resumed is true. Past the server's DisconnectTimeout (30 s by default), it starts a fresh session.
• disconnect() calls /abort, so the user leaves presence at once.
• Without a pingMethod, the server's own keep-alive (every 10 s by default) is what the heartbeat waits for. It waits for the longer of HeartbeatConfig.pongTimeout and the KeepAliveTimeout the server reports in negotiate (20 s by default), so a late keep-alive does not end the connection.
• If a resume fails and a fresh negotiate fails too, the server was unreachable rather than refusing, and the session is kept for the next attempt to resume.

Endpoint and token per attempt

When the hub address only comes with a token — a token endpoint that also returns the hub path, say — resolve both on every attempt instead of fixing uri up front:

RealtimeConfig(
  connectionProvider: () async {
    final grant = await api.getRealtimeToken();          // throw to fail the attempt
    return RealtimeEndpoint(uri: hubUri(grant.hubPath), token: grant.token);
  },
  auth: RealtimeAuthConfig(getToken: () async => null, tokenQueryParam: 't'),  // token placement only
)

It is called for the first connect, every reconnect, and the retry after a token refresh. A throw is an ordinary failed attempt and is retried; throw RealtimeFailure.unauthorized to get the onTokenRefresh path.

While a provider is set, auth.getToken is never called — not even by ASP.NET Core SignalR's per-request token lookup — so the provider's token is the only one sent. auth still decides where the token goes and how a refresh happens.

Receiving events

// The usual hub shape: Clients.All.SendAsync("ReceiveMessage", dto)
client.on('ReceiveMessage').listen((dto) => print(dto));

// Multi-parameter: Clients.Group(r).SendAsync("UserTyping", userId, isTyping)
client.onArgs('UserTyping').listen((args) => print('${args[0]} ${args[1]}'));

// Typed, with a parser
client.onAs('ReceiveMessage', parser: ChatMessage.fromJson).listen(_append);

// Everything, including targets nothing is subscribed to — good for debugging
client.messageStream.listen((m) => log('${m.target} ${m.arguments}'));

on gives you the first argument; onArgs gives you all of them. That is the only rule. Both are broadcast streams that stay alive across reconnects — cancel your own subscription to stop listening.

There are also optional typed streams for chat, typing, presence and WebRTC signaling (chatStream, typingStream, presenceStream, signalingStream). They assume particular hub method names and payload shapes, so if your hub uses its own names and shapes, stick with on / onAs.

Sending

// Fire-and-forget
await client.send('SendMessage', args: [chatId, payload]);

// Request/response — returns what the hub method returned
final result = await client.invoke('SendMessage', args: [chatId, payload]);
result.fold(
  (failure) => showError(failure.message),
  (data) => messageId = data['id'],
);

// Typed response
final history = await client.invokeAs(
  'GetHistory',
  args: [chatId],
  parser: listParser(ChatMessage.fromJson),
);

Sending while disconnected returns Left(RealtimeFailure.disconnected()) straight away. To buffer messages and flush them on the next connect instead, pass offlineQueue: const OfflineQueuePolicy().

Reliable sends

For messages that must arrive — a chat message, say — sendReliable keeps trying until the server acknowledges it:

final result = await client.sendReliable(
  'SendMessage',
  args: [{'conversationId': 9, 'text': 'hi'}],
  deliveryId: localMessage.id,   // optional; your id makes the updates easy to match
);

client.deliveryStream
    .where((d) => d.id == localMessage.id)
    .listen((d) => setState(() => localMessage.status = d.status));  // pending → sent → acknowledged / failed

• Every attempt carries the same key, written into the first argument when it is a map (clientMessageId by default). Have the server ignore a key it has already handled, and a retry after a lost acknowledgement becomes harmless.
• The acknowledgement is the hub method's answer. On a raw WebSocket, which has no answers, set ReliableDeliveryPolicy(ackEvent: 'MessageAck') and have the server send that event back with the key.
• While disconnected it waits for the connection, without spending attempts. Network failures and timeouts are retried with backoff; a hub that threw is not. It gives up after maxAttempts (5) or expireAfter (5 min), and disconnect() fails whatever is still in progress.

Rooms

await client.joinRoom('general');                        // JoinRoom("general")
await client.joinRoom(42, joinMethod: 'JoinChatGroup');  // JoinChatGroup(int chatId)
await client.leaveRoom('general');

The room is sent exactly as you pass it, so match the hub parameter's type. A string "42" will not bind to an int chatId. activeRooms lists rooms by their string form.

Joined rooms are tracked and re-joined automatically after a reconnect. Joining while disconnected is allowed; it takes effect on connect.

Errors

Nothing throws. Failures arrive in three places, and you generally want all three:

// 1. The call that caused it
final result = await client.invoke('CloseChat', args: [chatId]);
if (result.isLeft()) { /* ... */ }

// 2. Everything, including failures with no call site
client.errorStream.listen((f) {
  Sentry.captureException(f);
  if (f.errorType == RealtimeErrorSource.authentication) promptRelogin();
});

// 3. Connection state, for your UI
client.stateStream.listen((s) => setState(() => _offline = !s.isConnected));

RealtimeFailure carries errorType, a stable numeric code, message, data, details and stackTrace. Switch on errorType (network, timeout, handshake, authentication, protocol, serverError, parseError, offline, queueOverflow, cancelled, disconnected).

A 401 triggers auth.onTokenRefresh and one automatic retry before you ever see a failure.

Connection and reconnection

A failed attempt always schedules another one, including the very first connect():

final result = await client.connect();
// Left means *this attempt* failed. The client is now `reconnecting` and keeps
// trying in the background. Don't wrap connect() in your own retry loop.

• Retries back off exponentially from 500 ms to a 30 s cap, with jitter and no attempt limit.
• A dropped connection goes through the same retry loop, and joined rooms are rejoined afterwards.
• While the device is offline, the client checks again every 5 s. It reconnects the moment the OS reports the network is back, and these attempts don't count toward maxAttempts.
• A 401 calls auth.onTokenRefresh once and retries straight away.
• It stops only when you call disconnect() or dispose(), or when maxAttempts, retryIf or ReconnectionPolicy.disabled() says so.

Drive your UI from stateStream, not from what connect() returned.

connect() is safe to call from anywhere: it returns at once when already connected and joins an attempt that is already running.

Catching up after a reconnect

Events sent while the connection was down are gone. reconnectedStream fires each time it comes back — after rooms are rejoined and the offline queue is flushed — so that is where to refetch:

client.reconnectedStream.listen((r) {
  if (r.resumed) return;              // the server replayed everything (classic SignalR only)
  refreshConversations();             // r.downtime and r.attempts say how long it was out
});

The first connection of a client never fires it.

Connection quality

client.qualityStream.listen((q) => setState(() => _weak = q.level == ConnectionQualityLevel.poor));

quality smooths measured round trips — answered invocations, and heartbeat pings where the client sends them — into good, degraded or poor (thresholds in ConnectionQualityPolicy, 300 ms and 1 s by default). It is unknown until the first measurement and again after every drop. On ASP.NET Core SignalR only invocations are measured, since the protocol's pings are not answered.

Behind proxies and firewalls

Before each attempt the client checks whether the device is online. By default that check contacts public hosts (via internet_connection_checker_plus). If a network blocks those but allows your hub, the client decides it is offline and never tries the hub. On restrictive networks, check against your own server instead:

RealtimeConfig(
  uri: Uri.parse('https://api.example.com/chatHub'),
  connectivityProbe: () async {
    try {
      final r = await http.head(Uri.parse('https://api.example.com')).timeout(const Duration(seconds: 5));
      return r.statusCode < 500;
    } catch (_) {
      return false;
    }
  },
)

Or set bypassConnectivityCheck: true to always attempt the hub. You lose the 5 s offline polling, but the instant reconnect when the network returns still works.

A network that blocks WebSockets outright cannot be worked around. The client uses WebSockets only, because the Server-Sent Events fallback in signalr_netcore can crash the app (see SignalROptions). On such a network it keeps retrying and reports network failures on errorStream.

App lifecycle

By default (lifecycleAware: true) the client keeps the connection open in the background, for as long as the OS allows. When the app comes back to the foreground, it:

• notices a socket the OS closed while the app was away (iOS does this within seconds) and reconnects at once, instead of reporting connected until the server times it out;
• runs a pending reconnect whose timer the OS suspended, instead of waiting out the rest of the delay.

This needs a Flutter binding. Without one, call notifyAppResumed() yourself.

Going offline in the background (push notifications)

If your server sends push notifications only while the user is offline, it needs to be told when the app goes to the background. The socket also has to stay closed until the app comes back. Own the lifecycle yourself and turn the built-in handling off:

RealtimeConfig(
  uri: Uri.parse('https://api.example.com/chatHub'),
  lifecycleAware: false,
)

// AppLifecycleState.paused
await client.invoke('GoOffline', args: [chatId], timeout: const Duration(seconds: 3));
await client.disconnect();

// AppLifecycleState.resumed
await client.connect(); // rejoins your rooms; keeps retrying if the network is bad

• Await the offline call before disconnect(). If you don't, the disconnect can win, the server never hears the app went offline, and no pushes arrive.
• disconnect() guarantees nothing reconnects in the background. It cancels any pending retry and any attempt in flight.
• Keep the offline queue off (the default). With it on, an offline call that failed while disconnected would be sent right after the next resume, marking the user offline while they are in the app.
• Rooms survive disconnect(). connect() rejoins them, and so does every automatic reconnect. Call leaveRoom(room, leaveMethod: …) when a chat actually ends, or it will be rejoined next time.
• The OS can suspend the app before the offline call lands. No client can fully prevent that. Have the server also treat OnDisconnectedAsync as offline, so a dropped socket means the same thing.

Sharing one connection

When several features use the same hub, give each a lease on one shared client instead of a client each:

// in the chat feature
final lease = RealtimeClient.acquire('main-hub', config: () => hubConfig);
await lease.client.connect();
lease.client.on('conversationChanged').listen(_onChange);

// … and when the feature closes
await lease.release();   // the socket closes when the last lease is released

config (and transport) run only on the first acquire of a key. A holder must not call disconnect(), dispose() or off() on the shared client — those affect every holder. Release the lease and cancel your own subscriptions instead.

Configuration

Defaults are production-shaped; override only what you need.

RealtimeConfig(
  uri: Uri.parse('https://api.example.com/chatHub'),
  reconnect: const ReconnectionPolicy(baseDelay: Duration(seconds: 1)),
  heartbeat: const HeartbeatConfig(pingInterval: Duration(seconds: 15)),
  offlineQueue: const OfflineQueuePolicy(maxQueueSize: 200),  // off unless you pass one
  invokeTimeout: const Duration(seconds: 30),
  signalR: const SignalROptions(),   // negotiation, transport, negotiate headers, protocol
  reliableDelivery: const ReliableDeliveryPolicy(),  // sendReliable defaults
  quality: const ConnectionQualityPolicy(),          // latency thresholds
  observer: MyObserver(),            // lifecycle hooks for analytics
  interceptors: [MyInterceptor()],   // transform or drop messages
)

For smaller, faster frames on ASP.NET Core, enable MessagePack on the server (AddMessagePackProtocol()) and pass signalR: const SignalROptions(protocol: SignalRHubProtocol.messagePack). See its limitations below.

On a raw WebSocket the heartbeat sends {"type":"ping"}; change it with HeartbeatConfig(pingFrame: …) to whatever your server answers.

Custom transports

A RealtimeTransport can also implement any of InvokingTransport (request/response, so invoke returns the server's answer), KeepAliveTransport (its own ping, and its own proof of life) and SessionTransport (a server session that survives reconnects). The client hands each of those concerns to the transport instead of approximating it. ClassicSignalRTransport implements all three.

A KeepAliveTransport whose server sends keep-alives on its own schedule reports how long the server may go quiet as keepAliveTimeout; after a ping the server will not answer, the heartbeat waits at least that long. A ClassicSignalRTransport you construct yourself is yours to dispose(); the one RealtimeClient creates is disposed with the client.

Each of ReconnectionPolicy, HeartbeatConfig, OfflineQueuePolicy and RealtimeLogOptions has a .disabled() constructor. Logging masks bearer tokens and trims base64 blobs by default.

If your hub is WebSocket-only with negotiation disabled server-side, use signalR: const SignalROptions.webSocketsOnly(). Otherwise leave the default: a failed /negotiate automatically retries with it skipped, and the working route is remembered.

Testing

import 'package:baaba_realtime_client/testing.dart';

final client = FakeRealtimeClient();

client.simulateState(RealtimeConnectionState.connected);
client.simulateEvent('OrderUpdated', {'id': 1});
client.whenInvoke('GetHistory', [/* ... */]);
client.simulateError(RealtimeFailure.network(message: 'dropped'));
client.simulateReconnected(downtime: const Duration(seconds: 4));
client.simulateQuality(const RealtimeConnectionQuality(level: ConnectionQualityLevel.poor));

expect(client.recordedSends, hasLength(1));
expect(client.recordedReliableSends, hasLength(1));  // sendReliable resolves through whenInvoke stubs

No sockets, no servers, no timers.

Cleanup

await client.disconnect();  // keeps the instance usable
await client.dispose();     // releases its streams — for create() instances

Call disconnect() when the user logs out. Otherwise the client keeps reconnecting in the background with a token the server will reject.

Known limitations

• A rejected token is only recognised as one on the negotiate route. When negotiation is skipped (SignalROptions.webSocketsOnly(), or after the negotiate fallback kicks in), a 401 on the WebSocket upgrade surfaces as a network failure, because Dart's WebSocket error does not carry the status code. onTokenRefresh is not called there, but getToken is re-read on every reconnect attempt, so a token refreshed elsewhere in your app is picked up on the next retry.
• Reconnection does not stop on an authentication failure. With no onTokenRefresh, or after it returns false, the client keeps retrying with backoff. Use ReconnectionPolicy(retryIf: …) to stop earlier, or call disconnect() on logout.
• connectionId is synthetic when negotiation is skipped. The server only issues one in the negotiate response. Without it, the client fills in a local timestamp, which the server does not know. Don't send it to your backend on that route.
• No stateful reconnect on ASP.NET Core SignalR. .NET 8's stateful reconnect (replaying messages across a short drop) needs support in signalr_netcore, which does not have it. Refetch on reconnectedStream. Classic 2.x hubs do get replay, through session resumption.
• MessagePack: a server message with a null argument cannot be decoded by signalr_netcore. It is dropped and reported on errorStream as a parseError; the connection carries on. Sending null arguments works. Dates sent as MessagePack timestamps are not decoded either.
• Raw WebSocket (RealtimeTransportType.webSocket): headers are not sent on the upgrade request, on any platform. Put credentials in the URI (sendTokenAsQuery, the default).
• Classic SignalR on the web: headers reach the HTTP handshake requests, not the WebSocket upgrade — browsers cannot set them. Keep the token in the query string (tokenQueryParam, the default). On Android, iOS and desktop the upgrade carries them too.

Docs

architecture · public API · testing · known issues

License

MIT. See LICENSE.