Skip to content

Commit c04d5d1

Browse files
spydonspydon
authored
feat(realtime): add support for protocol format 2.0.0 (#1397)
## What Adds Realtime protocol **2.0.0** support to the Dart/Flutter SDK and makes the protocol version selectable. Closes [SDK-583](https://linear.app/supabase/issue/SDK-583/realtime-add-support-to-protocol-format-200-flutter). Protocol 2.0.0 changes the WebSocket frame format: - **Text frames** use the positional JSON array `[join_ref, ref, topic, event, payload]` instead of the 1.0.0 object layout, reducing JSON work on the backend (lower latency). - **Binary frames** allow raw binary payloads for broadcast user events. ## Changes ### Serializer (`serializer.dart`, new) - Text encode/decode as the positional array `[join_ref, ref, topic, event, payload]`. - Binary encode for broadcast pushes (`userBroadcastPush`, kind `3`). - Binary decode for incoming broadcasts (`userBroadcast`, kind `4`). - JSON/binary encoding flag and `allowedMetadataKeys`. - `decode` validates the text frame shape and throws a clear `FormatException` on malformed input; `onConnMessage` logs and drops such frames instead of crashing the connection. ### Selectable protocol version - New `RealtimeProtocolVersion` enum (`v1` → `1.0.0`, `v2` → `2.0.0`); each carries the `vsn` value sent as the connection parameter. - `RealtimeClient` takes a `version` parameter, **defaulting to `v2`**. `v2` uses the serializer; `v1` uses the legacy object-shaped JSON frames. ### Codec override - Optional `encode` / `decode` constructor arguments (and the `RealtimeEncode` / `RealtimeDecode` typedefs) let consumers swap in a custom serializer; they default to the codec for the selected `version`. - Return-based signatures: `RealtimeEncode` = `Object Function(Map<String, dynamic>)`, `RealtimeDecode` = `Map<String, dynamic> Function(Object)` (synchronous — async/isolate codec is a follow-up, see below). ### Binary broadcasts Incoming binary broadcast frames are decoded into the same map shape as JSON broadcasts. To send binary, provide a `Uint8List` (or any `TypedData`) under the `payload` key: \`\`\`dart channel.sendBroadcastMessage(event: 'file', payload: {'payload': myUint8List}); \`\`\` Binary frames are delivered as `Uint8List` on every platform; the web transport sets `binaryType` to arraybuffer explicitly. ### Misc - Expanded a few abbreviated identifiers (`obj`/`msg`/`res`) in touched code. ## Breaking changes - **Default Realtime protocol is now `2.0.0`** (pass `version: RealtimeProtocolVersion.v1` to keep the old behavior). - `RealtimeEncode` / `RealtimeDecode` changed from callback-based to return-based signatures. ## Related - The breaking field/method renames (`conn` → `connection`, `connState` → `connectionStatus`, `onConnMessage` → `onConnectionMessage`) are split into #1404, which stacks on this PR and will be retargeted to `v3` after this merges. - Follow-up: async codec path for isolate offloading — #1401. ## Tests - `realtime_client`: 121 passing (new `serializer_test.dart`, endpointURL `v1`, legacy encode, legacy decode + dispatch, custom-`encode` override, malformed-frame, binary-broadcast receive). - `supabase`: 83 passing. - `supabase_flutter/test/lifecycle_test.dart`: analyze clean.
1 parent 1c2e233 commit c04d5d1

10 files changed

Lines changed: 889 additions & 179 deletions

File tree

‎packages/realtime_client/lib/realtime_client.dart‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,11 @@
22
library;
33

44
export 'src/constants.dart'
5-
show RealtimeConstants, RealtimeLogLevel, SocketStates;
5+
show
6+
RealtimeConstants,
7+
RealtimeLogLevel,
8+
RealtimeProtocolVersion,
9+
SocketStates;
610
export 'src/realtime_channel.dart';
711
export 'src/realtime_client.dart';
812
export 'src/realtime_presence.dart';

‎packages/realtime_client/lib/src/constants.dart‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
import 'package:realtime_client/src/version.dart';
22

33
class Constants {
4-
static const String vsn = '1.0.0';
54
static const Duration defaultTimeout = Duration(milliseconds: 10000);
65
static const int defaultHeartbeatIntervalMs = 25000;
76
static const int wsCloseNormal = 1000;
@@ -12,6 +11,19 @@ class Constants {
1211

1312
typedef RealtimeConstants = Constants;
1413

14+
enum RealtimeProtocolVersion {
15+
/// Legacy protocol: object-shaped JSON text frames only.
16+
v1('1.0.0'),
17+
18+
/// Positional JSON array text frames plus binary frames.
19+
v2('2.0.0');
20+
21+
const RealtimeProtocolVersion(this.vsn);
22+
23+
/// The value sent as the `vsn` connection parameter.
24+
final String vsn;
25+
}
26+
1527
enum SocketStates {
1628
/// Client attempting to establish a connection
1729
connecting,

‎packages/realtime_client/lib/src/realtime_channel.dart‎

Lines changed: 19 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -565,7 +565,7 @@ class RealtimeChannel {
565565
]
566566
};
567567

568-
final res = await (socket.httpClient?.post ?? post)(
568+
final response = await (socket.httpClient?.post ?? post)(
569569
Uri.parse(broadcastEndpointURL),
570570
headers: headers,
571571
body: json.encode(body),
@@ -574,13 +574,13 @@ class RealtimeChannel {
574574
onTimeout: () => throw TimeoutException('Request timeout'),
575575
);
576576

577-
if (res.statusCode == 202) {
577+
if (response.statusCode == 202) {
578578
return;
579579
}
580580

581-
String errorMessage = res.reasonPhrase ?? 'Unknown error';
581+
String errorMessage = response.reasonPhrase ?? 'Unknown error';
582582
try {
583-
final errorBody = json.decode(res.body) as Map<String, dynamic>;
583+
final errorBody = json.decode(response.body) as Map<String, dynamic>;
584584
errorMessage = (errorBody['error'] ??
585585
errorBody['message'] ??
586586
errorMessage) as String;
@@ -592,6 +592,19 @@ class RealtimeChannel {
592592
}
593593

594594
/// Sends a realtime broadcast message.
595+
///
596+
/// With protocol `2.0.0` the message is sent as a positional JSON text frame.
597+
///
598+
/// To send a raw binary payload over a WebSocket binary frame (avoiding JSON
599+
/// encoding on the server), set a `Uint8List` (or any `TypedData`) under the
600+
/// `payload` key:
601+
///
602+
/// ```dart
603+
/// channel.sendBroadcastMessage(
604+
/// event: 'file',
605+
/// payload: {'payload': myUint8List},
606+
/// );
607+
/// ```
595608
Future<ChannelResponse> sendBroadcastMessage({
596609
required String event,
597610
required Map<String, dynamic> payload,
@@ -643,12 +656,12 @@ class RealtimeChannel {
643656
]
644657
};
645658
try {
646-
final res = await (socket.httpClient?.post ?? post)(
659+
final response = await (socket.httpClient?.post ?? post)(
647660
Uri.parse(broadcastEndpointURL),
648661
headers: headers,
649662
body: json.encode(body),
650663
);
651-
if (200 <= res.statusCode && res.statusCode < 300) {
664+
if (200 <= response.statusCode && response.statusCode < 300) {
652665
completer.complete(ChannelResponse.ok);
653666
} else {
654667
completer.complete(ChannelResponse.error);

‎packages/realtime_client/lib/src/realtime_client.dart‎

Lines changed: 80 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import 'package:realtime_client/realtime_client.dart';
1010
import 'package:realtime_client/src/constants.dart';
1111
import 'package:realtime_client/src/message.dart';
1212
import 'package:realtime_client/src/retry_timer.dart';
13+
import 'package:realtime_client/src/serializer.dart';
1314
import 'package:realtime_client/src/websocket/websocket.dart';
1415
import 'package:web_socket_channel/web_socket_channel.dart';
1516

@@ -18,15 +19,13 @@ typedef WebSocketTransport = WebSocketChannel Function(
1819
Map<String, String> headers,
1920
);
2021

21-
typedef RealtimeEncode = void Function(
22-
dynamic payload,
23-
void Function(String result) callback,
24-
);
22+
/// Serializes an outgoing message into the `String` or binary frame written to
23+
/// the WebSocket.
24+
typedef RealtimeEncode = Object Function(Map<String, dynamic> payload);
2525

26-
typedef RealtimeDecode = void Function(
27-
String payload,
28-
void Function(dynamic result) callback,
29-
);
26+
/// Deserializes a raw incoming WebSocket frame (`String` or binary) into a
27+
/// message map.
28+
typedef RealtimeDecode = Map<String, dynamic> Function(Object payload);
3029

3130
/// Event details for when the connection closed.
3231
class RealtimeCloseEvent {
@@ -89,13 +88,14 @@ class RealtimeCloseEvent {
8988
/// - Works on all Dart platforms (Flutter mobile/desktop, web, server).
9089
/// - On web, the underlying [WebSocketChannel] uses the browser WebSocket API.
9190
class RealtimeClient {
92-
// This is named `accessTokenValue` in supabase-js
9391
String? accessToken;
9492
List<RealtimeChannel> channels = [];
9593
final String endPoint;
9694

9795
final Map<String, String> headers;
9896
final Map<String, dynamic> params;
97+
98+
final RealtimeProtocolVersion version;
9999
final Duration timeout;
100100
final WebSocketTransport transport;
101101
final Client? httpClient;
@@ -111,9 +111,10 @@ class RealtimeClient {
111111
/// Unique reference ID for every heartbeat.
112112
int ref = 0;
113113
late RetryTimer reconnectTimer;
114-
void Function(String? kind, String? msg, dynamic data)? logger;
115-
late RealtimeEncode encode;
116-
late RealtimeDecode decode;
114+
void Function(String? kind, String? message, dynamic data)? logger;
115+
static final Serializer _serializer = Serializer();
116+
final RealtimeEncode encode;
117+
final RealtimeDecode decode;
117118
late TimerCalculation reconnectAfterMs;
118119
WebSocketChannel? conn;
119120
List sendBuffer = [];
@@ -127,7 +128,6 @@ class RealtimeClient {
127128
@Deprecated("No longer used. Will be removed in the next major version.")
128129
int longpollerTimeout = 20000;
129130
SocketStates? connState;
130-
// This is called `accessToken` in realtime-js
131131
Future<String?> Function()? customAccessToken;
132132

133133
/// Initializes the Socket
@@ -144,15 +144,21 @@ class RealtimeClient {
144144
///
145145
/// [heartbeatIntervalMs] The millisec interval to send a heartbeat message.
146146
///
147-
/// [logger] The optional function for specialized logging, ie: logger: (kind, msg, data) => { console.log(`$kind: $msg`, data) }
147+
/// [logger] The optional function for specialized logging, ie: logger: (kind, message, data) => { console.log(`$kind: $message`, data) }
148148
///
149-
/// [encode] The function to encode outgoing messages. Defaults to JSON: (payload, callback) => callback(JSON.stringify(payload))
149+
/// [encode] Overrides how outgoing messages are serialized, for example to
150+
/// use a faster JSON implementation. Defaults to the codec for [version].
150151
///
151-
/// [decode] The function to decode incoming messages. Defaults to JSON: (payload, callback) => callback(JSON.parse(payload))
152+
/// [decode] Overrides how incoming frames are deserialized. Defaults to the
153+
/// codec for [version].
152154
///
153155
/// [reconnectAfterMs] The optional function that returns the millsec reconnect interval. Defaults to stepped backoff off.
154156
///
155157
/// [logLevel] Specifies the log level for the connection on the server.
158+
///
159+
/// [version] The Realtime protocol version. Defaults to
160+
/// [RealtimeProtocolVersion.v2]; pass [RealtimeProtocolVersion.v1] for the
161+
/// legacy object-shaped JSON frames.
156162
RealtimeClient(
157163
String endPoint, {
158164
WebSocketTransport? transport,
@@ -168,6 +174,7 @@ class RealtimeClient {
168174
RealtimeLogLevel? logLevel,
169175
this.httpClient,
170176
this.customAccessToken,
177+
this.version = RealtimeProtocolVersion.v2,
171178
}) : endPoint = Uri.parse('$endPoint/${Transports.websocket}')
172179
.replace(
173180
queryParameters:
@@ -178,7 +185,15 @@ class RealtimeClient {
178185
...Constants.defaultHeaders,
179186
if (headers != null) ...headers,
180187
},
181-
transport = transport ?? createWebSocketClient {
188+
transport = transport ?? createWebSocketClient,
189+
encode = encode ??
190+
(version == RealtimeProtocolVersion.v1
191+
? _encodeLegacy
192+
: _serializer.encode),
193+
decode = decode ??
194+
(version == RealtimeProtocolVersion.v1
195+
? _decodeLegacy
196+
: _serializer.decode) {
182197
_log.config(
183198
'Initialize RealtimeClient with endpoint: $endPoint, timeout: $timeout, heartbeatIntervalMs: $heartbeatIntervalMs, logLevel: $logLevel');
184199
_log.finest('Initialize with headers: $headers, params: $params');
@@ -187,12 +202,6 @@ class RealtimeClient {
187202

188203
this.reconnectAfterMs =
189204
reconnectAfterMs ?? RetryTimer.createRetryFunction();
190-
this.encode = encode ??
191-
(dynamic payload, Function(String result) callback) =>
192-
callback(json.encode(payload));
193-
this.decode = decode ??
194-
(String payload, Function(dynamic result) callback) =>
195-
callback(json.decode(payload));
196205
reconnectTimer = RetryTimer(
197206
() async {
198207
await disconnect();
@@ -244,8 +253,7 @@ class RealtimeClient {
244253

245254
_onConnOpen();
246255
localConn.stream.listen(
247-
// incoming messages
248-
(message) => onConnMessage(message as String),
256+
(message) => onConnMessage(message),
249257
onError: _onConnError,
250258
onDone: () {
251259
// communication has been closed
@@ -322,9 +330,12 @@ class RealtimeClient {
322330
///
323331
/// [level] must be [Level.FINEST] for senitive data
324332
void log(
325-
[String? kind, String? msg, dynamic data, Level level = Level.FINEST]) {
326-
_log.log(level, '$kind: $msg', data);
327-
logger?.call(kind, msg, data);
333+
[String? kind,
334+
String? message,
335+
dynamic data,
336+
Level level = Level.FINEST]) {
337+
_log.log(level, '$kind: $message', data);
338+
logger?.call(kind, message, data);
328339
}
329340

330341
/// Registers callbacks for connection state change events
@@ -394,7 +405,7 @@ class RealtimeClient {
394405
/// If the socket is not connected, the message gets enqueued within a local buffer, and sent out when a connection is next established.
395406
String? push(Message message) {
396407
void callback() {
397-
encode(message.toJson(), (result) => conn?.sink.add(result));
408+
conn?.sink.add(encode(message.toJson()));
398409
}
399410

400411
log('push', '${message.topic} ${message.event} (${message.ref})',
@@ -408,39 +419,52 @@ class RealtimeClient {
408419
return null;
409420
}
410421

411-
void onConnMessage(String rawMessage) {
412-
decode(rawMessage, (msg) {
413-
final topic = msg['topic'] as String;
414-
final event = msg['event'] as String;
415-
final payload = msg['payload'];
416-
final ref = msg['ref'] as String?;
417-
if (ref != null && ref == pendingHeartbeatRef) {
418-
pendingHeartbeatRef = null;
419-
}
422+
void onConnMessage(Object rawMessage) {
423+
final Map<String, dynamic> message;
424+
try {
425+
message = decode(rawMessage);
426+
} catch (error) {
427+
log('transport', 'failed to decode message', error);
428+
return;
429+
}
420430

421-
log(
422-
'receive',
423-
"${payload['status'] ?? ''} $topic $event ${ref != null ? '($ref)' : ''}",
424-
payload,
425-
);
431+
final topic = message['topic'] as String;
432+
final event = message['event'] as String;
433+
final payload = message['payload'];
434+
final ref = message['ref'] as String?;
435+
if (ref != null && ref == pendingHeartbeatRef) {
436+
pendingHeartbeatRef = null;
437+
}
426438

427-
channels
428-
.where((channel) => channel.isMember(topic))
429-
.forEach((channel) => channel.trigger(
430-
event,
431-
payload,
432-
ref,
433-
));
434-
for (final callback in stateChangeCallbacks['message']!) {
435-
callback(msg);
436-
}
437-
});
439+
final status = payload is Map ? (payload['status'] ?? '') : '';
440+
log(
441+
'receive',
442+
"$status $topic $event ${ref != null ? '($ref)' : ''}",
443+
payload,
444+
);
445+
446+
channels.where((channel) => channel.isMember(topic)).forEach(
447+
(channel) => channel.trigger(
448+
event,
449+
payload,
450+
ref,
451+
),
452+
);
453+
for (final callback in stateChangeCallbacks['message']!) {
454+
callback(message);
455+
}
438456
}
439457

458+
static Object _encodeLegacy(Map<String, dynamic> message) =>
459+
jsonEncode(message);
460+
461+
static Map<String, dynamic> _decodeLegacy(Object rawMessage) =>
462+
Map<String, dynamic>.from(jsonDecode(rawMessage as String) as Map);
463+
440464
/// Returns the URL of the websocket.
441465
String get endPointURL {
442466
final params = Map<String, String>.from(this.params);
443-
params['vsn'] = Constants.vsn;
467+
params['vsn'] = version.vsn;
444468
return _appendParams(endPoint, params);
445469
}
446470

@@ -577,7 +601,7 @@ class RealtimeClient {
577601
pendingHeartbeatRef = null;
578602
log(
579603
'transport',
580-
'heartbeat timeout. Attempting to re-establish connection',
604+
'heartbeat timeout. Attempting to re-establish conn',
581605
);
582606
conn?.sink.close(Constants.wsCloseNormal, 'heartbeat timeout');
583607
return;

0 commit comments

Comments
 (0)