diff --git a/MIGRATION.md b/MIGRATION.md
index 8069dea2d..d92aa2939 100644
--- a/MIGRATION.md
+++ b/MIGRATION.md
@@ -473,6 +473,82 @@ final byKey = {
 
 The `Presence` payload class is unchanged and stays public.
 
+### `RealtimeEncode` and `RealtimeDecode` are asynchronous and typed
+
+The codec overrides on `RealtimeClient` were synchronous and worked on raw maps, so the JSON work
+always ran on the main isolate and a large payload could block the event loop. Both typedefs now
+work on a `RealtimeMessage` and return a `Future`, which lets you hand the work to a background
+isolate:
+
+| Before | After |
+| --- | --- |
+| `Object Function(Map<String, dynamic>)` | `Future<Object> Function(RealtimeMessage)` |
+| `Map<String, dynamic> Function(Object)` | `Future<RealtimeMessage> Function(Object)` |
+
+`RealtimeMessage` carries the `joinRef`, `ref`, `topic`, `event` and `payload` of a message, and
+converts to and from the shape a protocol version puts on the wire, so a codec only has to turn that
+shape into bytes and back. Outgoing frames are written in push order and incoming messages are
+dispatched in receive order, even when a later payload finishes encoding or decoding first. A codec
+call that never completes fails after `RealtimeClient.timeout` instead of stalling every write or
+dispatch queued behind it:
+
+```dart
+final isolate = YAJsonIsolate();
+
+final client = RealtimeClient(
+  'wss://project.supabase.co/realtime/v1',
+  encode: (message) => isolate.encode(message.toJson()),
+  decode: (frame) async =>
+      RealtimeMessage.fromJson(await isolate.decode(frame as String)),
+);
+```
+
+`RealtimeClient.onMessage` now emits `RealtimeMessage` instead of `Map<String, dynamic>`, so a
+listener that reads fields off the map needs to switch to properties:
+
+```dart
+// Before
+client.onMessage.listen((message) => print(message['event']));
+
+// After
+client.onMessage.listen((message) => print(message.event));
+```
+
+`toJson` and `RealtimeMessage.fromJson` default to protocol `2.0.0`; pass
+`RealtimeProtocolVersion.v1` to either when the client runs on the legacy protocol.
+
+`RealtimeClientOptions` takes the same two callbacks, so a codec can be set on `SupabaseClient` and
+on `Supabase.initialize` without constructing a `RealtimeClient` yourself:
+
+```dart
+await Supabase.initialize(
+  url: url,
+  anonKey: anonKey,
+  realtimeClientOptions: RealtimeClientOptions(
+    encode: (message) => isolate.encode(message.toJson()),
+    decode: (frame) async =>
+        RealtimeMessage.fromJson(await isolate.decode(frame as String)),
+  ),
+);
+```
+
+`encode` and `decode` are now `null` unless you pass one; `RealtimeClient` uses the built-in codec
+for whichever of the two is `null`. That codec is synchronous, so a client that overrides neither
+still writes and dispatches without a microtask hop. Reading a custom codec back off the client
+changed accordingly; the built-in codec has no public accessor, so this only applies when you
+passed your own:
+
+```dart
+// Before
+final Map<String, dynamic> message = client.decode(frame);
+
+// After
+final RealtimeMessage message = await client.decode!(frame);
+```
+
+A codec replaces the built-in one completely, so the one above handles text frames only. Handle a
+`Uint8List` frame as well if you send or receive binary broadcasts.
+
 ### Plural enum names singularized
 
 A Dart enum type names one value rather than the set, so its name should be singular. Five enums
diff --git a/packages/supabase/lib/src/realtime_client_options.dart b/packages/supabase/lib/src/realtime_client_options.dart
index a43f1b9e9..ef2e12ab6 100644
--- a/packages/supabase/lib/src/realtime_client_options.dart
+++ b/packages/supabase/lib/src/realtime_client_options.dart
@@ -26,6 +26,27 @@ class RealtimeClientOptions {
   /// heartbeat interval.
   final Duration? disconnectOnEmptyChannelsAfter;
 
+  /// Serializes outgoing messages, for example on a background isolate so
+  /// that a large payload does not block the event loop.
+  ///
+  /// Defaults to the built-in synchronous codec.
+  ///
+  /// ```dart
+  /// final isolate = YAJsonIsolate();
+  ///
+  /// RealtimeClientOptions(
+  ///   encode: (message) => isolate.encode(message.toJson()),
+  ///   decode: (frame) async =>
+  ///       RealtimeMessage.fromJson(await isolate.decode(frame as String)),
+  /// );
+  /// ```
+  final RealtimeEncode? encode;
+
+  /// Deserializes incoming frames, for example on a background isolate.
+  ///
+  /// Defaults to the built-in synchronous codec. See [encode] for an example.
+  final RealtimeDecode? decode;
+
   /// {@macro realtime_client_options}
   const RealtimeClientOptions({
     this.logLevel,
@@ -33,5 +54,7 @@ class RealtimeClientOptions {
     this.connectionCloseTimeout,
     this.transport,
     this.disconnectOnEmptyChannelsAfter,
+    this.encode,
+    this.decode,
   });
 }
diff --git a/packages/supabase/lib/src/supabase_client.dart b/packages/supabase/lib/src/supabase_client.dart
index f6416f675..270cd4952 100644
--- a/packages/supabase/lib/src/supabase_client.dart
+++ b/packages/supabase/lib/src/supabase_client.dart
@@ -399,6 +399,8 @@ class SupabaseClient {
       customAccessToken: accessToken,
       transport: options.transport,
       disconnectOnEmptyChannelsAfter: options.disconnectOnEmptyChannelsAfter,
+      encode: options.encode,
+      decode: options.decode,
     );
   }
 
diff --git a/packages/supabase/test/client_test.dart b/packages/supabase/test/client_test.dart
index f34a5e364..60ab9de8f 100644
--- a/packages/supabase/test/client_test.dart
+++ b/packages/supabase/test/client_test.dart
@@ -173,6 +173,29 @@ void main() {
       );
     });
 
