diff --git a/MIGRATION.md b/MIGRATION.md
index 0672c558a..44ddbe056 100644
--- a/MIGRATION.md
+++ b/MIGRATION.md
@@ -110,6 +110,149 @@ flushed once the join succeeds, so only channels that were never subscribed thro
 
 `httpSend()` requires a Realtime server running v2.97.0 or newer.
 
+### Realtime listener callbacks are now streams
+
+Every recurring-event listener in `realtime_client` is now a Dart `Stream` instead of a callback,
+following the shape `RealtimeClient.onHeartbeat` already had. Streams compose (`map`, `where`,
+`firstWhere`, `timeout`), support multiple listeners, and removing a listener is a
+`StreamSubscription.cancel()`, which the callback API had no public equivalent for.
+
+On `RealtimeClient`, the connection listeners are broadcast stream getters instead of
+callback-registration methods:
+
+```dart
+// Before
+client.onOpen(() => print('open'));
+client.onClose((event) => print('closed: $event'));
+client.onError((error) => print('error: $error'));
+client.onMessage((message) => print('message: $message'));
+
+// After
+client.onOpen.listen((_) => print('open'));
+client.onClose.listen((event) => print('closed: $event'));
+client.onError.listen((error) => print('error: $error'));
+client.onMessage.listen((message) => print('message: $message'));
+```
+
+On `RealtimeChannel`, `onPostgresChanges` and `onBroadcast` no longer take a `callback` parameter
+and return a typed stream instead of the channel, so they can no longer be chained. Repeated calls
+with the same arguments return the same stream. For `postgres_changes` the stream still has to be
+created before `subscribe()`, because the requested changes are part of the join payload, but it
+can be listened to at any point:
+
+```dart
+// Before
+supabase
+    .channel('room')
+    .onPostgresChanges(
+      event: PostgresChangeEvent.insert,
+      schema: 'public',
+      table: 'messages',
+      callback: (payload) => print(payload),
+    )
+    .onBroadcast(
+      event: 'cursor-pos',
+      callback: (payload) => print(payload),
+    )
+    .subscribe();
+
+// After
+final channel = supabase.channel('room');
+channel
+    .onPostgresChanges(
+      event: PostgresChangeEvent.insert,
+      schema: 'public',
+      table: 'messages',
+    )
+    .listen(print);
+channel.onBroadcast(event: 'cursor-pos').listen(print);
+channel.subscribe();
+```
+
+The presence and system listeners are stream getters, and `onSystemEvents` emits a typed
+`RealtimeSystemPayload` instead of a raw payload:
+
+```dart
+// Before
+channel.onPresenceSync((payload) { /* ... */ });
+channel.onPresenceJoin((payload) { /* ... */ });
+channel.onPresenceLeave((payload) { /* ... */ });
+channel.onSystemEvents((payload) {
+  final system = RealtimeSystemPayload.fromJson(
+    Map<String, dynamic>.from(payload as Map),
+  );
+});
+
+// After
+channel.onPresenceSync.listen((payload) { /* ... */ });
+channel.onPresenceJoin.listen((payload) { /* ... */ });
+channel.onPresenceLeave.listen((payload) { /* ... */ });
+channel.onSystemEvents.listen((system) { /* ... */ });
+```
+
+`subscribe()` no longer takes a status callback. Status changes are emitted on the new
+`RealtimeChannel.onStatusChange` stream as `RealtimeSubscribeStatusChange` values, which carry the
+`RealtimeSubscribeStatus` and, for `channelError`, the error that caused it. The optional timeout
+moved up to be the first positional parameter:
+
+```dart
+// Before
+channel.subscribe((status, [error]) {
+  if (status == RealtimeSubscribeStatus.subscribed) {
+    // ...
+  } else if (status == RealtimeSubscribeStatus.channelError) {
+    print('error: $error');
+  }
+}, const Duration(seconds: 10));
+
+// After
+channel.onStatusChange.listen((change) {
+  if (change.status == RealtimeSubscribeStatus.subscribed) {
+    // ...
+  } else if (change.status == RealtimeSubscribeStatus.channelError) {
+    print('error: ${change.error}');
+  }
+});
+channel.subscribe(const Duration(seconds: 10));
+```
+
+All channel streams complete when the channel closes, so `await for` loops and `onDone` handlers
+end on their own once the channel is gone.
+
+### `Binding` and `BindingCallback` are internal
+
+`Binding` and `BindingCallback` were the raw registration primitives underneath the channel
+listeners, exported by accident: their only consumers, `RealtimeChannel.onEvents` and
+`RealtimeChannel.off`, have always been internal. They are no longer exported. Use the typed
+channel streams (`onPostgresChanges`, `onBroadcast`, `onPresenceSync`, `onPresenceJoin`,
+`onPresenceLeave`, `onSystemEvents`) instead.
+
+### `RealtimePresence` is internal
+
+`RealtimePresence` and its helper types (`PresenceOptions`, `PresenceEvents`, `PresenceChooser`,
+`PresenceOnJoinCallback`, `PresenceOnLeaveCallback`) are now `@internal`, along with the
+`RealtimeChannel.presence` field. They were presence bookkeeping that leaked into the public API,
+and registering a callback through `channel.presence.onJoin(...)` silently disabled the channel's
+own presence events, because the channel's forwarders occupied the same single callback slot.
+
+Everything the class offered is available on the channel:
+
+```dart
+// Before
+channel.presence.onJoin((key, current, joined) { /* ... */ });
+channel.presence.onLeave((key, current, left) { /* ... */ });
+channel.presence.onSync(() { /* ... */ });
+final state = channel.presence.state;
+
+// After
+channel.onPresenceJoin.listen((payload) { /* ... */ });
+channel.onPresenceLeave.listen((payload) { /* ... */ });
+channel.onPresenceSync.listen((payload) { /* ... */ });
+final state = channel.presenceState();
+```
+
+The `Presence` payload class is unchanged and stays public.
+
 ### 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/examples/realtime_room/lib/room_channel.dart b/examples/realtime_room/lib/room_channel.dart
index d3c9de304..6f5ca5a84 100644
--- a/examples/realtime_room/lib/room_channel.dart
+++ b/examples/realtime_room/lib/room_channel.dart
@@ -15,7 +15,7 @@ import 'models.dart';
 ///   roster.
 ///
 /// The channel is created but not connected in the constructor; call
