From c08fc6e066f7f8a68e1b859cb0da21670906f98b Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 10:24:45 +0200 Subject: [PATCH 01/16] feat(realtime): add support for protocol format 2.0.0 Implements Realtime protocol 2.0.0 for the Dart/Flutter SDK (SDK-583). - Add a `Serializer` ported from the supabase-js reference that encodes text frames as the positional JSON array `[join_ref, ref, topic, event, payload]` and handles binary frames (`userBroadcastPush` kind 3 encode, `userBroadcast` kind 4 decode) with the JSON/binary encoding flag and `allowedMetadataKeys`. - Default the connection to `vsn=2.0.0` and wire the serializer as the default encode/decode. - Decode incoming binary broadcast frames and send broadcasts whose payload is a `Uint8List`/`TypedData` as binary frames. - Loosen `RealtimeEncode`/`RealtimeDecode` to carry `String` or bytes and let the connection listener accept binary frames. - Rename the `conn` family to avoid abbreviations: `conn` -> `connection`, `connState` -> `connectionStatus`, `onConnMessage` -> `onConnectionMessage`, `_onConn*` -> `_onConnection*`. - Update realtime frame mocks/tests to the 2.0.0 array format and add `serializer_test.dart`. BREAKING CHANGE: the default Realtime protocol is now 2.0.0, and the `RealtimeClient.conn`/`connState` fields are renamed to `connection`/`connectionStatus`. --- .../realtime_client/lib/src/constants.dart | 2 +- .../lib/src/realtime_channel.dart | 13 + .../lib/src/realtime_client.dart | 105 ++++---- .../realtime_client/lib/src/serializer.dart | 232 ++++++++++++++++ packages/realtime_client/test/mock_test.dart | 117 ++++---- .../realtime_client/test/serializer_test.dart | 252 ++++++++++++++++++ .../realtime_client/test/socket_test.dart | 110 ++++---- packages/supabase/test/mock_test.dart | 98 +++---- .../supabase_flutter/test/lifecycle_test.dart | 24 +- 9 files changed, 744 insertions(+), 209 deletions(-) create mode 100644 packages/realtime_client/lib/src/serializer.dart create mode 100644 packages/realtime_client/test/serializer_test.dart diff --git a/packages/realtime_client/lib/src/constants.dart b/packages/realtime_client/lib/src/constants.dart index 8e8a97ea0..7c7ad6f8c 100644 --- a/packages/realtime_client/lib/src/constants.dart +++ b/packages/realtime_client/lib/src/constants.dart @@ -1,7 +1,7 @@ import 'package:realtime_client/src/version.dart'; class Constants { - static const String vsn = '1.0.0'; + static const String vsn = '2.0.0'; static const Duration defaultTimeout = Duration(milliseconds: 10000); static const int defaultHeartbeatIntervalMs = 25000; static const int wsCloseNormal = 1000; diff --git a/packages/realtime_client/lib/src/realtime_channel.dart b/packages/realtime_client/lib/src/realtime_channel.dart index c509209e8..702a51522 100644 --- a/packages/realtime_client/lib/src/realtime_channel.dart +++ b/packages/realtime_client/lib/src/realtime_channel.dart @@ -592,6 +592,19 @@ class RealtimeChannel { } /// Sends a realtime broadcast message. + /// + /// With protocol `2.0.0` the message is sent as a positional JSON text frame. + /// + /// To send a raw binary payload over a WebSocket binary frame (avoiding JSON + /// encoding on the server), set a `Uint8List` (or any `TypedData`) under the + /// `payload` key, mirroring how supabase-js carries binary broadcast data: + /// + /// ```dart + /// channel.sendBroadcastMessage( + /// event: 'file', + /// payload: {'payload': myUint8List}, + /// ); + /// ``` Future sendBroadcastMessage({ required String event, required Map payload, diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index d9a0b26be..05308c7e2 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -1,5 +1,4 @@ import 'dart:async'; -import 'dart:convert'; import 'dart:core'; import 'package:collection/collection.dart'; @@ -10,6 +9,7 @@ import 'package:realtime_client/realtime_client.dart'; import 'package:realtime_client/src/constants.dart'; import 'package:realtime_client/src/message.dart'; import 'package:realtime_client/src/retry_timer.dart'; +import 'package:realtime_client/src/serializer.dart'; import 'package:realtime_client/src/websocket/websocket.dart'; import 'package:web_socket_channel/web_socket_channel.dart'; @@ -20,11 +20,11 @@ typedef WebSocketTransport = WebSocketChannel Function( typedef RealtimeEncode = void Function( dynamic payload, - void Function(String result) callback, + void Function(dynamic result) callback, ); typedef RealtimeDecode = void Function( - String payload, + dynamic payload, void Function(dynamic result) callback, ); @@ -115,7 +115,7 @@ class RealtimeClient { late RealtimeEncode encode; late RealtimeDecode decode; late TimerCalculation reconnectAfterMs; - WebSocketChannel? conn; + WebSocketChannel? connection; List sendBuffer = []; Map> stateChangeCallbacks = { 'open': [], @@ -126,7 +126,7 @@ class RealtimeClient { @Deprecated("No longer used. Will be removed in the next major version.") int longpollerTimeout = 20000; - SocketStates? connState; + SocketStates? connectionStatus; // This is called `accessToken` in realtime-js Future Function()? customAccessToken; @@ -187,12 +187,13 @@ class RealtimeClient { this.reconnectAfterMs = reconnectAfterMs ?? RetryTimer.createRetryFunction(); + final serializer = Serializer(); this.encode = encode ?? - (dynamic payload, Function(String result) callback) => - callback(json.encode(payload)); + (dynamic payload, Function(dynamic result) callback) => + serializer.encode(payload as Map, callback); this.decode = decode ?? - (String payload, Function(dynamic result) callback) => - callback(json.decode(payload)); + (dynamic payload, Function(dynamic result) callback) => + serializer.decode(payload, callback); reconnectTimer = RetryTimer( () async { await disconnect(); @@ -205,73 +206,74 @@ class RealtimeClient { /// Connects the socket. @internal Future connect() async { - if (conn != null) { + if (connection != null) { return; } try { log('transport', 'connecting to $endPointURL', null); log('transport', 'connecting', null, Level.FINE); - connState = SocketStates.connecting; - final WebSocketChannel localConn = transport(endPointURL, headers); - conn = localConn; + connectionStatus = SocketStates.connecting; + final WebSocketChannel localConnection = transport(endPointURL, headers); + connection = localConnection; try { - await localConn.ready; + await localConnection.ready; } catch (error) { // Bail out if disconnect() ran or a new connect() started during await - if (conn != localConn) { + if (connection != localConnection) { return; } // Don't schedule a reconnect and emit error if connection has been // closed by the user or [disconnect] waits for the connection to be // ready before closing it. - if (connState != SocketStates.disconnected && - connState != SocketStates.disconnecting) { - connState = SocketStates.closed; - _onConnError(error); + if (connectionStatus != SocketStates.disconnected && + connectionStatus != SocketStates.disconnecting) { + connectionStatus = SocketStates.closed; + _onConnectionError(error); reconnectTimer.scheduleTimeout(); } return; } // Guard: bail out if disconnect() ran during the await - if (conn != localConn || connState != SocketStates.connecting) { + if (connection != localConnection || + connectionStatus != SocketStates.connecting) { return; } - connState = SocketStates.open; + connectionStatus = SocketStates.open; - _onConnOpen(); - localConn.stream.listen( - // incoming messages - (message) => onConnMessage(message as String), - onError: _onConnError, + _onConnectionOpen(); + localConnection.stream.listen( + // incoming messages (text frames are `String`, binary frames are bytes) + (message) => onConnectionMessage(message), + onError: _onConnectionError, onDone: () { // communication has been closed - if (connState != SocketStates.disconnected && - connState != SocketStates.disconnecting) { - connState = SocketStates.closed; + if (connectionStatus != SocketStates.disconnected && + connectionStatus != SocketStates.disconnecting) { + connectionStatus = SocketStates.closed; } - _onConnClose(); + _onConnectionClose(); }, ); } catch (e) { /// General error handling - _onConnError(e); + _onConnectionError(e); } } /// Disconnects the socket with status [code] and [reason] for the disconnect Future disconnect({int? code, String? reason}) async { - final conn = this.conn; - if (conn != null) { - final oldState = connState; + final connection = this.connection; + if (connection != null) { + final oldState = connectionStatus; final shouldCloseSink = oldState == SocketStates.open || oldState == SocketStates.connecting; if (shouldCloseSink) { // Don't set the state to `disconnecting` if the connection is already closed. - connState = SocketStates.disconnecting; + connectionStatus = SocketStates.disconnecting; log('transport', 'disconnecting', {'code': code, 'reason': reason}, Level.FINE); } @@ -279,20 +281,20 @@ class RealtimeClient { // Connection cannot be closed while it's still connecting. Wait for connection to // be ready and then close it. if (oldState == SocketStates.connecting) { - await conn.ready.catchError((_) {}); + await connection.ready.catchError((_) {}); } if (shouldCloseSink) { if (code != null) { - await conn.sink.close(code, reason ?? ''); + await connection.sink.close(code, reason ?? ''); } else { - await conn.sink.close(); + await connection.sink.close(); } - connState = SocketStates.disconnected; + connectionStatus = SocketStates.disconnected; reconnectTimer.reset(); log('transport', 'disconnected', null, Level.FINE); } - this.conn = null; + this.connection = null; // remove open handles if (heartbeatTimer != null) heartbeatTimer?.cancel(); @@ -353,7 +355,7 @@ class RealtimeClient { /// Returns the current state of the socket. String get connectionState { - switch (connState) { + switch (connectionStatus) { case SocketStates.connecting: return 'connecting'; case SocketStates.open: @@ -369,7 +371,7 @@ class RealtimeClient { } /// Returns `true` is the connection is open. - bool get isConnected => connState == SocketStates.open; + bool get isConnected => connectionStatus == SocketStates.open; /// Removes a subscription from the socket. @internal @@ -394,7 +396,7 @@ class RealtimeClient { /// If the socket is not connected, the message gets enqueued within a local buffer, and sent out when a connection is next established. String? push(Message message) { void callback() { - encode(message.toJson(), (result) => conn?.sink.add(result)); + encode(message.toJson(), (result) => connection?.sink.add(result)); } log('push', '${message.topic} ${message.event} (${message.ref})', @@ -408,7 +410,7 @@ class RealtimeClient { return null; } - void onConnMessage(String rawMessage) { + void onConnectionMessage(dynamic rawMessage) { decode(rawMessage, (msg) { final topic = msg['topic'] as String; final event = msg['event'] as String; @@ -493,7 +495,7 @@ class RealtimeClient { } } - void _onConnOpen() { + void _onConnectionOpen() { log('transport', 'connected to $endPointURL'); log('transport', 'connected', null, Level.FINE); _flushSendBuffer(); @@ -509,17 +511,18 @@ class RealtimeClient { } /// communication has been closed - void _onConnClose() { - final statusCode = conn?.closeCode; + void _onConnectionClose() { + final statusCode = connection?.closeCode; RealtimeCloseEvent? event; if (statusCode != null) { - event = RealtimeCloseEvent(code: statusCode, reason: conn?.closeReason); + event = + RealtimeCloseEvent(code: statusCode, reason: connection?.closeReason); } log('transport', 'close', event, Level.FINE); /// SocketStates.disconnected: by user with socket.disconnect() /// SocketStates.closed: NOT by user, should try to reconnect - if (connState == SocketStates.closed) { + if (connectionStatus == SocketStates.closed) { _triggerChanError(event); reconnectTimer.scheduleTimeout(); } @@ -529,7 +532,7 @@ class RealtimeClient { } } - void _onConnError(dynamic error) { + void _onConnectionError(dynamic error) { log('transport', error.toString()); _triggerChanError(error); for (final callback in stateChangeCallbacks['error']!) { @@ -579,7 +582,7 @@ class RealtimeClient { 'transport', 'heartbeat timeout. Attempting to re-establish connection', ); - conn?.sink.close(Constants.wsCloseNormal, 'heartbeat timeout'); + connection?.sink.close(Constants.wsCloseNormal, 'heartbeat timeout'); return; } pendingHeartbeatRef = makeRef(); diff --git a/packages/realtime_client/lib/src/serializer.dart b/packages/realtime_client/lib/src/serializer.dart new file mode 100644 index 000000000..f3fcc0d1f --- /dev/null +++ b/packages/realtime_client/lib/src/serializer.dart @@ -0,0 +1,232 @@ +import 'dart:convert'; +import 'dart:typed_data'; + +/// Encodes and decodes Realtime protocol `2.0.0` frames. +/// +/// Text frames use the positional JSON array +/// `[joinRef, ref, topic, event, payload]` instead of the `1.0.0` object +/// layout. This lets the server skip part of the JSON encoding/decoding work, +/// lowering latency. +/// +/// Broadcast messages whose user payload is binary are sent as binary +/// WebSocket frames so that raw bytes can be forwarded without JSON encoding. +/// Incoming binary broadcast frames are decoded back into the same map shape as +/// their JSON counterparts. +/// +/// Ported from the supabase-js serializer: +/// https://github.com/supabase/supabase-js/blob/master/packages/core/realtime-js/src/lib/serializer.ts +class Serializer { + static const int headerLength = 1; + static const int userBroadcastPushMetaLength = 6; + + /// Binary frame sent by the client for a broadcast push. + static const int kindUserBroadcastPush = 3; + + /// Binary frame received from the server for a broadcast. + static const int kindUserBroadcast = 4; + + static const int binaryEncoding = 0; + static const int jsonEncoding = 1; + static const String broadcastEvent = 'broadcast'; + + /// Keys of the broadcast payload that are forwarded as frame metadata when + /// sending a binary broadcast push. + final List allowedMetadataKeys; + + Serializer({List? allowedMetadataKeys}) + : allowedMetadataKeys = allowedMetadataKeys ?? const []; + + /// Encodes a message map into the string or binary representation that is + /// written to the WebSocket. + /// + /// [message] is expected to hold `join_ref`, `ref`, `topic`, `event` and + /// `payload` keys, matching the output of `Message.toJson()`. + void encode( + Map message, + void Function(dynamic result) callback, + ) { + final payload = message['payload']; + if (message['event'] == broadcastEvent && + payload is Map && + payload['event'] is String && + _isBinary(payload['payload'])) { + callback(_encodeBinaryUserBroadcastPush(message, payload)); + return; + } + + callback(jsonEncode([ + message['join_ref'], + message['ref'], + message['topic'], + message['event'], + payload, + ])); + } + + /// Decodes a raw WebSocket frame into a message map with `join_ref`, `ref`, + /// `topic`, `event` and `payload` keys. + void decode( + dynamic rawPayload, + void Function(dynamic result) callback, + ) { + if (rawPayload is String) { + final decoded = jsonDecode(rawPayload) as List; + callback({ + 'join_ref': decoded[0], + 'ref': decoded[1], + 'topic': decoded[2], + 'event': decoded[3], + 'payload': decoded[4], + }); + return; + } + + final bytes = _asBytes(rawPayload); + if (bytes != null) { + callback(_binaryDecode(bytes)); + return; + } + + callback({}); + } + + Uint8List _encodeBinaryUserBroadcastPush( + Map message, + Map payload, + ) { + final topic = (message['topic'] ?? '') as String; + final ref = (message['ref'] ?? '') as String; + final joinRef = (message['join_ref'] ?? '') as String; + final userEvent = payload['event'] as String; + final encodedPayload = _asBytes(payload['payload'])!; + + final rest = allowedMetadataKeys.isEmpty + ? const {} + : _pick(payload, allowedMetadataKeys); + final metadata = rest.isEmpty ? '' : jsonEncode(rest); + + _checkLength('joinRef', joinRef.length); + _checkLength('ref', ref.length); + _checkLength('topic', topic.length); + _checkLength('userEvent', userEvent.length); + _checkLength('metadata', metadata.length); + + final metaLength = userBroadcastPushMetaLength + + joinRef.length + + ref.length + + topic.length + + userEvent.length + + metadata.length; + + final header = Uint8List(headerLength + metaLength); + var offset = 0; + header[offset++] = kindUserBroadcastPush; + header[offset++] = joinRef.length; + header[offset++] = ref.length; + header[offset++] = topic.length; + header[offset++] = userEvent.length; + header[offset++] = metadata.length; + header[offset++] = binaryEncoding; + offset = _writeString(header, offset, joinRef); + offset = _writeString(header, offset, ref); + offset = _writeString(header, offset, topic); + offset = _writeString(header, offset, userEvent); + offset = _writeString(header, offset, metadata); + + final combined = Uint8List(header.length + encodedPayload.length); + combined.setAll(0, header); + combined.setAll(header.length, encodedPayload); + return combined; + } + + Map _binaryDecode(Uint8List buffer) { + final view = ByteData.sublistView(buffer); + final kind = view.getUint8(0); + switch (kind) { + case kindUserBroadcast: + return _decodeUserBroadcast(buffer, view); + default: + return {}; + } + } + + Map _decodeUserBroadcast(Uint8List buffer, ByteData view) { + final topicSize = view.getUint8(1); + final userEventSize = view.getUint8(2); + final metadataSize = view.getUint8(3); + final payloadEncoding = view.getUint8(4); + + var offset = headerLength + 4; + final topic = utf8.decode(buffer.sublist(offset, offset + topicSize)); + offset += topicSize; + final userEvent = + utf8.decode(buffer.sublist(offset, offset + userEventSize)); + offset += userEventSize; + final metadata = metadataSize > 0 + ? utf8.decode(buffer.sublist(offset, offset + metadataSize)) + : ''; + offset += metadataSize; + + final payloadBytes = buffer.sublist(offset); + final dynamic parsedPayload = payloadEncoding == jsonEncoding + ? jsonDecode(utf8.decode(payloadBytes)) + : payloadBytes; + + final data = { + 'type': broadcastEvent, + 'event': userEvent, + 'payload': parsedPayload, + }; + if (metadataSize > 0) { + data['meta'] = jsonDecode(metadata); + } + + return { + 'join_ref': null, + 'ref': null, + 'topic': topic, + 'event': broadcastEvent, + 'payload': data, + }; + } + + void _checkLength(String field, int length) { + if (length > 255) { + throw ArgumentError('$field length $length exceeds maximum of 255'); + } + } + + int _writeString(Uint8List buffer, int offset, String value) { + for (final unit in value.codeUnits) { + buffer[offset++] = unit & 0xFF; + } + return offset; + } + + bool _isBinary(dynamic value) { + return value is Uint8List || value is ByteBuffer || value is TypedData; + } + + Uint8List? _asBytes(dynamic value) { + if (value is Uint8List) { + return value; + } + if (value is ByteBuffer) { + return value.asUint8List(); + } + if (value is TypedData) { + return value.buffer.asUint8List(value.offsetInBytes, value.lengthInBytes); + } + if (value is List) { + return Uint8List.fromList(value); + } + return null; + } + + Map _pick(Map obj, List keys) { + return { + for (final key in keys) + if (obj.containsKey(key)) key: obj[key], + }; + } +} diff --git a/packages/realtime_client/test/mock_test.dart b/packages/realtime_client/test/mock_test.dart index 89455fb75..de7510674 100644 --- a/packages/realtime_client/test/mock_test.dart +++ b/packages/realtime_client/test/mock_test.dart @@ -30,24 +30,28 @@ void main() { } hasSentData = true; + /// Protocol 2.0.0 text frames are positional arrays: + /// [join_ref, ref, topic, event, payload]. + /// /// `filter` might be there or not depending on whether is a filter set /// to the realtime subscription, so include the filter if the request /// includes a filter. - final requestJson = jsonDecode(request); - final String? postgresFilter = requestJson['payload']['config'] - ['postgres_changes'] - .first['filter']; + final requestJson = jsonDecode(request as String) as List; + final requestPayload = requestJson[4] as Map; + final String? postgresFilter = + requestPayload['config']['postgres_changes'].first['filter']; - final topic = (jsonDecode(request as String) as Map)['topic']; + final topic = requestJson[2]; // Send an insert event if (postgresFilter == null) { await Future.delayed(Duration(milliseconds: 300)); - final insertString = jsonEncode({ - 'topic': topic, - 'event': 'postgres_changes', - 'ref': null, - 'payload': { + final insertString = jsonEncode([ + null, + null, + topic, + 'postgres_changes', + { 'ids': [77086988], 'data': { 'commit_timestamp': '2021-08-01T08:00:20Z', @@ -75,17 +79,18 @@ void main() { ], }, }, - }); + ]); webSocket!.add(insertString); } // Send an update event for id = 2 await Future.delayed(Duration(milliseconds: 10)); - final updateString = jsonEncode({ - 'topic': topic, - 'ref': null, - 'event': 'postgres_changes', - 'payload': { + final updateString = jsonEncode([ + null, + null, + topic, + 'postgres_changes', + { 'ids': [25993878], 'data': { 'columns': [ @@ -107,16 +112,17 @@ void main() { if (postgresFilter != null) 'filter': postgresFilter, }, }, - }); + ]); webSocket!.add(updateString); // Send delete event for id=2 await Future.delayed(Duration(milliseconds: 10)); - final deleteString = jsonEncode({ - 'ref': null, - 'topic': topic, - 'event': 'postgres_changes', - 'payload': { + final deleteString = jsonEncode([ + null, + null, + topic, + 'postgres_changes', + { 'data': { 'columns': [ {'name': 'id', 'type': 'int4', 'type_modifier': 4294967295}, @@ -137,7 +143,7 @@ void main() { }, 'ids': [48673474] }, - }); + ]); webSocket!.add(deleteString); }); } else { @@ -309,31 +315,38 @@ void main() { } hasSentData = true; + /// Protocol 2.0.0 text frames are positional arrays: + /// [join_ref, ref, topic, event, payload]. + /// /// `filter` might be there or not depending on whether is a filter set /// to the realtime subscription, so include the filter if the request /// includes a filter. - final requestJson = jsonDecode(request); + final requestJson = jsonDecode(request as String) as List; + final requestPayload = requestJson[4] as Map; - final String? postgresFilter = requestJson['payload']['config'] - ['postgres_changes'] - .first['filter']; + final String? postgresFilter = + requestPayload['config']['postgres_changes'].first['filter']; - final topic = requestJson['topic']; + final topic = requestJson[2]; - final replyString = jsonEncode({ - "event": "phx_reply", - "payload": {"response": {}, "status": "ok"}, - "ref": "1", - "topic": topic - }); + final replyString = jsonEncode([ + null, + "1", + topic, + "phx_reply", + {"response": {}, "status": "ok"}, + ]); webSocket!.add(replyString); // Send an insert event if (postgresFilter == null) { await Future.delayed(Duration(milliseconds: 300)); - final insertString = jsonEncode({ - "event": "INSERT", - "payload": { + final insertString = jsonEncode([ + null, + null, + topic, + "INSERT", + { "columns": [ {"name": "id", "type": "int4"}, {"name": "task", "type": "text"}, @@ -346,17 +359,18 @@ void main() { "table": "todos", "type": "INSERT" }, - "ref": null, - "topic": topic - }); + ]); webSocket!.add(insertString); } // Send an update event for id = 2 await Future.delayed(Duration(milliseconds: 10)); - final updateString = jsonEncode({ - "event": "UPDATE", - "payload": { + final updateString = jsonEncode([ + null, + null, + topic, + "UPDATE", + { "columns": [ {"name": "id", "type": "int4"}, {"name": "task", "type": "text"}, @@ -370,16 +384,17 @@ void main() { "table": "todos", "type": "UPDATE" }, - "ref": null, - "topic": topic - }); + ]); webSocket!.add(updateString); // Send delete event for id=2 await Future.delayed(Duration(milliseconds: 10)); - final deleteString = jsonEncode({ - "event": "DELETE", - "payload": { + final deleteString = jsonEncode([ + null, + null, + topic, + "DELETE", + { "columns": [ {"name": "id", "type": "int4"}, {"name": "task", "type": "text"}, @@ -393,9 +408,7 @@ void main() { "table": "todos", "type": "DELETE" }, - "ref": null, - "topic": topic - }); + ]); webSocket!.add(deleteString); }); } else { diff --git a/packages/realtime_client/test/serializer_test.dart b/packages/realtime_client/test/serializer_test.dart new file mode 100644 index 000000000..2921e0ae7 --- /dev/null +++ b/packages/realtime_client/test/serializer_test.dart @@ -0,0 +1,252 @@ +import 'dart:convert'; +import 'dart:typed_data'; + +import 'package:realtime_client/src/serializer.dart'; +import 'package:test/test.dart'; + +/// Builds a `kind = userBroadcast` (4) binary frame the same way the server +/// does, so the decoder can be exercised in isolation. +Uint8List buildUserBroadcastFrame({ + required String topic, + required String event, + required Uint8List payload, + required int encoding, + String metadata = '', +}) { + final topicBytes = utf8.encode(topic); + final eventBytes = utf8.encode(event); + final metadataBytes = utf8.encode(metadata); + + final header = [ + Serializer.kindUserBroadcast, + topicBytes.length, + eventBytes.length, + metadataBytes.length, + encoding, + ]; + + return Uint8List.fromList([ + ...header, + ...topicBytes, + ...eventBytes, + ...metadataBytes, + ...payload, + ]); +} + +dynamic encodeSync(Serializer serializer, Map message) { + dynamic result; + serializer.encode(message, (r) => result = r); + return result; +} + +dynamic decodeSync(Serializer serializer, dynamic raw) { + dynamic result; + serializer.decode(raw, (r) => result = r); + return result; +} + +void main() { + late Serializer serializer; + + setUp(() { + serializer = Serializer(); + }); + + group('encode text frames', () { + test('encodes a message as a positional JSON array', () { + final result = encodeSync(serializer, { + 'join_ref': '1', + 'ref': '2', + 'topic': 'realtime:room', + 'event': 'phx_join', + 'payload': {'foo': 'bar'}, + }); + + expect(result, isA()); + expect( + jsonDecode(result as String), + equals([ + '1', + '2', + 'realtime:room', + 'phx_join', + {'foo': 'bar'}, + ]), + ); + }); + + test('preserves null join_ref and ref positionally', () { + final result = encodeSync(serializer, { + 'topic': 'phoenix', + 'event': 'heartbeat', + 'payload': {}, + }); + + expect( + jsonDecode(result as String), + equals([null, null, 'phoenix', 'heartbeat', {}]), + ); + }); + + test('encodes a non-binary broadcast as a text frame', () { + final result = encodeSync(serializer, { + 'join_ref': '1', + 'ref': '2', + 'topic': 'realtime:room', + 'event': 'broadcast', + 'payload': {'event': 'cursor', 'type': 'broadcast', 'x': 1}, + }); + + expect(result, isA()); + expect((jsonDecode(result as String) as List)[3], 'broadcast'); + }); + }); + + group('decode text frames', () { + test('decodes a positional JSON array into a message map', () { + final result = decodeSync( + serializer, + jsonEncode([ + '1', + '2', + 'realtime:room', + 'phx_reply', + {'status': 'ok'} + ]), + ); + + expect(result, { + 'join_ref': '1', + 'ref': '2', + 'topic': 'realtime:room', + 'event': 'phx_reply', + 'payload': {'status': 'ok'}, + }); + }); + }); + + group('decode binary frames', () { + test('decodes a JSON-encoded user broadcast', () { + final frame = buildUserBroadcastFrame( + topic: 'realtime:room', + event: 'cursor', + payload: Uint8List.fromList(utf8.encode(jsonEncode({'x': 1, 'y': 2}))), + encoding: Serializer.jsonEncoding, + metadata: jsonEncode({'replayed': true}), + ); + + final result = decodeSync(serializer, frame) as Map; + + expect(result['join_ref'], isNull); + expect(result['ref'], isNull); + expect(result['topic'], 'realtime:room'); + expect(result['event'], 'broadcast'); + expect(result['payload'], { + 'type': 'broadcast', + 'event': 'cursor', + 'payload': {'x': 1, 'y': 2}, + 'meta': {'replayed': true}, + }); + }); + + test('decodes a binary user broadcast payload as raw bytes', () { + final rawPayload = Uint8List.fromList([1, 2, 3, 4, 255]); + final frame = buildUserBroadcastFrame( + topic: 'realtime:room', + event: 'file', + payload: rawPayload, + encoding: Serializer.binaryEncoding, + ); + + final result = decodeSync(serializer, frame) as Map; + final payload = result['payload'] as Map; + + expect(payload['event'], 'file'); + expect(payload['payload'], rawPayload); + expect(payload.containsKey('meta'), isFalse); + }); + + test('returns an empty map for unknown binary kinds', () { + final result = decodeSync(serializer, Uint8List.fromList([99, 0, 0])); + expect(result, {}); + }); + }); + + group('encode binary broadcast push', () { + test('encodes a broadcast with a binary payload as a binary frame', () { + final payload = Uint8List.fromList([10, 20, 30]); + final result = encodeSync(serializer, { + 'join_ref': '7', + 'ref': '8', + 'topic': 'realtime:room', + 'event': 'broadcast', + 'payload': { + 'type': 'broadcast', + 'event': 'file', + 'payload': payload, + }, + }); + + expect(result, isA()); + final bytes = result as Uint8List; + + expect(bytes[0], Serializer.kindUserBroadcastPush); + expect(bytes[1], '7'.length); // joinRef length + expect(bytes[2], '8'.length); // ref length + expect(bytes[3], 'realtime:room'.length); // topic length + expect(bytes[4], 'file'.length); // userEvent length + expect(bytes[5], 0); // metadata length (no allowed keys) + expect(bytes[6], Serializer.binaryEncoding); // encoding + + // The user payload is appended verbatim at the end of the frame. + expect(bytes.sublist(bytes.length - payload.length), payload); + }); + + test('forwards allowed metadata keys', () { + final serializerWithMeta = Serializer(allowedMetadataKeys: ['trace_id']); + final result = encodeSync(serializerWithMeta, { + 'topic': 'realtime:room', + 'event': 'broadcast', + 'payload': { + 'type': 'broadcast', + 'event': 'file', + 'trace_id': 'abc', + 'ignored': 'nope', + 'payload': Uint8List.fromList([1]), + }, + }); + + final bytes = result as Uint8List; + final metadataLength = bytes[5]; + expect(metadataLength, greaterThan(0)); + + // metadata sits after the header and the joinRef/ref/topic/event strings. + final metadataStart = Serializer.headerLength + + Serializer.userBroadcastPushMetaLength + + 0 + // joinRef + 0 + // ref + 'realtime:room'.length + + 'file'.length; + final metadata = utf8 + .decode(bytes.sublist(metadataStart, metadataStart + metadataLength)); + expect(jsonDecode(metadata), {'trace_id': 'abc'}); + }); + + test('throws when a frame field exceeds 255 bytes', () { + final longTopic = 'a' * 256; + expect( + () => encodeSync(serializer, { + 'topic': longTopic, + 'event': 'broadcast', + 'payload': { + 'type': 'broadcast', + 'event': 'file', + 'payload': Uint8List.fromList([1]), + }, + }), + throwsArgumentError, + ); + }); + }); +} diff --git a/packages/realtime_client/test/socket_test.dart b/packages/realtime_client/test/socket_test.dart index 67d4a638f..cbf95e3bd 100644 --- a/packages/realtime_client/test/socket_test.dart +++ b/packages/realtime_client/test/socket_test.dart @@ -1,6 +1,7 @@ import 'dart:async'; import 'dart:convert'; import 'dart:io'; +import 'dart:typed_data'; import 'package:crypto/crypto.dart'; import 'package:mocktail/mocktail.dart'; @@ -137,7 +138,7 @@ void main() { final socket = RealtimeClient('wss://example.org/chat'); expect( socket.endPointURL, - 'wss://example.org/chat/websocket?vsn=1.0.0', + 'wss://example.org/chat/websocket?vsn=2.0.0', ); }); @@ -146,7 +147,7 @@ void main() { RealtimeClient('ws://example.org/chat', params: {'foo': 'bar'}); expect( socket.endPointURL, - 'ws://example.org/chat/websocket?foo=bar&vsn=1.0.0', + 'ws://example.org/chat/websocket?foo=bar&vsn=2.0.0', ); }); @@ -159,7 +160,7 @@ void main() { ); expect( socket.endPointURL, - 'ws://example.org/chat/websocket?apikey=123456789&vsn=1.0.0', + 'ws://example.org/chat/websocket?apikey=123456789&vsn=2.0.0', ); }); }); @@ -177,14 +178,14 @@ void main() { test('establishes websocket connection with endpoint', () async { final connFuture = socket.connect(); - expect(socket.connState, SocketStates.connecting); + expect(socket.connectionStatus, SocketStates.connecting); - final conn = socket.conn; + final connection = socket.connection; await connFuture; - expect(socket.connState, SocketStates.open); + expect(socket.connectionStatus, SocketStates.open); - expect(conn, isA()); + expect(connection, isA()); //! Not verifying connection url }); @@ -230,9 +231,9 @@ void main() { test('is idempotent', () { socket.connect(); - final conn = socket.conn; + final connection = socket.connection; socket.connect(); - expect(socket.conn, conn); + expect(socket.connection, connection); }); }); @@ -249,10 +250,10 @@ void main() { test('removes existing connection', () async { await socket.connect(); - expect(socket.conn, isNotNull); + expect(socket.connection, isNotNull); await socket.disconnect(); - expect(socket.conn, isNull); + expect(socket.connection, isNull); }); test('calls callback', () async { @@ -282,7 +283,7 @@ void main() { const tReason = 'reason'; mockedSocket.connect(); - mockedSocket.connState = SocketStates.open; + mockedSocket.connectionStatus = SocketStates.open; await Future.delayed(const Duration(milliseconds: 200)); mockedSocket.disconnect(code: tCode, reason: tReason); await Future.delayed(const Duration(milliseconds: 200)); @@ -297,32 +298,32 @@ void main() { test('disconnecting a closed connections stays closed', () async { await socket.connect(); - expect(socket.connState, SocketStates.open); + expect(socket.connectionStatus, SocketStates.open); await mockServer.close(); await Future.delayed(const Duration(milliseconds: 200)); - expect(socket.connState, SocketStates.closed); - expect(socket.conn, isNotNull); + expect(socket.connectionStatus, SocketStates.closed); + expect(socket.connection, isNotNull); final disconnectFuture = socket.disconnect(); - // `connState` stays `closed` during disconnect - expect(socket.connState, SocketStates.closed); + // `connectionStatus` stays `closed` during disconnect + expect(socket.connectionStatus, SocketStates.closed); await disconnectFuture; - expect(socket.connState, SocketStates.closed); - expect(socket.conn, isNull); + expect(socket.connectionStatus, SocketStates.closed); + expect(socket.connection, isNull); }); test('disconnecting an open connection', () async { await socket.connect(); - expect(socket.connState, SocketStates.open); + expect(socket.connectionStatus, SocketStates.open); final disconnectFuture = socket.disconnect(); - // `connState` stays `closed` during disconnect - expect(socket.connState, SocketStates.disconnecting); + // `connectionStatus` stays `closed` during disconnect + expect(socket.connectionStatus, SocketStates.disconnecting); await disconnectFuture; - expect(socket.connState, SocketStates.disconnected); - expect(socket.conn, isNull); + expect(socket.connectionStatus, SocketStates.disconnected); + expect(socket.connection, isNull); }); test('does not throw when no connection', () { @@ -409,12 +410,10 @@ void main() { const event = ChannelEvents.join; const payload = 'payload'; const ref = 'ref'; - final jsonData = json.encode({ - 'topic': topic, - 'event': event.eventName(), - 'payload': payload, - 'ref': ref - }); + // Protocol 2.0.0 text frames are positional arrays: + // [join_ref, ref, topic, event, payload]. + final jsonData = + json.encode([null, ref, topic, event.eventName(), payload]); IOWebSocketChannel mockedSocketChannel; late RealtimeClient mockedSocket; @@ -437,7 +436,7 @@ void main() { test('sends data to connection when connected', () { mockedSocket.connect(); - mockedSocket.connState = SocketStates.open; + mockedSocket.connectionStatus = SocketStates.open; final message = Message(topic: topic, payload: payload, event: event, ref: ref); @@ -449,7 +448,7 @@ void main() { test('buffers data when not connected', () async { mockedSocket.connect(); - mockedSocket.connState = SocketStates.connecting; + mockedSocket.connectionStatus = SocketStates.connecting; expect(mockedSocket.sendBuffer.length, 0); @@ -465,6 +464,26 @@ void main() { verify(() => mockedSink.add(captureAny(that: equals(jsonData)))) .called(1); }); + + test('sends a broadcast with a binary payload as a binary frame', () { + mockedSocket.connect(); + mockedSocket.connectionStatus = SocketStates.open; + + final binaryPayload = Uint8List.fromList([1, 2, 3]); + final message = Message( + topic: 'realtime:room', + event: ChannelEvents.broadcast, + payload: { + 'type': 'broadcast', + 'event': 'file', + 'payload': binaryPayload, + }, + ); + mockedSocket.push(message); + + verify(() => mockedSink.add(captureAny(that: isA()))) + .called(1); + }); }); group('makeRef', () { @@ -603,12 +622,7 @@ void main() { IOWebSocketChannel mockedSocketChannel; late RealtimeClient mockedSocket; late WebSocketSink mockedSink; - final data = json.encode({ - 'topic': 'phoenix', - 'event': 'heartbeat', - 'payload': {}, - 'ref': '1', - }); + final data = json.encode([null, '1', 'phoenix', 'heartbeat', {}]); setUp(() { mockedSocketChannel = MockIOWebSocketChannel(); @@ -630,7 +644,7 @@ void main() { //! Unimplemented Test: closes socket when heartbeat is not ack'd within heartbeat window test('pushes heartbeat data when connected', () async { - mockedSocket.connState = SocketStates.open; + mockedSocket.connectionStatus = SocketStates.open; await mockedSocket.sendHeartbeat(); @@ -638,7 +652,7 @@ void main() { }); test('no ops when not connected', () async { - mockedSocket.connState = SocketStates.connecting; + mockedSocket.connectionStatus = SocketStates.connecting; await mockedSocket.sendHeartbeat(); verifyNever(() => mockedSink.add(any())); @@ -647,7 +661,7 @@ void main() { group('connect/disconnect race condition', () { test( - 'connect does not crash if disconnect nullifies conn during await ready', + 'connect does not crash if disconnect nullifies connection during await ready', () async { final readyCompleter = Completer(); final mockedSocketChannel = MockIOWebSocketChannel(); @@ -676,12 +690,12 @@ void main() { await disconnectFuture; await connectFuture; - // Should NOT have transitioned to open because disconnect nullified conn - expect(socket.connState, isNot(SocketStates.open)); - expect(socket.conn, isNull); + // Should NOT have transitioned to open because disconnect nullified connection + expect(socket.connectionStatus, isNot(SocketStates.open)); + expect(socket.connection, isNull); }); - test('connect bails out when connState changes during await ready', + test('connect bails out when connectionStatus changes during await ready', () async { final readyCompleter = Completer(); final mockedSocketChannel = MockIOWebSocketChannel(); @@ -710,7 +724,7 @@ void main() { await disconnectFuture; await connectFuture; - expect(socket.connState, isNot(SocketStates.open)); + expect(socket.connectionStatus, isNot(SocketStates.open)); }); test('rapid connect-disconnect-connect cycle does not crash', () async { @@ -764,8 +778,8 @@ void main() { readyCompleter2.complete(); await socket.connect(); - expect(socket.connState, SocketStates.open); - expect(socket.conn, mockedSocketChannel2); + expect(socket.connectionStatus, SocketStates.open); + expect(socket.connection, mockedSocketChannel2); await socket.disconnect(); await streamController2.close(); diff --git a/packages/supabase/test/mock_test.dart b/packages/supabase/test/mock_test.dart index a37952098..baee930c2 100644 --- a/packages/supabase/test/mock_test.dart +++ b/packages/supabase/test/mock_test.dart @@ -108,13 +108,17 @@ void main() { } hasListener = true; listener = webSocket!.listen((request) async { + /// Protocol 2.0.0 text frames are positional arrays: + /// [join_ref, ref, topic, event, payload]. + /// /// `filter` might be there or not depending on whether is a filter set /// to the realtime subscription, so include the filter if the request /// includes a filter. - final requestJson = jsonDecode(request); - final topic = requestJson['topic']; - final ref = requestJson["ref"]; - final event = requestJson['event']; + final requestJson = jsonDecode(request as String) as List; + final ref = requestJson[1]; + final topic = requestJson[2]; + final event = requestJson[3]; + final requestPayload = requestJson[4] as Map; if (event == 'phx_leave') { listeners.remove(topic); @@ -125,11 +129,9 @@ void main() { } listeners.add(topic); - final String? realtimeFilter = requestJson['payload']['config'] - ['postgres_changes'] - .first['filter']; - final bool isPrivate = - requestJson['payload']['config']['private'] as bool; + final String? realtimeFilter = + requestPayload['config']['postgres_changes'].first['filter']; + final bool isPrivate = requestPayload['config']['private'] as bool; if (expectedFilter != null) { expect(realtimeFilter, expectedFilter); @@ -138,9 +140,12 @@ void main() { expect(isPrivate, expectedPrivate); } - final replyString = jsonEncode({ - 'event': 'phx_reply', - 'payload': { + final replyString = jsonEncode([ + null, + ref, + topic, + 'phx_reply', + { 'response': { 'postgres_changes': [ { @@ -154,18 +159,17 @@ void main() { }, 'status': 'ok' }, - 'ref': ref, - 'topic': topic - }); + ]); webSocket!.add(replyString); // Send an insert event await Future.delayed(Duration(milliseconds: 10)); - final insertString = jsonEncode({ - 'topic': topic, - 'event': 'postgres_changes', - 'ref': null, - 'payload': { + final insertString = jsonEncode([ + null, + null, + topic, + 'postgres_changes', + { 'ids': [77086988], 'data': { 'commit_timestamp': '2021-08-01T08:00:20Z', @@ -193,16 +197,17 @@ void main() { ], }, }, - }); + ]); webSocket!.add(insertString); // Send an update event for id = 2 await Future.delayed(Duration(milliseconds: 10)); - final updateString = jsonEncode({ - 'topic': topic, - 'ref': null, - 'event': 'postgres_changes', - 'payload': { + final updateString = jsonEncode([ + null, + null, + topic, + 'postgres_changes', + { 'ids': [77086988], 'data': { 'columns': [ @@ -224,16 +229,17 @@ void main() { if (realtimeFilter != null) 'filter': realtimeFilter, }, }, - }); + ]); webSocket!.add(updateString); // Send delete event for id=2 await Future.delayed(Duration(milliseconds: 10)); - final deleteString = jsonEncode({ - 'ref': null, - 'topic': topic, - 'event': 'postgres_changes', - 'payload': { + final deleteString = jsonEncode([ + null, + null, + topic, + 'postgres_changes', + { 'data': { 'columns': [ {'name': 'id', 'type': 'int4', 'type_modifier': 4294967295}, @@ -254,18 +260,19 @@ void main() { }, 'ids': [77086988] }, - }); + ]); webSocket!.add(deleteString); /// Send an update event for id = 4 /// Record with id = 4 did not exist in the initial data fetch, /// so the SDK should insert the record in the in memory cache await Future.delayed(Duration(milliseconds: 10)); - final updateId4 = jsonEncode({ - 'topic': topic, - 'ref': null, - 'event': 'postgres_changes', - 'payload': { + final updateId4 = jsonEncode([ + null, + null, + topic, + 'postgres_changes', + { 'ids': [77086988], 'data': { 'columns': [ @@ -287,17 +294,18 @@ void main() { if (realtimeFilter != null) 'filter': realtimeFilter, }, }, - }); + ]); webSocket!.add(updateId4); // Send delete event for id=5 /// Should be ignored by the SDK await Future.delayed(Duration(milliseconds: 10)); - final ignoredDeleteString = jsonEncode({ - 'ref': null, - 'topic': topic, - 'event': 'postgres_changes', - 'payload': { + final ignoredDeleteString = jsonEncode([ + null, + null, + topic, + 'postgres_changes', + { 'data': { 'columns': [ {'name': 'id', 'type': 'int4', 'type_modifier': 4294967295}, @@ -318,7 +326,7 @@ void main() { }, 'ids': [77086988] }, - }); + ]); webSocket!.add(ignoredDeleteString); }); } else { diff --git a/packages/supabase_flutter/test/lifecycle_test.dart b/packages/supabase_flutter/test/lifecycle_test.dart index d94cb3c44..4d74dc1ec 100644 --- a/packages/supabase_flutter/test/lifecycle_test.dart +++ b/packages/supabase_flutter/test/lifecycle_test.dart @@ -138,7 +138,7 @@ void main() { // Connect with ready completed immediately await connectAndReady(realtime); - expect(realtime.connState, SocketStates.open); + expect(realtime.connectionStatus, SocketStates.open); // paused → triggers disconnect binding.handleAppLifecycleStateChanged(AppLifecycleState.inactive); @@ -152,8 +152,8 @@ void main() { // Complete any pending ready futures (reconnect) await settleLifecycle(); - expect(realtime.connState, SocketStates.open); - expect(realtime.conn, isNotNull); + expect(realtime.connectionStatus, SocketStates.open); + expect(realtime.connection, isNotNull); }); test( @@ -165,7 +165,7 @@ void main() { realtime.channel('test'); await connectAndReady(realtime); - expect(realtime.connState, SocketStates.open); + expect(realtime.connectionStatus, SocketStates.open); // paused → starts disconnect binding.handleAppLifecycleStateChanged(AppLifecycleState.inactive); @@ -186,8 +186,8 @@ void main() { await settleLifecycle(); // Should have reconnected, not stuck disconnecting - expect(realtime.connState, SocketStates.open); - expect(realtime.conn, isNotNull); + expect(realtime.connectionStatus, SocketStates.open); + expect(realtime.connection, isNotNull); }); test( @@ -199,7 +199,7 @@ void main() { realtime.channel('test'); await connectAndReady(realtime); - expect(realtime.connState, SocketStates.open); + expect(realtime.connectionStatus, SocketStates.open); // Rapid lifecycle flapping binding.handleAppLifecycleStateChanged(AppLifecycleState.inactive); @@ -219,8 +219,8 @@ void main() { // Complete all pending ready futures as they appear await settleLifecycle(); - expect(realtime.connState, SocketStates.open); - expect(realtime.conn, isNotNull); + expect(realtime.connectionStatus, SocketStates.open); + expect(realtime.connection, isNotNull); }); test( @@ -232,7 +232,7 @@ void main() { realtime.channel('test'); await connectAndReady(realtime); - expect(realtime.connState, SocketStates.open); + expect(realtime.connectionStatus, SocketStates.open); // paused → triggers disconnect binding.handleAppLifecycleStateChanged(AppLifecycleState.inactive); @@ -253,8 +253,8 @@ void main() { await settleLifecycle(); // Should be disconnected since the last event was paused - expect(realtime.connState, SocketStates.disconnected); - expect(realtime.conn, isNull); + expect(realtime.connectionStatus, SocketStates.disconnected); + expect(realtime.connection, isNull); }); }); } From 190e6cf0d68861f0160c142b7a3830a6f338428b Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 10:33:49 +0200 Subject: [PATCH 02/16] refactor(realtime): return encode/decode results instead of using a callback The callback form was a port artifact from the realtime-js/Phoenix serializer. Both encode and decode are synchronous here, so return the result directly: RealtimeEncode is now Object Function(Map) and RealtimeDecode is Map Function(Object). --- .../lib/src/realtime_client.dart | 79 +++++++++---------- .../realtime_client/lib/src/serializer.dart | 27 +++---- .../realtime_client/test/serializer_test.dart | 33 +++----- 3 files changed, 56 insertions(+), 83 deletions(-) diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index 05308c7e2..7c67df5f6 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -18,15 +18,13 @@ typedef WebSocketTransport = WebSocketChannel Function( Map headers, ); -typedef RealtimeEncode = void Function( - dynamic payload, - void Function(dynamic result) callback, -); +/// Encodes an outgoing message into the `String` or binary frame to write to +/// the WebSocket. +typedef RealtimeEncode = Object Function(Map payload); -typedef RealtimeDecode = void Function( - dynamic payload, - void Function(dynamic result) callback, -); +/// Decodes a raw incoming WebSocket frame (`String` or binary) into a message +/// map. +typedef RealtimeDecode = Map Function(Object payload); /// Event details for when the connection closed. class RealtimeCloseEvent { @@ -146,9 +144,11 @@ class RealtimeClient { /// /// [logger] The optional function for specialized logging, ie: logger: (kind, msg, data) => { console.log(`$kind: $msg`, data) } /// - /// [encode] The function to encode outgoing messages. Defaults to JSON: (payload, callback) => callback(JSON.stringify(payload)) + /// [encode] The function to encode outgoing messages. Defaults to the + /// protocol `2.0.0` serializer. /// - /// [decode] The function to decode incoming messages. Defaults to JSON: (payload, callback) => callback(JSON.parse(payload)) + /// [decode] The function to decode incoming messages. Defaults to the + /// protocol `2.0.0` serializer. /// /// [reconnectAfterMs] The optional function that returns the millsec reconnect interval. Defaults to stepped backoff off. /// @@ -188,12 +188,8 @@ class RealtimeClient { this.reconnectAfterMs = reconnectAfterMs ?? RetryTimer.createRetryFunction(); final serializer = Serializer(); - this.encode = encode ?? - (dynamic payload, Function(dynamic result) callback) => - serializer.encode(payload as Map, callback); - this.decode = decode ?? - (dynamic payload, Function(dynamic result) callback) => - serializer.decode(payload, callback); + this.encode = encode ?? serializer.encode; + this.decode = decode ?? serializer.decode; reconnectTimer = RetryTimer( () async { await disconnect(); @@ -396,7 +392,7 @@ class RealtimeClient { /// If the socket is not connected, the message gets enqueued within a local buffer, and sent out when a connection is next established. String? push(Message message) { void callback() { - encode(message.toJson(), (result) => connection?.sink.add(result)); + connection?.sink.add(encode(message.toJson())); } log('push', '${message.topic} ${message.event} (${message.ref})', @@ -410,33 +406,32 @@ class RealtimeClient { return null; } - void onConnectionMessage(dynamic rawMessage) { - decode(rawMessage, (msg) { - final topic = msg['topic'] as String; - final event = msg['event'] as String; - final payload = msg['payload']; - final ref = msg['ref'] as String?; - if (ref != null && ref == pendingHeartbeatRef) { - pendingHeartbeatRef = null; - } + void onConnectionMessage(Object rawMessage) { + final msg = decode(rawMessage); + final topic = msg['topic'] as String; + final event = msg['event'] as String; + final payload = msg['payload']; + final ref = msg['ref'] as String?; + if (ref != null && ref == pendingHeartbeatRef) { + pendingHeartbeatRef = null; + } - log( - 'receive', - "${payload['status'] ?? ''} $topic $event ${ref != null ? '($ref)' : ''}", - payload, - ); + log( + 'receive', + "${payload['status'] ?? ''} $topic $event ${ref != null ? '($ref)' : ''}", + payload, + ); - channels - .where((channel) => channel.isMember(topic)) - .forEach((channel) => channel.trigger( - event, - payload, - ref, - )); - for (final callback in stateChangeCallbacks['message']!) { - callback(msg); - } - }); + channels.where((channel) => channel.isMember(topic)).forEach( + (channel) => channel.trigger( + event, + payload, + ref, + ), + ); + for (final callback in stateChangeCallbacks['message']!) { + callback(msg); + } } /// Returns the URL of the websocket. diff --git a/packages/realtime_client/lib/src/serializer.dart b/packages/realtime_client/lib/src/serializer.dart index f3fcc0d1f..771bf31a8 100644 --- a/packages/realtime_client/lib/src/serializer.dart +++ b/packages/realtime_client/lib/src/serializer.dart @@ -41,53 +41,44 @@ class Serializer { /// /// [message] is expected to hold `join_ref`, `ref`, `topic`, `event` and /// `payload` keys, matching the output of `Message.toJson()`. - void encode( - Map message, - void Function(dynamic result) callback, - ) { + Object encode(Map message) { final payload = message['payload']; if (message['event'] == broadcastEvent && payload is Map && payload['event'] is String && _isBinary(payload['payload'])) { - callback(_encodeBinaryUserBroadcastPush(message, payload)); - return; + return _encodeBinaryUserBroadcastPush(message, payload); } - callback(jsonEncode([ + return jsonEncode([ message['join_ref'], message['ref'], message['topic'], message['event'], payload, - ])); + ]); } /// Decodes a raw WebSocket frame into a message map with `join_ref`, `ref`, /// `topic`, `event` and `payload` keys. - void decode( - dynamic rawPayload, - void Function(dynamic result) callback, - ) { + Map decode(Object rawPayload) { if (rawPayload is String) { final decoded = jsonDecode(rawPayload) as List; - callback({ + return { 'join_ref': decoded[0], 'ref': decoded[1], 'topic': decoded[2], 'event': decoded[3], 'payload': decoded[4], - }); - return; + }; } final bytes = _asBytes(rawPayload); if (bytes != null) { - callback(_binaryDecode(bytes)); - return; + return _binaryDecode(bytes); } - callback({}); + return {}; } Uint8List _encodeBinaryUserBroadcastPush( diff --git a/packages/realtime_client/test/serializer_test.dart b/packages/realtime_client/test/serializer_test.dart index 2921e0ae7..60e6993af 100644 --- a/packages/realtime_client/test/serializer_test.dart +++ b/packages/realtime_client/test/serializer_test.dart @@ -34,18 +34,6 @@ Uint8List buildUserBroadcastFrame({ ]); } -dynamic encodeSync(Serializer serializer, Map message) { - dynamic result; - serializer.encode(message, (r) => result = r); - return result; -} - -dynamic decodeSync(Serializer serializer, dynamic raw) { - dynamic result; - serializer.decode(raw, (r) => result = r); - return result; -} - void main() { late Serializer serializer; @@ -55,7 +43,7 @@ void main() { group('encode text frames', () { test('encodes a message as a positional JSON array', () { - final result = encodeSync(serializer, { + final result = serializer.encode({ 'join_ref': '1', 'ref': '2', 'topic': 'realtime:room', @@ -77,7 +65,7 @@ void main() { }); test('preserves null join_ref and ref positionally', () { - final result = encodeSync(serializer, { + final result = serializer.encode({ 'topic': 'phoenix', 'event': 'heartbeat', 'payload': {}, @@ -90,7 +78,7 @@ void main() { }); test('encodes a non-binary broadcast as a text frame', () { - final result = encodeSync(serializer, { + final result = serializer.encode({ 'join_ref': '1', 'ref': '2', 'topic': 'realtime:room', @@ -105,8 +93,7 @@ void main() { group('decode text frames', () { test('decodes a positional JSON array into a message map', () { - final result = decodeSync( - serializer, + final result = serializer.decode( jsonEncode([ '1', '2', @@ -136,7 +123,7 @@ void main() { metadata: jsonEncode({'replayed': true}), ); - final result = decodeSync(serializer, frame) as Map; + final result = serializer.decode(frame); expect(result['join_ref'], isNull); expect(result['ref'], isNull); @@ -159,7 +146,7 @@ void main() { encoding: Serializer.binaryEncoding, ); - final result = decodeSync(serializer, frame) as Map; + final result = serializer.decode(frame); final payload = result['payload'] as Map; expect(payload['event'], 'file'); @@ -168,7 +155,7 @@ void main() { }); test('returns an empty map for unknown binary kinds', () { - final result = decodeSync(serializer, Uint8List.fromList([99, 0, 0])); + final result = serializer.decode(Uint8List.fromList([99, 0, 0])); expect(result, {}); }); }); @@ -176,7 +163,7 @@ void main() { group('encode binary broadcast push', () { test('encodes a broadcast with a binary payload as a binary frame', () { final payload = Uint8List.fromList([10, 20, 30]); - final result = encodeSync(serializer, { + final result = serializer.encode({ 'join_ref': '7', 'ref': '8', 'topic': 'realtime:room', @@ -205,7 +192,7 @@ void main() { test('forwards allowed metadata keys', () { final serializerWithMeta = Serializer(allowedMetadataKeys: ['trace_id']); - final result = encodeSync(serializerWithMeta, { + final result = serializerWithMeta.encode({ 'topic': 'realtime:room', 'event': 'broadcast', 'payload': { @@ -236,7 +223,7 @@ void main() { test('throws when a frame field exceeds 255 bytes', () { final longTopic = 'a' * 256; expect( - () => encodeSync(serializer, { + () => serializer.encode({ 'topic': longTopic, 'event': 'broadcast', 'payload': { From 598ee20ce4b8934cda0f5eeea954b34679e3da2d Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 10:38:38 +0200 Subject: [PATCH 03/16] refactor(realtime)!: always use the 2.0.0 serializer Remove the custom encode/decode constructor hook and the RealtimeEncode/ RealtimeDecode typedefs. With the protocol pinned to 2.0.0, a custom codec that does not emit valid frames would only break the connection, and the hook was unused. The client now holds a private Serializer directly. BREAKING CHANGE: the RealtimeClient encode/decode constructor parameters, the encode/decode fields, and the RealtimeEncode/RealtimeDecode typedefs are removed. --- .../lib/src/realtime_client.dart | 26 +++---------------- 1 file changed, 3 insertions(+), 23 deletions(-) diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index 7c67df5f6..64e348cfa 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -18,14 +18,6 @@ typedef WebSocketTransport = WebSocketChannel Function( Map headers, ); -/// Encodes an outgoing message into the `String` or binary frame to write to -/// the WebSocket. -typedef RealtimeEncode = Object Function(Map payload); - -/// Decodes a raw incoming WebSocket frame (`String` or binary) into a message -/// map. -typedef RealtimeDecode = Map Function(Object payload); - /// Event details for when the connection closed. class RealtimeCloseEvent { /// Web socket protocol status codes for when a connection is closed. @@ -110,8 +102,7 @@ class RealtimeClient { int ref = 0; late RetryTimer reconnectTimer; void Function(String? kind, String? msg, dynamic data)? logger; - late RealtimeEncode encode; - late RealtimeDecode decode; + final Serializer _serializer = Serializer(); late TimerCalculation reconnectAfterMs; WebSocketChannel? connection; List sendBuffer = []; @@ -144,12 +135,6 @@ class RealtimeClient { /// /// [logger] The optional function for specialized logging, ie: logger: (kind, msg, data) => { console.log(`$kind: $msg`, data) } /// - /// [encode] The function to encode outgoing messages. Defaults to the - /// protocol `2.0.0` serializer. - /// - /// [decode] The function to decode incoming messages. Defaults to the - /// protocol `2.0.0` serializer. - /// /// [reconnectAfterMs] The optional function that returns the millsec reconnect interval. Defaults to stepped backoff off. /// /// [logLevel] Specifies the log level for the connection on the server. @@ -159,8 +144,6 @@ class RealtimeClient { this.timeout = Constants.defaultTimeout, this.heartbeatIntervalMs = Constants.defaultHeartbeatIntervalMs, this.logger, - RealtimeEncode? encode, - RealtimeDecode? decode, TimerCalculation? reconnectAfterMs, Map? headers, this.params = const {}, @@ -187,9 +170,6 @@ class RealtimeClient { this.reconnectAfterMs = reconnectAfterMs ?? RetryTimer.createRetryFunction(); - final serializer = Serializer(); - this.encode = encode ?? serializer.encode; - this.decode = decode ?? serializer.decode; reconnectTimer = RetryTimer( () async { await disconnect(); @@ -392,7 +372,7 @@ class RealtimeClient { /// If the socket is not connected, the message gets enqueued within a local buffer, and sent out when a connection is next established. String? push(Message message) { void callback() { - connection?.sink.add(encode(message.toJson())); + connection?.sink.add(_serializer.encode(message.toJson())); } log('push', '${message.topic} ${message.event} (${message.ref})', @@ -407,7 +387,7 @@ class RealtimeClient { } void onConnectionMessage(Object rawMessage) { - final msg = decode(rawMessage); + final msg = _serializer.decode(rawMessage); final topic = msg['topic'] as String; final event = msg['event'] as String; final payload = msg['payload']; From d4921964c8c6187a6291dae766f59b3f7f4e442f Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 10:48:27 +0200 Subject: [PATCH 04/16] refactor(realtime): rename decoded message variable from msg to message --- .../realtime_client/lib/src/realtime_client.dart | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index 64e348cfa..070981712 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -387,11 +387,11 @@ class RealtimeClient { } void onConnectionMessage(Object rawMessage) { - final msg = _serializer.decode(rawMessage); - final topic = msg['topic'] as String; - final event = msg['event'] as String; - final payload = msg['payload']; - final ref = msg['ref'] as String?; + final message = _serializer.decode(rawMessage); + final topic = message['topic'] as String; + final event = message['event'] as String; + final payload = message['payload']; + final ref = message['ref'] as String?; if (ref != null && ref == pendingHeartbeatRef) { pendingHeartbeatRef = null; } @@ -410,7 +410,7 @@ class RealtimeClient { ), ); for (final callback in stateChangeCallbacks['message']!) { - callback(msg); + callback(message); } } From aa7a78842b40cf39df397ae08a8434285c6fc2c1 Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 11:01:42 +0200 Subject: [PATCH 05/16] feat(realtime): make the protocol version selectable via a version enum Add a RealtimeProtocolVersion enum (v1/v2) and a version constructor parameter on RealtimeClient, defaulting to v2. v1 uses the legacy object-shaped JSON frames; v2 uses the serializer. The enum carries the wire vsn string sent as the connection parameter. Also drop the porting references from the doc comments. --- .../realtime_client/lib/realtime_client.dart | 6 ++- .../realtime_client/lib/src/constants.dart | 15 ++++++- .../lib/src/realtime_channel.dart | 2 +- .../lib/src/realtime_client.dart | 33 +++++++++++++-- .../realtime_client/lib/src/serializer.dart | 3 -- .../realtime_client/test/socket_test.dart | 41 +++++++++++++++++++ 6 files changed, 91 insertions(+), 9 deletions(-) diff --git a/packages/realtime_client/lib/realtime_client.dart b/packages/realtime_client/lib/realtime_client.dart index e36275c18..2682de885 100644 --- a/packages/realtime_client/lib/realtime_client.dart +++ b/packages/realtime_client/lib/realtime_client.dart @@ -2,7 +2,11 @@ library; export 'src/constants.dart' - show RealtimeConstants, RealtimeLogLevel, SocketStates; + show + RealtimeConstants, + RealtimeLogLevel, + RealtimeProtocolVersion, + SocketStates; export 'src/realtime_channel.dart'; export 'src/realtime_client.dart'; export 'src/realtime_presence.dart'; diff --git a/packages/realtime_client/lib/src/constants.dart b/packages/realtime_client/lib/src/constants.dart index 7c7ad6f8c..3b055691f 100644 --- a/packages/realtime_client/lib/src/constants.dart +++ b/packages/realtime_client/lib/src/constants.dart @@ -1,7 +1,6 @@ import 'package:realtime_client/src/version.dart'; class Constants { - static const String vsn = '2.0.0'; static const Duration defaultTimeout = Duration(milliseconds: 10000); static const int defaultHeartbeatIntervalMs = 25000; static const int wsCloseNormal = 1000; @@ -12,6 +11,20 @@ class Constants { typedef RealtimeConstants = Constants; +/// Realtime protocol version, sent as the `vsn` connection parameter. +enum RealtimeProtocolVersion { + /// Legacy protocol: object-shaped JSON text frames only. + v1('1.0.0'), + + /// Positional JSON array text frames plus binary frames. + v2('2.0.0'); + + const RealtimeProtocolVersion(this.vsn); + + /// The value sent as the `vsn` connection parameter. + final String vsn; +} + enum SocketStates { /// Client attempting to establish a connection connecting, diff --git a/packages/realtime_client/lib/src/realtime_channel.dart b/packages/realtime_client/lib/src/realtime_channel.dart index 702a51522..717a6f767 100644 --- a/packages/realtime_client/lib/src/realtime_channel.dart +++ b/packages/realtime_client/lib/src/realtime_channel.dart @@ -597,7 +597,7 @@ class RealtimeChannel { /// /// To send a raw binary payload over a WebSocket binary frame (avoiding JSON /// encoding on the server), set a `Uint8List` (or any `TypedData`) under the - /// `payload` key, mirroring how supabase-js carries binary broadcast data: + /// `payload` key: /// /// ```dart /// channel.sendBroadcastMessage( diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index 070981712..b16570fbd 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:convert'; import 'dart:core'; import 'package:collection/collection.dart'; @@ -86,6 +87,9 @@ class RealtimeClient { final Map headers; final Map params; + + /// The Realtime protocol version sent as the `vsn` connection parameter. + final RealtimeProtocolVersion version; final Duration timeout; final WebSocketTransport transport; final Client? httpClient; @@ -138,6 +142,10 @@ class RealtimeClient { /// [reconnectAfterMs] The optional function that returns the millsec reconnect interval. Defaults to stepped backoff off. /// /// [logLevel] Specifies the log level for the connection on the server. + /// + /// [version] The Realtime protocol version. Defaults to + /// [RealtimeProtocolVersion.v2]; pass [RealtimeProtocolVersion.v1] for the + /// legacy object-shaped JSON frames. RealtimeClient( String endPoint, { WebSocketTransport? transport, @@ -151,6 +159,7 @@ class RealtimeClient { RealtimeLogLevel? logLevel, this.httpClient, this.customAccessToken, + this.version = RealtimeProtocolVersion.v2, }) : endPoint = Uri.parse('$endPoint/${Transports.websocket}') .replace( queryParameters: @@ -372,7 +381,7 @@ class RealtimeClient { /// If the socket is not connected, the message gets enqueued within a local buffer, and sent out when a connection is next established. String? push(Message message) { void callback() { - connection?.sink.add(_serializer.encode(message.toJson())); + connection?.sink.add(_encode(message.toJson())); } log('push', '${message.topic} ${message.event} (${message.ref})', @@ -387,7 +396,7 @@ class RealtimeClient { } void onConnectionMessage(Object rawMessage) { - final message = _serializer.decode(rawMessage); + final message = _decode(rawMessage); final topic = message['topic'] as String; final event = message['event'] as String; final payload = message['payload']; @@ -414,10 +423,28 @@ class RealtimeClient { } } + /// Encodes an outgoing message for the negotiated protocol [version]. + Object _encode(Map message) { + if (version == RealtimeProtocolVersion.v1) { + return jsonEncode(message); + } + return _serializer.encode(message); + } + + /// Decodes a raw incoming frame for the negotiated protocol [version]. + Map _decode(Object rawMessage) { + if (version == RealtimeProtocolVersion.v1) { + return Map.from( + jsonDecode(rawMessage as String) as Map, + ); + } + return _serializer.decode(rawMessage); + } + /// Returns the URL of the websocket. String get endPointURL { final params = Map.from(this.params); - params['vsn'] = Constants.vsn; + params['vsn'] = version.vsn; return _appendParams(endPoint, params); } diff --git a/packages/realtime_client/lib/src/serializer.dart b/packages/realtime_client/lib/src/serializer.dart index 771bf31a8..63f4957ef 100644 --- a/packages/realtime_client/lib/src/serializer.dart +++ b/packages/realtime_client/lib/src/serializer.dart @@ -12,9 +12,6 @@ import 'dart:typed_data'; /// WebSocket frames so that raw bytes can be forwarded without JSON encoding. /// Incoming binary broadcast frames are decoded back into the same map shape as /// their JSON counterparts. -/// -/// Ported from the supabase-js serializer: -/// https://github.com/supabase/supabase-js/blob/master/packages/core/realtime-js/src/lib/serializer.ts class Serializer { static const int headerLength = 1; static const int userBroadcastPushMetaLength = 6; diff --git a/packages/realtime_client/test/socket_test.dart b/packages/realtime_client/test/socket_test.dart index cbf95e3bd..1344827e8 100644 --- a/packages/realtime_client/test/socket_test.dart +++ b/packages/realtime_client/test/socket_test.dart @@ -163,6 +163,17 @@ void main() { 'ws://example.org/chat/websocket?apikey=123456789&vsn=2.0.0', ); }); + + test('uses the legacy vsn when version is v1', () { + final socket = RealtimeClient( + 'wss://example.org/chat', + version: RealtimeProtocolVersion.v1, + ); + expect( + socket.endPointURL, + 'wss://example.org/chat/websocket?vsn=1.0.0', + ); + }); }); group('connect with Websocket', () { @@ -484,6 +495,36 @@ void main() { verify(() => mockedSink.add(captureAny(that: isA()))) .called(1); }); + + test('encodes with the legacy object format when version is v1', () { + final legacyChannel = MockIOWebSocketChannel(); + final legacySink = MockWebSocketSink(); + when(() => legacyChannel.sink).thenReturn(legacySink); + when(() => legacyChannel.ready).thenAnswer((_) => Future.value()); + when(() => legacySink.close()).thenAnswer((_) => Future.value()); + + final legacySocket = RealtimeClient( + socketEndpoint, + transport: (url, headers) => legacyChannel, + version: RealtimeProtocolVersion.v1, + ); + legacySocket.connect(); + legacySocket.connectionStatus = SocketStates.open; + + final legacyData = json.encode({ + 'topic': topic, + 'event': event.eventName(), + 'payload': payload, + 'ref': ref, + }); + + final message = + Message(topic: topic, payload: payload, event: event, ref: ref); + legacySocket.push(message); + + verify(() => legacySink.add(captureAny(that: equals(legacyData)))) + .called(1); + }); }); group('makeRef', () { From 16a4e606534dd717a0234e9ddff5bbf349d75d5b Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 11:03:33 +0200 Subject: [PATCH 06/16] docs(realtime): remove comments referencing the JS SDKs --- packages/realtime_client/lib/src/realtime_client.dart | 2 -- 1 file changed, 2 deletions(-) diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index b16570fbd..842573a06 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -80,7 +80,6 @@ class RealtimeCloseEvent { /// - Works on all Dart platforms (Flutter mobile/desktop, web, server). /// - On web, the underlying [WebSocketChannel] uses the browser WebSocket API. class RealtimeClient { - // This is named `accessTokenValue` in supabase-js String? accessToken; List channels = []; final String endPoint; @@ -120,7 +119,6 @@ class RealtimeClient { @Deprecated("No longer used. Will be removed in the next major version.") int longpollerTimeout = 20000; SocketStates? connectionStatus; - // This is called `accessToken` in realtime-js Future Function()? customAccessToken; /// Initializes the Socket From 27205f8556e4c7515c6987b28bb1e74084509f51 Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 11:07:40 +0200 Subject: [PATCH 07/16] docs(realtime): drop low-value comments --- packages/realtime_client/lib/src/constants.dart | 1 - packages/realtime_client/lib/src/realtime_client.dart | 4 ---- 2 files changed, 5 deletions(-) diff --git a/packages/realtime_client/lib/src/constants.dart b/packages/realtime_client/lib/src/constants.dart index 3b055691f..61b432333 100644 --- a/packages/realtime_client/lib/src/constants.dart +++ b/packages/realtime_client/lib/src/constants.dart @@ -11,7 +11,6 @@ class Constants { typedef RealtimeConstants = Constants; -/// Realtime protocol version, sent as the `vsn` connection parameter. enum RealtimeProtocolVersion { /// Legacy protocol: object-shaped JSON text frames only. v1('1.0.0'), diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index 842573a06..7ce7fe2c7 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -87,7 +87,6 @@ class RealtimeClient { final Map headers; final Map params; - /// The Realtime protocol version sent as the `vsn` connection parameter. final RealtimeProtocolVersion version; final Duration timeout; final WebSocketTransport transport; @@ -229,7 +228,6 @@ class RealtimeClient { _onConnectionOpen(); localConnection.stream.listen( - // incoming messages (text frames are `String`, binary frames are bytes) (message) => onConnectionMessage(message), onError: _onConnectionError, onDone: () { @@ -421,7 +419,6 @@ class RealtimeClient { } } - /// Encodes an outgoing message for the negotiated protocol [version]. Object _encode(Map message) { if (version == RealtimeProtocolVersion.v1) { return jsonEncode(message); @@ -429,7 +426,6 @@ class RealtimeClient { return _serializer.encode(message); } - /// Decodes a raw incoming frame for the negotiated protocol [version]. Map _decode(Object rawMessage) { if (version == RealtimeProtocolVersion.v1) { return Map.from( From 23b01907d720aa1e5a55c1cd695a5e884320b71b Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 11:14:03 +0200 Subject: [PATCH 08/16] feat(realtime): allow overriding the encode/decode codec Re-add optional encode/decode constructor arguments (and the RealtimeEncode/RealtimeDecode typedefs). They default to the codec selected by version, so consumers can swap in a faster serializer. --- .../lib/src/realtime_client.dart | 24 +++++++++++++++++-- .../realtime_client/test/socket_test.dart | 23 ++++++++++++++++++ 2 files changed, 45 insertions(+), 2 deletions(-) diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index 7ce7fe2c7..36bacd3c3 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -19,6 +19,14 @@ typedef WebSocketTransport = WebSocketChannel Function( Map headers, ); +/// Serializes an outgoing message into the `String` or binary frame written to +/// the WebSocket. +typedef RealtimeEncode = Object Function(Map payload); + +/// Deserializes a raw incoming WebSocket frame (`String` or binary) into a +/// message map. +typedef RealtimeDecode = Map Function(Object payload); + /// Event details for when the connection closed. class RealtimeCloseEvent { /// Web socket protocol status codes for when a connection is closed. @@ -105,6 +113,8 @@ class RealtimeClient { late RetryTimer reconnectTimer; void Function(String? kind, String? msg, dynamic data)? logger; final Serializer _serializer = Serializer(); + late final RealtimeEncode encode; + late final RealtimeDecode decode; late TimerCalculation reconnectAfterMs; WebSocketChannel? connection; List sendBuffer = []; @@ -136,6 +146,12 @@ class RealtimeClient { /// /// [logger] The optional function for specialized logging, ie: logger: (kind, msg, data) => { console.log(`$kind: $msg`, data) } /// + /// [encode] Overrides how outgoing messages are serialized, for example to + /// use a faster JSON implementation. Defaults to the codec for [version]. + /// + /// [decode] Overrides how incoming frames are deserialized. Defaults to the + /// codec for [version]. + /// /// [reconnectAfterMs] The optional function that returns the millsec reconnect interval. Defaults to stepped backoff off. /// /// [logLevel] Specifies the log level for the connection on the server. @@ -149,6 +165,8 @@ class RealtimeClient { this.timeout = Constants.defaultTimeout, this.heartbeatIntervalMs = Constants.defaultHeartbeatIntervalMs, this.logger, + RealtimeEncode? encode, + RealtimeDecode? decode, TimerCalculation? reconnectAfterMs, Map? headers, this.params = const {}, @@ -176,6 +194,8 @@ class RealtimeClient { this.reconnectAfterMs = reconnectAfterMs ?? RetryTimer.createRetryFunction(); + this.encode = encode ?? _encode; + this.decode = decode ?? _decode; reconnectTimer = RetryTimer( () async { await disconnect(); @@ -377,7 +397,7 @@ class RealtimeClient { /// If the socket is not connected, the message gets enqueued within a local buffer, and sent out when a connection is next established. String? push(Message message) { void callback() { - connection?.sink.add(_encode(message.toJson())); + connection?.sink.add(encode(message.toJson())); } log('push', '${message.topic} ${message.event} (${message.ref})', @@ -392,7 +412,7 @@ class RealtimeClient { } void onConnectionMessage(Object rawMessage) { - final message = _decode(rawMessage); + final message = decode(rawMessage); final topic = message['topic'] as String; final event = message['event'] as String; final payload = message['payload']; diff --git a/packages/realtime_client/test/socket_test.dart b/packages/realtime_client/test/socket_test.dart index 1344827e8..e5190b1dc 100644 --- a/packages/realtime_client/test/socket_test.dart +++ b/packages/realtime_client/test/socket_test.dart @@ -525,6 +525,29 @@ void main() { verify(() => legacySink.add(captureAny(that: equals(legacyData)))) .called(1); }); + + test('uses a custom encode override when provided', () { + final customChannel = MockIOWebSocketChannel(); + final customSink = MockWebSocketSink(); + when(() => customChannel.sink).thenReturn(customSink); + when(() => customChannel.ready).thenAnswer((_) => Future.value()); + when(() => customSink.close()).thenAnswer((_) => Future.value()); + + final customSocket = RealtimeClient( + socketEndpoint, + transport: (url, headers) => customChannel, + encode: (payload) => 'custom-frame', + ); + customSocket.connect(); + customSocket.connectionStatus = SocketStates.open; + + customSocket.push( + Message(topic: topic, payload: payload, event: event, ref: ref), + ); + + verify(() => customSink.add(captureAny(that: equals('custom-frame')))) + .called(1); + }); }); group('makeRef', () { From 930104e9c899d1921037c370040d64be3ed577d4 Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 11:58:16 +0200 Subject: [PATCH 09/16] fix(realtime): harden frame handling from review feedback - decode() validates the 2.0.0 text frame shape and throws a clear FormatException instead of an obscure cast error - onConnectionMessage drops malformed frames instead of crashing the connection, and guards the status log against non-map payloads - set the web binaryType explicitly to arraybuffer/list - document the ASCII assumption for binary frame string fields --- .../realtime_client/lib/src/realtime_client.dart | 12 ++++++++++-- packages/realtime_client/lib/src/serializer.dart | 9 ++++++++- .../lib/src/websocket/websocket_web.dart | 4 +++- packages/realtime_client/test/serializer_test.dart | 11 +++++++++++ packages/realtime_client/test/socket_test.dart | 10 ++++++++++ 5 files changed, 42 insertions(+), 4 deletions(-) diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index 36bacd3c3..5030bee38 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -412,7 +412,14 @@ class RealtimeClient { } void onConnectionMessage(Object rawMessage) { - final message = decode(rawMessage); + final Map message; + try { + message = decode(rawMessage); + } catch (error) { + log('transport', 'failed to decode message', error); + return; + } + final topic = message['topic'] as String; final event = message['event'] as String; final payload = message['payload']; @@ -421,9 +428,10 @@ class RealtimeClient { pendingHeartbeatRef = null; } + final status = payload is Map ? (payload['status'] ?? '') : ''; log( 'receive', - "${payload['status'] ?? ''} $topic $event ${ref != null ? '($ref)' : ''}", + "$status $topic $event ${ref != null ? '($ref)' : ''}", payload, ); diff --git a/packages/realtime_client/lib/src/serializer.dart b/packages/realtime_client/lib/src/serializer.dart index 63f4957ef..1376d287b 100644 --- a/packages/realtime_client/lib/src/serializer.dart +++ b/packages/realtime_client/lib/src/serializer.dart @@ -60,7 +60,10 @@ class Serializer { /// `topic`, `event` and `payload` keys. Map decode(Object rawPayload) { if (rawPayload is String) { - final decoded = jsonDecode(rawPayload) as List; + final decoded = jsonDecode(rawPayload); + if (decoded is! List || decoded.length < 5) { + throw FormatException('Invalid 2.0.0 text frame', rawPayload); + } return { 'join_ref': decoded[0], 'ref': decoded[1], @@ -184,6 +187,10 @@ class Serializer { } } + // Writes one byte per UTF-16 code unit, matching the size byte (which is the + // code-unit count). Frame string fields (joinRef, ref, topic, userEvent, + // metadata) are therefore assumed to be ASCII; non-ASCII characters would be + // truncated and not round-trip through the utf8 decode on the other side. int _writeString(Uint8List buffer, int offset, String value) { for (final unit in value.codeUnits) { buffer[offset++] = unit & 0xFF; diff --git a/packages/realtime_client/lib/src/websocket/websocket_web.dart b/packages/realtime_client/lib/src/websocket/websocket_web.dart index e83174ece..ab78d5586 100644 --- a/packages/realtime_client/lib/src/websocket/websocket_web.dart +++ b/packages/realtime_client/lib/src/websocket/websocket_web.dart @@ -5,5 +5,7 @@ WebSocketChannel createWebSocketClient( String url, Map headers, ) { - return HtmlWebSocketChannel.connect(url); + // Deliver binary frames as `Uint8List` (arraybuffer) so protocol 2.0.0 + // binary frames can be decoded synchronously. + return HtmlWebSocketChannel.connect(url, binaryType: BinaryType.list); } diff --git a/packages/realtime_client/test/serializer_test.dart b/packages/realtime_client/test/serializer_test.dart index 60e6993af..1bc16655f 100644 --- a/packages/realtime_client/test/serializer_test.dart +++ b/packages/realtime_client/test/serializer_test.dart @@ -111,6 +111,17 @@ void main() { 'payload': {'status': 'ok'}, }); }); + + test('throws a FormatException on a malformed text frame', () { + expect( + () => serializer.decode('{"not": "an array"}'), + throwsFormatException, + ); + expect( + () => serializer.decode(jsonEncode(['too', 'short'])), + throwsFormatException, + ); + }); }); group('decode binary frames', () { diff --git a/packages/realtime_client/test/socket_test.dart b/packages/realtime_client/test/socket_test.dart index e5190b1dc..2f5cb9394 100644 --- a/packages/realtime_client/test/socket_test.dart +++ b/packages/realtime_client/test/socket_test.dart @@ -550,6 +550,16 @@ void main() { }); }); + group('onConnectionMessage', () { + test('drops a malformed frame without throwing', () { + final socket = RealtimeClient(socketEndpoint); + expect( + () => socket.onConnectionMessage('{"not": "an array"}'), + returnsNormally, + ); + }); + }); + group('makeRef', () { late RealtimeClient socket; setUp(() { From a32b4b0ab502c17cfa3153915076b9546349fa28 Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 13:21:42 +0200 Subject: [PATCH 10/16] refactor(realtime): use sublistView in binary decode and a named offset constant Avoid copying frame slices during binary decode, name the userBroadcast header offset, and add an end-to-end test for dispatching a received binary broadcast to onBroadcast. --- .../realtime_client/lib/src/serializer.dart | 18 ++++++---- .../realtime_client/test/socket_test.dart | 33 +++++++++++++++++++ 2 files changed, 45 insertions(+), 6 deletions(-) diff --git a/packages/realtime_client/lib/src/serializer.dart b/packages/realtime_client/lib/src/serializer.dart index 1376d287b..81d6e12fa 100644 --- a/packages/realtime_client/lib/src/serializer.dart +++ b/packages/realtime_client/lib/src/serializer.dart @@ -16,6 +16,10 @@ class Serializer { static const int headerLength = 1; static const int userBroadcastPushMetaLength = 6; + /// Size bytes after the kind byte of a `userBroadcast` frame: + /// topic size, user event size, metadata size and payload encoding. + static const int userBroadcastMetaLength = 4; + /// Binary frame sent by the client for a broadcast push. static const int kindUserBroadcastPush = 3; @@ -147,18 +151,20 @@ class Serializer { final metadataSize = view.getUint8(3); final payloadEncoding = view.getUint8(4); - var offset = headerLength + 4; - final topic = utf8.decode(buffer.sublist(offset, offset + topicSize)); + var offset = headerLength + userBroadcastMetaLength; + final topic = + utf8.decode(Uint8List.sublistView(buffer, offset, offset + topicSize)); offset += topicSize; - final userEvent = - utf8.decode(buffer.sublist(offset, offset + userEventSize)); + final userEvent = utf8 + .decode(Uint8List.sublistView(buffer, offset, offset + userEventSize)); offset += userEventSize; final metadata = metadataSize > 0 - ? utf8.decode(buffer.sublist(offset, offset + metadataSize)) + ? utf8.decode( + Uint8List.sublistView(buffer, offset, offset + metadataSize)) : ''; offset += metadataSize; - final payloadBytes = buffer.sublist(offset); + final payloadBytes = Uint8List.sublistView(buffer, offset); final dynamic parsedPayload = payloadEncoding == jsonEncoding ? jsonDecode(utf8.decode(payloadBytes)) : payloadBytes; diff --git a/packages/realtime_client/test/socket_test.dart b/packages/realtime_client/test/socket_test.dart index 2f5cb9394..25024a33a 100644 --- a/packages/realtime_client/test/socket_test.dart +++ b/packages/realtime_client/test/socket_test.dart @@ -558,6 +558,39 @@ void main() { returnsNormally, ); }); + + test('dispatches a received binary broadcast to onBroadcast', () { + final socket = RealtimeClient(socketEndpoint); + final channel = socket.channel('room'); + + Map? received; + channel.onBroadcast( + event: 'cursor', + callback: (payload) => received = payload, + ); + + final topic = utf8.encode('realtime:room'); + final event = utf8.encode('cursor'); + final payload = utf8.encode(json.encode({'x': 1})); + final frame = Uint8List.fromList([ + 4, // kind: userBroadcast + topic.length, + event.length, + 0, // metadata size + 1, // payload encoding: json + ...topic, + ...event, + ...payload, + ]); + + socket.onConnectionMessage(frame); + + expect(received, { + 'type': 'broadcast', + 'event': 'cursor', + 'payload': {'x': 1}, + }); + }); }); group('makeRef', () { From cdee562b29bde2918e94ea6fdacccd58b6e1ec10 Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 13:21:42 +0200 Subject: [PATCH 11/16] fix(supabase_flutter): use renamed connectionStatus getter --- packages/supabase_flutter/lib/src/supabase.dart | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/supabase_flutter/lib/src/supabase.dart b/packages/supabase_flutter/lib/src/supabase.dart index c2ce4e3c3..d2836b157 100644 --- a/packages/supabase_flutter/lib/src/supabase.dart +++ b/packages/supabase_flutter/lib/src/supabase.dart @@ -299,7 +299,7 @@ class Supabase { } else { // paused or detached — disconnect the WebSocket if it is active. if (realtime.isConnected || - realtime.connState == SocketStates.connecting) { + realtime.connectionStatus == SocketStates.connecting) { await realtime.disconnect(); } } From 0f50101583d2044ce008b699f698ff64f7e47dee Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 13:43:27 +0200 Subject: [PATCH 12/16] perf(realtime): build binary frames in a single buffer Write the header directly into the final frame buffer instead of allocating a separate header and copying it, removing one allocation and one copy per binary broadcast push. --- .../realtime_client/lib/src/serializer.dart | 34 +++++++++---------- 1 file changed, 16 insertions(+), 18 deletions(-) diff --git a/packages/realtime_client/lib/src/serializer.dart b/packages/realtime_client/lib/src/serializer.dart index 81d6e12fa..a8d8f6281 100644 --- a/packages/realtime_client/lib/src/serializer.dart +++ b/packages/realtime_client/lib/src/serializer.dart @@ -113,25 +113,23 @@ class Serializer { userEvent.length + metadata.length; - final header = Uint8List(headerLength + metaLength); + final frame = Uint8List(headerLength + metaLength + encodedPayload.length); var offset = 0; - header[offset++] = kindUserBroadcastPush; - header[offset++] = joinRef.length; - header[offset++] = ref.length; - header[offset++] = topic.length; - header[offset++] = userEvent.length; - header[offset++] = metadata.length; - header[offset++] = binaryEncoding; - offset = _writeString(header, offset, joinRef); - offset = _writeString(header, offset, ref); - offset = _writeString(header, offset, topic); - offset = _writeString(header, offset, userEvent); - offset = _writeString(header, offset, metadata); - - final combined = Uint8List(header.length + encodedPayload.length); - combined.setAll(0, header); - combined.setAll(header.length, encodedPayload); - return combined; + frame[offset++] = kindUserBroadcastPush; + frame[offset++] = joinRef.length; + frame[offset++] = ref.length; + frame[offset++] = topic.length; + frame[offset++] = userEvent.length; + frame[offset++] = metadata.length; + frame[offset++] = binaryEncoding; + offset = _writeString(frame, offset, joinRef); + offset = _writeString(frame, offset, ref); + offset = _writeString(frame, offset, topic); + offset = _writeString(frame, offset, userEvent); + offset = _writeString(frame, offset, metadata); + + frame.setAll(offset, encodedPayload); + return frame; } Map _binaryDecode(Uint8List buffer) { From 06f375d510975778e0c3f60262b28c18a3b161d0 Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 14:07:48 +0200 Subject: [PATCH 13/16] refactor(realtime): resolve encode/decode in the initializer list Make encode/decode plain final fields resolved in the initializer list instead of late, using a shared static serializer and static legacy codec helpers. Removes the late modifier and a per-client serializer allocation. --- .../lib/src/realtime_client.dart | 36 +++++++++---------- 1 file changed, 16 insertions(+), 20 deletions(-) diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index 5030bee38..ab6ca12c9 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -112,9 +112,9 @@ class RealtimeClient { int ref = 0; late RetryTimer reconnectTimer; void Function(String? kind, String? msg, dynamic data)? logger; - final Serializer _serializer = Serializer(); - late final RealtimeEncode encode; - late final RealtimeDecode decode; + static final Serializer _serializer = Serializer(); + final RealtimeEncode encode; + final RealtimeDecode decode; late TimerCalculation reconnectAfterMs; WebSocketChannel? connection; List sendBuffer = []; @@ -185,7 +185,15 @@ class RealtimeClient { ...Constants.defaultHeaders, if (headers != null) ...headers, }, - transport = transport ?? createWebSocketClient { + transport = transport ?? createWebSocketClient, + encode = encode ?? + (version == RealtimeProtocolVersion.v1 + ? _encodeLegacy + : _serializer.encode), + decode = decode ?? + (version == RealtimeProtocolVersion.v1 + ? _decodeLegacy + : _serializer.decode) { _log.config( 'Initialize RealtimeClient with endpoint: $endPoint, timeout: $timeout, heartbeatIntervalMs: $heartbeatIntervalMs, logLevel: $logLevel'); _log.finest('Initialize with headers: $headers, params: $params'); @@ -194,8 +202,6 @@ class RealtimeClient { this.reconnectAfterMs = reconnectAfterMs ?? RetryTimer.createRetryFunction(); - this.encode = encode ?? _encode; - this.decode = decode ?? _decode; reconnectTimer = RetryTimer( () async { await disconnect(); @@ -447,21 +453,11 @@ class RealtimeClient { } } - Object _encode(Map message) { - if (version == RealtimeProtocolVersion.v1) { - return jsonEncode(message); - } - return _serializer.encode(message); - } + static Object _encodeLegacy(Map message) => + jsonEncode(message); - Map _decode(Object rawMessage) { - if (version == RealtimeProtocolVersion.v1) { - return Map.from( - jsonDecode(rawMessage as String) as Map, - ); - } - return _serializer.decode(rawMessage); - } + static Map _decodeLegacy(Object rawMessage) => + Map.from(jsonDecode(rawMessage as String) as Map); /// Returns the URL of the websocket. String get endPointURL { From b9f9fc4e36847f9dd2e56a4c290e53221f0091ec Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 14:20:56 +0200 Subject: [PATCH 14/16] refactor(realtime): expand abbreviated identifiers Rename obj -> source in the serializer, msg -> message in the logger and log(), and res -> response in the channel HTTP helpers. --- .../realtime_client/lib/src/realtime_channel.dart | 12 ++++++------ .../realtime_client/lib/src/realtime_client.dart | 13 ++++++++----- packages/realtime_client/lib/src/serializer.dart | 4 ++-- 3 files changed, 16 insertions(+), 13 deletions(-) diff --git a/packages/realtime_client/lib/src/realtime_channel.dart b/packages/realtime_client/lib/src/realtime_channel.dart index 717a6f767..ada8240b8 100644 --- a/packages/realtime_client/lib/src/realtime_channel.dart +++ b/packages/realtime_client/lib/src/realtime_channel.dart @@ -565,7 +565,7 @@ class RealtimeChannel { ] }; - final res = await (socket.httpClient?.post ?? post)( + final response = await (socket.httpClient?.post ?? post)( Uri.parse(broadcastEndpointURL), headers: headers, body: json.encode(body), @@ -574,13 +574,13 @@ class RealtimeChannel { onTimeout: () => throw TimeoutException('Request timeout'), ); - if (res.statusCode == 202) { + if (response.statusCode == 202) { return; } - String errorMessage = res.reasonPhrase ?? 'Unknown error'; + String errorMessage = response.reasonPhrase ?? 'Unknown error'; try { - final errorBody = json.decode(res.body) as Map; + final errorBody = json.decode(response.body) as Map; errorMessage = (errorBody['error'] ?? errorBody['message'] ?? errorMessage) as String; @@ -656,12 +656,12 @@ class RealtimeChannel { ] }; try { - final res = await (socket.httpClient?.post ?? post)( + final response = await (socket.httpClient?.post ?? post)( Uri.parse(broadcastEndpointURL), headers: headers, body: json.encode(body), ); - if (200 <= res.statusCode && res.statusCode < 300) { + if (200 <= response.statusCode && response.statusCode < 300) { completer.complete(ChannelResponse.ok); } else { completer.complete(ChannelResponse.error); diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index ab6ca12c9..7a0c7bf12 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -111,7 +111,7 @@ class RealtimeClient { /// Unique reference ID for every heartbeat. int ref = 0; late RetryTimer reconnectTimer; - void Function(String? kind, String? msg, dynamic data)? logger; + void Function(String? kind, String? message, dynamic data)? logger; static final Serializer _serializer = Serializer(); final RealtimeEncode encode; final RealtimeDecode decode; @@ -144,7 +144,7 @@ class RealtimeClient { /// /// [heartbeatIntervalMs] The millisec interval to send a heartbeat message. /// - /// [logger] The optional function for specialized logging, ie: logger: (kind, msg, data) => { console.log(`$kind: $msg`, data) } + /// [logger] The optional function for specialized logging, ie: logger: (kind, message, data) => { console.log(`$kind: $message`, data) } /// /// [encode] Overrides how outgoing messages are serialized, for example to /// use a faster JSON implementation. Defaults to the codec for [version]. @@ -331,9 +331,12 @@ class RealtimeClient { /// /// [level] must be [Level.FINEST] for senitive data void log( - [String? kind, String? msg, dynamic data, Level level = Level.FINEST]) { - _log.log(level, '$kind: $msg', data); - logger?.call(kind, msg, data); + [String? kind, + String? message, + dynamic data, + Level level = Level.FINEST]) { + _log.log(level, '$kind: $message', data); + logger?.call(kind, message, data); } /// Registers callbacks for connection state change events diff --git a/packages/realtime_client/lib/src/serializer.dart b/packages/realtime_client/lib/src/serializer.dart index a8d8f6281..eae36ae34 100644 --- a/packages/realtime_client/lib/src/serializer.dart +++ b/packages/realtime_client/lib/src/serializer.dart @@ -222,10 +222,10 @@ class Serializer { return null; } - Map _pick(Map obj, List keys) { + Map _pick(Map source, List keys) { return { for (final key in keys) - if (obj.containsKey(key)) key: obj[key], + if (source.containsKey(key)) key: source[key], }; } } From 3482a48a0c133ec10d0d1fc47bd34d8827cb68f5 Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 14:44:01 +0200 Subject: [PATCH 15/16] test(realtime): cover v1 legacy decode and dispatch The mock_test suites now exercise v2; add a v1 decode + dispatch test so the legacy object-frame path keeps coverage alongside the existing v1 endpointURL and encode tests. --- .../realtime_client/test/socket_test.dart | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/packages/realtime_client/test/socket_test.dart b/packages/realtime_client/test/socket_test.dart index 25024a33a..16fd25fa5 100644 --- a/packages/realtime_client/test/socket_test.dart +++ b/packages/realtime_client/test/socket_test.dart @@ -591,6 +591,38 @@ void main() { 'payload': {'x': 1}, }); }); + + test('decodes a legacy object frame and dispatches it when version is v1', + () { + final socket = RealtimeClient( + socketEndpoint, + version: RealtimeProtocolVersion.v1, + ); + final channel = socket.channel('room'); + + Map? received; + channel.onBroadcast( + event: 'cursor', + callback: (payload) => received = payload, + ); + + socket.onConnectionMessage(json.encode({ + 'topic': 'realtime:room', + 'event': 'broadcast', + 'payload': { + 'type': 'broadcast', + 'event': 'cursor', + 'payload': {'x': 1}, + }, + 'ref': null, + })); + + expect(received, { + 'type': 'broadcast', + 'event': 'cursor', + 'payload': {'x': 1}, + }); + }); }); group('makeRef', () { From fc96b81c2666dfcf312fb50d037534a63f23de9b Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 12 Jun 2026 16:22:02 +0200 Subject: [PATCH 16/16] refactor(realtime): defer conn/connState/onConnMessage renames to v3 Keep the original conn/connState fields, the onConnMessage method, and the _onConn* handlers in this PR so it carries only the protocol 2.0.0 feature and RealtimeProtocolVersion. The breaking renames move to a follow-up PR targeting v3. --- .../lib/src/realtime_client.dart | 90 +++++++++---------- .../realtime_client/test/socket_test.dart | 76 ++++++++-------- .../supabase_flutter/lib/src/supabase.dart | 2 +- .../supabase_flutter/test/lifecycle_test.dart | 24 ++--- 4 files changed, 95 insertions(+), 97 deletions(-) diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart index 7a0c7bf12..a343f2670 100644 --- a/packages/realtime_client/lib/src/realtime_client.dart +++ b/packages/realtime_client/lib/src/realtime_client.dart @@ -116,7 +116,7 @@ class RealtimeClient { final RealtimeEncode encode; final RealtimeDecode decode; late TimerCalculation reconnectAfterMs; - WebSocketChannel? connection; + WebSocketChannel? conn; List sendBuffer = []; Map> stateChangeCallbacks = { 'open': [], @@ -127,7 +127,7 @@ class RealtimeClient { @Deprecated("No longer used. Will be removed in the next major version.") int longpollerTimeout = 20000; - SocketStates? connectionStatus; + SocketStates? connState; Future Function()? customAccessToken; /// Initializes the Socket @@ -214,73 +214,72 @@ class RealtimeClient { /// Connects the socket. @internal Future connect() async { - if (connection != null) { + if (conn != null) { return; } try { log('transport', 'connecting to $endPointURL', null); log('transport', 'connecting', null, Level.FINE); - connectionStatus = SocketStates.connecting; - final WebSocketChannel localConnection = transport(endPointURL, headers); - connection = localConnection; + connState = SocketStates.connecting; + final WebSocketChannel localConn = transport(endPointURL, headers); + conn = localConn; try { - await localConnection.ready; + await localConn.ready; } catch (error) { // Bail out if disconnect() ran or a new connect() started during await - if (connection != localConnection) { + if (conn != localConn) { return; } // Don't schedule a reconnect and emit error if connection has been // closed by the user or [disconnect] waits for the connection to be // ready before closing it. - if (connectionStatus != SocketStates.disconnected && - connectionStatus != SocketStates.disconnecting) { - connectionStatus = SocketStates.closed; - _onConnectionError(error); + if (connState != SocketStates.disconnected && + connState != SocketStates.disconnecting) { + connState = SocketStates.closed; + _onConnError(error); reconnectTimer.scheduleTimeout(); } return; } // Guard: bail out if disconnect() ran during the await - if (connection != localConnection || - connectionStatus != SocketStates.connecting) { + if (conn != localConn || connState != SocketStates.connecting) { return; } - connectionStatus = SocketStates.open; + connState = SocketStates.open; - _onConnectionOpen(); - localConnection.stream.listen( - (message) => onConnectionMessage(message), - onError: _onConnectionError, + _onConnOpen(); + localConn.stream.listen( + (message) => onConnMessage(message), + onError: _onConnError, onDone: () { // communication has been closed - if (connectionStatus != SocketStates.disconnected && - connectionStatus != SocketStates.disconnecting) { - connectionStatus = SocketStates.closed; + if (connState != SocketStates.disconnected && + connState != SocketStates.disconnecting) { + connState = SocketStates.closed; } - _onConnectionClose(); + _onConnClose(); }, ); } catch (e) { /// General error handling - _onConnectionError(e); + _onConnError(e); } } /// Disconnects the socket with status [code] and [reason] for the disconnect Future disconnect({int? code, String? reason}) async { - final connection = this.connection; - if (connection != null) { - final oldState = connectionStatus; + final conn = this.conn; + if (conn != null) { + final oldState = connState; final shouldCloseSink = oldState == SocketStates.open || oldState == SocketStates.connecting; if (shouldCloseSink) { // Don't set the state to `disconnecting` if the connection is already closed. - connectionStatus = SocketStates.disconnecting; + connState = SocketStates.disconnecting; log('transport', 'disconnecting', {'code': code, 'reason': reason}, Level.FINE); } @@ -288,20 +287,20 @@ class RealtimeClient { // Connection cannot be closed while it's still connecting. Wait for connection to // be ready and then close it. if (oldState == SocketStates.connecting) { - await connection.ready.catchError((_) {}); + await conn.ready.catchError((_) {}); } if (shouldCloseSink) { if (code != null) { - await connection.sink.close(code, reason ?? ''); + await conn.sink.close(code, reason ?? ''); } else { - await connection.sink.close(); + await conn.sink.close(); } - connectionStatus = SocketStates.disconnected; + connState = SocketStates.disconnected; reconnectTimer.reset(); log('transport', 'disconnected', null, Level.FINE); } - this.connection = null; + this.conn = null; // remove open handles if (heartbeatTimer != null) heartbeatTimer?.cancel(); @@ -365,7 +364,7 @@ class RealtimeClient { /// Returns the current state of the socket. String get connectionState { - switch (connectionStatus) { + switch (connState) { case SocketStates.connecting: return 'connecting'; case SocketStates.open: @@ -381,7 +380,7 @@ class RealtimeClient { } /// Returns `true` is the connection is open. - bool get isConnected => connectionStatus == SocketStates.open; + bool get isConnected => connState == SocketStates.open; /// Removes a subscription from the socket. @internal @@ -406,7 +405,7 @@ class RealtimeClient { /// If the socket is not connected, the message gets enqueued within a local buffer, and sent out when a connection is next established. String? push(Message message) { void callback() { - connection?.sink.add(encode(message.toJson())); + conn?.sink.add(encode(message.toJson())); } log('push', '${message.topic} ${message.event} (${message.ref})', @@ -420,7 +419,7 @@ class RealtimeClient { return null; } - void onConnectionMessage(Object rawMessage) { + void onConnMessage(Object rawMessage) { final Map message; try { message = decode(rawMessage); @@ -518,7 +517,7 @@ class RealtimeClient { } } - void _onConnectionOpen() { + void _onConnOpen() { log('transport', 'connected to $endPointURL'); log('transport', 'connected', null, Level.FINE); _flushSendBuffer(); @@ -534,18 +533,17 @@ class RealtimeClient { } /// communication has been closed - void _onConnectionClose() { - final statusCode = connection?.closeCode; + void _onConnClose() { + final statusCode = conn?.closeCode; RealtimeCloseEvent? event; if (statusCode != null) { - event = - RealtimeCloseEvent(code: statusCode, reason: connection?.closeReason); + event = RealtimeCloseEvent(code: statusCode, reason: conn?.closeReason); } log('transport', 'close', event, Level.FINE); /// SocketStates.disconnected: by user with socket.disconnect() /// SocketStates.closed: NOT by user, should try to reconnect - if (connectionStatus == SocketStates.closed) { + if (connState == SocketStates.closed) { _triggerChanError(event); reconnectTimer.scheduleTimeout(); } @@ -555,7 +553,7 @@ class RealtimeClient { } } - void _onConnectionError(dynamic error) { + void _onConnError(dynamic error) { log('transport', error.toString()); _triggerChanError(error); for (final callback in stateChangeCallbacks['error']!) { @@ -603,9 +601,9 @@ class RealtimeClient { pendingHeartbeatRef = null; log( 'transport', - 'heartbeat timeout. Attempting to re-establish connection', + 'heartbeat timeout. Attempting to re-establish conn', ); - connection?.sink.close(Constants.wsCloseNormal, 'heartbeat timeout'); + conn?.sink.close(Constants.wsCloseNormal, 'heartbeat timeout'); return; } pendingHeartbeatRef = makeRef(); diff --git a/packages/realtime_client/test/socket_test.dart b/packages/realtime_client/test/socket_test.dart index 16fd25fa5..bca2d1a35 100644 --- a/packages/realtime_client/test/socket_test.dart +++ b/packages/realtime_client/test/socket_test.dart @@ -189,14 +189,14 @@ void main() { test('establishes websocket connection with endpoint', () async { final connFuture = socket.connect(); - expect(socket.connectionStatus, SocketStates.connecting); + expect(socket.connState, SocketStates.connecting); - final connection = socket.connection; + final conn = socket.conn; await connFuture; - expect(socket.connectionStatus, SocketStates.open); + expect(socket.connState, SocketStates.open); - expect(connection, isA()); + expect(conn, isA()); //! Not verifying connection url }); @@ -242,9 +242,9 @@ void main() { test('is idempotent', () { socket.connect(); - final connection = socket.connection; + final conn = socket.conn; socket.connect(); - expect(socket.connection, connection); + expect(socket.conn, conn); }); }); @@ -261,10 +261,10 @@ void main() { test('removes existing connection', () async { await socket.connect(); - expect(socket.connection, isNotNull); + expect(socket.conn, isNotNull); await socket.disconnect(); - expect(socket.connection, isNull); + expect(socket.conn, isNull); }); test('calls callback', () async { @@ -294,7 +294,7 @@ void main() { const tReason = 'reason'; mockedSocket.connect(); - mockedSocket.connectionStatus = SocketStates.open; + mockedSocket.connState = SocketStates.open; await Future.delayed(const Duration(milliseconds: 200)); mockedSocket.disconnect(code: tCode, reason: tReason); await Future.delayed(const Duration(milliseconds: 200)); @@ -309,32 +309,32 @@ void main() { test('disconnecting a closed connections stays closed', () async { await socket.connect(); - expect(socket.connectionStatus, SocketStates.open); + expect(socket.connState, SocketStates.open); await mockServer.close(); await Future.delayed(const Duration(milliseconds: 200)); - expect(socket.connectionStatus, SocketStates.closed); - expect(socket.connection, isNotNull); + expect(socket.connState, SocketStates.closed); + expect(socket.conn, isNotNull); final disconnectFuture = socket.disconnect(); - // `connectionStatus` stays `closed` during disconnect - expect(socket.connectionStatus, SocketStates.closed); + // `connState` stays `closed` during disconnect + expect(socket.connState, SocketStates.closed); await disconnectFuture; - expect(socket.connectionStatus, SocketStates.closed); - expect(socket.connection, isNull); + expect(socket.connState, SocketStates.closed); + expect(socket.conn, isNull); }); test('disconnecting an open connection', () async { await socket.connect(); - expect(socket.connectionStatus, SocketStates.open); + expect(socket.connState, SocketStates.open); final disconnectFuture = socket.disconnect(); - // `connectionStatus` stays `closed` during disconnect - expect(socket.connectionStatus, SocketStates.disconnecting); + // `connState` stays `closed` during disconnect + expect(socket.connState, SocketStates.disconnecting); await disconnectFuture; - expect(socket.connectionStatus, SocketStates.disconnected); - expect(socket.connection, isNull); + expect(socket.connState, SocketStates.disconnected); + expect(socket.conn, isNull); }); test('does not throw when no connection', () { @@ -447,7 +447,7 @@ void main() { test('sends data to connection when connected', () { mockedSocket.connect(); - mockedSocket.connectionStatus = SocketStates.open; + mockedSocket.connState = SocketStates.open; final message = Message(topic: topic, payload: payload, event: event, ref: ref); @@ -459,7 +459,7 @@ void main() { test('buffers data when not connected', () async { mockedSocket.connect(); - mockedSocket.connectionStatus = SocketStates.connecting; + mockedSocket.connState = SocketStates.connecting; expect(mockedSocket.sendBuffer.length, 0); @@ -478,7 +478,7 @@ void main() { test('sends a broadcast with a binary payload as a binary frame', () { mockedSocket.connect(); - mockedSocket.connectionStatus = SocketStates.open; + mockedSocket.connState = SocketStates.open; final binaryPayload = Uint8List.fromList([1, 2, 3]); final message = Message( @@ -509,7 +509,7 @@ void main() { version: RealtimeProtocolVersion.v1, ); legacySocket.connect(); - legacySocket.connectionStatus = SocketStates.open; + legacySocket.connState = SocketStates.open; final legacyData = json.encode({ 'topic': topic, @@ -539,7 +539,7 @@ void main() { encode: (payload) => 'custom-frame', ); customSocket.connect(); - customSocket.connectionStatus = SocketStates.open; + customSocket.connState = SocketStates.open; customSocket.push( Message(topic: topic, payload: payload, event: event, ref: ref), @@ -550,11 +550,11 @@ void main() { }); }); - group('onConnectionMessage', () { + group('onConnMessage', () { test('drops a malformed frame without throwing', () { final socket = RealtimeClient(socketEndpoint); expect( - () => socket.onConnectionMessage('{"not": "an array"}'), + () => socket.onConnMessage('{"not": "an array"}'), returnsNormally, ); }); @@ -583,7 +583,7 @@ void main() { ...payload, ]); - socket.onConnectionMessage(frame); + socket.onConnMessage(frame); expect(received, { 'type': 'broadcast', @@ -606,7 +606,7 @@ void main() { callback: (payload) => received = payload, ); - socket.onConnectionMessage(json.encode({ + socket.onConnMessage(json.encode({ 'topic': 'realtime:room', 'event': 'broadcast', 'payload': { @@ -783,7 +783,7 @@ void main() { //! Unimplemented Test: closes socket when heartbeat is not ack'd within heartbeat window test('pushes heartbeat data when connected', () async { - mockedSocket.connectionStatus = SocketStates.open; + mockedSocket.connState = SocketStates.open; await mockedSocket.sendHeartbeat(); @@ -791,7 +791,7 @@ void main() { }); test('no ops when not connected', () async { - mockedSocket.connectionStatus = SocketStates.connecting; + mockedSocket.connState = SocketStates.connecting; await mockedSocket.sendHeartbeat(); verifyNever(() => mockedSink.add(any())); @@ -830,11 +830,11 @@ void main() { await connectFuture; // Should NOT have transitioned to open because disconnect nullified connection - expect(socket.connectionStatus, isNot(SocketStates.open)); - expect(socket.connection, isNull); + expect(socket.connState, isNot(SocketStates.open)); + expect(socket.conn, isNull); }); - test('connect bails out when connectionStatus changes during await ready', + test('connect bails out when connState changes during await ready', () async { final readyCompleter = Completer(); final mockedSocketChannel = MockIOWebSocketChannel(); @@ -863,7 +863,7 @@ void main() { await disconnectFuture; await connectFuture; - expect(socket.connectionStatus, isNot(SocketStates.open)); + expect(socket.connState, isNot(SocketStates.open)); }); test('rapid connect-disconnect-connect cycle does not crash', () async { @@ -917,8 +917,8 @@ void main() { readyCompleter2.complete(); await socket.connect(); - expect(socket.connectionStatus, SocketStates.open); - expect(socket.connection, mockedSocketChannel2); + expect(socket.connState, SocketStates.open); + expect(socket.conn, mockedSocketChannel2); await socket.disconnect(); await streamController2.close(); diff --git a/packages/supabase_flutter/lib/src/supabase.dart b/packages/supabase_flutter/lib/src/supabase.dart index d2836b157..c2ce4e3c3 100644 --- a/packages/supabase_flutter/lib/src/supabase.dart +++ b/packages/supabase_flutter/lib/src/supabase.dart @@ -299,7 +299,7 @@ class Supabase { } else { // paused or detached — disconnect the WebSocket if it is active. if (realtime.isConnected || - realtime.connectionStatus == SocketStates.connecting) { + realtime.connState == SocketStates.connecting) { await realtime.disconnect(); } } diff --git a/packages/supabase_flutter/test/lifecycle_test.dart b/packages/supabase_flutter/test/lifecycle_test.dart index 4d74dc1ec..d94cb3c44 100644 --- a/packages/supabase_flutter/test/lifecycle_test.dart +++ b/packages/supabase_flutter/test/lifecycle_test.dart @@ -138,7 +138,7 @@ void main() { // Connect with ready completed immediately await connectAndReady(realtime); - expect(realtime.connectionStatus, SocketStates.open); + expect(realtime.connState, SocketStates.open); // paused → triggers disconnect binding.handleAppLifecycleStateChanged(AppLifecycleState.inactive); @@ -152,8 +152,8 @@ void main() { // Complete any pending ready futures (reconnect) await settleLifecycle(); - expect(realtime.connectionStatus, SocketStates.open); - expect(realtime.connection, isNotNull); + expect(realtime.connState, SocketStates.open); + expect(realtime.conn, isNotNull); }); test( @@ -165,7 +165,7 @@ void main() { realtime.channel('test'); await connectAndReady(realtime); - expect(realtime.connectionStatus, SocketStates.open); + expect(realtime.connState, SocketStates.open); // paused → starts disconnect binding.handleAppLifecycleStateChanged(AppLifecycleState.inactive); @@ -186,8 +186,8 @@ void main() { await settleLifecycle(); // Should have reconnected, not stuck disconnecting - expect(realtime.connectionStatus, SocketStates.open); - expect(realtime.connection, isNotNull); + expect(realtime.connState, SocketStates.open); + expect(realtime.conn, isNotNull); }); test( @@ -199,7 +199,7 @@ void main() { realtime.channel('test'); await connectAndReady(realtime); - expect(realtime.connectionStatus, SocketStates.open); + expect(realtime.connState, SocketStates.open); // Rapid lifecycle flapping binding.handleAppLifecycleStateChanged(AppLifecycleState.inactive); @@ -219,8 +219,8 @@ void main() { // Complete all pending ready futures as they appear await settleLifecycle(); - expect(realtime.connectionStatus, SocketStates.open); - expect(realtime.connection, isNotNull); + expect(realtime.connState, SocketStates.open); + expect(realtime.conn, isNotNull); }); test( @@ -232,7 +232,7 @@ void main() { realtime.channel('test'); await connectAndReady(realtime); - expect(realtime.connectionStatus, SocketStates.open); + expect(realtime.connState, SocketStates.open); // paused → triggers disconnect binding.handleAppLifecycleStateChanged(AppLifecycleState.inactive); @@ -253,8 +253,8 @@ void main() { await settleLifecycle(); // Should be disconnected since the last event was paused - expect(realtime.connectionStatus, SocketStates.disconnected); - expect(realtime.connection, isNull); + expect(realtime.connState, SocketStates.disconnected); + expect(realtime.conn, isNull); }); }); }