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 8e8a97ea0..61b432333 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 = '1.0.0';
   static const Duration defaultTimeout = Duration(milliseconds: 10000);
   static const int defaultHeartbeatIntervalMs = 25000;
   static const int wsCloseNormal = 1000;
@@ -12,6 +11,19 @@ class Constants {
 
 typedef RealtimeConstants = Constants;
 
+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 c509209e8..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<String, dynamic>;
+      final errorBody = json.decode(response.body) as Map<String, dynamic>;
       errorMessage = (errorBody['error'] ??
           errorBody['message'] ??
           errorMessage) as String;
@@ -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:
+  ///
+  /// ```dart
+  /// channel.sendBroadcastMessage(
+  ///   event: 'file',
+  ///   payload: {'payload': myUint8List},
+  /// );
+  /// ```
   Future<ChannelResponse> sendBroadcastMessage({
     required String event,
     required Map<String, dynamic> payload,
@@ -643,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 d9a0b26be..a343f2670 100644
--- a/packages/realtime_client/lib/src/realtime_client.dart
+++ b/packages/realtime_client/lib/src/realtime_client.dart
@@ -10,6 +10,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';
 
@@ -18,15 +19,13 @@ typedef WebSocketTransport = WebSocketChannel Function(
   Map<String, String> headers,
 );
 
-typedef RealtimeEncode = void Function(
-  dynamic payload,
-  void Function(String result) callback,
-);
+/// Serializes an outgoing message into the `String` or binary frame written to
+/// the WebSocket.
+typedef RealtimeEncode = Object Function(Map<String, dynamic> payload);
 
-typedef RealtimeDecode = void Function(
-  String payload,
-  void Function(dynamic result) callback,
-);
+/// Deserializes a raw incoming WebSocket frame (`String` or binary) into a
+/// message map.
+typedef RealtimeDecode = Map<String, dynamic> Function(Object payload);
 
 /// Event details for when the connection closed.
 class RealtimeCloseEvent {
@@ -89,13 +88,14 @@ 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<RealtimeChannel> channels = [];
   final String endPoint;
 
   final Map<String, String> headers;
   final Map<String, dynamic> params;
+
+  final RealtimeProtocolVersion version;
   final Duration timeout;
   final WebSocketTransport transport;
   final Client? httpClient;
@@ -111,9 +111,10 @@ class RealtimeClient {
   /// Unique reference ID for every heartbeat.
   int ref = 0;
   late RetryTimer reconnectTimer;
-  void Function(String? kind, String? msg, dynamic data)? logger;
-  late RealtimeEncode encode;
-  late RealtimeDecode decode;
+  void Function(String? kind, String? message, dynamic data)? logger;
+  static final Serializer _serializer = Serializer();
+  final RealtimeEncode encode;
+  final RealtimeDecode decode;
   late TimerCalculation reconnectAfterMs;
   WebSocketChannel? conn;
   List sendBuffer = [];
@@ -127,7 +128,6 @@ class RealtimeClient {
   @Deprecated("No longer used. Will be removed in the next major version.")
   int longpollerTimeout = 20000;
   SocketStates? connState;
-  // This is called `accessToken` in realtime-js
   Future<String?> Function()? customAccessToken;
 
   /// Initializes the Socket
@@ -144,15 +144,21 @@ 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] The function to encode outgoing messages. Defaults to JSON: (payload, callback) => callback(JSON.stringify(payload))
+  /// [encode] Overrides how outgoing messages are serialized, for example to
+  /// use a faster JSON implementation. Defaults to the codec for [version].
   ///
-  /// [decode] The function to decode incoming messages. Defaults to JSON: (payload, callback) => callback(JSON.parse(payload))
+  /// [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.
+  ///
+  /// [version] The Realtime protocol version. Defaults to
+  /// [RealtimeProtocolVersion.v2]; pass [RealtimeProtocolVersion.v1] for the
+  /// legacy object-shaped JSON frames.
   RealtimeClient(
     String endPoint, {
     WebSocketTransport? transport,
@@ -168,6 +174,7 @@ class RealtimeClient {
     RealtimeLogLevel? logLevel,
     this.httpClient,
     this.customAccessToken,
+    this.version = RealtimeProtocolVersion.v2,
   })  : endPoint = Uri.parse('$endPoint/${Transports.websocket}')
             .replace(
               queryParameters:
@@ -178,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');
@@ -187,12 +202,6 @@ class RealtimeClient {
 
     this.reconnectAfterMs =
         reconnectAfterMs ?? RetryTimer.createRetryFunction();
-    this.encode = encode ??
-        (dynamic payload, Function(String result) callback) =>
-            callback(json.encode(payload));
-    this.decode = decode ??
-        (String payload, Function(dynamic result) callback) =>
-            callback(json.decode(payload));
     reconnectTimer = RetryTimer(
       () async {
         await disconnect();
@@ -244,8 +253,7 @@ class RealtimeClient {
 
       _onConnOpen();
       localConn.stream.listen(
-        // incoming messages
-        (message) => onConnMessage(message as String),
+        (message) => onConnMessage(message),
         onError: _onConnError,
         onDone: () {
           // communication has been closed
@@ -322,9 +330,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
@@ -394,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() {
-      encode(message.toJson(), (result) => conn?.sink.add(result));
+      conn?.sink.add(encode(message.toJson()));
     }
 
     log('push', '${message.topic} ${message.event} (${message.ref})',
@@ -408,39 +419,52 @@ class RealtimeClient {
     return null;
   }
 
-  void onConnMessage(String 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 onConnMessage(Object rawMessage) {
+    final Map<String, dynamic> message;
+    try {
+      message = decode(rawMessage);
+    } catch (error) {
+      log('transport', 'failed to decode message', error);
+      return;
+    }
 
-      log(
-        'receive',
-        "${payload['status'] ?? ''} $topic $event ${ref != null ? '($ref)' : ''}",
-        payload,
-      );
+    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;
+    }
 
-      channels
-          .where((channel) => channel.isMember(topic))
-          .forEach((channel) => channel.trigger(
-                event,
-                payload,
-                ref,
-              ));
-      for (final callback in stateChangeCallbacks['message']!) {
-        callback(msg);
-      }
-    });
+    final status = payload is Map ? (payload['status'] ?? '') : '';
+    log(
+      'receive',
+      "$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(message);
+    }
   }
 
+  static Object _encodeLegacy(Map<String, dynamic> message) =>
+      jsonEncode(message);
+
+  static Map<String, dynamic> _decodeLegacy(Object rawMessage) =>
+      Map<String, dynamic>.from(jsonDecode(rawMessage as String) as Map);
+
   /// Returns the URL of the websocket.
   String get endPointURL {
     final params = Map<String, String>.from(this.params);
-    params['vsn'] = Constants.vsn;
+    params['vsn'] = version.vsn;
     return _appendParams(endPoint, params);
   }
 
@@ -577,7 +601,7 @@ class RealtimeClient {
       pendingHeartbeatRef = null;
       log(
         'transport',
-        'heartbeat timeout. Attempting to re-establish connection',
+        'heartbeat timeout. Attempting to re-establish conn',
       );
       conn?.sink.close(Constants.wsCloseNormal, 'heartbeat timeout');
       return;
diff --git a/packages/realtime_client/lib/src/serializer.dart b/packages/realtime_client/lib/src/serializer.dart
new file mode 100644
index 000000000..eae36ae34
--- /dev/null
+++ b/packages/realtime_client/lib/src/serializer.dart
@@ -0,0 +1,231 @@
+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.
+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;
+
+  /// 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<String> allowedMetadataKeys;
+
+  Serializer({List<String>? 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()`.
+  Object encode(Map<String, dynamic> message) {
+    final payload = message['payload'];
+    if (message['event'] == broadcastEvent &&
+        payload is Map &&
+        payload['event'] is String &&
+        _isBinary(payload['payload'])) {
+      return _encodeBinaryUserBroadcastPush(message, payload);
+    }
+
+    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.
+  Map<String, dynamic> decode(Object rawPayload) {
+    if (rawPayload is String) {
+      final decoded = jsonDecode(rawPayload);
+      if (decoded is! List || decoded.length < 5) {
+        throw FormatException('Invalid 2.0.0 text frame', rawPayload);
+      }
+      return <String, dynamic>{
+        'join_ref': decoded[0],
+        'ref': decoded[1],
+        'topic': decoded[2],
+        'event': decoded[3],
+        'payload': decoded[4],
+      };
+    }
+
+    final bytes = _asBytes(rawPayload);
+    if (bytes != null) {
+      return _binaryDecode(bytes);
+    }
+
+    return <String, dynamic>{};
+  }
+
+  Uint8List _encodeBinaryUserBroadcastPush(
+    Map<String, dynamic> message,
+    Map<dynamic, dynamic> 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 <String, dynamic>{}
+        : _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 frame = Uint8List(headerLength + metaLength + encodedPayload.length);
+    var offset = 0;
+    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<String, dynamic> _binaryDecode(Uint8List buffer) {
+    final view = ByteData.sublistView(buffer);
+    final kind = view.getUint8(0);
+    switch (kind) {
+      case kindUserBroadcast:
+        return _decodeUserBroadcast(buffer, view);
+      default:
+        return <String, dynamic>{};
+    }
+  }
+
+  Map<String, dynamic> _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 + userBroadcastMetaLength;
+    final topic =
+        utf8.decode(Uint8List.sublistView(buffer, offset, offset + topicSize));
+    offset += topicSize;
+    final userEvent = utf8
+        .decode(Uint8List.sublistView(buffer, offset, offset + userEventSize));
+    offset += userEventSize;
+    final metadata = metadataSize > 0
+        ? utf8.decode(
+            Uint8List.sublistView(buffer, offset, offset + metadataSize))
+        : '';
+    offset += metadataSize;
+
+    final payloadBytes = Uint8List.sublistView(buffer, offset);
+    final dynamic parsedPayload = payloadEncoding == jsonEncoding
+        ? jsonDecode(utf8.decode(payloadBytes))
+        : payloadBytes;
+
+    final data = <String, dynamic>{
+      'type': broadcastEvent,
+      'event': userEvent,
+      'payload': parsedPayload,
+    };
+    if (metadataSize > 0) {
+      data['meta'] = jsonDecode(metadata);
+    }
+
+    return <String, dynamic>{
+      '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');
+    }
+  }
+
+  // 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;
+    }
+    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<int>) {
+      return Uint8List.fromList(value);
+    }
+    return null;
+  }
+
+  Map<String, dynamic> _pick(Map<dynamic, dynamic> source, List<String> keys) {
+    return <String, dynamic>{
+      for (final key in keys)
+        if (source.containsKey(key)) key: source[key],
+    };
+  }
+}
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<String, String> 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/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..1bc16655f
--- /dev/null
+++ b/packages/realtime_client/test/serializer_test.dart
@@ -0,0 +1,250 @@
+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 = <int>[
+    Serializer.kindUserBroadcast,
+    topicBytes.length,
+    eventBytes.length,
+    metadataBytes.length,
+    encoding,
+  ];
+
+  return Uint8List.fromList([
+    ...header,
+    ...topicBytes,
+    ...eventBytes,
+    ...metadataBytes,
+    ...payload,
+  ]);
+}
+
+void main() {
+  late Serializer serializer;
+
+  setUp(() {
+    serializer = Serializer();
+  });
+
+  group('encode text frames', () {
+    test('encodes a message as a positional JSON array', () {
+      final result = serializer.encode({
+        'join_ref': '1',
+        'ref': '2',
+        'topic': 'realtime:room',
+        'event': 'phx_join',
+        'payload': {'foo': 'bar'},
+      });
+
+      expect(result, isA<String>());
+      expect(
+        jsonDecode(result as String),
+        equals([
+          '1',
+          '2',
+          'realtime:room',
+          'phx_join',
+          {'foo': 'bar'},
+        ]),
+      );
+    });
+
+    test('preserves null join_ref and ref positionally', () {
+      final result = serializer.encode({
+        'topic': 'phoenix',
+        'event': 'heartbeat',
+        'payload': <String, dynamic>{},
+      });
+
+      expect(
+        jsonDecode(result as String),
+        equals([null, null, 'phoenix', 'heartbeat', <String, dynamic>{}]),
+      );
+    });
+
+    test('encodes a non-binary broadcast as a text frame', () {
+      final result = serializer.encode({
+        'join_ref': '1',
+        'ref': '2',
+        'topic': 'realtime:room',
+        'event': 'broadcast',
+        'payload': {'event': 'cursor', 'type': 'broadcast', 'x': 1},
+      });
+
+      expect(result, isA<String>());
+      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 = serializer.decode(
+        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'},
+      });
+    });
+
+    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', () {
+    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 = serializer.decode(frame);
+
+      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 = serializer.decode(frame);
+      final payload = result['payload'] as Map<String, dynamic>;
+
+      expect(payload['event'], 'file');
+      expect(payload['payload'], rawPayload);
+      expect(payload.containsKey('meta'), isFalse);
+    });
+
+    test('returns an empty map for unknown binary kinds', () {
+      final result = serializer.decode(Uint8List.fromList([99, 0, 0]));
+      expect(result, <String, dynamic>{});
+    });
+  });
+
+  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 = serializer.encode({
+        'join_ref': '7',
+        'ref': '8',
+        'topic': 'realtime:room',
+        'event': 'broadcast',
+        'payload': {
+          'type': 'broadcast',
+          'event': 'file',
+          'payload': payload,
+        },
+      });
+
+      expect(result, isA<Uint8List>());
+      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 = serializerWithMeta.encode({
+        '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(
+        () => serializer.encode({
+          '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..bca2d1a35 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,18 @@ 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',
+      );
+    });
+
+    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',
       );
     });
   });
@@ -409,12 +421,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;
@@ -465,6 +475,154 @@ 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.connState = 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<Uint8List>())))
+          .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.connState = 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);
+    });
+
+    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.connState = SocketStates.open;
+
+      customSocket.push(
+        Message(topic: topic, payload: payload, event: event, ref: ref),
+      );
+
+      verify(() => customSink.add(captureAny(that: equals('custom-frame'))))
+          .called(1);
+    });
+  });
+
+  group('onConnMessage', () {
+    test('drops a malformed frame without throwing', () {
+      final socket = RealtimeClient(socketEndpoint);
+      expect(
+        () => socket.onConnMessage('{"not": "an array"}'),
+        returnsNormally,
+      );
+    });
+
+    test('dispatches a received binary broadcast to onBroadcast', () {
+      final socket = RealtimeClient(socketEndpoint);
+      final channel = socket.channel('room');
+
+      Map<String, dynamic>? 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.onConnMessage(frame);
+
+      expect(received, {
+        'type': 'broadcast',
+        'event': 'cursor',
+        '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<String, dynamic>? received;
+      channel.onBroadcast(
+        event: 'cursor',
+        callback: (payload) => received = payload,
+      );
+
+      socket.onConnMessage(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', () {
@@ -603,12 +761,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();
@@ -647,7 +800,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<void>();
       final mockedSocketChannel = MockIOWebSocketChannel();
@@ -676,7 +829,7 @@ void main() {
       await disconnectFuture;
       await connectFuture;
 
-      // Should NOT have transitioned to open because disconnect nullified conn
+      // Should NOT have transitioned to open because disconnect nullified connection
       expect(socket.connState, isNot(SocketStates.open));
       expect(socket.conn, isNull);
     });
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 {