-/// [subscribe] to join and [dispose] to leave and release the streams.
+/// [subscribe] to join and [dispose] to leave, which also closes the streams.
 class RoomChannel {
   RoomChannel({
     required SupabaseClient client,
@@ -32,7 +32,36 @@ class RoomChannel {
          // inserted right after joining can be missed, because replication is
          // set up asynchronously after the join is confirmed.
          opts: const RealtimeChannelConfig(self: true, replicationReady: true),
-       );
+       ) {
+    // Postgres Changes: new rows in the `messages` table.
+    onMessageInserted = _channel
+        .onPostgresChanges(
+          event: PostgresChangeEvent.insert,
+          schema: 'public',
+          table: 'messages',
+        )
+        .map((payload) => Message.fromJson(payload.newRecord));
+
+    // Postgres Changes: deleted rows. The delete payload carries the removed
+    // row under `oldRecord`.
+    onMessageDeleted = _channel
+        .onPostgresChanges(
+          event: PostgresChangeEvent.delete,
+          schema: 'public',
+          table: 'messages',
+        )
+        .map((payload) => payload.oldRecord['id'] as String);
+
+    // Broadcast: transient "typing" pings. Our own pings are filtered out.
+    onTyping = _channel
+        .onBroadcast(event: _typingEvent)
+        .map((payload) => payload['username'] as String)
+        .where((name) => name != username);
+
+    // Presence: fires whenever the roster changes. Read the full state back
+    // and map it to the connected users.
+    onlineUsers = _channel.onPresenceSync.map((_) => _currentUsers());
+  }
 
   final SupabaseClient _client;
   final RealtimeChannel _channel;
@@ -44,93 +73,62 @@ class RoomChannel {
   /// messages, typing pings and presence.
   final String roomName;
 
-  final _messageInserted = StreamController<Message>.broadcast();
-  final _messageDeleted = StreamController<String>.broadcast();
-  final _typing = StreamController<String>.broadcast();
-  final _onlineUsers = StreamController<List<OnlineUser>>.broadcast();
-
   /// A message someone added to the room (from a Postgres Changes insert
   /// event).
-  Stream<Message> get onMessageInserted => _messageInserted.stream;
+  late final Stream<Message> onMessageInserted;
 
   /// The id of a message someone removed (from a Postgres Changes delete
   /// event).
-  Stream<String> get onMessageDeleted => _messageDeleted.stream;
+  late final Stream<String> onMessageDeleted;
 
   /// The username of another client that is currently typing (from a broadcast
-  /// event). Our own typing pings are filtered out.
-  Stream<String> get onTyping => _typing.stream;
+  /// event).
+  late final Stream<String> onTyping;
 
   /// The current room roster (recomputed on every presence sync event).
-  Stream<List<OnlineUser>> get onlineUsers => _onlineUsers.stream;
+  late final Stream<List<OnlineUser>> onlineUsers;
 
   static const _typingEvent = 'typing';
 
-  /// Registers the realtime listeners and joins the channel. Completes once the
-  /// server confirms both the subscription and that Postgres Changes
-  /// replication is live, so a message sent right afterwards is guaranteed to
-  /// stream back.
+  /// Joins the channel. Completes once the server confirms both the
+  /// subscription and that Postgres Changes replication is live, so a message
+  /// sent right afterwards is guaranteed to stream back.
   Future<void> subscribe() {
     final ready = Completer<void>();
 
-    _channel
-        // Postgres Changes: new rows in the `messages` table.
-        .onPostgresChanges(
-          event: PostgresChangeEvent.insert,
-          schema: 'public',
-          table: 'messages',
-          callback: (payload) =>
-              _messageInserted.add(Message.fromJson(payload.newRecord)),
-        )
-        // Postgres Changes: deleted rows. The delete payload carries the
-        // removed row under `oldRecord`.
-        .onPostgresChanges(
-          event: PostgresChangeEvent.delete,
-          schema: 'public',
-          table: 'messages',
-          callback: (payload) =>
-              _messageDeleted.add(payload.oldRecord['id'] as String),
-        )
-        // Broadcast: a transient "typing" ping from another client.
-        .onBroadcast(
-          event: _typingEvent,
-          callback: (payload) {
-            final name = payload['username'] as String;
-            if (name != username) _typing.add(name);
-          },
-        )
-        // Presence: fires whenever the roster changes. Read the full state back
-        // and map it to the connected users.
-        .onPresenceSync((_) => _onlineUsers.add(_currentUsers()))
-        // The replication-ready signal requested with `replicationReady: true`.
-        // It arrives after the join, once Postgres Changes is actually
-        // streaming, so this is what completes [subscribe].
-        .onSystemEvents((payload) {
-          final system = RealtimeSystemPayload.fromJson(
-            Map<String, dynamic>.from(payload as Map),
-          );
-          if (system.extension != 'system' || ready.isCompleted) return;
-          if (system.status == 'ok') {
-            ready.complete();
-          } else {
-            ready.completeError(Exception(system.message));
-          }
-        })
-        .subscribe((status, error) {
-          if (status == RealtimeSubscribeStatus.subscribed) {
-            // Announce ourselves to the room now that we're connected. The
-            // payload is arbitrary JSON the other clients read back as
-            // presence.
-            unawaited(
-              _channel.track({
-                'username': username,
-                'online_at': DateTime.now().toIso8601String(),
-              }),
-            );
-          } else if (error != null && !ready.isCompleted) {
-            ready.completeError(error);
-          }
-        });
+    // The replication-ready signal requested with `replicationReady: true`.
+    // It arrives after the join, once Postgres Changes is actually streaming,
+    // so this is what completes [subscribe].
+    _channel.onSystemEvents.listen((system) {
+      if (system.extension != 'system' || ready.isCompleted) return;
+      if (system.status == 'ok') {
+        ready.complete();
+      } else {
+        ready.completeError(Exception(system.message));
+      }
+    });
+
+    _channel.onStatusChange.listen((change) {
+      if (change.status == RealtimeSubscribeStatus.subscribed) {
+        // Announce ourselves to the room now that we're connected. The
+        // payload is arbitrary JSON the other clients read back as presence.
+        unawaited(
+          _channel.track({
+            'username': username,
+            'online_at': DateTime.now().toIso8601String(),
+          }),
+        );
+      } else if (change.error != null && !ready.isCompleted) {
+        ready.completeError(change.error!);
+      } else if (change.status == RealtimeSubscribeStatus.closed &&
+          !ready.isCompleted) {
+        ready.completeError(
+          StateError('channel closed before the subscription completed'),
+        );
+      }
+    });
+
+    _channel.subscribe();
 
     return ready.future;
   }
@@ -152,12 +150,9 @@ class RoomChannel {
         .toList();
   }
 
-  /// Leaves the room (which also untracks our presence) and closes the streams.
+  /// Leaves the room (which also untracks our presence). Closing the channel
+  /// also closes all of its streams.
   Future<void> dispose() async {
     await _client.removeChannel(_channel);
-    await _messageInserted.close();
-    await _messageDeleted.close();
-    await _typing.close();
-    await _onlineUsers.close();
   }
 }
diff --git a/packages/realtime_client/example/main.dart b/packages/realtime_client/example/main.dart
index fe9bdf1ce..05dbe7d94 100644
--- a/packages/realtime_client/example/main.dart
+++ b/packages/realtime_client/example/main.dart
@@ -12,35 +12,41 @@ Future<void> main() async {
   );
 
   final channel = socket.channel('realtime:public');
-  channel.onPostgresChanges(
-    event: PostgresChangeEvent.all,
-    filter: PostgresChangeFilter(
-      type: PostgresChangeFilterType.eq,
-      column: 'column',
-      value: 'value',
-    ),
-    callback: (payload) {},
-  );
-  channel.onPostgresChanges(
-    event: PostgresChangeEvent.delete,
-    schema: 'public',
-    callback: (payload) {
-      print('channel delete payload: ${payload.toString()}');
-    },
-  );
-  channel.onPostgresChanges(
-    event: PostgresChangeEvent.insert,
-    schema: 'public',
-    callback: (payload) {
-      print('channel insert payload: ${payload.toString()}');
-    },
-  );
+  channel
+      .onPostgresChanges(
+        event: PostgresChangeEvent.all,
+        filter: PostgresChangeFilter(
+          type: PostgresChangeFilterType.eq,
+          column: 'column',
+          value: 'value',
+        ),
+      )
+      .listen((payload) {});
+  channel
+      .onPostgresChanges(
+        event: PostgresChangeEvent.delete,
+        schema: 'public',
+      )
+      .listen((payload) {
+        print('channel delete payload: ${payload.toString()}');
+      });
+  channel
+      .onPostgresChanges(
+        event: PostgresChangeEvent.insert,
+        schema: 'public',
+      )
+      .listen((payload) {
+        print('channel insert payload: ${payload.toString()}');
+      });
 
-  socket.onMessage((message) => print('MESSAGE $message'));
+  socket.onMessage.listen((message) => print('MESSAGE $message'));
 
   // on connect and subscribe
   await socket.connect();
-  channel.subscribe((a, [_]) => print('SUBSCRIBED'));
+  channel.onStatusChange.listen(
+    (change) => print('STATUS ${change.status.name}'),
+  );
+  channel.subscribe();
 
   // delay 20s to receive events from server
   await Future.delayed(const Duration(seconds: 20));
diff --git a/packages/realtime_client/lib/realtime_client.dart b/packages/realtime_client/lib/realtime_client.dart
index 47fa2f73a..47a4498a2 100644
--- a/packages/realtime_client/lib/realtime_client.dart
+++ b/packages/realtime_client/lib/realtime_client.dart
@@ -10,6 +10,7 @@ export 'src/constants.dart'
         SocketState;
 export 'src/realtime_channel.dart';
 export 'src/realtime_client.dart';
-export 'src/realtime_presence.dart';
+export 'src/realtime_presence.dart' show Presence;
 export 'src/transformers.dart' show PostgresColumn, PostgresType;
-export 'src/types.dart' hide ChannelFilter, RealtimeListenType;
+export 'src/types.dart'
+    hide Binding, BindingCallback, ChannelFilter, RealtimeListenType;
diff --git a/packages/realtime_client/lib/src/message.dart b/packages/realtime_client/lib/src/message.dart
index 8a54dcc51..2dab5e8b1 100644
--- a/packages/realtime_client/lib/src/message.dart
+++ b/packages/realtime_client/lib/src/message.dart
@@ -1,6 +1,6 @@
-import 'package:realtime_client/realtime_client.dart';
-import 'package:realtime_client/src/constants.dart';
 import 'package:meta/meta.dart';
+import 'package:realtime_client/src/constants.dart';
+import 'package:realtime_client/src/types.dart';
 
 @internal
 class Message {
diff --git a/packages/realtime_client/lib/src/realtime_channel.dart b/packages/realtime_client/lib/src/realtime_channel.dart
index d952038b6..85bb9915a 100644
--- a/packages/realtime_client/lib/src/realtime_channel.dart
+++ b/packages/realtime_client/lib/src/realtime_channel.dart
@@ -8,6 +8,7 @@ import 'package:meta/meta.dart';
 import 'package:realtime_client/realtime_client.dart';
 import 'package:realtime_client/src/constants.dart';
 import 'package:realtime_client/src/push.dart';
+import 'package:realtime_client/src/realtime_presence.dart';
 import 'package:realtime_client/src/retry_timer.dart';
 import 'package:realtime_client/src/transformers.dart';
 import 'package:realtime_client/src/types.dart';
@@ -22,6 +23,7 @@ class RealtimeChannel {
   late Push joinPush;
   late RetryTimer _rejoinTimer;
   List<Push> _pushBuffer = [];
+  @internal
   late RealtimePresence presence;
   @internal
   late final String broadcastEndpointURL;
@@ -38,6 +40,26 @@ class RealtimeChannel {
   /// to check data
   late final bool _private;
 
+  final _statusController =
+      StreamController<RealtimeSubscribeStatusChange>.broadcast();
+
+  /// Controllers backing the event streams handed out by the `on*` members,
+  /// closed when the channel closes.
+  final List<StreamController<dynamic>> _eventControllers = [];
+
+  Stream<RealtimePresenceSyncPayload>? _presenceSyncStream;
+  Stream<RealtimePresenceJoinPayload>? _presenceJoinStream;
+  Stream<RealtimePresenceLeavePayload>? _presenceLeaveStream;
+  Stream<RealtimeSystemPayload>? _systemEventsStream;
+
+  /// Streams handed out by [onPostgresChanges], keyed by the encoded filter,
+  /// so that repeated calls with the same arguments reuse one binding.
+  final Map<String, Stream<PostgresChangePayload>> _postgresChangeStreams = {};
+
+  /// Streams handed out by [onBroadcast], keyed by the event name, so that
+  /// repeated calls with the same event reuse one binding.
+  final Map<String, Stream<Map<String, dynamic>>> _broadcastStreams = {};
+
   RealtimeChannel(
     this.topic,
     this.socket, {
@@ -75,6 +97,10 @@ class RealtimeChannel {
       socket.log('channel', 'close $topic $joinRef');
       _state = ChannelState.closed;
       socket.remove(this);
+      // Deferred so that bindings registered after this one (such as the
+      // status forwarding set up in [subscribe]) still run for this close
+      // event before their controllers are closed.
+      scheduleMicrotask(_closeStreamControllers);
     });
 
     _onError((reason) {
@@ -136,14 +162,11 @@ class RealtimeChannel {
 
   /// Subscribes to receive real-time changes
   ///
-  /// Pass a [callback] to react to different status changes.
+  /// Listen to [onStatusChange] to react to different status changes.
   ///
   /// [timeout] parameter can be used to override the default timeout set on
   /// [RealtimeClient].
-  RealtimeChannel subscribe([
-    void Function(RealtimeSubscribeStatus status, Object? error)? callback,
-    Duration? timeout,
-  ]) {
+  RealtimeChannel subscribe([Duration? timeout]) {
     if (!socket.isConnected) {
       unawaited(socket.connect());
     }
@@ -156,10 +179,10 @@ class RealtimeChannel {
     final isPrivate = params['config']['private'];
 
     _onError((e) {
-      if (callback != null) callback(RealtimeSubscribeStatus.channelError, e);
+      _addStatus(RealtimeSubscribeStatus.channelError, e);
     });
     _onClose(() {
-      if (callback != null) callback(RealtimeSubscribeStatus.closed, null);
+      _addStatus(RealtimeSubscribeStatus.closed);
     });
 
     // A `postgres_changes` subscription's replication setup happens
@@ -170,20 +193,18 @@ class RealtimeChannel {
     // under a stale/expired token, or the server declines), the verdict
     // arrives on a later `system` event with status `error` -- which is
     // otherwise only exposed via `onSystemEvents` (for debugging) and never
-    // reaches this status callback. The result is a channel that reports
+    // reaches the status stream. The result is a channel that reports
     // `subscribed` yet delivers nothing. Forward it as `channelError` so
     // consumers can detect and recover from the failed subscription.
-    onSystemEvents((payload) {
+    onEvents('system', ChannelFilter(), (payload, [ref]) {
       if (payload is Map && payload['status'] == 'error') {
-        if (callback != null) {
-          callback(
-            RealtimeSubscribeStatus.channelError,
-            Exception(
-              payload['message']?.toString() ??
-                  'postgres_changes subscription failed',
-            ),
-          );
-        }
+        _addStatus(
+          RealtimeSubscribeStatus.channelError,
+          Exception(
+            payload['message']?.toString() ??
+                'postgres_changes subscription failed',
+          ),
+        );
       }
     });
 
@@ -211,35 +232,53 @@ class RealtimeChannel {
         .receive(
           'ok',
           (response) => unawaited(
-            _handleJoinOk(response as Map<String, dynamic>, callback),
+            _handleJoinOk(response as Map<String, dynamic>),
           ),
         )
         .receive('error', (error) {
-          if (callback != null) {
-            callback(
-              RealtimeSubscribeStatus.channelError,
-              Exception(
-                jsonEncode(
-                  (error as Map<String, dynamic>).isNotEmpty
-                      ? (error).values.join(', ')
-                      : 'error',
-                ),
+          _addStatus(
+            RealtimeSubscribeStatus.channelError,
+            Exception(
+              jsonEncode(
+                (error as Map<String, dynamic>).isNotEmpty
+                    ? (error).values.join(', ')
+                    : 'error',
               ),
-            );
-          }
+            ),
+          );
         })
         .receive('timeout', (_) {
-          if (callback != null) {
-            callback(RealtimeSubscribeStatus.timedOut, null);
-          }
+          _addStatus(RealtimeSubscribeStatus.timedOut);
         });
     return this;
   }
 
-  Future<void> _handleJoinOk(
-    Map<String, dynamic> response,
-    void Function(RealtimeSubscribeStatus status, Object? error)? callback,
-  ) async {
+  /// Emits every status change of this channel's subscription, along with the
+  /// error that caused it for [RealtimeSubscribeStatus.channelError].
+  ///
+  /// ```dart
+  /// channel.onStatusChange.listen((change) {
+  ///   print('channel status: ${change.status}');
+  /// });
+  /// channel.subscribe();
+  /// ```
+  Stream<RealtimeSubscribeStatusChange> get onStatusChange =>
+      _statusController.stream;
+
+  void _addStatus(RealtimeSubscribeStatus status, [Object? error]) {
+    if (!_statusController.isClosed) {
+      _statusController.add(RealtimeSubscribeStatusChange(status, error));
+    }
+  }
+
+  void _closeStreamControllers() {
+    unawaited(_statusController.close());
+    for (final controller in _eventControllers) {
+      unawaited(controller.close());
+    }
+  }
+
+  Future<void> _handleJoinOk(Map<String, dynamic> response) async {
     final serverPostgresFilters = response['postgres_changes'];
     final accessToken = socket.accessToken;
     if (accessToken != null) {
@@ -259,9 +298,7 @@ class RealtimeChannel {
     }
 
     if (serverPostgresFilters == null) {
-      if (callback != null) {
-        callback(RealtimeSubscribeStatus.subscribed, null);
-      }
+      _addStatus(RealtimeSubscribeStatus.subscribed);
       return;
     }
     final clientPostgresBindings = _bindings['postgres_changes'];
@@ -293,24 +330,20 @@ class RealtimeChannel {
         );
       } else {
         unawaited(unsubscribe());
-        if (callback != null) {
-          callback(
-            RealtimeSubscribeStatus.channelError,
-            Exception(
-              'mismatch between server and client bindings for postgres '
-              'changes',
-            ),
-          );
-        }
+        _addStatus(
+          RealtimeSubscribeStatus.channelError,
+          Exception(
+            'mismatch between server and client bindings for postgres '
+            'changes',
+          ),
+        );
         return;
       }
     }
 
     _bindings['postgres_changes'] = newPostgresBindings;
 
-    if (callback != null) {
-      callback(RealtimeSubscribeStatus.subscribed, null);
-    }
+    _addStatus(RealtimeSubscribeStatus.subscribed);
   }
 
   List<SinglePresenceState> presenceState() {
@@ -392,28 +425,36 @@ class RealtimeChannel {
   /// the full row (reducing payload size). The listed columns must be
   /// selectable by the subscribing role.
   ///
+  /// Returns a broadcast stream of the matching changes. The stream has to be
+  /// created before calling [subscribe], because the requested changes are
+  /// part of the subscription setup, but it can be listened to at any point.
+  /// Repeated calls with the same arguments return the same stream.
+  ///
   /// ```dart
-  /// supabase.channel('my_channel').onPostgresChanges(
-  ///     event: PostgresChangeEvent.all,
-  ///     schema: 'public',
-  ///     table: 'messages',
-  ///     filter: PostgresChangeFilter(
-  ///       type: PostgresChangeFilterType.eq,
-  ///       column: 'room_id',
-  ///       value: 200,
-  ///     ),
-  ///     callback: (payload) {
+  /// final channel = supabase.channel('my_channel');
+  /// channel
+  ///     .onPostgresChanges(
+  ///       event: PostgresChangeEvent.all,
+  ///       schema: 'public',
+  ///       table: 'messages',
+  ///       filter: PostgresChangeFilter(
+  ///         type: PostgresChangeFilterType.eq,
+  ///         column: 'room_id',
+  ///         value: 200,
+  ///       ),
+  ///     )
+  ///     .listen((payload) {
   ///       print(payload);
-  ///     }).subscribe();
+  ///     });
+  /// channel.subscribe();
   /// ```
-  RealtimeChannel onPostgresChanges({
+  Stream<PostgresChangePayload> onPostgresChanges({
     required PostgresChangeEvent event,
     String? schema,
     String? table,
     PostgresChangeFilter? filter,
     List<PostgresChangeFilter>? filters,
     List<String>? select,
-    required void Function(PostgresChangePayload payload) callback,
   }) {
     assert(
       filter == null || filters == null,
@@ -426,16 +467,21 @@ class RealtimeChannel {
     ];
     final filterString = allFilters.isEmpty ? null : allFilters.join(',');
 
-    return onEvents(
-      'postgres_changes',
-      ChannelFilter(
-        event: event.toRealtimeEvent(),
-        schema: schema,
-        table: table,
-        filter: filterString,
-        select: select,
+    final channelFilter = ChannelFilter(
+      event: event.toRealtimeEvent(),
+      schema: schema,
+      table: table,
+      filter: filterString,
+      select: select,
+    );
+
+    return _postgresChangeStreams.putIfAbsent(
+      jsonEncode(channelFilter.toMap()),
+      () => _eventStream(
+        'postgres_changes',
+        channelFilter,
+        (payload) => PostgresChangePayload.fromPayload(payload),
       ),
-      (payload, [ref]) => callback(PostgresChangePayload.fromPayload(payload)),
     );
   }
 
@@ -443,121 +489,91 @@ class RealtimeChannel {
   ///
   /// [event] is the broadcast event name to which you want to listen.
   ///
+  /// Returns a broadcast stream of the matching messages. Repeated calls with
+  /// the same [event] return the same stream.
+  ///
   /// ```dart
-  /// supabase.channel('my_channel').onBroadcast(
-  ///     event: 'position',
-  ///     callback: (payload) {
-  ///       print(payload);
-  ///     }).subscribe();
+  /// final channel = supabase.channel('my_channel');
+  /// channel.onBroadcast(event: 'position').listen((payload) {
+  ///   print(payload);
+  /// });
+  /// channel.subscribe();
   /// ```
-  RealtimeChannel onBroadcast({
-    required String event,
-    required void Function(Map<String, dynamic> payload) callback,
-  }) {
-    return onEvents(
-      'broadcast',
-      ChannelFilter(event: event),
-      (payload, [ref]) => callback(Map<String, dynamic>.from(payload)),
+  Stream<Map<String, dynamic>> onBroadcast({required String event}) {
+    return _broadcastStreams.putIfAbsent(
+      event,
+      () => _eventStream(
+        'broadcast',
+        ChannelFilter(event: event),
+        (payload) => Map<String, dynamic>.from(payload),
+      ),
     );
   }
 
-  /// Sets up a listener for realtime presence sync event.
+  /// Emits whenever the presence state of the channel is synced.
   ///
   /// ```dart
   /// final channel = supabase.channel('my_channel');
-  /// channel
-  ///     .onPresenceSync(
-  ///         (RealtimePresenceSyncPayload payload) {
-  ///           print('Synced presence state: ${channel.presenceState()}');
-  ///         })
-  ///     .subscribe();
+  /// channel.onPresenceSync.listen((payload) {
+  ///   print('Synced presence state: ${channel.presenceState()}');
+  /// });
+  /// channel.subscribe();
   /// ```
-  RealtimeChannel onPresenceSync(
-    void Function(RealtimePresenceSyncPayload payload) callback,
-  ) {
-    final result = onEvents(
-      'presence',
-      ChannelFilter(
-        event: PresenceEvent.sync.name,
-      ),
-      (payload, [ref]) {
-        callback(
-          RealtimePresenceSyncPayload.fromJson(
-            Map<String, dynamic>.from(payload),
-          ),
-        );
-      },
+  Stream<RealtimePresenceSyncPayload> get onPresenceSync {
+    return _presenceSyncStream ??= _presenceStream(
+      PresenceEvent.sync,
+      RealtimePresenceSyncPayload.fromJson,
     );
-    _handlePresenceUpdate();
-    return result;
   }
 
-  /// Sets up a listener for realtime presence join event.
+  /// Emits whenever clients join the presence state of the channel.
   ///
   /// ```dart
   /// final channel = supabase.channel('my_channel');
-  /// channel
-  ///     .onPresenceJoin(
-  ///         (RealtimePresenceJoinPayload payload) {
-  ///           print('Newly joined Presence: ${channel.presenceState()}');
-  ///         })
-  ///     .subscribe();
+  /// channel.onPresenceJoin.listen((payload) {
+  ///   print('Newly joined presences: ${payload.newPresences}');
+  /// });
+  /// channel.subscribe();
   /// ```
-  RealtimeChannel onPresenceJoin(
-    void Function(RealtimePresenceJoinPayload payload) callback,
-  ) {
-    final result = onEvents(
-      'presence',
-      ChannelFilter(
-        event: PresenceEvent.join.name,
-      ),
-      (payload, [ref]) {
-        callback(
-          RealtimePresenceJoinPayload.fromJson(
-            Map<String, dynamic>.from(payload),
-          ),
-        );
-      },
+  Stream<RealtimePresenceJoinPayload> get onPresenceJoin {
+    return _presenceJoinStream ??= _presenceStream(
+      PresenceEvent.join,
+      RealtimePresenceJoinPayload.fromJson,
     );
-    _handlePresenceUpdate();
-    return result;
   }
 
-  /// Sets up a listener for realtime presence leave event.
+  /// Emits whenever clients leave the presence state of the channel.
   ///
   /// ```dart
   /// final channel = supabase.channel('my_channel');
-  /// channel
-  ///     .onPresenceLeave(
-  ///         (RealtimePresenceLeavePayload payload) {
-  ///           print('Newly left Presence: ${channel.presenceState()}');
-  ///         })
-  ///     .subscribe();
+  /// channel.onPresenceLeave.listen((payload) {
+  ///   print('Newly left presences: ${payload.leftPresences}');
+  /// });
+  /// channel.subscribe();
   /// ```
-  RealtimeChannel onPresenceLeave(
-    void Function(RealtimePresenceLeavePayload payload) callback,
+  Stream<RealtimePresenceLeavePayload> get onPresenceLeave {
+    return _presenceLeaveStream ??= _presenceStream(
+      PresenceEvent.leave,
+      RealtimePresenceLeavePayload.fromJson,
+    );
+  }
+
+  /// Creates the stream backing one of the presence getters and enables
+  /// presence on the channel if it isn't already.
+  Stream<T> _presenceStream<T extends Object>(
+    PresenceEvent event,
+    T Function(Map<String, dynamic> json) fromJson,
   ) {
-    final result = onEvents(
+    final stream = _eventStream(
       'presence',
-      ChannelFilter(
-        event: PresenceEvent.leave.name,
-      ),
-      (payload, [ref]) {
-        callback(
-          RealtimePresenceLeavePayload.fromJson(
-            Map<String, dynamic>.from(payload),
-          ),
-        );
-      },
+      ChannelFilter(event: event.name),
+      (payload) => fromJson(Map<String, dynamic>.from(payload)),
     );
     _handlePresenceUpdate();
-    return result;
+    return stream;
   }
 
-  /// Sets up a listener for realtime `system` events.
-  ///
-  /// The [callback] receives the raw payload (typically a `Map`). To work with
-  /// it as a typed value, parse it with [RealtimeSystemPayload.fromJson].
+  /// Emits realtime `system` events.
   ///
   /// Opt in to the replication-ready notification with
   /// [RealtimeChannelConfig.replicationReady] when creating the channel, then
@@ -569,33 +585,54 @@ class RealtimeChannel {
   ///   'room1',
   ///   opts: const RealtimeChannelConfig(replicationReady: true),
   /// );
-  /// channel
-  ///     .onPostgresChanges(
-  ///       event: PostgresChangeEvent.all,
-  ///       schema: 'public',
-  ///       table: 'messages',
-  ///       callback: (payload) => print('Change received! $payload'),
-  ///     )
-  ///     .onSystemEvents((payload) {
-  ///       final system = RealtimeSystemPayload.fromJson(
-  ///         Map<String, dynamic>.from(payload as Map),
-  ///       );
-  ///       if (system.extension == 'system' && system.status == 'ok') {
-  ///         print('Replication connection is ready: ${system.message}');
-  ///       }
-  ///     })
-  ///     .subscribe();
+  /// channel.onSystemEvents.listen((payload) {
+  ///   if (payload.extension == 'system' && payload.status == 'ok') {
+  ///     print('Replication connection is ready: ${payload.message}');
+  ///   }
+  /// });
+  /// channel.subscribe();
   /// ```
-  RealtimeChannel onSystemEvents(
-    void Function(dynamic payload) callback,
-  ) {
-    return onEvents(
+  Stream<RealtimeSystemPayload> get onSystemEvents {
+    return _systemEventsStream ??= _eventStream(
       'system',
       ChannelFilter(),
-      (payload, [ref]) => callback(payload),
+      (payload) => payload is Map
+          ? RealtimeSystemPayload.fromJson(Map.from(payload))
+          : null,
     );
   }
 
+  /// Registers a binding that forwards matching events into a broadcast
+  /// stream.
+  ///
+  /// [transform] converts the raw payload into the emitted value; returning
+  /// `null` drops the event. The controller is tracked so it can be closed
+  /// when the channel closes.
+  Stream<T> _eventStream<T extends Object>(
+    String type,
+    ChannelFilter filter,
+    T? Function(dynamic payload) transform,
+  ) {
+    final controller = StreamController<T>.broadcast();
+
+    // Once a subscribed channel has closed it cannot be joined again, and the
+    // one-shot cleanup that completes the event streams has already run, so
+    // hand out an already-closed stream instead of one that never completes.
+    if (joinedOnce && isClosed) {
+      unawaited(controller.close());
+      return controller.stream;
+    }
+
+    onEvents(type, filter, (payload, [ref]) {
+      final value = transform(payload);
+      if (value != null && !controller.isClosed) {
+        controller.add(value);
+      }
+    });
+    _eventControllers.add(controller);
+    return controller.stream;
+  }
+
   @internal
   RealtimeChannel onEvents(
     String type,
diff --git a/packages/realtime_client/lib/src/realtime_client.dart b/packages/realtime_client/lib/src/realtime_client.dart
index 078767cbc..6b694af2f 100644
--- a/packages/realtime_client/lib/src/realtime_client.dart
+++ b/packages/realtime_client/lib/src/realtime_client.dart
@@ -148,13 +148,11 @@ class RealtimeClient {
   StreamSubscription<dynamic>? _connectionSubscription;
   @internal
   List<dynamic> sendBuffer = [];
-  @internal
-  Map<String, List<Function>> stateChangeCallbacks = {
-    'open': [],
-    'close': [],
-    'error': [],
-    'message': [],
-  };
+
+  final _openController = StreamController<void>.broadcast();
+  final _closeController = StreamController<RealtimeCloseEvent?>.broadcast();
+  final _errorController = StreamController<Object>.broadcast();
+  final _messageController = StreamController<Map<String, dynamic>>.broadcast();
 
   final _heartbeatController =
       StreamController<RealtimeHeartbeatStatus>.broadcast();
@@ -432,29 +430,26 @@ class RealtimeClient {
     logger?.call(kind, message, data);
   }
 
-  /// Registers callbacks for connection state change events
+  /// Emits whenever the WebSocket connection is opened.
   ///
-  /// Examples
-  /// socket.onOpen(() {print("Socket opened.");});
+  /// ```dart
+  /// final subscription = client.onOpen.listen((_) {
+  ///   print('Socket opened.');
+  /// });
+  /// ```
+  Stream<void> get onOpen => _openController.stream;
+
+  /// Emits whenever the WebSocket connection is closed.
   ///
-  void onOpen(void Function() callback) {
-    stateChangeCallbacks['open']!.add(callback);
-  }
+  /// The emitted [RealtimeCloseEvent] carries the close code and reason sent
+  /// by the server, or `null` when the connection closed without one.
+  Stream<RealtimeCloseEvent?> get onClose => _closeController.stream;
 
-  /// Registers a callbacks for connection state change events.
-  void onClose(void Function(dynamic) callback) {
-    stateChangeCallbacks['close']!.add(callback);
-  }
+  /// Emits whenever the WebSocket connection reports an error.
+  Stream<Object> get onError => _errorController.stream;
 
-  /// Registers a callbacks for connection state change events.
-  void onError(void Function(dynamic) callback) {
-    stateChangeCallbacks['error']!.add(callback);
-  }
-
-  /// Calls a function any time a message is received.
-  void onMessage(void Function(dynamic) callback) {
-    stateChangeCallbacks['message']!.add(callback);
-  }
+  /// Emits every decoded message received over the WebSocket.
+  Stream<Map<String, dynamic>> get onMessage => _messageController.stream;
 
   /// Emits a status whenever a heartbeat is sent, acknowledged, errors, or
   /// times out.
@@ -591,9 +586,7 @@ class RealtimeClient {
             messageRef,
           ),
         );
-    for (final callback in stateChangeCallbacks['message']!) {
-      callback(message);
-    }
+    _messageController.add(message);
   }
 
   static Object _encodeLegacy(Map<String, dynamic> message) =>
@@ -681,9 +674,7 @@ class RealtimeClient {
       log('transport', 'error while rejoining channels', e, Level.WARNING);
     }
 
-    for (final callback in stateChangeCallbacks['open']!) {
-      callback();
-    }
+    _openController.add(null);
   }
 
   /// communication has been closed
@@ -705,17 +696,13 @@ class RealtimeClient {
       reconnectTimer.scheduleTimeout();
     }
     if (heartbeatTimer != null) heartbeatTimer!.cancel();
-    for (final callback in stateChangeCallbacks['close']!) {
-      callback(event);
-    }
+    _closeController.add(event);
   }
 
-  void _onConnectionError(dynamic error) {
+  void _onConnectionError(Object error) {
     log('transport', error.toString());
     _triggerChanError(error);
-    for (final callback in stateChangeCallbacks['error']!) {
-      callback(error);
-    }
+    _errorController.add(error);
   }
 
   void _triggerChanError([dynamic error]) {
diff --git a/packages/realtime_client/lib/src/realtime_presence.dart b/packages/realtime_client/lib/src/realtime_presence.dart
index 26b2e8f9c..469311242 100644
--- a/packages/realtime_client/lib/src/realtime_presence.dart
+++ b/packages/realtime_client/lib/src/realtime_presence.dart
@@ -1,4 +1,5 @@
 // ignore_for_file: public_member_api_docs, sort_constructors_first
+import 'package:meta/meta.dart';
 import 'package:realtime_client/realtime_client.dart';
 import 'package:realtime_client/src/types.dart';
 
@@ -36,20 +37,25 @@ class Presence {
   String toString() => 'Presence(presenceRef: $presenceRef, payload: $payload)';
 }
 
+@internal
 typedef PresenceChooser<T> = T Function(String key, dynamic presence);
 
+@internal
 typedef PresenceOnJoinCallback =
     void Function(String? key, dynamic currentPresences, dynamic newPresences);
 
+@internal
 typedef PresenceOnLeaveCallback =
     void Function(String? key, dynamic currentPresences, dynamic newPresences);
 
-class PresenceOpts {
+@internal
+class PresenceOptions {
   final PresenceEvents events;
 
-  const PresenceOpts({required this.events});
+  const PresenceOptions({required this.events});
 }
 
+@internal
 class PresenceEvents {
   final String state;
   final String diff;
@@ -57,6 +63,17 @@ class PresenceEvents {
   const PresenceEvents({required this.state, required this.diff});
 }
 
+/// Internal bookkeeping for the presence state of a [RealtimeChannel].
+///
+/// Not part of the public API: the [onJoin], [onLeave], and [onSync] setters
+/// hold a single callback slot each, and the [RealtimePresence] constructor
+/// installs the forwarders that feed the channel presence streams through
+/// them, so replacing a callback silently disables those streams.
+///
+/// To observe presence, listen to [RealtimeChannel.onPresenceSync],
+/// [RealtimeChannel.onPresenceJoin], and [RealtimeChannel.onPresenceLeave],
+/// and read the current state with [RealtimeChannel.presenceState].
+@internal
 class RealtimePresence {
   Map<String, List<Presence>> state = <String, List<Presence>>{};
   List<Map<String, dynamic>> pendingDiffs = [];
@@ -73,11 +90,11 @@ class RealtimePresence {
   ///
   /// `channel` - The RealtimeChannel
   ///
-  /// `opts` - The options, for example `PresenceOpts(events:
+  /// `options` - The options, for example `PresenceOptions(events:
   /// PresenceEvents(state: 'state', diff: 'diff'))`
-  RealtimePresence(this.channel, [PresenceOpts? opts]) {
+  RealtimePresence(this.channel, [PresenceOptions? options]) {
     final events =
-        opts?.events ??
+        options?.events ??
         PresenceEvents(state: 'presence_state', diff: 'presence_diff');
 
     channel.onEvents(events.state, ChannelFilter(), (newState, [_]) {
diff --git a/packages/realtime_client/lib/src/types.dart b/packages/realtime_client/lib/src/types.dart
index 13e9e5d97..1bca9d777 100644
--- a/packages/realtime_client/lib/src/types.dart
+++ b/packages/realtime_client/lib/src/types.dart
@@ -3,8 +3,10 @@ import 'package:collection/collection.dart';
 import 'package:meta/meta.dart';
 import 'package:realtime_client/realtime_client.dart';
 
+@internal
 typedef BindingCallback = void Function(dynamic payload, [dynamic ref]);
 
+@internal
 class Binding {
   String type;
   Map<String, dynamic> filter;
@@ -147,6 +149,22 @@ enum PresenceEvent {
 
 enum RealtimeSubscribeStatus { subscribed, channelError, closed, timedOut }
 
+/// A subscription status change emitted by [RealtimeChannel.onStatusChange].
+class RealtimeSubscribeStatusChange {
+  /// The new status of the channel subscription.
+  final RealtimeSubscribeStatus status;
+
+  /// The error that caused a [RealtimeSubscribeStatus.channelError] status,
+  /// `null` for other statuses.
+  final Object? error;
+
+  const RealtimeSubscribeStatusChange(this.status, [this.error]);
+
+  @override
+  String toString() =>
+      'RealtimeSubscribeStatusChange(status: ${status.name}, error: $error)';
+}
+
 /// Configuration for broadcast replay feature.
 /// Allows replaying broadcast messages from a specific timestamp.
 class ReplayOption {
@@ -497,7 +515,7 @@ class PostgresChangeFilter {
   }
 }
 
-/// Base class for the payload in `.onPresence()` callback functions.
+/// Base class for the payloads emitted by the presence streams.
 abstract class RealtimePresencePayload {
   /// Name of the presence event.
   final PresenceEvent event;
diff --git a/packages/realtime_client/test/channel_test.dart b/packages/realtime_client/test/channel_test.dart
index 4950387d9..23fe9a9c6 100644
--- a/packages/realtime_client/test/channel_test.dart
+++ b/packages/realtime_client/test/channel_test.dart
@@ -117,7 +117,7 @@ void main() {
 
       expect(joinPush.timeout, Constants.defaultTimeout);
 
-      channel.subscribe((_, [_]) {}, newTimeout);
+      channel.subscribe(newTimeout);
 
       expect(joinPush.timeout, newTimeout);
     });
@@ -143,12 +143,13 @@ void main() {
       final localChannel = throwingSocket.channel('topic');
 
       RealtimeSubscribeStatus? status;
-      localChannel.subscribe((s, _) => status = s);
+      localChannel.onStatusChange.listen((change) => status = change.status);
+      localChannel.subscribe();
       localChannel.joinPush.trigger('ok', {});
 
-      // Drain the microtask queue so the async 'ok' callback completes.
-      await Future<void>.value();
-      await Future<void>.value();
+      // Drain the event queue so the async 'ok' handler completes and the
+      // status stream delivers.
+      await Future<void>.delayed(Duration.zero);
 
       expect(throwingSocket.setAuthCalls, 1);
       expect(
@@ -175,10 +176,12 @@ void main() {
       // the test runner zone.
       await runZonedGuarded(
         () async {
-          localChannel.subscribe((s, _) => status = s);
+          localChannel.onStatusChange.listen(
+            (change) => status = change.status,
+          );
+          localChannel.subscribe();
           localChannel.joinPush.trigger('ok', {});
-          await Future<void>.value();
-          await Future<void>.value();
+          await Future<void>.delayed(Duration.zero);
         },
         (_, _) {
           /* expected: rethrown FormatException */
@@ -203,7 +206,7 @@ void main() {
     });
 
     test('subscribes when the server echoes back an `in` filter with escaped '
-        'quotes and backslashes', () {
+        'quotes and backslashes', () async {
       RealtimeSubscribeStatus? status;
       channel.onPostgresChanges(
         event: PostgresChangeEvent.all,
@@ -214,10 +217,10 @@ void main() {
           column: 'name',
           value: [r'a"b\c'],
         ),
-        callback: (_) {},
       );
 
-      channel.subscribe((newStatus, _) => status = newStatus);
+      channel.onStatusChange.listen((change) => status = change.status);
+      channel.subscribe();
 
       final sentFilter =
           (channel.joinPush.payload['config']['postgres_changes'] as List)[0]
@@ -229,12 +232,13 @@ void main() {
           {'id': 1, ...sentFilter},
         ],
       });
+      await Future<void>.delayed(Duration.zero);
 
       expect(status, RealtimeSubscribeStatus.subscribed);
     });
 
     test('reports a channelError when the server echoes back a different '
-        'filter', () {
+        'filter', () async {
       RealtimeSubscribeStatus? status;
       channel.onPostgresChanges(
         event: PostgresChangeEvent.all,
@@ -245,10 +249,10 @@ void main() {
           column: 'name',
           value: [r'a"b\c'],
         ),
-        callback: (_) {},
       );
 
-      channel.subscribe((newStatus, _) => status = newStatus);
+      channel.onStatusChange.listen((change) => status = change.status);
+      channel.subscribe();
 
       final sentFilter =
           (channel.joinPush.payload['config']['postgres_changes'] as List)[0]
@@ -259,6 +263,7 @@ void main() {
           {...sentFilter, 'id': 1, 'filter': 'name=in.("a"b\\c")'},
         ],
       });
+      await Future<void>.delayed(Duration.zero);
 
       expect(status, RealtimeSubscribeStatus.channelError);
     });
@@ -269,7 +274,6 @@ void main() {
         schema: 'public',
         table: 'users',
         select: ['id', 'first_name'],
-        callback: (_) {},
       );
 
       channel.subscribe();
@@ -297,7 +301,6 @@ void main() {
             value: ['open', 'pending'],
           ),
         ],
-        callback: (_) {},
       );
 
       channel.subscribe();
@@ -348,19 +351,21 @@ void main() {
     });
 
     test(
-      'forwards a system error to the subscribe callback as channelError',
-      () {
+      'forwards a system error to the status stream as channelError',
+      () async {
         RealtimeSubscribeStatus? status;
         Object? error;
-        channel.subscribe((newStatus, newError) {
-          status = newStatus;
-          error = newError;
+        channel.onStatusChange.listen((change) {
+          status = change.status;
+          error = change.error;
         });
+        channel.subscribe();
 
         channel.trigger('system', {
           'status': 'error',
           'message': 'Unable to subscribe to changes with given parameters',
         });
+        await Future<void>.delayed(Duration.zero);
 
         expect(status, RealtimeSubscribeStatus.channelError);
         expect(error, isA<Exception>());
@@ -371,55 +376,56 @@ void main() {
       },
     );
 
-    test('falls back to a default message when the system error has none', () {
-      Object? error;
-      channel.subscribe((_, newError) => error = newError);
+    test(
+      'falls back to a default message when the system error has none',
+      () async {
+        Object? error;
+        channel.onStatusChange.listen((change) => error = change.error);
+        channel.subscribe();
 
-      channel.trigger('system', {'status': 'error'});
+        channel.trigger('system', {'status': 'error'});
+        await Future<void>.delayed(Duration.zero);
 
-      expect(error, isA<Exception>());
-      expect(
-        error?.toString(),
-        contains('postgres_changes subscription failed'),
-      );
-    });
+        expect(error, isA<Exception>());
+        expect(
+          error?.toString(),
+          contains('postgres_changes subscription failed'),
+        );
+      },
+    );
 
-    test('does not surface a system ok event as an error', () {
+    test('does not surface a system ok event as an error', () async {
       RealtimeSubscribeStatus? status;
-      channel.subscribe((newStatus, _) => status = newStatus);
+      channel.onStatusChange.listen((change) => status = change.status);
+      channel.subscribe();
 
       channel.trigger('system', {
         'status': 'ok',
         'message': 'Subscribed to PostgreSQL',
       });
+      await Future<void>.delayed(Duration.zero);
 
       expect(status, isNot(RealtimeSubscribeStatus.channelError));
     });
 
-    test(
-      'forwards the raw payload, parseable into a RealtimeSystemPayload',
-      () {
-        dynamic received;
-        channel.onSystemEvents((payload) => received = payload);
-
-        channel.trigger('system', {
-          'extension': 'system',
-          'status': 'ok',
-          'message': 'Replication connection established',
-          'channel': 'topic',
-        });
+    test('emits a typed RealtimeSystemPayload', () async {
+      RealtimeSystemPayload? received;
+      channel.onSystemEvents.listen((payload) => received = payload);
 
-        expect(received, isA<Map<dynamic, dynamic>>());
+      channel.trigger('system', {
+        'extension': 'system',
+        'status': 'ok',
+        'message': 'Replication connection established',
+        'channel': 'topic',
+      });
+      await Future<void>.delayed(Duration.zero);
 
-        final system = RealtimeSystemPayload.fromJson(
-          Map<String, dynamic>.from(received),
-        );
-        expect(system.extension, 'system');
-        expect(system.status, 'ok');
-        expect(system.message, 'Replication connection established');
-        expect(system.channel, 'topic');
-      },
-    );
+      expect(received, isA<RealtimeSystemPayload>());
+      expect(received!.extension, 'system');
+      expect(received!.status, 'ok');
+      expect(received!.message, 'Replication connection established');
+      expect(received!.channel, 'topic');
+    });
   });
 
   group('onMessage', () {
@@ -487,6 +493,77 @@ void main() {
     });
   });
 
+  group('event stream reuse', () {
+    setUp(() {
+      socket = RealtimeClient('wss://example.com/socket');
+      channel = socket.channel('topic');
+    });
+
+    test(
+      'returns the same stream for identical postgres_changes arguments',
+      () {
+        final first = channel.onPostgresChanges(
+          event: PostgresChangeEvent.insert,
+          schema: 'public',
+          table: 'todos',
+        );
+        final second = channel.onPostgresChanges(
+          event: PostgresChangeEvent.insert,
+          schema: 'public',
+          table: 'todos',
+        );
+        final other = channel.onPostgresChanges(
+          event: PostgresChangeEvent.delete,
+          schema: 'public',
+          table: 'todos',
+        );
+
+        expect(identical(first, second), isTrue);
+        expect(identical(first, other), isFalse);
+      },
+    );
+
+    test('returns the same stream for the same broadcast event', () {
+      final first = channel.onBroadcast(event: 'cursor');
+      final second = channel.onBroadcast(event: 'cursor');
+      final other = channel.onBroadcast(event: 'position');
+
+      expect(identical(first, second), isTrue);
+      expect(identical(first, other), isFalse);
+    });
+
+    test(
+      'streams created after the channel closed complete immediately',
+      () async {
+        channel.subscribe();
+        channel.trigger('phx_close');
+        // Wait for the deferred controller cleanup to run.
+        await Future<void>.delayed(Duration.zero);
+        expect(channel.isClosed, isTrue);
+
+        await expectLater(channel.onBroadcast(event: 'cursor'), emitsDone);
+        await expectLater(channel.onPresenceSync, emitsDone);
+      },
+    );
+
+    test('reused streams deliver events to every listener', () async {
+      var firstEvents = 0;
+      var secondEvents = 0;
+      channel.onBroadcast(event: 'cursor').listen((_) => firstEvents++);
+      channel.onBroadcast(event: 'cursor').listen((_) => secondEvents++);
+      channel.subscribe();
+
+      channel.trigger('broadcast', {
+        'event': 'cursor',
+        'payload': {'x': 1},
+      });
+      await Future<void>.delayed(Duration.zero);
+
+      expect(firstEvents, 1);
+      expect(secondEvents, 1);
+    });
+  });
+
   group('blocking listeners after subscribe', () {
     setUp(() {
       socket = RealtimeClient('wss://example.com/socket');
@@ -498,10 +575,7 @@ void main() {
       expect(channel.isJoining, isTrue);
 
       expect(
-        () => channel.onPostgresChanges(
-          event: PostgresChangeEvent.all,
-          callback: (_) {},
-        ),
+        () => channel.onPostgresChanges(event: PostgresChangeEvent.all),
         throwsA(
           allOf(
             isA<String>(),
@@ -517,10 +591,7 @@ void main() {
       expect(channel.isJoined, isTrue);
 
       expect(
-        () => channel.onPostgresChanges(
-          event: PostgresChangeEvent.all,
-          callback: (_) {},
-        ),
+        () => channel.onPostgresChanges(event: PostgresChangeEvent.all),
         throwsA(
           allOf(
             isA<String>(),
@@ -532,10 +603,7 @@ void main() {
 
     test('allows adding postgres_changes listener before subscribe', () {
       expect(
-        () => channel.onPostgresChanges(
-          event: PostgresChangeEvent.all,
-          callback: (_) {},
-        ),
+        () => channel.onPostgresChanges(event: PostgresChangeEvent.all),
         returnsNormally,
       );
     });
@@ -545,7 +613,7 @@ void main() {
       expect(channel.isJoining, isTrue);
 
       expect(
-        () => channel.onPresenceSync((_) {}),
+        () => channel.onPresenceSync,
         returnsNormally,
       );
     });
@@ -555,7 +623,7 @@ void main() {
       expect(channel.isJoining, isTrue);
 
       expect(
-        () => channel.onBroadcast(event: 'test', callback: (_) {}),
+        () => channel.onBroadcast(event: 'test'),
         returnsNormally,
       );
     });
@@ -720,12 +788,10 @@ void main() {
     });
 
     test('send message via WebSocket when subscribed to channel', () async {
-      final subscribed = Completer<void>();
-      channel.subscribe((status, [error]) {
-        if (status == RealtimeSubscribeStatus.subscribed) {
-          subscribed.complete();
-        }
-      });
+      final subscribed = channel.onStatusChange.firstWhere(
+        (change) => change.status == RealtimeSubscribeStatus.subscribed,
+      );
+      channel.subscribe();
 
       // Accept the websocket the client opens on subscribe, then reply to the
       // channel join so it transitions to subscribed.
@@ -750,7 +816,7 @@ void main() {
             broadcast.complete(message);
         }
       });
-      await subscribed.future;
+      await subscribed;
 
       // Once subscribed, broadcasts are pushed over the websocket instead of
       // falling back to the REST endpoint.
@@ -822,21 +888,21 @@ void main() {
       );
     });
 
-    test('description', () async {
+    test('emits presence events on the presence streams', () async {
       bool syncCalled = false, joinCalled = false, leaveCalled = false;
-      channel
-          .onPresenceSync((payload) {
-            syncCalled = true;
-          })
-          .onPresenceJoin((payload) {
-            joinCalled = true;
-          })
-          .onPresenceLeave((payload) {
-            leaveCalled = true;
-          })
-          .subscribe();
+      channel.onPresenceSync.listen((payload) {
+        syncCalled = true;
+      });
+      channel.onPresenceJoin.listen((payload) {
+        joinCalled = true;
+      });
+      channel.onPresenceLeave.listen((payload) {
+        leaveCalled = true;
+      });
+      channel.subscribe();
 
       channel.trigger('presence', {'event': 'sync'}, '1');
+      await Future<void>.delayed(Duration.zero);
       expect(syncCalled, isTrue);
       channel.trigger('presence', {
         'event': 'join',
@@ -844,6 +910,7 @@ void main() {
         'newPresences': <Presence>[],
         'currentPresences': <Presence>[],
       }, '2');
+      await Future<void>.delayed(Duration.zero);
       expect(joinCalled, isTrue);
       channel.trigger('presence', {
         'event': 'leave',
@@ -851,6 +918,7 @@ void main() {
         'leftPresences': <Presence>[],
         'currentPresences': <Presence>[],
       }, '3');
+      await Future<void>.delayed(Duration.zero);
       expect(leaveCalled, isTrue);
     });
   });
@@ -929,7 +997,7 @@ void main() {
         params: const RealtimeChannelConfig(),
       );
 
-      channel.onPresenceSync((payload) {});
+      channel.onPresenceSync.listen((payload) {});
       channel.subscribe();
 
       final joinPayload = channel.joinPush.payload;
@@ -946,7 +1014,7 @@ void main() {
           params: const RealtimeChannelConfig(enabled: true),
         );
 
-        channel.onPresenceSync((payload) {});
+        channel.onPresenceSync.listen((payload) {});
         channel.subscribe();
 
         final joinPayload = channel.joinPush.payload;
@@ -978,7 +1046,7 @@ void main() {
         params: const RealtimeChannelConfig(),
       );
 
-      channel.onPresenceJoin((payload) {});
+      channel.onPresenceJoin.listen((payload) {});
       channel.subscribe();
 
       final joinPayload = channel.joinPush.payload;
@@ -992,7 +1060,7 @@ void main() {
         params: const RealtimeChannelConfig(),
       );
 
-      channel.onPresenceLeave((payload) {});
+      channel.onPresenceLeave.listen((payload) {});
       channel.subscribe();
 
       final joinPayload = channel.joinPush.payload;
@@ -1019,7 +1087,7 @@ void main() {
         channel.joinPush.trigger('ok', {});
         expect(channel.params['config']['presence']['enabled'], isFalse);
 
-        channel.onPresenceSync((payload) {});
+        channel.onPresenceSync.listen((payload) {});
 
         expect(channel.params['config']['presence']['enabled'], isTrue);
       },
@@ -1039,7 +1107,7 @@ void main() {
         channel.joinPush.trigger('ok', {});
         final initialPayload = Map.from(channel.params);
 
-        channel.onPresenceSync((payload) {});
+        channel.onPresenceSync.listen((payload) {});
 
         expect(channel.params['config']['presence']['enabled'], isTrue);
         expect(channel.params, equals(initialPayload));
@@ -1059,13 +1127,13 @@ void main() {
         channel.joinPush.trigger('ok', {});
         expect(channel.params['config']['presence']['enabled'], isFalse);
 
-        channel.onPresenceSync((payload) {});
+        channel.onPresenceSync.listen((payload) {});
         expect(channel.params['config']['presence']['enabled'], isTrue);
 
         final payloadAfterFirst = Map.from(channel.params);
 
-        channel.onPresenceJoin((payload) {});
-        channel.onPresenceLeave((payload) {});
+        channel.onPresenceJoin.listen((payload) {});
+        channel.onPresenceLeave.listen((payload) {});
 
         expect(channel.params, equals(payloadAfterFirst));
       },
@@ -1083,7 +1151,7 @@ void main() {
 
         expect(channel.joinedOnce, isFalse);
 
-        channel.onPresenceSync((payload) {});
+        channel.onPresenceSync.listen((payload) {});
 
         expect(channel.params['config']['presence']['enabled'], isFalse);
       },
@@ -1092,7 +1160,7 @@ void main() {
     test(
       'should receive presence events after resubscription triggered by adding '
       'callback',
-      () {
+      () async {
         channel = RealtimeChannel(
           'topic',
           socket,
@@ -1103,11 +1171,12 @@ void main() {
         channel.joinPush.trigger('ok', {});
 
         bool syncCalled = false;
-        channel.onPresenceSync((payload) {
+        channel.onPresenceSync.listen((payload) {
           syncCalled = true;
         });
 
         channel.trigger('presence', {'event': 'sync'}, '1');
+        await Future<void>.delayed(Duration.zero);
 
         expect(syncCalled, isTrue);
       },
@@ -1124,7 +1193,7 @@ void main() {
       channel.joinPush.trigger('ok', {});
       expect(channel.params['config']['presence']['enabled'], isFalse);
 
-      channel.onPresenceJoin((payload) {});
+      channel.onPresenceJoin.listen((payload) {});
 
       expect(channel.params['config']['presence']['enabled'], isTrue);
     });
@@ -1140,7 +1209,7 @@ void main() {
       channel.joinPush.trigger('ok', {});
       expect(channel.params['config']['presence']['enabled'], isFalse);
 
-      channel.onPresenceLeave((payload) {});
+      channel.onPresenceLeave.listen((payload) {});
 
       expect(channel.params['config']['presence']['enabled'], isTrue);
     });
diff --git a/packages/realtime_client/test/mock_test.dart b/packages/realtime_client/test/mock_test.dart
index d3e16635f..05814ac39 100644
--- a/packages/realtime_client/test/mock_test.dart
+++ b/packages/realtime_client/test/mock_test.dart
@@ -188,22 +188,16 @@ void main() {
     });
 
     test('.on()', () {
-      final streamController = StreamController<PostgresChangePayload>();
-
-      client
-          .channel('public:todos')
-          .onPostgresChanges(
-            event: PostgresChangeEvent.all,
-            schema: 'public',
-            table: 'todos',
-            callback: (payload) {
-              streamController.add(payload);
-            },
-          )
-          .subscribe();
+      final channel = client.channel('public:todos');
+      final changes = channel.onPostgresChanges(
+        event: PostgresChangeEvent.all,
+        schema: 'public',
+        table: 'todos',
+      );
+      channel.subscribe();
 
       expect(
-        streamController.stream,
+        changes,
         emitsInOrder([
           PostgresChangePayload.fromPayload({
             'schema': 'public',
@@ -237,27 +231,21 @@ void main() {
     });
 
     test('.on() with filter', () {
-      final streamController = StreamController<PostgresChangePayload>();
-
-      client
-          .channel('public:todos')
-          .onPostgresChanges(
-            event: PostgresChangeEvent.all,
-            schema: 'public',
-            table: 'todos',
-            filter: PostgresChangeFilter(
-              type: PostgresChangeFilterType.eq,
-              column: 'id',
-              value: 2,
-            ),
-            callback: (payload) {
-              streamController.add(payload);
-            },
-          )
-          .subscribe();
+      final channel = client.channel('public:todos');
+      final changes = channel.onPostgresChanges(
+        event: PostgresChangeEvent.all,
+        schema: 'public',
+        table: 'todos',
+        filter: PostgresChangeFilter(
+          type: PostgresChangeFilterType.eq,
+          column: 'id',
+          value: 2,
+        ),
+      );
+      channel.subscribe();
 
       expect(
-        streamController.stream,
+        changes,
         emitsInOrder([
           PostgresChangePayload.fromPayload({
             'schema': 'public',
@@ -282,34 +270,32 @@ void main() {
     });
 
     test("correct CHANNEL_ERROR data on heartbeat timeout", () async {
-      final subscribeCallback = expectAsync2((
-        RealtimeSubscribeStatus event,
-        error,
+      final statusListener = expectAsync1((
+        RealtimeSubscribeStatusChange change,
       ) {
-        if (event == RealtimeSubscribeStatus.channelError) {
-          expect(error, isA<RealtimeCloseEvent>());
-          error as RealtimeCloseEvent;
+        if (change.status == RealtimeSubscribeStatus.channelError) {
+          expect(change.error, isA<RealtimeCloseEvent>());
+          final error = change.error as RealtimeCloseEvent;
           expect(error.reason, "heartbeat timeout");
         } else {
-          expect(event, RealtimeSubscribeStatus.closed);
+          expect(change.status, RealtimeSubscribeStatus.closed);
         }
       }, count: 2);
 
-      final channel = client
-          .channel('public:todos')
-          .onPostgresChanges(
-            event: PostgresChangeEvent.all,
-            schema: 'public',
-            table: 'todos',
-            filter: PostgresChangeFilter(
-              type: PostgresChangeFilterType.eq,
-              column: 'id',
-              value: 2,
-            ),
-            callback: (payload) {},
-          );
-
-      channel.subscribe(subscribeCallback);
+      final channel = client.channel('public:todos');
+      channel.onPostgresChanges(
+        event: PostgresChangeEvent.all,
+        schema: 'public',
+        table: 'todos',
+        filter: PostgresChangeFilter(
+          type: PostgresChangeFilterType.eq,
+          column: 'id',
+          value: 2,
+        ),
+      );
+
+      channel.onStatusChange.listen(statusListener);
+      channel.subscribe();
 
       await Future.delayed(Duration(milliseconds: 200));
       await webSocket?.close(Constants.wsCloseNormal, "heartbeat timeout");
@@ -451,22 +437,16 @@ void main() {
     });
 
     test('.on()', () {
-      final streamController = StreamController<PostgresChangePayload>();
-
-      client
-          .channel('public:todos')
-          .onPostgresChanges(
-            event: PostgresChangeEvent.all,
-            schema: 'public',
-            table: 'todos',
-            callback: (payload) {
-              streamController.add(payload);
-            },
-          )
-          .subscribe();
+      final channel = client.channel('public:todos');
+      final changes = channel.onPostgresChanges(
+        event: PostgresChangeEvent.all,
+        schema: 'public',
+        table: 'todos',
+      );
+      channel.subscribe();
 
       expect(
-        streamController.stream,
+        changes,
         emitsInOrder([
           PostgresChangePayload.fromPayload({
             'schema': 'public',
@@ -500,27 +480,21 @@ void main() {
     });
 
     test('.on() with filter', () {
-      final streamController = StreamController<PostgresChangePayload>();
-
-      client
-          .channel('public:todos')
-          .onPostgresChanges(
-            event: PostgresChangeEvent.all,
-            schema: 'public',
-            table: 'todos',
-            filter: PostgresChangeFilter(
-              type: PostgresChangeFilterType.eq,
-              column: 'id',
-              value: 2,
-            ),
-            callback: (payload) {
-              streamController.add(payload);
-            },
-          )
-          .subscribe();
+      final channel = client.channel('public:todos');
+      final changes = channel.onPostgresChanges(
+        event: PostgresChangeEvent.all,
+        schema: 'public',
+        table: 'todos',
+        filter: PostgresChangeFilter(
+          type: PostgresChangeFilterType.eq,
+          column: 'id',
+          value: 2,
+        ),
+      );
+      channel.subscribe();
 
       expect(
-        streamController.stream,
+        changes,
         emitsInOrder([
           PostgresChangePayload.fromPayload({
             'schema': 'public',
diff --git a/packages/realtime_client/test/realtime_integration_test.dart b/packages/realtime_client/test/realtime_integration_test.dart
index f4eac9c98..e68b7272f 100644
--- a/packages/realtime_client/test/realtime_integration_test.dart
+++ b/packages/realtime_client/test/realtime_integration_test.dart
@@ -36,7 +36,7 @@ void main() {
 
       test('connects and reports the open state', () async {
         final opened = Completer<void>();
-        client.onOpen(() {
+        client.onOpen.listen((_) {
           if (!opened.isCompleted) opened.complete();
         });
         await client.connect();
@@ -93,12 +93,9 @@ void main() {
           const RealtimeChannelConfig(self: true),
         );
         final received = Completer<Map<String, dynamic>>();
-        channel.onBroadcast(
-          event: 'ping',
-          callback: (payload) {
-            if (!received.isCompleted) received.complete(payload);
-          },
-        );
+        channel.onBroadcast(event: 'ping').listen((payload) {
+          if (!received.isCompleted) received.complete(payload);
+        });
         await _subscribe(channel);
         await channel.sendBroadcastMessage(
           event: 'ping',
@@ -125,12 +122,9 @@ void main() {
 
         final received = Completer<Map<String, dynamic>>();
         final receiverChannel = receiver.channel(topic);
-        receiverChannel.onBroadcast(
-          event: 'cursor',
-          callback: (payload) {
-            if (!received.isCompleted) received.complete(payload);
-          },
-        );
+        receiverChannel.onBroadcast(event: 'cursor').listen((payload) {
+          if (!received.isCompleted) received.complete(payload);
+        });
         await _subscribe(receiverChannel);
 
         final senderChannel = sender.channel(topic);
@@ -155,14 +149,14 @@ void main() {
         );
 
         final synced = Completer<void>();
-        channel.onPresenceSync((_) {
+        channel.onPresenceSync.listen((_) {
           if (channel.presenceState().isNotEmpty && !synced.isCompleted) {
             synced.complete();
           }
         });
 
         final joined = Completer<RealtimePresenceJoinPayload>();
-        channel.onPresenceJoin((payload) {
+        channel.onPresenceJoin.listen((payload) {
           if (!joined.isCompleted) joined.complete(payload);
         });
 
@@ -192,7 +186,7 @@ void main() {
         );
 
         final left = Completer<RealtimePresenceLeavePayload>();
-        channel.onPresenceLeave((payload) {
+        channel.onPresenceLeave.listen((payload) {
           if (!left.isCompleted) left.complete(payload);
         });
 
@@ -227,23 +221,24 @@ void main() {
           final deletes = Completer<PostgresChangePayload>();
 
           final channel = client.channel('db-changes-${version.vsn}');
-          channel.onPostgresChanges(
-            event: PostgresChangeEvent.all,
-            schema: 'public',
-            table: 'todos',
-            callback: (payload) {
-              switch (payload.eventType) {
-                case PostgresChangeEvent.insert:
-                  if (!inserts.isCompleted) inserts.complete(payload);
-                case PostgresChangeEvent.update:
-                  if (!updates.isCompleted) updates.complete(payload);
-                case PostgresChangeEvent.delete:
-                  if (!deletes.isCompleted) deletes.complete(payload);
-                case PostgresChangeEvent.all:
-                  break;
-              }
-            },
-          );
+          channel
+              .onPostgresChanges(
+                event: PostgresChangeEvent.all,
+                schema: 'public',
+                table: 'todos',
+              )
+              .listen((payload) {
+                switch (payload.eventType) {
+                  case PostgresChangeEvent.insert:
+                    if (!inserts.isCompleted) inserts.complete(payload);
+                  case PostgresChangeEvent.update:
+                    if (!updates.isCompleted) updates.complete(payload);
+                  case PostgresChangeEvent.delete:
+                    if (!deletes.isCompleted) deletes.complete(payload);
+                  case PostgresChangeEvent.all:
+                    break;
+                }
+              });
 
           await _subscribe(channel);
           await Future<void>.delayed(const Duration(seconds: 2));
@@ -282,19 +277,20 @@ void main() {
           final matched = Completer<PostgresChangePayload>();
 
           final channel = client.channel('db-filter-${version.vsn}');
-          channel.onPostgresChanges(
-            event: PostgresChangeEvent.insert,
-            schema: 'public',
-            table: 'todos',
-            filter: PostgresChangeFilter(
-              type: PostgresChangeFilterType.eq,
-              column: 'is_complete',
-              value: true,
-            ),
-            callback: (payload) {
-              if (!matched.isCompleted) matched.complete(payload);
-            },
-          );
+          channel
+              .onPostgresChanges(
+                event: PostgresChangeEvent.insert,
+                schema: 'public',
+                table: 'todos',
+                filter: PostgresChangeFilter(
+                  type: PostgresChangeFilterType.eq,
+                  column: 'is_complete',
+                  value: true,
+                ),
+              )
+              .listen((payload) {
+                if (!matched.isCompleted) matched.complete(payload);
+              });
 
           await _subscribe(channel);
           await Future<void>.delayed(const Duration(seconds: 2));
@@ -330,18 +326,22 @@ void main() {
 /// Subscribes to [channel] and resolves with the terminal subscribe status.
 Future<RealtimeSubscribeStatus> _subscribe(RealtimeChannel channel) {
   final completer = Completer<RealtimeSubscribeStatus>();
-  channel.subscribe((status, error) {
+  channel.onStatusChange.listen((change) {
     if (completer.isCompleted) return;
-    if (status == RealtimeSubscribeStatus.subscribed) {
-      completer.complete(status);
-    } else if (status == RealtimeSubscribeStatus.channelError ||
-        status == RealtimeSubscribeStatus.timedOut) {
+    if (change.status == RealtimeSubscribeStatus.subscribed) {
+      completer.complete(change.status);
+    } else if (change.status == RealtimeSubscribeStatus.channelError ||
+        change.status == RealtimeSubscribeStatus.timedOut ||
+        change.status == RealtimeSubscribeStatus.closed) {
       completer.completeError(
-        StateError('Failed to subscribe: ${status.name} ($error)'),
+        StateError(
+          'Failed to subscribe: ${change.status.name} (${change.error})',
+        ),
         StackTrace.current,
       );
     }
   });
+  channel.subscribe();
   return completer.future.timeout(const Duration(seconds: 15));
 }
 
diff --git a/packages/realtime_client/test/socket_test.dart b/packages/realtime_client/test/socket_test.dart
index 174711eb0..7ca2dd9f5 100644
--- a/packages/realtime_client/test/socket_test.dart
+++ b/packages/realtime_client/test/socket_test.dart
@@ -84,12 +84,6 @@ void main() {
       expect(socket.sendBuffer, isEmpty);
       expect(socket.ref, 0);
       expect(socket.endPoint, 'wss://example.com/socket/websocket');
-      expect(socket.stateChangeCallbacks, {
-        'open': [],
-        'close': [],
-        'error': [],
-        'message': [],
-      });
       expect(socket.timeout, const Duration(milliseconds: 10000));
       expect(socket.heartbeatIntervalMs, Constants.defaultHeartbeatIntervalMs);
       expect(
@@ -121,12 +115,6 @@ void main() {
       expect(socket.sendBuffer, isEmpty);
       expect(socket.ref, 0);
       expect(socket.endPoint, 'wss://example.com/socket/websocket');
-      expect(socket.stateChangeCallbacks, {
-        'open': [],
-        'close': [],
-        'error': [],
-        'message': [],
-      });
       expect(socket.timeout, const Duration(milliseconds: 40000));
       expect(socket.heartbeatIntervalMs, 60000);
       expect(
@@ -211,17 +199,17 @@ void main() {
       //! Not verifying connection url
     });
 
-    test('sets callbacks for connection', () async {
+    test('emits connection state events on the streams', () async {
       int opens = 0;
-      socket.onOpen(() {
+      socket.onOpen.listen((_) {
         opens += 1;
       });
       int closes = 0;
-      socket.onClose((_) {
+      socket.onClose.listen((_) {
         closes += 1;
       });
       late dynamic lastMessage;
-      socket.onMessage((message) {
+      socket.onMessage.listen((message) {
         lastMessage = message;
       });
 
@@ -239,16 +227,13 @@ void main() {
       expect(closes, 1);
     });
 
-    test('sets callback for errors', () {
-      dynamic lastError;
-      final RealtimeClient erroneousSocket = RealtimeClient('badurl')
-        ..onError((error) {
-          lastError = error;
-        });
+    test('emits errors on the onError stream', () async {
+      final RealtimeClient erroneousSocket = RealtimeClient('badurl');
+      final errorFuture = erroneousSocket.onError.first;
 
       unawaited(erroneousSocket.connect());
 
-      expect(lastError, isA<WebSocketException>());
+      expect(await errorFuture, isA<WebSocketException>());
     });
 
     test('is idempotent', () {
@@ -518,8 +503,8 @@ void main() {
         socketEndpoint,
         transport: (url, headers) => mockedSocketChannel,
       );
-      var closeCallbacks = 0;
-      mockedSocket.onClose((_) => closeCallbacks += 1);
+      var closeEvents = 0;
+      mockedSocket.onClose.listen((_) => closeEvents += 1);
 
       when(() => mockedSocketChannel.ready).thenAnswer((_) => Future.value());
       when(() => mockedSocketChannel.sink).thenReturn(mockedSink);
@@ -532,9 +517,11 @@ void main() {
       expect(mockedSocket.connectionState, SocketState.open);
 
       await mockedSocket.disconnect();
+      // Wait for the async stream delivery of the close event.
+      await Future<void>.delayed(Duration.zero);
       expect(mockedSocket.connectionState, SocketState.disconnected);
       expect(mockedSocket.connection, isNull);
-      expect(closeCallbacks, 1);
+      expect(closeEvents, 1);
       verify(() => mockedSink.close()).called(1);
 
       await streamController.close();
@@ -969,15 +956,14 @@ void main() {
       );
     });
 
-    test('dispatches a received binary broadcast to onBroadcast', () {
+    test('dispatches a received binary broadcast to onBroadcast', () async {
       final socket = RealtimeClient(socketEndpoint);
       final channel = socket.channel('room');
 
       Map<String, dynamic>? received;
-      channel.onBroadcast(
-        event: 'cursor',
-        callback: (payload) => received = payload,
-      );
+      channel.onBroadcast(event: 'cursor').listen((payload) {
+        received = payload;
+      });
 
       final topic = utf8.encode('realtime:room');
       final event = utf8.encode('cursor');
@@ -994,6 +980,8 @@ void main() {
       ]);
 
       socket.onConnectionMessage(frame);
+      // Wait for the async stream delivery of the broadcast event.
+      await Future<void>.delayed(Duration.zero);
 
       expect(received, {
         'type': 'broadcast',
@@ -1004,7 +992,7 @@ void main() {
 
     test(
       'decodes a legacy object frame and dispatches it when version is v1',
-      () {
+      () async {
         final socket = RealtimeClient(
           socketEndpoint,
           version: RealtimeProtocolVersion.v1,
@@ -1012,10 +1000,9 @@ void main() {
         final channel = socket.channel('room');
 
         Map<String, dynamic>? received;
-        channel.onBroadcast(
-          event: 'cursor',
-          callback: (payload) => received = payload,
-        );
+        channel.onBroadcast(event: 'cursor').listen((payload) {
+          received = payload;
+        });
 
         socket.onConnectionMessage(
           json.encode({
@@ -1029,6 +1016,8 @@ void main() {
             'ref': null,
           }),
         );
+        // Wait for the async stream delivery of the broadcast event.
+        await Future<void>.delayed(Duration.zero);
 
         expect(received, {
           'type': 'broadcast',
@@ -1204,7 +1193,7 @@ void main() {
         transport: (url, headers) => mockedSocketChannel,
       );
       var opens = 0;
-      socket.onOpen(() => opens += 1);
+      socket.onOpen.listen((_) => opens += 1);
 
       when(() => mockedSocketChannel.ready).thenAnswer((_) => Future.value());
       when(() => mockedSocketChannel.sink).thenReturn(mockedSink);
@@ -1221,6 +1210,8 @@ void main() {
 
       verify(() => erroredChannel.rejoin()).called(1);
       verifyNever(() => healthyChannel.rejoin());
+      // Wait for the async stream delivery of the open event.
+      await Future<void>.delayed(Duration.zero);
       expect(opens, 1);
       expect(socket.connectionState, SocketState.open);
 
diff --git a/packages/realtime_client/test/utils/realtime_test_utils.dart b/packages/realtime_client/test/utils/realtime_test_utils.dart
index 0e692082d..7a0a6c8f2 100644
--- a/packages/realtime_client/test/utils/realtime_test_utils.dart
+++ b/packages/realtime_client/test/utils/realtime_test_utils.dart
@@ -116,28 +116,31 @@ Future<void> primePostgresChanges({
   final received = Completer<void>();
 
   final channel = client.channel('postgres-changes-warmup');
-  channel.onPostgresChanges(
-    event: PostgresChangeEvent.insert,
-    schema: 'public',
-    table: 'todos',
-    callback: (_) {
-      if (!received.isCompleted) received.complete();
-    },
-  );
+  channel
+      .onPostgresChanges(
+        event: PostgresChangeEvent.insert,
+        schema: 'public',
+        table: 'todos',
+      )
+      .listen((_) {
+        if (!received.isCompleted) received.complete();
+      });
 
   final subscribed = Completer<void>();
-  channel.subscribe((status, error) {
+  channel.onStatusChange.listen((change) {
     if (subscribed.isCompleted) return;
-    if (status == RealtimeSubscribeStatus.subscribed) {
+    if (change.status == RealtimeSubscribeStatus.subscribed) {
       subscribed.complete();
-    } else if (status == RealtimeSubscribeStatus.channelError ||
-        status == RealtimeSubscribeStatus.timedOut) {
+    } else if (change.status == RealtimeSubscribeStatus.channelError ||
+        change.status == RealtimeSubscribeStatus.timedOut ||
+        change.status == RealtimeSubscribeStatus.closed) {
       subscribed.completeError(
-        StateError('warmup subscribe failed: ${status.name}'),
+        StateError('warmup subscribe failed: ${change.status.name}'),
         StackTrace.current,
       );
     }
   });
+  channel.subscribe();
 
   try {
     await subscribed.future.timeout(const Duration(seconds: 15));
@@ -182,21 +185,23 @@ Future<void> waitForRealtimeServer({
     httpReachable = await _isRealtimeHttpReachable();
 
     final client = createRealtimeClient(RealtimeProtocolVersion.v1);
-    client.onError((error) => lastError = error);
+    client.onError.listen((error) => lastError = error);
 
     final completer = Completer<bool>();
     final channel = client.channel('readiness-check');
-    channel.subscribe((status, error) {
+    channel.onStatusChange.listen((change) {
       if (completer.isCompleted) return;
-      lastStatus = status.name;
-      if (error != null) lastError = error;
-      if (status == RealtimeSubscribeStatus.subscribed) {
+      lastStatus = change.status.name;
+      if (change.error != null) lastError = change.error;
+      if (change.status == RealtimeSubscribeStatus.subscribed) {
         completer.complete(true);
-      } else if (status == RealtimeSubscribeStatus.channelError ||
-          status == RealtimeSubscribeStatus.timedOut) {
+      } else if (change.status == RealtimeSubscribeStatus.channelError ||
+          change.status == RealtimeSubscribeStatus.timedOut ||
+          change.status == RealtimeSubscribeStatus.closed) {
         completer.complete(false);
       }
     });
+    channel.subscribe();
 
     var ready = false;
     try {
diff --git a/packages/supabase/example/web/main.dart b/packages/supabase/example/web/main.dart
index a0a5d241a..3eefaf359 100644
--- a/packages/supabase/example/web/main.dart
+++ b/packages/supabase/example/web/main.dart
@@ -41,9 +41,9 @@ void exampleUsage(SupabaseClient supabase) async {
         event: PostgresChangeEvent.all,
         schema: 'public',
         table: 'countries',
-        callback: (payload) {},
       )
-      .subscribe();
+      .listen((payload) {});
+  realtimeChannel.subscribe();
 
   // remember to remove channel when no longer needed
   unawaited(supabase.removeChannel(realtimeChannel));
diff --git a/packages/supabase/lib/src/supabase_stream_builder.dart b/packages/supabase/lib/src/supabase_stream_builder.dart
index 28a842f71..9b3af96d8 100644
--- a/packages/supabase/lib/src/supabase_stream_builder.dart
+++ b/packages/supabase/lib/src/supabase_stream_builder.dart
@@ -56,6 +56,12 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   /// StreamController for `stream()` method.
   ReplaySubject<SupabaseStreamEvent>? _streamController;
 
+  /// Subscription on the channel's postgres changes stream.
+  StreamSubscription<PostgresChangePayload>? _changesSubscription;
+
+  /// Subscription on the channel's subscription status stream.
+  StreamSubscription<RealtimeSubscribeStatusChange>? _statusSubscription;
+
   /// Contains the combined data of postgrest and realtime to emit as stream.
   SupabaseStreamEvent _streamData = [];
 
@@ -144,6 +150,10 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
       },
       onCancel: () {
         _log.fine('stream controller for table: $_table got closed');
+        unawaited(_changesSubscription?.cancel());
+        unawaited(_statusSubscription?.cancel());
+        _changesSubscription = null;
+        _statusSubscription = null;
         unawaited(_channel?.unsubscribe());
         unawaited(_streamController?.close());
         _streamController = null;
@@ -170,64 +180,65 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
       ),
     );
 
-    _channel!
+    _changesSubscription = _channel!
         .onPostgresChanges(
           event: PostgresChangeEvent.all,
           schema: _schema,
           table: _table,
           filters: realtimeFilters,
-          callback: (payload) {
-            switch (payload.eventType) {
-              case PostgresChangeEvent.insert:
-                final newRecord = payload.newRecord;
-                _streamData.add(newRecord);
-                _addStream();
-              case PostgresChangeEvent.update:
-                final updatedIndex = _streamData.indexWhere(
-                  (element) =>
-                      _isTargetRecord(record: element, payload: payload),
-                );
-
-                final updatedRecord = payload.newRecord;
-                if (updatedIndex >= 0) {
-                  _streamData[updatedIndex] = updatedRecord;
-                } else {
-                  _streamData.add(updatedRecord);
-                }
-                _addStream();
-              case PostgresChangeEvent.delete:
-                final deletedIndex = _streamData.indexWhere(
-                  (element) =>
-                      _isTargetRecord(record: element, payload: payload),
-                );
-                if (deletedIndex >= 0) {
-                  /// Delete the data from in memory cache if it was found
-                  _streamData.removeAt(deletedIndex);
-                  _addStream();
-                }
-              case PostgresChangeEvent.all:
-                break;
-            }
-          },
         )
-        .subscribe((status, [error]) {
-          switch (status) {
-            case RealtimeSubscribeStatus.subscribed:
-              // Reload all data from PostgREST after a realtime reconnect, so
-              // that changes missed while the socket was down are picked up.
-              // The first subscribe is skipped because the initial load is
-              // already started below, right after subscribing.
-              if (_wasSubscribed) {
-                unawaited(_getPostgrestData());
+        .listen((payload) {
+          switch (payload.eventType) {
+            case PostgresChangeEvent.insert:
+              final newRecord = payload.newRecord;
+              _streamData.add(newRecord);
+              _addStream();
+            case PostgresChangeEvent.update:
+              final updatedIndex = _streamData.indexWhere(
+                (element) => _isTargetRecord(record: element, payload: payload),
+              );
+
+              final updatedRecord = payload.newRecord;
+              if (updatedIndex >= 0) {
+                _streamData[updatedIndex] = updatedRecord;
+              } else {
+                _streamData.add(updatedRecord);
+              }
+              _addStream();
+            case PostgresChangeEvent.delete:
+              final deletedIndex = _streamData.indexWhere(
+                (element) => _isTargetRecord(record: element, payload: payload),
+              );
+              if (deletedIndex >= 0) {
+                /// Delete the data from in memory cache if it was found
+                _streamData.removeAt(deletedIndex);
+                _addStream();
               }
-              _wasSubscribed = true;
-            case RealtimeSubscribeStatus.closed:
-              unawaited(_streamController?.close());
-            case RealtimeSubscribeStatus.timedOut:
-            case RealtimeSubscribeStatus.channelError:
-              _addException(RealtimeSubscribeException(status, error));
+            case PostgresChangeEvent.all:
+              break;
           }
         });
+    _statusSubscription = _channel!.onStatusChange.listen((change) {
+      switch (change.status) {
+        case RealtimeSubscribeStatus.subscribed:
+          // Reload all data from PostgREST after a realtime reconnect, so
+          // that changes missed while the socket was down are picked up.
+          // The first subscribe is skipped because the initial load is
+          // already started below, right after subscribing.
+          if (_wasSubscribed) {
+            unawaited(_getPostgrestData());
+          }
+          _wasSubscribed = true;
+        case RealtimeSubscribeStatus.closed:
+          unawaited(_streamController?.close());
+        case RealtimeSubscribeStatus.timedOut:
+        case RealtimeSubscribeStatus.channelError:
+          _addException(
+            RealtimeSubscribeException(change.status, change.error),
+          );
+      }
+    });
+    _channel!.subscribe();
     unawaited(_getPostgrestData());
   }
 
diff --git a/packages/supabase/test/mock_test.dart b/packages/supabase/test/mock_test.dart
index 314b5fd87..c3ece9f0c 100644
--- a/packages/supabase/test/mock_test.dart
+++ b/packages/supabase/test/mock_test.dart
@@ -695,17 +695,17 @@ void main() {
       /// exception
       /// https://github.com/supabase-community/supabase-flutter/issues/81
       test('Calling Postgrest within realtime callback', () async {
-        supabase
-            .channel('todos')
+        final channel = supabase.channel('todos');
+        channel
             .onPostgresChanges(
               event: PostgresChangeEvent.all,
               schema: 'public',
               table: 'todos',
-              callback: (payload) {
-                unawaited(supabase.from('todos'));
-              },
             )
-            .subscribe();
+            .listen((payload) {
+              unawaited(supabase.from('todos'));
+            });
+        channel.subscribe();
 
         await Future.delayed(const Duration(milliseconds: 700));
 
diff --git a/packages/supabase/test/realtime_test.dart b/packages/supabase/test/realtime_test.dart
index 7e7d19de7..774842aa5 100644
--- a/packages/supabase/test/realtime_test.dart
+++ b/packages/supabase/test/realtime_test.dart
@@ -50,16 +50,12 @@ void main() {
     /// expectation:
     /// - error
     test('subscribe on existing subscription fail', () {
-      channel
-          .onPostgresChanges(
-            event: PostgresChangeEvent.insert,
-            schema: 'public',
-            table: 'countries',
-            callback: (payload) {},
-          )
-          .subscribe(
-            (event, [errorMessage]) {},
-          );
+      channel.onPostgresChanges(
+        event: PostgresChangeEvent.insert,
+        schema: 'public',
+        table: 'countries',
+      );
+      channel.subscribe();
       expect(
         () => channel.subscribe(),
         throwsA(isA<String>()),
diff --git a/packages/supabase_common/test/timestamp_test.dart b/packages/supabase_common/test/timestamp_test.dart
index e49f7fe7d..91be069b4 100644
--- a/packages/supabase_common/test/timestamp_test.dart
+++ b/packages/supabase_common/test/timestamp_test.dart
@@ -32,7 +32,7 @@ void main() {
 
     test('throws when the value is missing', () {
       expect(
-        () => parseIso8601(<String, dynamic>{}, 'created_at'),
+        () => parseIso8601({}, 'created_at'),
         throwsA(
           isA<FormatException>().having(
             (exception) => exception.message,
@@ -146,7 +146,7 @@ void main() {
     });
 
     test('returns null when the value is absent', () {
-      expect(tryParseIso8601(<String, dynamic>{}, 'created_at'), isNull);
+      expect(tryParseIso8601({}, 'created_at'), isNull);
     });
 
     test('parses a present value', () {
diff --git a/packages/supabase_flutter/README.md b/packages/supabase_flutter/README.md
index 8ede9e767..9474228eb 100644
--- a/packages/supabase_flutter/README.md
+++ b/packages/supabase_flutter/README.md
@@ -381,11 +381,12 @@ myChannel
       event: PostgresChangeEvent.all,
       schema: 'public',
       table: 'countries',
-      callback: (payload) {
-        // Do something fun or interesting when there is an change on the database
-      },
     )
-    .subscribe();
+    .listen((payload) {
+      // Do something fun or interesting when there is a change on the database
+    });
+
+myChannel.subscribe();
 ```
 
 #### <a id="broadcast"></a>[Broadcast](https://supabase.com/docs/guides/realtime#broadcast)
@@ -395,14 +396,12 @@ Broadcast lets you send and receive low latency messages between connected clien
 ```dart
 final myChannel = supabase.channel('my_channel');
 
-// Subscribe to `cursor-pos` broadcast event
-final myChannel = supabase.channel('my_channel');
+// Listen to `cursor-pos` broadcast events
+myChannel.onBroadcast(event: 'cursor-pos').listen((payload) {
+  // Do something fun or interesting with the received message
+});
 
-myChannel
-    .onBroadcast(event: 'cursor-pos', callback: (payload) {}
-        // Do something fun or interesting when there is an change on the database
-        )
-    .subscribe();
+myChannel.subscribe();
 
 // Send a broadcast message to other connected clients
 await myChannel.sendBroadcastMessage(
@@ -418,31 +417,26 @@ Presence let's you easily create "I'm online" feature.
 ```dart
 final myChannel = supabase.channel('my_channel');
 
-// Subscribe to presence events
-myChannel
-    .onPresence(
-        event: PresenceEvent.sync,
-        callback: (payload) {
-          final onlineUsers = myChannel.presenceState();
-          // handle sync event
-        })
-    .onPresence(
-        event: PresenceEvent.join,
-        callback: (payload) {
-          // New users have joined
-        })
-    .onPresence(
-        event: PresenceEvent.leave,
-        callback: (payload) {
-          // Users have left
-        })
-    .subscribe(((status, [_]) async {
-  if (status == RealtimeSubscribeStatus.subscribed) {
+// Listen to presence events
+myChannel.onPresenceSync.listen((payload) {
+  final onlineUsers = myChannel.presenceState();
+  // handle sync event
+});
+myChannel.onPresenceJoin.listen((payload) {
+  // New users have joined
+});
+myChannel.onPresenceLeave.listen((payload) {
+  // Users have left
+});
+
+myChannel.onStatusChange.listen((change) async {
+  if (change.status == RealtimeSubscribeStatus.subscribed) {
     // Send the current user's state upon subscribing
-    final status = await myChannel
-        .track({'online_at': DateTime.now().toIso8601String()});
+    await myChannel.track({'online_at': DateTime.now().toIso8601String()});
   }
-}));
+});
+
+myChannel.subscribe();
 ```
 
 ### <a id="storage"></a>[Storage](https://supabase.com/docs/guides/storage)
diff --git a/sdk-compliance.yaml b/sdk-compliance.yaml
index f1a8a57d0..fcb966eda 100644
--- a/sdk-compliance.yaml
+++ b/sdk-compliance.yaml
@@ -1620,7 +1620,13 @@ features:
     symbols:
       - RealtimeChannel.subscribe
     supporting_symbols:
+      - RealtimeChannel.onStatusChange
       - RealtimeSubscribeStatus
+      - RealtimeSubscribeStatusChange
+      - RealtimeSubscribeStatusChange.RealtimeSubscribeStatusChange
+      - RealtimeSubscribeStatusChange.error
+      - RealtimeSubscribeStatusChange.status
+      - RealtimeSubscribeStatusChange.toString
   realtime.channel.unsubscribe:
     status: implemented
     symbols:
@@ -1763,17 +1769,7 @@ features:
       - RealtimeChannel.onPresenceJoin
       - RealtimeChannel.onPresenceLeave
     supporting_symbols:
-      - PresenceChooser
       - PresenceEvent
-      - PresenceEvents
-      - PresenceEvents.PresenceEvents
-      - PresenceEvents.diff
-      - PresenceEvents.state
-      - PresenceOnJoinCallback
-      - PresenceOnLeaveCallback
-      - PresenceOpts
-      - PresenceOpts.PresenceOpts
-      - PresenceOpts.events
       - RealtimePresenceJoinPayload
       - RealtimePresenceJoinPayload.RealtimePresenceJoinPayload
       - RealtimePresenceJoinPayload.currentPresences
@@ -1841,21 +1837,6 @@ features:
       - Presence.payload
       - Presence.presenceRef
       - Presence.toString
-      - RealtimeChannel.presence
-      - RealtimePresence
-      - RealtimePresence.RealtimePresence
-      - RealtimePresence.caller
-      - RealtimePresence.channel
-      - RealtimePresence.inPendingSyncState
-      - RealtimePresence.joinRef
-      - RealtimePresence.list
-      - RealtimePresence.onJoin
-      - RealtimePresence.onLeave
-      - RealtimePresence.onSync
-      - RealtimePresence.pendingDiffs
-      - RealtimePresence.state
-      - RealtimePresence.syncDiff
-      - RealtimePresence.syncState
       - SinglePresenceState
       - SinglePresenceState.SinglePresenceState
       - SinglePresenceState.key
@@ -2121,14 +2102,6 @@ supporting_symbols:
   - AuthWeakPasswordException.AuthWeakPasswordException
   - AuthWeakPasswordException.reasons
   - AuthWeakPasswordException.toString
-  - Binding
-  - Binding.Binding
-  - Binding.callback
-  - Binding.copyWith
-  - Binding.filter
-  - Binding.id
-  - Binding.type
-  - BindingCallback
   - BroadcastChannel
   - Bucket
   - Bucket.Bucket