+    test('codec overrides are handed to the realtime client', () async {
+      Future<Object> encode(RealtimeMessage message) => Future.value('');
+      Future<RealtimeMessage> decode(Object frame) =>
+          Future.value(const RealtimeMessage(topic: '', event: ''));
+
+      supabase = SupabaseClient(
+        supabaseUrl,
+        supabaseKey,
+        realtimeClientOptions: RealtimeClientOptions(
+          encode: encode,
+          decode: decode,
+        ),
+      );
+
+      expect(supabase.realtime.encode, same(encode));
+      expect(supabase.realtime.decode, same(decode));
+    });
+
+    test('the realtime client uses the built-in codec by default', () async {
+      expect(supabase.realtime.encode, isNull);
+      expect(supabase.realtime.decode, isNull);
+    });
+
     test('realtime access token is set properly', () async {
       final request = await getRealtimeRequest(
         server: mockServer,
diff --git a/packages/supabase_realtime/lib/src/message.dart b/packages/supabase_realtime/lib/src/message.dart
deleted file mode 100644
index ad8933824..000000000
--- a/packages/supabase_realtime/lib/src/message.dart
+++ /dev/null
@@ -1,58 +0,0 @@
-import 'package:meta/meta.dart';
-import 'package:supabase_realtime/src/constants.dart';
-import 'package:supabase_realtime/src/types.dart';
-
-@internal
-class Message {
-  final String topic;
-  final ChannelEvent event;
-  final dynamic payload;
-  final String? ref;
-  final String? joinRef;
-
-  const Message({
-    required this.topic,
-    required this.event,
-    required this.payload,
-    this.ref,
-    this.joinRef,
-  });
-
-  /// Converting to JSON while removing functions
-  Map<String, dynamic> toJson() {
-    final dynamic processedPayload;
-    if (payload is Map) {
-      processedPayload = <String, dynamic>{};
-      for (final outerKey in payload.keys) {
-        final outerValue = payload[outerKey];
-        if (outerValue is Map) {
-          processedPayload[outerKey] = {};
-          for (final innerKey in outerValue.keys) {
-            final innerValue = outerValue[innerKey];
-            if (innerValue is Binding) {
-              processedPayload[outerKey][innerKey] = <String, dynamic>{
-                'type': innerValue.type,
-                'filter': innerValue.filter,
-              };
-            } else {
-              processedPayload[outerKey][innerKey] = innerValue;
-            }
-          }
-        } else {
-          processedPayload[outerKey] = outerValue;
-        }
-      }
-    } else {
-      processedPayload = payload;
-    }
-    return {
-      'topic': topic,
-      'event': event != ChannelEvent.heartbeat
-          ? event.eventName()
-          : 'heartbeat',
-      'payload': processedPayload,
-      'ref': ?ref,
-      'join_ref': ?joinRef,
-    };
-  }
-}
diff --git a/packages/supabase_realtime/lib/src/push.dart b/packages/supabase_realtime/lib/src/push.dart
index f6e8c2d04..2d671ffae 100644
--- a/packages/supabase_realtime/lib/src/push.dart
+++ b/packages/supabase_realtime/lib/src/push.dart
@@ -2,7 +2,6 @@ import 'dart:async';
 
 import 'package:supabase_realtime/supabase_realtime.dart';
 import 'package:supabase_realtime/src/constants.dart';
-import 'package:supabase_realtime/src/message.dart';
 import 'package:supabase_realtime/src/types.dart';
 import 'package:meta/meta.dart';
 
@@ -62,7 +61,7 @@ class Push {
     startTimeout();
     sent = true;
     _channel.socket.push(
-      Message(
+      RealtimeMessage.outgoing(
         topic: _channel.topic,
         event: _event,
         payload: payload,
diff --git a/packages/supabase_realtime/lib/src/realtime_client.dart b/packages/supabase_realtime/lib/src/realtime_client.dart
index 78f8b984e..57f82b098 100644
--- a/packages/supabase_realtime/lib/src/realtime_client.dart
+++ b/packages/supabase_realtime/lib/src/realtime_client.dart
@@ -9,7 +9,6 @@ import 'package:supabase_common/supabase_common.dart';
 import 'package:supabase_realtime/supabase_realtime.dart';
 import 'package:supabase_realtime/src/constants.dart';
 import 'package:supabase_realtime/src/logger.dart';
-import 'package:supabase_realtime/src/message.dart';
 import 'package:supabase_realtime/src/retry_timer.dart';
 import 'package:supabase_realtime/src/serializer.dart';
 import 'package:supabase_realtime/src/websocket/websocket.dart';
@@ -23,11 +22,19 @@ typedef WebSocketTransport =
 
 /// Serializes an outgoing message into the `String` or binary frame written to
 /// the WebSocket.
-typedef RealtimeEncode = Object Function(Map<String, dynamic> payload);
+///
+/// The serialization can run on a background isolate: frames are written to
+/// the socket in the order the messages were pushed, even when a later encode
+/// completes first.
+typedef RealtimeEncode = Future<Object> Function(RealtimeMessage message);
 
 /// Deserializes a raw incoming WebSocket frame (`String` or binary) into a
-/// message map.
-typedef RealtimeDecode = Map<String, dynamic> Function(Object payload);
+/// message.
+///
+/// The deserialization can run on a background isolate: messages are
+/// dispatched to the channels in the order the frames were received, even when
+/// a later decode completes first.
+typedef RealtimeDecode = Future<RealtimeMessage> Function(Object frame);
 
 /// Event details for when the connection closed.
 class RealtimeCloseEvent {
@@ -169,8 +176,21 @@ class RealtimeClient {
   @internal
   late RetryTimer reconnectTimer;
   static final Serializer _serializer = Serializer();
-  final RealtimeEncode encode;
-  final RealtimeDecode decode;
+
+  /// Serializes outgoing messages, or `null` to use the built-in codec for
+  /// [version].
+  final RealtimeEncode? encode;
+
+  /// Deserializes incoming frames, or `null` to use the built-in codec for
+  /// [version].
+  final RealtimeDecode? decode;
+
+  /// Codec used while [encode] and [decode] are `null`.
+  ///
+  /// It is synchronous, so a client that does not override the codec writes
+  /// and dispatches without a microtask hop.
+  final Object Function(RealtimeMessage) _builtInEncode;
+  final RealtimeMessage Function(Object) _builtInDecode;
   late TimerCalculation reconnectAfter;
   WebSocketChannel? connection;
   StreamSubscription<dynamic>? _connectionSubscription;
@@ -179,11 +199,23 @@ class RealtimeClient {
 
   final _statusController =
       StreamController<RealtimeConnectionStatusChange>.broadcast();
-  final _messageController = StreamController<Map<String, dynamic>>.broadcast();
+  final _messageController = StreamController<RealtimeMessage>.broadcast();
 
   final _heartbeatController =
       StreamController<RealtimeHeartbeatStatus>.broadcast();
 
+  /// The most recent write that is waiting on an asynchronous [encode], or
+  /// `null` when every pushed message has reached the socket.
+  ///
+  /// Encoding starts as soon as the message is pushed, only the write to the
+  /// sink is chained, so that a slow encode does not hold back the ones after
+  /// it any longer than the ordering requires.
+  Future<void>? _pendingWrite;
+
+  /// The most recent dispatch that is waiting on an asynchronous [decode], or
+  /// `null` when every received frame has been dispatched.
+  Future<void>? _pendingDispatch;
+
   /// The current state of the socket, or `null` before the first [connect].
   SocketState? connectionState;
   Future<String?> Function()? customAccessToken;
@@ -195,7 +227,10 @@ class RealtimeClient {
   ///
   /// [transport] The Websocket Transport, for example WebSocket.
   ///
-  /// [timeout] The default timeout to trigger push timeouts.
+  /// [timeout] The default timeout to trigger push timeouts. Also bounds a
+  /// custom [encode] or [decode] call, so one that never completes fails
+  /// after [timeout] instead of stalling every write or dispatch chained
+  /// after it.
   ///
   /// [connectionCloseTimeout] The timeout to wait for the connection to close
   /// before dismissing the result. Defaults to 6 seconds.
@@ -213,7 +248,7 @@ class RealtimeClient {
   /// the heartbeat interval.
   ///
   /// [encode] Overrides how outgoing messages are serialized, for example to
-  /// use a faster JSON implementation. Defaults to the codec for [version].
+  /// serialize on a background isolate. Defaults to the codec for [version].
   ///
   /// [decode] Overrides how incoming frames are deserialized. Defaults to the
   /// codec for [version].
@@ -235,8 +270,8 @@ class RealtimeClient {
         RealtimeConstants.defaultConnectionCloseTimeout,
     this.heartbeatInterval = RealtimeConstants.defaultHeartbeatInterval,
     Duration? disconnectOnEmptyChannelsAfter,
-    RealtimeEncode? encode,
-    RealtimeDecode? decode,
+    this.encode,
+    this.decode,
     TimerCalculation? reconnectAfter,
     Map<String, String>? headers,
     this.parameters = const {},
@@ -256,16 +291,12 @@ class RealtimeClient {
          ...?headers,
        },
        transport = transport ?? createWebSocketClient,
-       encode =
-           encode ??
-           (version == RealtimeProtocolVersion.v1
-               ? _encodeLegacy
-               : _serializer.encode),
-       decode =
-           decode ??
-           (version == RealtimeProtocolVersion.v1
-               ? _decodeLegacy
-               : _serializer.decode) {
+       _builtInEncode = version == RealtimeProtocolVersion.v1
+           ? _encodeLegacy
+           : _serializer.encode,
+       _builtInDecode = version == RealtimeProtocolVersion.v1
+           ? _decodeLegacy
+           : _serializer.decode {
     realtimeLogger.config(
       'Initialize RealtimeClient with endpoint: '
       '${Uri.parse(this.endpoint).redacted}, timeout: $timeout, '
@@ -411,6 +442,12 @@ class RealtimeClient {
       await _connectionSubscription?.cancel();
       _connectionSubscription = null;
 
+      // Drop the chain so a write or dispatch from the closed connection does
+      // not hold back the next session's; each is dropped on its own once it
+      // resolves, since it no longer matches the current connection.
+      _pendingWrite = null;
+      _pendingDispatch = null;
+
       // remove open handles
       if (heartbeatTimer != null) heartbeatTimer?.cancel();
     }
@@ -453,7 +490,7 @@ class RealtimeClient {
       _statusController.stream;
 
   /// Emits every decoded message received over the WebSocket.
-  Stream<Map<String, dynamic>> get onMessage => _messageController.stream;
+  Stream<RealtimeMessage> get onMessage => _messageController.stream;
 
   /// Emits a status whenever a heartbeat is sent, acknowledged, errors, or
   /// times out.
@@ -534,36 +571,192 @@ 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.
   @internal
-  void push(Message message) {
-    void callback() {
-      connection?.sink.add(encode(message.toJson()));
-    }
-
+  void push(RealtimeMessage message) {
     realtimeLogger.finest(
-      'Push ${message.topic} ${message.event.name} (${message.ref}): '
+      'Push ${message.topic} ${message.event} (${message.ref}): '
       '${redactedPayload(message.payload)}',
     );
 
     if (isConnected) {
-      callback();
+      _write(message);
     } else {
-      sendBuffer.add(callback);
+      sendBuffer.add(() => _write(message));
     }
   }
 
+  /// Encodes [message] and writes it to the socket.
+  ///
+  /// The built-in codec writes straight to the sink. A custom [encode] is
+  /// awaited first, bounded by [timeout] so it cannot stall the chain
+  /// forever, and its write is chained onto [_pendingWrite] so that a fast
+  /// encode never overtakes a slow one that was pushed before it.
+  void _write(RealtimeMessage message) {
+    final connection = this.connection;
+    final encode = this.encode;
+    if (encode == null) {
+      Object frame;
+      try {
+        frame = _builtInEncode(message);
+      } catch (error) {
+        realtimeLogger.warning('Failed to encode message', error);
+        return;
+      }
+      try {
+        connection?.sink.add(frame);
+      } catch (error) {
+        realtimeLogger.warning('Failed to write message', error);
+      }
+      return;
+    }
+
+    final Future<Object> encoded;
+    try {
+      encoded = encode(message).timeout(timeout);
+    } catch (error) {
+      realtimeLogger.warning('Failed to encode message', error);
+      return;
+    }
+
+    final write = _writeWhenReady(connection, _pendingWrite, encoded);
+    _pendingWrite = write;
+    unawaited(
+      write.whenComplete(() {
+        if (identical(_pendingWrite, write)) {
+          _pendingWrite = null;
+        }
+      }),
+    );
+  }
+
+  /// Awaits [encoded] and every write pushed before it, then writes the frame
+  /// to [originConnection] if it is still the current one.
+  ///
+  /// Never completes with an error, so that a failed encode does not stall the
+  /// writes chained after it. A frame whose connection was replaced by a
+  /// reconnect in the meantime is dropped rather than written to the new
+  /// connection.
+  Future<void> _writeWhenReady(
+    WebSocketChannel? originConnection,
+    Future<void>? previousWrite,
+    Future<Object> encoded,
+  ) async {
+    Object? frame;
+    try {
+      frame = await encoded;
+    } catch (error) {
+      realtimeLogger.warning('Failed to encode message', error);
+    }
+
+    // Awaited even when the encode failed, otherwise the next write would
+    // chain onto this one alone and could overtake `previousWrite`.
+    await previousWrite;
+    if (frame == null) {
+      return;
+    }
+
+    if (!identical(originConnection, connection)) {
+      realtimeLogger.finest(
+        'Dropping an encoded frame from a superseded connection',
+      );
+      return;
+    }
+
+    try {
+      originConnection?.sink.add(frame);
+    } catch (error) {
+      realtimeLogger.warning('Failed to write message', error);
+    }
+  }
+
+  /// Decodes [rawMessage] and dispatches it to the channels it belongs to.
+  ///
+  /// The built-in codec dispatches straight away. A custom [decode] is
+  /// awaited first, bounded by [timeout] so it cannot stall the chain
+  /// forever, and its dispatch is chained onto [_pendingDispatch] so that a
+  /// fast decode never overtakes a slow one that was received before it.
   void onConnectionMessage(Object rawMessage) {
-    final Map<String, dynamic> message;
+    final connection = this.connection;
+    final decode = this.decode;
+    if (decode == null) {
+      final RealtimeMessage message;
+      try {
+        message = _builtInDecode(rawMessage);
+      } catch (error) {
+        realtimeLogger.warning('Failed to decode message', error);
+        return;
+      }
+      try {
+        _dispatch(message);
+      } catch (error) {
+        realtimeLogger.warning('Failed to dispatch message', error);
+      }
+      return;
+    }
+
+    final Future<RealtimeMessage> decoded;
+    try {
+      decoded = decode(rawMessage).timeout(timeout);
+    } catch (error) {
+      realtimeLogger.warning('Failed to decode message', error);
+      return;
+    }
+
+    final dispatch = _dispatchWhenReady(connection, _pendingDispatch, decoded);
+    _pendingDispatch = dispatch;
+    unawaited(
+      dispatch.whenComplete(() {
+        if (identical(_pendingDispatch, dispatch)) {
+          _pendingDispatch = null;
+        }
+      }),
+    );
+  }
+
+  /// Awaits [decoded] and every message received before it, then dispatches it
+  /// if [originConnection] is still the current one.
+  ///
+  /// Never completes with an error, so that a failed decode does not stall the
+  /// messages chained after it. A message received on a connection that a
+  /// reconnect has since replaced is dropped rather than dispatched into the
+  /// new session.
+  Future<void> _dispatchWhenReady(
+    WebSocketChannel? originConnection,
+    Future<void>? previousDispatch,
+    Future<RealtimeMessage> decoded,
+  ) async {
+    RealtimeMessage? message;
     try {
-      message = decode(rawMessage);
+      message = await decoded;
     } catch (error) {
       realtimeLogger.warning('Failed to decode message', error);
+    }
+
+    // Awaited even when the decode failed, otherwise the next message would
+    // chain onto this one alone and could overtake `previousDispatch`.
+    await previousDispatch;
+    if (message == null) {
       return;
     }
 
-    final topic = message['topic'] as String;
-    final event = message['event'] as String;
-    final payload = message['payload'];
-    final messageRef = message['ref'] as String?;
+    if (!identical(originConnection, connection)) {
+      realtimeLogger.finest(
+        'Dropping a decoded message from a superseded connection',
+      );
+      return;
+    }
+
+    try {
+      _dispatch(message);
+    } catch (error) {
+      realtimeLogger.warning('Failed to dispatch message', error);
+    }
+  }
+
+  void _dispatch(RealtimeMessage message) {
+    final topic = message.topic;
+    final event = message.event;
+    final payload = message.payload;
+    final messageRef = message.ref;
     if (messageRef != null && messageRef == pendingHeartbeatRef) {
       pendingHeartbeatRef = null;
       final heartbeatStatus = payload is Map ? payload['status'] : null;
@@ -593,11 +786,14 @@ class RealtimeClient {
     _messageController.add(message);
   }
 
-  static Object _encodeLegacy(Map<String, dynamic> message) =>
-      jsonEncode(message);
+  static Object _encodeLegacy(RealtimeMessage message) =>
+      jsonEncode(message.toJson(RealtimeProtocolVersion.v1));
 
-  static Map<String, dynamic> _decodeLegacy(Object rawMessage) =>
-      Map.from(jsonDecode(rawMessage as String) as Map);
+  static RealtimeMessage _decodeLegacy(Object frame) =>
+      RealtimeMessage.fromJson(
+        jsonDecode(frame as String),
+        RealtimeProtocolVersion.v1,
+      );
 
   /// Returns the URL of the websocket.
   String get endpointUrl {
@@ -806,7 +1002,7 @@ class RealtimeClient {
     }
     pendingHeartbeatRef = makeRef();
     push(
-      Message(
+      RealtimeMessage.outgoing(
         topic: 'phoenix',
         event: ChannelEvent.heartbeat,
         payload: {},
diff --git a/packages/supabase_realtime/lib/src/realtime_message.dart b/packages/supabase_realtime/lib/src/realtime_message.dart
new file mode 100644
index 000000000..3d32bac6b
--- /dev/null
+++ b/packages/supabase_realtime/lib/src/realtime_message.dart
@@ -0,0 +1,198 @@
+import 'dart:typed_data';
+
+import 'package:collection/collection.dart';
+import 'package:meta/meta.dart';
+import 'package:supabase_realtime/src/constants.dart';
+import 'package:supabase_realtime/src/types.dart';
+
+/// A message as it travels over the Realtime WebSocket connection.
+///
+/// This is what a custom `RealtimeEncode` is handed and what a custom
+/// `RealtimeDecode` returns. [toJson] and [RealtimeMessage.fromJson] convert
+/// between a message and its shape on the wire for a protocol version, so a
+/// codec only has to turn that shape into bytes and back:
+///
+/// ```dart
+/// RealtimeClient(
+///   url,
+///   encode: (message) => isolate.encode(message.toJson()),
+///   decode: (frame) async =>
+///       RealtimeMessage.fromJson(await isolate.decode(frame as String)),
+/// );
+/// ```
+class RealtimeMessage {
+  /// Reference of the channel join this message belongs to.
+  ///
+  /// `null` for messages that are not tied to a join, such as heartbeats.
+  final String? joinRef;
+
+  /// Reference tying a reply to the message that triggered it.
+  final String? ref;
+
+  /// Topic the message belongs to, for example `realtime:room`.
+  final String topic;
+
+  /// Event name as it appears on the wire, for example `phx_join`.
+  final String event;
+
+  /// Body of the message, `null` when the event carries none.
+  final Object? payload;
+
+  const RealtimeMessage({
+    required this.topic,
+    required this.event,
+    this.payload,
+    this.ref,
+    this.joinRef,
+  });
+
+  /// Builds an outgoing message for a channel [event].
+  ///
+  /// [Binding]s in the payload are replaced by their serializable shape, both
+  /// because they hold a callback that cannot be encoded and because a codec
+  /// running on a background isolate cannot be handed a closure.
+  @internal
+  static RealtimeMessage outgoing({
+    required String topic,
+    required ChannelEvent event,
+    required Object? payload,
+    String? ref,
+    String? joinRef,
+  }) {
+    return RealtimeMessage(
+      topic: topic,
+      // The heartbeat is the one event the server does not expect a `phx_`
+      // prefix on.
+      event: event == ChannelEvent.heartbeat ? 'heartbeat' : event.eventName(),
+      payload: _withoutBindings(payload),
+      ref: ref,
+      joinRef: joinRef,
+    );
+  }
+
+  /// Reads a message from the JSON structure of a [version] frame.
+  ///
+  /// Throws a [FormatException] when [json] does not have the shape [version]
+  /// prescribes.
+  factory RealtimeMessage.fromJson(
+    Object? json, [
+    RealtimeProtocolVersion version = RealtimeProtocolVersion.v2,
+  ]) {
+    return switch (version) {
+      RealtimeProtocolVersion.v1 => _fromObject(json),
+      RealtimeProtocolVersion.v2 => _fromArray(json),
+    };
+  }
+
+  /// The JSON structure of this message for [version].
+  ///
+  /// Protocol `2.0.0` puts the fields in a positional array, the legacy
+  /// `1.0.0` protocol uses an object and leaves out the references it has none
+  /// of.
+  Object toJson([
+    RealtimeProtocolVersion version = RealtimeProtocolVersion.v2,
+  ]) {
+    return switch (version) {
+      RealtimeProtocolVersion.v1 => {
+        'topic': topic,
+        'event': event,
+        'payload': payload,
+        'ref': ?ref,
+        'join_ref': ?joinRef,
+      },
+      RealtimeProtocolVersion.v2 => [joinRef, ref, topic, event, payload],
+    };
+  }
+
+  static RealtimeMessage _fromObject(Object? json) {
+    if (json is! Map) {
+      throw FormatException('Invalid 1.0.0 message', json);
+    }
+    final topic = json['topic'];
+    final event = json['event'];
+    final ref = json['ref'];
+    final joinRef = json['join_ref'];
+    if (topic is! String ||
+        event is! String ||
+        ref is! String? ||
+        joinRef is! String?) {
+      throw FormatException('Invalid 1.0.0 message', json);
+    }
+    return RealtimeMessage(
+      topic: topic,
+      event: event,
+      payload: json['payload'],
+      ref: ref,
+      joinRef: joinRef,
+    );
+  }
+
+  static RealtimeMessage _fromArray(Object? json) {
+    if (json is! List || json.length < 5) {
+      throw FormatException('Invalid 2.0.0 message', json);
+    }
+    final joinRef = json[0];
+    final ref = json[1];
+    final topic = json[2];
+    final event = json[3];
+    if (joinRef is! String? ||
+        ref is! String? ||
+        topic is! String ||
+        event is! String) {
+      throw FormatException('Invalid 2.0.0 message', json);
+    }
+    return RealtimeMessage(
+      joinRef: joinRef,
+      ref: ref,
+      topic: topic,
+      event: event,
+      payload: json[4],
+    );
+  }
+
+  static Object? _withoutBindings(Object? payload) {
+    if (payload is Binding) {
+      return {'type': payload.type, 'filter': payload.filter};
+    }
+    // Binary payloads (Uint8List and other TypedData/ByteBuffer) are opaque
+    // bytes for a binary broadcast, not a JSON list to recurse into.
+    if (payload is TypedData || payload is ByteBuffer) {
+      return payload;
+    }
+    if (payload is Map) {
+      return {
+        for (final entry in payload.entries)
+          entry.key: _withoutBindings(entry.value),
+      };
+    }
+    if (payload is List) {
+      return [for (final item in payload) _withoutBindings(item)];
+    }
+    return payload;
+  }
+
+  @override
+  bool operator ==(Object other) {
+    return other is RealtimeMessage &&
+        other.joinRef == joinRef &&
+        other.ref == ref &&
+        other.topic == topic &&
+        other.event == event &&
+        const DeepCollectionEquality().equals(other.payload, payload);
+  }
+
+  @override
+  int get hashCode => Object.hash(
+    joinRef,
+    ref,
+    topic,
+    event,
+    const DeepCollectionEquality().hash(payload),
+  );
+
+  @override
+  String toString() {
+    return 'RealtimeMessage(joinRef: $joinRef, ref: $ref, topic: $topic, '
+        'event: $event, payload: $payload)';
+  }
+}
diff --git a/packages/supabase_realtime/lib/src/serializer.dart b/packages/supabase_realtime/lib/src/serializer.dart
index c4096e88f..c98aab4b6 100644
--- a/packages/supabase_realtime/lib/src/serializer.dart
+++ b/packages/supabase_realtime/lib/src/serializer.dart
@@ -1,6 +1,8 @@
 import 'dart:convert';
 import 'dart:typed_data';
+
 import 'package:meta/meta.dart';
+import 'package:supabase_realtime/src/realtime_message.dart';
 
 /// Encodes and decodes Realtime protocol `2.0.0` frames.
 ///
@@ -40,44 +42,24 @@ class Serializer {
   const Serializer({List<String>? allowedMetadataKeys})
     : allowedMetadataKeys = allowedMetadataKeys ?? const [];
 
-  /// Encodes a message map into the string or binary representation that is
+  /// Encodes a message 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 &&
+  Object encode(RealtimeMessage 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,
-    ]);
+    return jsonEncode(message.toJson());
   }
 
-  /// Decodes a raw WebSocket frame into a message map with `join_ref`, `ref`,
-  /// `topic`, `event` and `payload` keys.
-  Map<String, dynamic> decode(Object rawPayload) {
+  /// Decodes a raw WebSocket frame into a message.
+  RealtimeMessage 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 {
-        'join_ref': decoded[0],
-        'ref': decoded[1],
-        'topic': decoded[2],
-        'event': decoded[3],
-        'payload': decoded[4],
-      };
+      return RealtimeMessage.fromJson(jsonDecode(rawPayload));
     }
 
     final bytes = _asBytes(rawPayload);
@@ -85,11 +67,11 @@ class Serializer {
       return _binaryDecode(bytes);
     }
 
-    return {};
+    throw FormatException('Unsupported 2.0.0 frame', rawPayload);
   }
 
   Uint8List _encodeBinaryUserBroadcastPush(
-    Map<String, dynamic> message,
+    RealtimeMessage message,
     Map<dynamic, dynamic> payload,
   ) {
     final encodedPayload = _asBytes(payload['payload'])!;
@@ -103,9 +85,9 @@ class Serializer {
     // and the decode side uses utf8.decode, so measuring with String.length
     // (UTF-16 code units) and writing one byte per unit would corrupt any
     // multi-byte character (e.g. accents or emoji) and desynchronize the frame.
-    final topic = utf8.encode((message['topic'] ?? '') as String);
-    final ref = utf8.encode((message['ref'] ?? '') as String);
-    final joinRef = utf8.encode((message['join_ref'] ?? '') as String);
+    final topic = utf8.encode(message.topic);
+    final ref = utf8.encode(message.ref ?? '');
+    final joinRef = utf8.encode(message.joinRef ?? '');
     final userEvent = utf8.encode(payload['event'] as String);
     final metadata = utf8.encode(metadataString);
 
@@ -142,16 +124,16 @@ class Serializer {
     return frame;
   }
 
-  Map<String, dynamic> _binaryDecode(Uint8List buffer) {
+  RealtimeMessage _binaryDecode(Uint8List buffer) {
     final view = ByteData.sublistView(buffer);
     final kind = view.getUint8(0);
     return switch (kind) {
       kindUserBroadcast => _decodeUserBroadcast(buffer, view),
-      _ => {},
+      _ => throw FormatException('Unknown 2.0.0 binary frame kind $kind'),
     };
   }
 
-  Map<String, dynamic> _decodeUserBroadcast(Uint8List buffer, ByteData view) {
+  RealtimeMessage _decodeUserBroadcast(Uint8List buffer, ByteData view) {
     final topicSize = view.getUint8(1);
     final userEventSize = view.getUint8(2);
     final metadataSize = view.getUint8(3);
@@ -187,13 +169,11 @@ class Serializer {
       data['meta'] = jsonDecode(metadata);
     }
 
-    return {
-      'join_ref': null,
-      'ref': null,
-      'topic': topic,
-      'event': broadcastEvent,
-      'payload': data,
-    };
+    return RealtimeMessage(
+      topic: topic,
+      event: broadcastEvent,
+      payload: data,
+    );
   }
 
   void _checkLength(String field, int length) {
diff --git a/packages/supabase_realtime/lib/supabase_realtime.dart b/packages/supabase_realtime/lib/supabase_realtime.dart
index 2321e38a5..829a0cad9 100644
--- a/packages/supabase_realtime/lib/supabase_realtime.dart
+++ b/packages/supabase_realtime/lib/supabase_realtime.dart
@@ -7,6 +7,7 @@ export 'src/constants.dart'
 export 'src/realtime_channel.dart';
 export 'src/realtime_client.dart';
 export 'src/realtime_constants.dart';
+export 'src/realtime_message.dart';
 export 'src/realtime_presence.dart' show Presence;
 export 'src/transformers.dart' show PostgresColumn, PostgresType;
 export 'src/types.dart'
diff --git a/packages/supabase_realtime/test/async_codec_test.dart b/packages/supabase_realtime/test/async_codec_test.dart
new file mode 100644
index 000000000..d8c79bbf9
--- /dev/null
+++ b/packages/supabase_realtime/test/async_codec_test.dart
@@ -0,0 +1,491 @@
+import 'dart:async';
+
+import 'package:mocktail/mocktail.dart';
+import 'package:supabase_realtime/supabase_realtime.dart';
+import 'package:test/test.dart';
+
+import 'socket_test_stubs.dart';
+
+void main() {
+  const socketEndpoint = 'wss://localhost:0/';
+
+  late MockIOWebSocketChannel mockedChannel;
+  late MockWebSocketSink mockedSink;
+  late List<Object?> written;
+
+  setUp(() {
+    mockedChannel = MockIOWebSocketChannel();
+    mockedSink = MockWebSocketSink();
+    written = [];
+
+    when(() => mockedChannel.sink).thenReturn(mockedSink);
+    when(() => mockedChannel.ready).thenAnswer((_) => Future.value());
+    when(() => mockedSink.close()).thenAnswer((_) => Future.value());
+    when(() => mockedSink.add(any())).thenAnswer((invocation) {
+      written.add(invocation.positionalArguments.first);
+    });
+  });
+
+  RealtimeClient createClient({
+    RealtimeEncode? encode,
+    RealtimeDecode? decode,
+    Duration? timeout,
+  }) {
+    final client = RealtimeClient(
+      socketEndpoint,
+      transport: (url, headers) => mockedChannel,
+      encode: encode,
+      decode: decode,
+      timeout: timeout ?? RealtimeConstants.defaultTimeout,
+    );
+    unawaited(client.connect());
+    client.connectionState = SocketState.open;
+    return client;
+  }
+
+  RealtimeMessage messageWithRef(String ref) => RealtimeMessage(
+    topic: 'realtime:room',
+    event: 'broadcast',
+    payload: const {'type': 'broadcast', 'event': 'cursor'},
+    ref: ref,
+  );
+
+  group('asynchronous encode', () {
+    test('writes the frame once the encode completes', () async {
+      final completer = Completer<Object>();
+      final client = createClient(encode: (_) => completer.future);
+
+      client.push(messageWithRef('1'));
+      await pumpEventQueue();
+
+      expect(written, isEmpty);
+
+      completer.complete('frame-1');
+      await pumpEventQueue();
+
+      expect(written, ['frame-1']);
+    });
+
+    test('writes in push order when a later encode completes first', () async {
+      final completers = {
+        '1': Completer<Object>(),
+        '2': Completer<Object>(),
+        '3': Completer<Object>(),
+      };
+      final client = createClient(
+        encode: (message) => completers[message.ref]!.future,
+      );
+
+      for (final ref in completers.keys) {
+        client.push(messageWithRef(ref));
+      }
+
+      completers['3']!.complete('frame-3');
+      completers['2']!.complete('frame-2');
+      await pumpEventQueue();
+
+      expect(written, isEmpty);
+
+      completers['1']!.complete('frame-1');
+      await pumpEventQueue();
+
+      expect(written, ['frame-1', 'frame-2', 'frame-3']);
+    });
+
+    test('an immediate encode waits for the messages before it', () async {
+      final completer = Completer<Object>();
+      final client = createClient(
+        encode: (message) => message.ref == '1'
+            ? completer.future
+            : Future.value('frame-${message.ref}'),
+      );
+
+      client.push(messageWithRef('1'));
+      client.push(messageWithRef('2'));
+      await pumpEventQueue();
+
+      expect(written, isEmpty);
+
+      completer.complete('frame-1');
+      await pumpEventQueue();
+
+      expect(written, ['frame-1', 'frame-2']);
+    });
+
+    test('a failed encode drops only its own message', () async {
+      final client = createClient(
+        encode: (message) => message.ref == '1'
+            ? Future<Object>.error(StateError('encode failed'))
+            : Future<Object>.value('frame-${message.ref}'),
+      );
+
+      client.push(messageWithRef('1'));
+      client.push(messageWithRef('2'));
+      await pumpEventQueue();
+
+      expect(written, ['frame-2']);
+    });
+
+    test(
+      'a synchronously throwing encode drops only its own message',
+      () async {
+        final client = createClient(
+          encode: (message) => message.ref == '1'
+              ? throw StateError('encode failed')
+              : Future<Object>.value('frame-${message.ref}'),
+        );
+
+        client.push(messageWithRef('1'));
+        client.push(messageWithRef('2'));
+        await pumpEventQueue();
+
+        expect(written, ['frame-2']);
+      },
+    );
+
+    test('buffered messages are written in order once connected', () async {
+      final completer = Completer<Object>();
+      final client = createClient(
+        encode: (message) => message.ref == '1'
+            ? completer.future
+            : Future.value('frame-${message.ref}'),
+      );
+      client.connectionState = SocketState.connecting;
+
+      client.push(messageWithRef('1'));
+      client.push(messageWithRef('2'));
+
+      expect(client.sendBuffer, hasLength(2));
+
+      client.connectionState = SocketState.open;
+      for (final callback in client.sendBuffer) {
+        callback();
+      }
+      completer.complete('frame-1');
+      await pumpEventQueue();
+
+      expect(written, ['frame-1', 'frame-2']);
+    });
+
+    test(
+      'a custom encode that never completes fails after timeout, without '
+      'stalling later writes',
+      () async {
+        final neverCompletes = Completer<Object>();
+        final client = createClient(
+          encode: (message) => message.ref == '1'
+              ? neverCompletes.future
+              : Future.value('frame-${message.ref}'),
+          timeout: const Duration(milliseconds: 20),
+        );
+
+        client.push(messageWithRef('1'));
+        await pumpEventQueue();
+
+        client.push(messageWithRef('2'));
+        await Future<void>.delayed(const Duration(milliseconds: 100));
+
+        expect(written, ['frame-2']);
+      },
+    );
+
+    test('the built-in codec writes without a microtask hop', () {
+      final client = createClient();
+
+      client.push(messageWithRef('1'));
+
+      expect(written, hasLength(1));
+    });
+
+    test('the built-in codec logs and swallows an encode failure', () {
+      final client = createClient();
+
+      expect(
+        () => client.push(
+          RealtimeMessage(
+            topic: 'realtime:room',
+            event: 'broadcast',
+            payload: {'bad': Object()},
+          ),
+        ),
+        returnsNormally,
+      );
+      expect(written, isEmpty);
+    });
+
+    test('the built-in codec logs and swallows a write failure', () {
+      final client = createClient();
+      when(() => mockedSink.add(any())).thenThrow(StateError('boom'));
+
+      expect(() => client.push(messageWithRef('1')), returnsNormally);
+    });
+
+    test(
+      'drops a pending write if the connection changes before it completes',
+      () async {
+        final completer = Completer<Object>();
+        final writtenPerConnection = <List<Object?>>[];
+
+        final client = RealtimeClient(
+          socketEndpoint,
+          transport: (url, headers) {
+            final channel = MockIOWebSocketChannel();
+            final sink = MockWebSocketSink();
+            final writtenHere = <Object?>[];
+            when(() => channel.sink).thenReturn(sink);
+            when(() => channel.ready).thenAnswer((_) => Future.value());
+            when(() => sink.close()).thenAnswer((_) => Future.value());
+            when(() => sink.add(any())).thenAnswer((invocation) {
+              writtenHere.add(invocation.positionalArguments.first);
+            });
+            writtenPerConnection.add(writtenHere);
+            return channel;
+          },
+          encode: (message) => message.ref == '1'
+              ? completer.future
+              : Future.value('frame-${message.ref}'),
+        );
+        unawaited(client.connect());
+        client.connectionState = SocketState.open;
+
+        client.push(messageWithRef('1'));
+        await pumpEventQueue();
+
+        // Reconnect to a new connection while the encode above is pending.
+        await client.disconnect();
+        unawaited(client.connect());
+        client.connectionState = SocketState.open;
+
+        // The new connection keeps working while the stale encode is
+        // pending, so it is not the encode itself stalling forever.
+        client.push(messageWithRef('2'));
+        await pumpEventQueue();
+
+        completer.complete('frame-1');
+        await pumpEventQueue();
+
+        expect(writtenPerConnection, hasLength(2));
+        expect(writtenPerConnection[0], isEmpty);
+        expect(writtenPerConnection[1], ['frame-2']);
+      },
+    );
+  });
+
+  group('asynchronous decode', () {
+    RealtimeMessage frameWithRef(String ref) => RealtimeMessage(
+      topic: 'realtime:room',
+      event: 'broadcast',
+      payload: const {'type': 'broadcast', 'event': 'cursor'},
+      ref: ref,
+    );
+
+    test('dispatches the message once the decode completes', () async {
+      final completer = Completer<RealtimeMessage>();
+      final client = createClient(decode: (_) => completer.future);
+
+      final received = <String?>[];
+      client.onMessage.listen((message) => received.add(message.ref));
+
+      client.onConnectionMessage('raw-1');
+      await pumpEventQueue();
+
+      expect(received, isEmpty);
+
+      completer.complete(frameWithRef('1'));
+      await pumpEventQueue();
+
+      expect(received, ['1']);
+    });
+
+    test(
+      'dispatches in receive order when a later decode completes first',
+      () async {
+        final completers = {
+          '1': Completer<RealtimeMessage>(),
+          '2': Completer<RealtimeMessage>(),
+          '3': Completer<RealtimeMessage>(),
+        };
+        final client = createClient(
+          decode: (rawMessage) => completers[rawMessage]!.future,
+        );
+
+        final received = <String?>[];
+        client.onMessage.listen((message) => received.add(message.ref));
+
+        for (final ref in completers.keys) {
+          client.onConnectionMessage(ref);
+        }
+
+        completers['3']!.complete(frameWithRef('3'));
+        completers['2']!.complete(frameWithRef('2'));
+        await pumpEventQueue();
+
+        expect(received, isEmpty);
+
+        completers['1']!.complete(frameWithRef('1'));
+        await pumpEventQueue();
+
+        expect(received, ['1', '2', '3']);
+      },
+    );
+
+    test('an immediate decode waits for the frames before it', () async {
+      final completer = Completer<RealtimeMessage>();
+      final client = createClient(
+        decode: (rawMessage) => rawMessage == '1'
+            ? completer.future
+            : Future.value(frameWithRef(rawMessage as String)),
+      );
+
+      final received = <String?>[];
+      client.onMessage.listen((message) => received.add(message.ref));
+
+      client.onConnectionMessage('1');
+      client.onConnectionMessage('2');
+      await pumpEventQueue();
+
+      expect(received, isEmpty);
+
+      completer.complete(frameWithRef('1'));
+      await pumpEventQueue();
+
+      expect(received, ['1', '2']);
+    });
+
+    test('a failed decode drops only its own frame', () async {
+      final client = createClient(
+        decode: (rawMessage) => rawMessage == '1'
+            ? Future<RealtimeMessage>.error(FormatException('decode failed'))
+            : Future.value(frameWithRef(rawMessage as String)),
+      );
+
+      final received = <String?>[];
+      client.onMessage.listen((message) => received.add(message.ref));
+
+      client.onConnectionMessage('1');
+      client.onConnectionMessage('2');
+      await pumpEventQueue();
+
+      expect(received, ['2']);
+    });
+
+    test('a synchronously throwing decode drops only its own frame', () async {
+      final client = createClient(
+        decode: (rawMessage) => rawMessage == '1'
+            ? throw FormatException('decode failed')
+            : Future.value(frameWithRef(rawMessage as String)),
+      );
+
+      final received = <String?>[];
+      client.onMessage.listen((message) => received.add(message.ref));
+
+      client.onConnectionMessage('1');
+      client.onConnectionMessage('2');
+      await pumpEventQueue();
+
+      expect(received, ['2']);
+    });
+
+    test(
+      'a custom decode that never completes fails after timeout, without '
+      'stalling later dispatches',
+      () async {
+        final neverCompletes = Completer<RealtimeMessage>();
+        final client = createClient(
+          decode: (rawMessage) => rawMessage == '1'
+              ? neverCompletes.future
+              : Future.value(frameWithRef(rawMessage as String)),
+          timeout: const Duration(milliseconds: 20),
+        );
+
+        final received = <String?>[];
+        client.onMessage.listen((message) => received.add(message.ref));
+
+        client.onConnectionMessage('1');
+        await pumpEventQueue();
+
+        client.onConnectionMessage('2');
+        await Future<void>.delayed(const Duration(milliseconds: 100));
+
+        expect(received, ['2']);
+      },
+    );
+
+    test('the built-in codec dispatches without a microtask hop', () {
+      final client = createClient();
+
+      final channel = MockChannel();
+      when(() => channel.isMember('realtime:room')).thenReturn(true);
+      client.channels.add(channel);
+
+      client.onConnectionMessage('[null,"1","realtime:room","broadcast",{}]');
+
+      verify(() => channel.trigger('broadcast', any(), '1')).called(1);
+    });
+
+    test('the built-in codec logs and swallows a dispatch failure', () {
+      final client = createClient();
+
+      final channel = MockChannel();
+      when(() => channel.isMember('realtime:room')).thenReturn(true);
+      when(
+        () => channel.trigger(any(), any(), any()),
+      ).thenThrow(StateError('boom'));
+      client.channels.add(channel);
+
+      expect(
+        () => client.onConnectionMessage(
+          '[null,"1","realtime:room","broadcast",{}]',
+        ),
+        returnsNormally,
+      );
+    });
+
+    test(
+      'drops a pending dispatch if the connection changes before it '
+      'completes',
+      () async {
+        final completer = Completer<RealtimeMessage>();
+
+        final client = RealtimeClient(
+          socketEndpoint,
+          transport: (url, headers) {
+            final channel = MockIOWebSocketChannel();
+            final sink = MockWebSocketSink();
+            when(() => channel.sink).thenReturn(sink);
+            when(() => channel.ready).thenAnswer((_) => Future.value());
+            when(() => sink.close()).thenAnswer((_) => Future.value());
+            when(() => sink.add(any())).thenAnswer((_) {});
+            return channel;
+          },
+          decode: (rawMessage) => rawMessage == 'raw-1'
+              ? completer.future
+              : Future.value(frameWithRef(rawMessage as String)),
+        );
+        unawaited(client.connect());
+        client.connectionState = SocketState.open;
+
+        final received = <String?>[];
+        client.onMessage.listen((message) => received.add(message.ref));
+
+        client.onConnectionMessage('raw-1');
+        await pumpEventQueue();
+
+        // Reconnect to a new connection while the decode above is pending.
+        await client.disconnect();
+        unawaited(client.connect());
+        client.connectionState = SocketState.open;
+
+        // The new connection keeps dispatching while the stale decode is
+        // pending, so it is not the decode itself stalling forever.
+        client.onConnectionMessage('2');
+        await pumpEventQueue();
+
+        completer.complete(frameWithRef('1'));
+        await pumpEventQueue();
+
+        expect(received, ['2']);
+      },
+    );
+  });
+}
diff --git a/packages/supabase_realtime/test/heartbeat_test.dart b/packages/supabase_realtime/test/heartbeat_test.dart
index 7776bdf28..feaef0410 100644
--- a/packages/supabase_realtime/test/heartbeat_test.dart
+++ b/packages/supabase_realtime/test/heartbeat_test.dart
@@ -12,8 +12,10 @@ void main() {
   setUp(() {
     client = RealtimeClient(
       'wss://localhost:0/',
-      decode: (rawMessage) =>
-          Map<String, dynamic>.from(jsonDecode(rawMessage as String) as Map),
+      decode: (rawMessage) async => RealtimeMessage.fromJson(
+        jsonDecode(rawMessage as String),
+        RealtimeProtocolVersion.v1,
+      ),
     );
     statuses = [];
     subscription = client.onHeartbeat.listen(statuses.add);
diff --git a/packages/supabase_realtime/test/message_test.dart b/packages/supabase_realtime/test/message_test.dart
deleted file mode 100644
index c9808bd25..000000000
--- a/packages/supabase_realtime/test/message_test.dart
+++ /dev/null
@@ -1,123 +0,0 @@
-import 'package:supabase_realtime/src/constants.dart';
-import 'package:supabase_realtime/src/message.dart';
-import 'package:test/test.dart';
-
-void main() {
-  group('Message', () {
-    test('message with null refs serializes correctly', () {
-      final message = Message(
-        topic: 'phoenix',
-        event: ChannelEvent.heartbeat,
-        payload: {},
-        ref: null,
-        joinRef: null,
-      );
-      final json = message.toJson();
-      expect(json['ref'], isNull);
-      expect(json['join_ref'], isNull);
-      expect(json['topic'], equals('phoenix'));
-      expect(json['event'], equals('heartbeat'));
-      expect(json['payload'], equals({}));
-    });
-
-    test('heartbeat message with null joinRef', () {
-      final message = Message(
-        topic: 'phoenix',
-        event: ChannelEvent.heartbeat,
-        payload: {},
-        ref: '123',
-        joinRef: null,
-      );
-      final json = message.toJson();
-      expect(json['ref'], equals('123'));
-      expect(json.containsKey('join_ref'), isFalse);
-      expect(json['topic'], equals('phoenix'));
-      expect(json['event'], equals('heartbeat'));
-    });
-
-    test('message with null ref but valid joinRef', () {
-      final message = Message(
-        topic: 'room:lobby',
-        event: ChannelEvent.join,
-        payload: {'user_id': '123'},
-        ref: null,
-        joinRef: 'join-456',
-      );
-      final json = message.toJson();
-      expect(json.containsKey('ref'), isFalse);
-      expect(json['join_ref'], equals('join-456'));
-      expect(json['topic'], equals('room:lobby'));
-      expect(json['payload'], equals({'user_id': '123'}));
-    });
-
-    test('message with both ref and joinRef', () {
-      final message = Message(
-        topic: 'room:lobby',
-        event: ChannelEvent.join,
-        payload: {'user_id': '123'},
-        ref: 'ref-789',
-        joinRef: 'join-456',
-      );
-      final json = message.toJson();
-      expect(json['ref'], equals('ref-789'));
-      expect(json['join_ref'], equals('join-456'));
-      expect(json['topic'], equals('room:lobby'));
-      expect(json['payload'], equals({'user_id': '123'}));
-    });
-
-    test('message with ref but null joinRef', () {
-      final message = Message(
-        topic: 'room:lobby',
-        event: ChannelEvent.leave,
-        payload: {},
-        ref: 'ref-999',
-        joinRef: null,
-      );
-      final json = message.toJson();
-      expect(json['ref'], equals('ref-999'));
-      expect(json.containsKey('join_ref'), isFalse);
-      expect(json['topic'], equals('room:lobby'));
-    });
-
-    test('ref parameter is optional in constructor', () {
-      final message = Message(
-        topic: 'phoenix',
-        event: ChannelEvent.heartbeat,
-        payload: {},
-        // ref is not provided, should be null
-      );
-      expect(message.ref, isNull);
-      final json = message.toJson();
-      expect(json.containsKey('ref'), isFalse);
-    });
-
-    test('a nested empty map is preserved, not dropped', () {
-      final message = Message(
-        topic: 'room:lobby',
-        event: ChannelEvent.presence,
-        payload: {'event': 'track', 'payload': <String, dynamic>{}},
-      );
-      final json = message.toJson();
-      expect(json['payload'], equals({'event': 'track', 'payload': {}}));
-    });
-
-    test('nested non-empty maps still serialize correctly', () {
-      final message = Message(
-        topic: 'room:lobby',
-        event: ChannelEvent.presence,
-        payload: {
-          'event': 'track',
-          'payload': {'name': 'alice'},
-        },
-      );
-      final json = message.toJson();
-      expect(
-        json['payload'],
-        equals({
-          'event': 'track',
-          'payload': {'name': 'alice'},
-        }),
-      );
-    });
-  });
-}
diff --git a/packages/supabase_realtime/test/realtime_integration_test.dart b/packages/supabase_realtime/test/realtime_integration_test.dart
index d6a808e96..7e80fd006 100644
--- a/packages/supabase_realtime/test/realtime_integration_test.dart
+++ b/packages/supabase_realtime/test/realtime_integration_test.dart
@@ -2,6 +2,8 @@
 library;
 
 import 'dart:async';
+import 'dart:convert';
+import 'dart:isolate';
 
 import 'package:postgres/postgres.dart';
 import 'package:supabase_realtime/supabase_realtime.dart';
@@ -324,6 +326,91 @@ void main() {
       });
     });
   }
+
+  group('asynchronous codec', () {
+    late RealtimeClient client;
+
+    setUp(() {
+      client = RealtimeClient(
+        localStackRealtimeUrl,
+        parameters: {'apikey': generateRealtimeToken()},
+        heartbeatInterval: const Duration(seconds: 5),
+        encode: _encodeOnBackgroundIsolate,
+        decode: _decodeOnBackgroundIsolate,
+      );
+    });
+
+    tearDown(() async {
+      await client.removeAllChannels();
+      await client.disconnect();
+    });
+
+    test('round trips broadcasts in order', () async {
+      final channel = client.channel(
+        'async-codec',
+        const RealtimeChannelConfig(self: true),
+      );
+
+      final received = <int>[];
+      final allReceived = Completer<void>();
+      channel.onBroadcast(event: 'tick').listen((payload) {
+        received.add(payload['payload']['index'] as int);
+        if (received.length == _asyncCodecMessageCount &&
+            !allReceived.isCompleted) {
+          allReceived.complete();
+        }
+      });
+      await _subscribe(channel);
+
+      for (var index = 0; index < _asyncCodecMessageCount; index++) {
+        await channel.sendBroadcastMessage(
+          event: 'tick',
+          payload: {
+            'payload': {'index': index},
+          },
+        );
+      }
+
+      await allReceived.future.timeout(const Duration(seconds: 20));
+      expect(
+        received,
+        List.generate(_asyncCodecMessageCount, (index) => index),
+      );
+    });
+  });
+}
+
+/// Number of broadcasts the asynchronous codec test sends.
+const _asyncCodecMessageCount = 10;
+
+/// Encodes a protocol `2.0.0` text frame on a background isolate.
+Future<Object> _encodeOnBackgroundIsolate(RealtimeMessage message) async {
+  final json = message.toJson();
+  await _delayInReverseOrder(message.payload);
+  return Isolate.run(() => jsonEncode(json));
+}
+
+/// Decodes a protocol `2.0.0` text frame on a background isolate.
+Future<RealtimeMessage> _decodeOnBackgroundIsolate(Object frame) async {
+  final json = await Isolate.run(() => jsonDecode(frame as String));
+  final message = RealtimeMessage.fromJson(json);
+  await _delayInReverseOrder(message.payload);
+  return message;
+}
+
+/// Holds back a broadcast for longer the earlier its index is.
+///
+/// The codec calls then complete in the opposite order of the messages, so the
+/// client has to restore the push order before writing to the socket and the
+/// receive order before dispatching to the channel.
+Future<void> _delayInReverseOrder(Object? payload) async {
+  final index = payload is Map ? (payload['payload']?['index'] as int?) : null;
+  if (index == null) {
+    return;
+  }
+  await Future<void>.delayed(
+    Duration(milliseconds: 20 * (_asyncCodecMessageCount - index)),
+  );
 }
 
 /// Subscribes to [channel] and resolves with the terminal subscribe status.
diff --git a/packages/supabase_realtime/test/realtime_message_test.dart b/packages/supabase_realtime/test/realtime_message_test.dart
new file mode 100644
index 000000000..8085752d5
--- /dev/null
+++ b/packages/supabase_realtime/test/realtime_message_test.dart
@@ -0,0 +1,293 @@
+import 'dart:typed_data';
+
+import 'package:supabase_realtime/supabase_realtime.dart';
+import 'package:supabase_realtime/src/constants.dart';
+import 'package:supabase_realtime/src/types.dart';
+import 'package:test/test.dart';
+
+void main() {
+  group('outgoing', () {
+    test('names a heartbeat without the phx prefix', () {
+      final message = RealtimeMessage.outgoing(
+        topic: 'phoenix',
+        event: ChannelEvent.heartbeat,
+        payload: <String, dynamic>{},
+        ref: '1',
+      );
+
+      expect(message.event, 'heartbeat');
+      expect(message.topic, 'phoenix');
+      expect(message.ref, '1');
+      expect(message.joinRef, isNull);
+    });
+
+    test('names every other event as the server expects it', () {
+      final message = RealtimeMessage.outgoing(
+        topic: 'realtime:room',
+        event: ChannelEvent.join,
+        payload: {'user_id': '123'},
+        joinRef: 'join-456',
+      );
+
+      expect(message.event, 'phx_join');
+      expect(message.payload, {'user_id': '123'});
+      expect(message.joinRef, 'join-456');
+      expect(message.ref, isNull);
+    });
+
+    test('replaces a nested binding with its serializable shape', () {
+      final binding = Binding('postgres_changes', {'event': '*'}, (_, [_]) {});
+      final message = RealtimeMessage.outgoing(
+        topic: 'realtime:room',
+        event: ChannelEvent.join,
+        payload: {
+          'config': {'postgres_changes': binding},
+        },
+      );
+
+      expect(message.payload, {
+        'config': {
+          'postgres_changes': {
+            'type': 'postgres_changes',
+            'filter': {'event': '*'},
+          },
+        },
+      });
+    });
+
+    test('preserves a nested empty map', () {
+      final message = RealtimeMessage.outgoing(
+        topic: 'realtime:room',
+        event: ChannelEvent.presence,
+        payload: {'event': 'track', 'payload': <String, dynamic>{}},
+      );
+
+      expect(message.payload, {'event': 'track', 'payload': {}});
+    });
+
+    test('leaves a binary payload as Uint8List', () {
+      final bytes = Uint8List.fromList([1, 2, 3]);
+      final message = RealtimeMessage.outgoing(
+        topic: 'realtime:room',
+        event: ChannelEvent.broadcast,
+        payload: {'type': 'broadcast', 'event': 'file', 'payload': bytes},
+      );
+
+      expect((message.payload as Map)['payload'], isA<Uint8List>());
+      expect((message.payload as Map)['payload'], same(bytes));
+    });
+
+    test('replaces a binding nested inside a list', () {
+      final binding = Binding('postgres_changes', {'event': '*'}, (_, [_]) {});
+      final message = RealtimeMessage.outgoing(
+        topic: 'realtime:room',
+        event: ChannelEvent.join,
+        payload: {
+          'config': {
+            'postgres_changes': [binding],
+          },
+        },
+      );
+
+      expect(message.payload, {
+        'config': {
+          'postgres_changes': [
+            {
+              'type': 'postgres_changes',
+              'filter': {'event': '*'},
+            },
+          ],
+        },
+      });
+    });
+
+    test('replaces a binding nested three levels deep', () {
+      final binding = Binding('postgres_changes', {'event': '*'}, (_, [_]) {});
+      final message = RealtimeMessage.outgoing(
+        topic: 'realtime:room',
+        event: ChannelEvent.join,
+        payload: {
+          'config': {
+            'postgres_changes': {'0': binding},
+          },
+        },
+      );
+
+      expect(message.payload, {
+        'config': {
+          'postgres_changes': {
+            '0': {
+              'type': 'postgres_changes',
+              'filter': {'event': '*'},
+            },
+          },
+        },
+      });
+    });
+  });
+
+  group('toJson', () {
+    test('lays the fields out positionally for 2.0.0', () {
+      const message = RealtimeMessage(
+        topic: 'realtime:room',
+        event: 'phx_join',
+        payload: {'foo': 'bar'},
+        ref: '2',
+        joinRef: '1',
+      );
+
+      expect(message.toJson(), [
+        '1',
+        '2',
+        'realtime:room',
+        'phx_join',
+        {'foo': 'bar'},
+      ]);
+    });
+
+    test('keeps missing references positional for 2.0.0', () {
+      const message = RealtimeMessage(
+        topic: 'phoenix',
+        event: 'heartbeat',
+        payload: <String, dynamic>{},
+      );
+
+      expect(message.toJson(), [null, null, 'phoenix', 'heartbeat', {}]);
+    });
+
+    test('uses an object for 1.0.0', () {
+      const message = RealtimeMessage(
+        topic: 'realtime:room',
+        event: 'phx_join',
+        payload: {'foo': 'bar'},
+        ref: '2',
+        joinRef: '1',
+      );
+
+      expect(message.toJson(RealtimeProtocolVersion.v1), {
+        'topic': 'realtime:room',
+        'event': 'phx_join',
+        'payload': {'foo': 'bar'},
+        'ref': '2',
+        'join_ref': '1',
+      });
+    });
+
+    test('leaves out the references it has none of for 1.0.0', () {
+      const message = RealtimeMessage(
+        topic: 'phoenix',
+        event: 'heartbeat',
+        payload: <String, dynamic>{},
+        ref: '1',
+      );
+
+      final json = message.toJson(RealtimeProtocolVersion.v1) as Map;
+
+      expect(json['ref'], '1');
+      expect(json.containsKey('join_ref'), isFalse);
+    });
+  });
+
+  group('fromJson', () {
+    test('reads a positional 2.0.0 frame', () {
+      final message = RealtimeMessage.fromJson([
+        '1',
+        '2',
+        'realtime:room',
+        'phx_reply',
+        {'status': 'ok'},
+      ]);
+
+      expect(message.joinRef, '1');
+      expect(message.ref, '2');
+      expect(message.topic, 'realtime:room');
+      expect(message.event, 'phx_reply');
+      expect(message.payload, {'status': 'ok'});
+    });
+
+    test('reads an object 1.0.0 frame', () {
+      final message = RealtimeMessage.fromJson({
+        'topic': 'realtime:room',
+        'event': 'phx_reply',
+        'payload': {'status': 'ok'},
+        'ref': '2',
+      }, RealtimeProtocolVersion.v1);
+
+      expect(message.topic, 'realtime:room');
+      expect(message.event, 'phx_reply');
+      expect(message.payload, {'status': 'ok'});
+      expect(message.ref, '2');
+      expect(message.joinRef, isNull);
+    });
+
+    test('throws when the frame does not match the protocol version', () {
+      expect(
+        () => RealtimeMessage.fromJson({'not': 'an array'}),
+        throwsFormatException,
+      );
+      expect(
+        () => RealtimeMessage.fromJson(['too', 'short']),
+        throwsFormatException,
+      );
+      expect(
+        () => RealtimeMessage.fromJson([], RealtimeProtocolVersion.v1),
+        throwsFormatException,
+      );
+    });
+
+    test('throws a FormatException when a field has the wrong type', () {
+      expect(
+        () => RealtimeMessage.fromJson({
+          'topic': 1,
+          'event': 'phx_reply',
+        }, RealtimeProtocolVersion.v1),
+        throwsFormatException,
+      );
+      expect(
+        () => RealtimeMessage.fromJson({
+          'topic': 'realtime:room',
+          'event': 'phx_reply',
+          'ref': 2,
+        }, RealtimeProtocolVersion.v1),
+        throwsFormatException,
+      );
+      expect(
+        () => RealtimeMessage.fromJson([
+          '1',
+          '2',
+          'realtime:room',
+          3,
+          {'status': 'ok'},
+        ]),
+        throwsFormatException,
+      );
+      expect(
+        () => RealtimeMessage.fromJson([
+          1,
+          '2',
+          'realtime:room',
+          'phx_reply',
+          {'status': 'ok'},
+        ]),
+        throwsFormatException,
+      );
+    });
+
+    test('round trips through toJson for every protocol version', () {
+      const message = RealtimeMessage(
+        topic: 'realtime:room',
+        event: 'broadcast',
+        payload: {'event': 'cursor'},
+        ref: '2',
+        joinRef: '1',
+      );
+
+      for (final version in RealtimeProtocolVersion.values) {
+        expect(
+          RealtimeMessage.fromJson(message.toJson(version), version),
+          message,
+        );
+      }
+    });
+  });
+}
diff --git a/packages/supabase_realtime/test/serializer_test.dart b/packages/supabase_realtime/test/serializer_test.dart
index 2c13ff7b7..658011c02 100644
--- a/packages/supabase_realtime/test/serializer_test.dart
+++ b/packages/supabase_realtime/test/serializer_test.dart
@@ -2,6 +2,7 @@ import 'dart:convert';
 import 'dart:typed_data';
 
 import 'package:supabase_realtime/src/serializer.dart';
+import 'package:supabase_realtime/supabase_realtime.dart';
 import 'package:test/test.dart';
 
 /// Builds a `kind = userBroadcast` (4) binary frame the same way the server
@@ -43,13 +44,15 @@ void main() {
 
   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'},
-      });
+      final result = serializer.encode(
+        const RealtimeMessage(
+          joinRef: '1',
+          ref: '2',
+          topic: 'realtime:room',
+          event: 'phx_join',
+          payload: {'foo': 'bar'},
+        ),
+      );
 
       expect(result, isA<String>());
       expect(
@@ -65,11 +68,13 @@ void main() {
     });
 
     test('preserves null join_ref and ref positionally', () {
-      final result = serializer.encode({
-        'topic': 'phoenix',
-        'event': 'heartbeat',
-        'payload': <String, dynamic>{},
-      });
+      final result = serializer.encode(
+        const RealtimeMessage(
+          topic: 'phoenix',
+          event: 'heartbeat',
+          payload: <String, dynamic>{},
+        ),
+      );
 
       expect(
         jsonDecode(result as String),
@@ -78,13 +83,15 @@ void main() {
     });
 
     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},
-      });
+      final result = serializer.encode(
+        const RealtimeMessage(
+          joinRef: '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');
@@ -92,7 +99,7 @@ void main() {
   });
 
   group('decode text frames', () {
-    test('decodes a positional JSON array into a message map', () {
+    test('decodes a positional JSON array into a message', () {
       final result = serializer.decode(
         jsonEncode([
           '1',
@@ -103,13 +110,16 @@ void main() {
         ]),
       );
 
-      expect(result, {
-        'join_ref': '1',
-        'ref': '2',
-        'topic': 'realtime:room',
-        'event': 'phx_reply',
-        'payload': {'status': 'ok'},
-      });
+      expect(
+        result,
+        const RealtimeMessage(
+          joinRef: '1',
+          ref: '2',
+          topic: 'realtime:room',
+          event: 'phx_reply',
+          payload: {'status': 'ok'},
+        ),
+      );
     });
 
     test('throws a FormatException on a malformed text frame', () {
@@ -136,11 +146,11 @@ void main() {
 
       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'], {
+      expect(result.joinRef, 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},
@@ -158,33 +168,37 @@ void main() {
       );
 
       final result = serializer.decode(frame);
-      final payload = result['payload'] as Map<String, dynamic>;
+      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>{});
+    test('throws a FormatException on an unknown binary kind', () {
+      expect(
+        () => serializer.decode(Uint8List.fromList([99, 0, 0])),
+        throwsFormatException,
+      );
     });
   });
 
   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,
-        },
-      });
+      final result = serializer.encode(
+        RealtimeMessage(
+          joinRef: '7',
+          ref: '8',
+          topic: 'realtime:room',
+          event: 'broadcast',
+          payload: {
+            'type': 'broadcast',
+            'event': 'file',
+            'payload': payload,
+          },
+        ),
+      );
 
       expect(result, isA<Uint8List>());
       final bytes = result as Uint8List;
@@ -203,17 +217,19 @@ void main() {
 
     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 result = serializerWithMeta.encode(
+        RealtimeMessage(
+          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];
@@ -238,18 +254,20 @@ void main() {
       final topic = 'realtime:café';
       final userEvent = 'café-🎉';
       final payload = Uint8List.fromList([1, 2, 3]);
-      final result = serializerWithMeta.encode({
-        'join_ref': '10',
-        'ref': '1',
-        'topic': topic,
-        'event': 'broadcast',
-        'payload': {
-          'type': 'broadcast',
-          'event': userEvent,
-          'label': 'naïve',
-          'payload': payload,
-        },
-      });
+      final result = serializerWithMeta.encode(
+        RealtimeMessage(
+          joinRef: '10',
+          ref: '1',
+          topic: topic,
+          event: 'broadcast',
+          payload: {
+            'type': 'broadcast',
+            'event': userEvent,
+            'label': 'naïve',
+            'payload': payload,
+          },
+        ),
+      );
 
       final bytes = result as Uint8List;
       final joinRefBytes = utf8.encode('10');
@@ -294,15 +312,17 @@ void main() {
     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]),
-          },
-        }),
+        () => serializer.encode(
+          RealtimeMessage(
+            topic: longTopic,
+            event: 'broadcast',
+            payload: {
+              'type': 'broadcast',
+              'event': 'file',
+              'payload': Uint8List.fromList([1]),
+            },
+          ),
+        ),
         throwsArgumentError,
       );
     });
diff --git a/packages/supabase_realtime/test/socket_test.dart b/packages/supabase_realtime/test/socket_test.dart
index 929760e4e..88c77d931 100644
--- a/packages/supabase_realtime/test/socket_test.dart
+++ b/packages/supabase_realtime/test/socket_test.dart
@@ -7,7 +7,6 @@ import 'package:mocktail/mocktail.dart';
 import 'package:supabase_common/testing.dart';
 import 'package:supabase_realtime/supabase_realtime.dart';
 import 'package:supabase_realtime/src/constants.dart';
-import 'package:supabase_realtime/src/message.dart';
 import 'package:test/test.dart';
 import 'package:web_socket_channel/io.dart';
 import 'package:web_socket_channel/web_socket_channel.dart';
@@ -173,7 +172,7 @@ void main() {
       socket.onStatusChange.listen((change) {
         statuses.add(change.status);
       });
-      late dynamic lastMessage;
+      late RealtimeMessage lastMessage;
       socket.onMessage.listen((message) {
         lastMessage = message;
       });
@@ -185,7 +184,7 @@ void main() {
       await socket.sendHeartbeat();
       // need to wait for event to trigger
       await Future.delayed(const Duration(seconds: 1));
-      expect(lastMessage['event'], 'heartbeat');
+      expect(lastMessage.event, 'heartbeat');
 
       await socket.disconnect();
       await Future.delayed(const Duration(seconds: 1));
@@ -831,7 +830,7 @@ void main() {
       unawaited(mockedSocket.connect());
       mockedSocket.connectionState = SocketState.open;
 
-      final message = Message(
+      final message = RealtimeMessage.outgoing(
         topic: topic,
         payload: payload,
         event: event,
@@ -850,7 +849,7 @@ void main() {
 
       expect(mockedSocket.sendBuffer, isEmpty);
 
-      final message = Message(
+      final message = RealtimeMessage.outgoing(
         topic: topic,
         payload: payload,
         event: event,
@@ -873,7 +872,7 @@ void main() {
       mockedSocket.connectionState = SocketState.open;
 
       final binaryPayload = Uint8List.fromList([1, 2, 3]);
-      final message = Message(
+      final message = RealtimeMessage.outgoing(
         topic: 'realtime:room',
         event: ChannelEvent.broadcast,
         payload: {
@@ -911,7 +910,7 @@ void main() {
         'ref': ref,
       });
 
-      final message = Message(
+      final message = RealtimeMessage.outgoing(
         topic: topic,
         payload: payload,
         event: event,
@@ -924,7 +923,7 @@ void main() {
       ).called(1);
     });
 
-    test('uses a custom encode override when provided', () {
+    test('uses a custom encode override when provided', () async {
       final customChannel = MockIOWebSocketChannel();
       final customSink = MockWebSocketSink();
       when(() => customChannel.sink).thenReturn(customSink);
@@ -934,14 +933,20 @@ void main() {
       final customSocket = RealtimeClient(
         socketEndpoint,
         transport: (url, headers) => customChannel,
-        encode: (_) => 'custom-frame',
+        encode: (_) async => 'custom-frame',
       );
       unawaited(customSocket.connect());
       customSocket.connectionState = SocketState.open;
 
       customSocket.push(
-        Message(topic: topic, payload: payload, event: event, ref: ref),
+        RealtimeMessage.outgoing(
+          topic: topic,
+          payload: payload,
+          event: event,
+          ref: ref,
+        ),
       );
+      await pumpEventQueue();
 
       verify(
         () => customSink.add(captureAny(that: equals('custom-frame'))),
diff --git a/sdk-compliance.yaml b/sdk-compliance.yaml
index d7c18a2e3..631f41fe4 100644
--- a/sdk-compliance.yaml
+++ b/sdk-compliance.yaml
@@ -1942,9 +1942,23 @@ features:
     symbols:
       - RealtimeClient.encode
       - RealtimeClient.decode
+      - RealtimeClientOptions.encode
+      - RealtimeClientOptions.decode
     supporting_symbols:
       - RealtimeDecode
       - RealtimeEncode
+      - RealtimeMessage
+      - RealtimeMessage.RealtimeMessage
+      - RealtimeMessage.==
+      - RealtimeMessage.event
+      - RealtimeMessage.fromJson
+      - RealtimeMessage.hashCode
+      - RealtimeMessage.joinRef
+      - RealtimeMessage.payload
+      - RealtimeMessage.ref
+      - RealtimeMessage.toJson
+      - RealtimeMessage.toString
+      - RealtimeMessage.topic
       - RealtimeProtocolVersion
       - RealtimeProtocolVersion.RealtimeProtocolVersion
       - RealtimeProtocolVersion.wireVersion