From 69e838742dad61efc956992aa92598b7d6fabca4 Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Thu, 13 Aug 2026 15:08:41 +0200 Subject: [PATCH 1/8] refactor(realtime)!: replace listener callbacks with streams BREAKING CHANGE: RealtimeClient.onOpen/onClose/onError/onMessage are now broadcast Stream getters, RealtimeChannel.onPostgresChanges/onBroadcast return typed streams, onPresenceSync/onPresenceJoin/onPresenceLeave and onSystemEvents are stream getters, and subscribe() no longer takes a status callback; listen to the new RealtimeChannel.onStatusChange stream instead. --- examples/realtime_room/lib/room_channel.dart | 148 ++++---- packages/supabase/example/web/main.dart | 4 +- .../lib/src/supabase_stream_builder.dart | 109 +++--- packages/supabase/test/mock_test.dart | 12 +- packages/supabase/test/realtime_test.dart | 16 +- packages/supabase_flutter/README.md | 59 ++- packages/supabase_realtime/example/main.dart | 54 +-- .../lib/src/realtime_channel.dart | 352 +++++++++--------- .../lib/src/realtime_client.dart | 63 ++-- packages/supabase_realtime/lib/src/types.dart | 18 +- .../supabase_realtime/test/channel_test.dart | 202 +++++----- .../supabase_realtime/test/mock_test.dart | 150 +++----- .../test/realtime_integration_test.dart | 103 +++-- .../supabase_realtime/test/socket_test.dart | 65 ++-- .../test/utils/realtime_test_utils.dart | 43 ++- sdk-compliance.yaml | 6 + 16 files changed, 691 insertions(+), 713 deletions(-) diff --git a/examples/realtime_room/lib/room_channel.dart b/examples/realtime_room/lib/room_channel.dart index 843c5bc1e..6e13dfd25 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, @@ -35,7 +35,36 @@ class RoomChannel { 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; @@ -47,93 +76,57 @@ class RoomChannel { /// messages, typing pings and presence. final String roomName; - final _messageInserted = StreamController.broadcast(); - final _messageDeleted = StreamController.broadcast(); - final _typing = StreamController.broadcast(); - final _onlineUsers = StreamController>.broadcast(); - /// A message someone added to the room (from a Postgres Changes insert /// event). - Stream get onMessageInserted => _messageInserted.stream; + late final Stream onMessageInserted; /// The id of a message someone removed (from a Postgres Changes delete /// event). - Stream get onMessageDeleted => _messageDeleted.stream; + late final Stream onMessageDeleted; /// The username of another client that is currently typing (from a broadcast - /// event). Our own typing pings are filtered out. - Stream get onTyping => _typing.stream; + /// event). + late final Stream onTyping; /// The current room roster (recomputed on every presence sync event). - Stream> get onlineUsers => _onlineUsers.stream; + late final Stream> 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 subscribe() { final ready = Completer(); - _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.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!); + } + }); + + _channel.subscribe(); return ready.future; } @@ -155,12 +148,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 dispose() async { await _client.removeChannel(_channel); - await _messageInserted.close(); - await _messageDeleted.close(); - await _typing.close(); - await _onlineUsers.close(); } } 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 { /// StreamController for `stream()` method. ReplaySubject? _streamController; + /// Subscription on the channel's postgres changes stream. + StreamSubscription? _changesSubscription; + + /// Subscription on the channel's subscription status stream. + StreamSubscription? _statusSubscription; + /// Contains the combined data of postgrest and realtime to emit as stream. SupabaseStreamEvent _streamData = []; @@ -144,6 +150,10 @@ class SupabaseStreamBuilder extends Stream { }, 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 { ), ); - _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()), diff --git a/packages/supabase_flutter/README.md b/packages/supabase_flutter/README.md index 35274012d..48cb7c6d5 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(); ``` #### [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,27 @@ 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()}); } -})); +}); + +myChannel.subscribe(); ``` ### [Storage](https://supabase.com/docs/guides/storage) diff --git a/packages/supabase_realtime/example/main.dart b/packages/supabase_realtime/example/main.dart index df504a75f..35d2ccc7a 100644 --- a/packages/supabase_realtime/example/main.dart +++ b/packages/supabase_realtime/example/main.dart @@ -12,35 +12,39 @@ Future 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}')); + channel.subscribe(); // delay 20s to receive events from server await Future.delayed(const Duration(seconds: 20)); diff --git a/packages/supabase_realtime/lib/src/realtime_channel.dart b/packages/supabase_realtime/lib/src/realtime_channel.dart index 44d0663df..5aed8428e 100644 --- a/packages/supabase_realtime/lib/src/realtime_channel.dart +++ b/packages/supabase_realtime/lib/src/realtime_channel.dart @@ -38,6 +38,18 @@ class RealtimeChannel { /// to check data late final bool _private; + final _statusController = + StreamController.broadcast(); + + /// Controllers backing the event streams handed out by the `on*` members, + /// closed when the channel closes. + final List> _eventControllers = []; + + Stream? _presenceSyncStream; + Stream? _presenceJoinStream; + Stream? _presenceLeaveStream; + Stream? _systemEventsStream; + RealtimeChannel( this.topic, this.socket, { @@ -75,6 +87,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) { @@ -137,14 +153,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()); } @@ -157,10 +170,10 @@ class RealtimeChannel { final isPrivate = parameters['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 @@ -171,20 +184,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', + ), + ); } }); @@ -212,35 +223,53 @@ class RealtimeChannel { .receive( 'ok', (response) => unawaited( - _handleJoinOk(response as Map, callback), + _handleJoinOk(response as Map), ), ) .receive('error', (error) { - if (callback != null) { - callback( - RealtimeSubscribeStatus.channelError, - Exception( - jsonEncode( - (error as Map).isNotEmpty - ? (error).values.join(', ') - : 'error', - ), + _addStatus( + RealtimeSubscribeStatus.channelError, + Exception( + jsonEncode( + (error as Map).isNotEmpty + ? (error).values.join(', ') + : 'error', ), - ); - } + ), + ); }) .receive('timeout', (_) { - if (callback != null) { - callback(RealtimeSubscribeStatus.timedOut, null); - } + _addStatus(RealtimeSubscribeStatus.timedOut); }); return this; } - Future _handleJoinOk( - Map 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 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 _handleJoinOk(Map response) async { final serverPostgresFilters = response['postgres_changes']; final accessToken = socket.accessToken; if (accessToken != null) { @@ -260,9 +289,7 @@ class RealtimeChannel { } if (serverPostgresFilters == null) { - if (callback != null) { - callback(RealtimeSubscribeStatus.subscribed, null); - } + _addStatus(RealtimeSubscribeStatus.subscribed); return; } final clientPostgresBindings = _bindings['postgres_changes']; @@ -294,24 +321,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 presenceState() { @@ -393,28 +416,35 @@ 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. + /// /// ```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 onPostgresChanges({ required PostgresChangeEvent event, String? schema, String? table, PostgresChangeFilter? filter, List? filters, List? select, - required void Function(PostgresChangePayload payload) callback, }) { assert( filter == null || filters == null, @@ -427,7 +457,7 @@ class RealtimeChannel { ]; final filterString = allFilters.isEmpty ? null : allFilters.join(','); - return onEvents( + return _eventStream( 'postgres_changes', ChannelFilter( event: event.toRealtimeEvent(), @@ -436,7 +466,7 @@ class RealtimeChannel { filter: filterString, select: select, ), - (payload, [ref]) => callback(PostgresChangePayload.fromPayload(payload)), + (payload) => PostgresChangePayload.fromPayload(payload), ); } @@ -444,121 +474,87 @@ class RealtimeChannel { /// /// [event] is the broadcast event name to which you want to listen. /// + /// Returns a broadcast stream of the matching messages. + /// /// ```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 payload) callback, - }) { - return onEvents( + Stream> onBroadcast({required String event}) { + return _eventStream( 'broadcast', ChannelFilter(event: event), - (payload, [ref]) => callback(Map.from(payload)), + (payload) => Map.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.from(payload), - ), - ); - }, + Stream 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.from(payload), - ), - ); - }, + Stream 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 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 _presenceStream( + PresenceEvent event, + T Function(Map json) fromJson, ) { - final result = onEvents( + final stream = _eventStream( 'presence', - ChannelFilter( - event: PresenceEvent.leave.name, - ), - (payload, [ref]) { - callback( - RealtimePresenceLeavePayload.fromJson( - Map.from(payload), - ), - ); - }, + ChannelFilter(event: event.name), + (payload) => fromJson(Map.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 @@ -570,33 +566,45 @@ class RealtimeChannel { /// 'room1', /// options: 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.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 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 _eventStream( + String type, + ChannelFilter filter, + T? Function(dynamic payload) transform, + ) { + final controller = StreamController.broadcast(); + 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/supabase_realtime/lib/src/realtime_client.dart b/packages/supabase_realtime/lib/src/realtime_client.dart index 9b3e86b3f..22503175d 100644 --- a/packages/supabase_realtime/lib/src/realtime_client.dart +++ b/packages/supabase_realtime/lib/src/realtime_client.dart @@ -148,13 +148,11 @@ class RealtimeClient { StreamSubscription? _connectionSubscription; @internal List sendBuffer = []; - @internal - Map> stateChangeCallbacks = { - 'open': [], - 'close': [], - 'error': [], - 'message': [], - }; + + final _openController = StreamController.broadcast(); + final _closeController = StreamController.broadcast(); + final _errorController = StreamController.broadcast(); + final _messageController = StreamController>.broadcast(); final _heartbeatController = StreamController.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 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 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 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> 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 message) => @@ -681,9 +674,7 @@ class RealtimeClient { log('transport', 'error while rejoining channels', error, 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) { log('transport', error.toString()); _triggerChanError(error); - for (final callback in stateChangeCallbacks['error']!) { - callback(error); - } + _errorController.add(error as Object); } void _triggerChanError([dynamic error]) { diff --git a/packages/supabase_realtime/lib/src/types.dart b/packages/supabase_realtime/lib/src/types.dart index 8b9213296..670e08a06 100644 --- a/packages/supabase_realtime/lib/src/types.dart +++ b/packages/supabase_realtime/lib/src/types.dart @@ -147,6 +147,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 +513,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/supabase_realtime/test/channel_test.dart b/packages/supabase_realtime/test/channel_test.dart index 471901e14..f570cd8ea 100644 --- a/packages/supabase_realtime/test/channel_test.dart +++ b/packages/supabase_realtime/test/channel_test.dart @@ -116,7 +116,7 @@ void main() { expect(joinPush.timeout, RealtimeConstants.defaultTimeout); - channel.subscribe((_, [_]) {}, newTimeout); + channel.subscribe(newTimeout); expect(joinPush.timeout, newTimeout); }); @@ -144,12 +144,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.value(); - await Future.value(); + // Drain the event queue so the async 'ok' handler completes and the + // status stream delivers. + await Future.delayed(Duration.zero); expect(throwingSocket.setAuthCalls, 1); expect( @@ -179,10 +180,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.value(); - await Future.value(); + await Future.delayed(Duration.zero); }, (_, _) { /* expected: rethrown FormatException */ @@ -208,7 +211,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, @@ -219,10 +222,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] @@ -234,12 +237,13 @@ void main() { {'id': 1, ...sentFilter}, ], }); + await Future.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, @@ -250,10 +254,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] @@ -264,6 +268,7 @@ void main() { {...sentFilter, 'id': 1, 'filter': 'name=in.("a"b\\c")'}, ], }); + await Future.delayed(Duration.zero); expect(status, RealtimeSubscribeStatus.channelError); }); @@ -274,7 +279,6 @@ void main() { schema: 'public', table: 'users', select: ['id', 'first_name'], - callback: (_) {}, ); channel.subscribe(); @@ -302,7 +306,6 @@ void main() { value: ['open', 'pending'], ), ], - callback: (_) {}, ); channel.subscribe(); @@ -353,19 +356,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.delayed(Duration.zero); expect(status, RealtimeSubscribeStatus.channelError); expect(error, isA()); @@ -376,55 +381,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.delayed(Duration.zero); - expect(error, isA()); - expect( - error?.toString(), - contains('postgres_changes subscription failed'), - ); - }); + expect(error, isA()); + 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.delayed(Duration.zero); expect(status, isNot(RealtimeSubscribeStatus.channelError)); }); - test( - 'forwards the raw payload, parseable into a RealtimeSystemPayload', - () { - dynamic received; - channel.onSystemEvents((payload) => received = payload); + test('emits a typed RealtimeSystemPayload', () async { + RealtimeSystemPayload? received; + channel.onSystemEvents.listen((payload) => received = payload); - channel.trigger('system', { - 'extension': 'system', - 'status': 'ok', - 'message': 'Replication connection established', - 'channel': 'topic', - }); - - expect(received, isA>()); + channel.trigger('system', { + 'extension': 'system', + 'status': 'ok', + 'message': 'Replication connection established', + 'channel': 'topic', + }); + await Future.delayed(Duration.zero); - final system = RealtimeSystemPayload.fromJson( - Map.from(received), - ); - expect(system.extension, 'system'); - expect(system.status, 'ok'); - expect(system.message, 'Replication connection established'); - expect(system.channel, 'topic'); - }, - ); + expect(received, isA()); + expect(received!.extension, 'system'); + expect(received!.status, 'ok'); + expect(received!.message, 'Replication connection established'); + expect(received!.channel, 'topic'); + }); }); group('onMessage', () { @@ -503,10 +509,7 @@ void main() { expect(channel.isJoining, isTrue); expect( - () => channel.onPostgresChanges( - event: PostgresChangeEvent.all, - callback: (_) {}, - ), + () => channel.onPostgresChanges(event: PostgresChangeEvent.all), throwsA( allOf( isA(), @@ -522,10 +525,7 @@ void main() { expect(channel.isJoined, isTrue); expect( - () => channel.onPostgresChanges( - event: PostgresChangeEvent.all, - callback: (_) {}, - ), + () => channel.onPostgresChanges(event: PostgresChangeEvent.all), throwsA( allOf( isA(), @@ -537,10 +537,7 @@ void main() { test('allows adding postgres_changes listener before subscribe', () { expect( - () => channel.onPostgresChanges( - event: PostgresChangeEvent.all, - callback: (_) {}, - ), + () => channel.onPostgresChanges(event: PostgresChangeEvent.all), returnsNormally, ); }); @@ -550,7 +547,7 @@ void main() { expect(channel.isJoining, isTrue); expect( - () => channel.onPresenceSync((_) {}), + () => channel.onPresenceSync, returnsNormally, ); }); @@ -560,7 +557,7 @@ void main() { expect(channel.isJoining, isTrue); expect( - () => channel.onBroadcast(event: 'test', callback: (_) {}), + () => channel.onBroadcast(event: 'test'), returnsNormally, ); }); @@ -725,12 +722,10 @@ void main() { }); test('send message via WebSocket when subscribed to channel', () async { - final subscribed = Completer(); - 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. @@ -755,7 +750,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. @@ -827,21 +822,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.delayed(Duration.zero); expect(syncCalled, isTrue); channel.trigger('presence', { 'event': 'join', @@ -849,6 +844,7 @@ void main() { 'newPresences': [], 'currentPresences': [], }, '2'); + await Future.delayed(Duration.zero); expect(joinCalled, isTrue); channel.trigger('presence', { 'event': 'leave', @@ -856,6 +852,7 @@ void main() { 'leftPresences': [], 'currentPresences': [], }, '3'); + await Future.delayed(Duration.zero); expect(leaveCalled, isTrue); }); }); @@ -934,7 +931,7 @@ void main() { config: const RealtimeChannelConfig(), ); - channel.onPresenceSync((payload) {}); + channel.onPresenceSync.listen((payload) {}); channel.subscribe(); final joinPayload = channel.joinPush.payload; @@ -951,7 +948,7 @@ void main() { config: const RealtimeChannelConfig(enabled: true), ); - channel.onPresenceSync((payload) {}); + channel.onPresenceSync.listen((payload) {}); channel.subscribe(); final joinPayload = channel.joinPush.payload; @@ -983,7 +980,7 @@ void main() { config: const RealtimeChannelConfig(), ); - channel.onPresenceJoin((payload) {}); + channel.onPresenceJoin.listen((payload) {}); channel.subscribe(); final joinPayload = channel.joinPush.payload; @@ -997,7 +994,7 @@ void main() { config: const RealtimeChannelConfig(), ); - channel.onPresenceLeave((payload) {}); + channel.onPresenceLeave.listen((payload) {}); channel.subscribe(); final joinPayload = channel.joinPush.payload; @@ -1024,7 +1021,7 @@ void main() { channel.joinPush.trigger('ok', {}); expect(channel.parameters['config']['presence']['enabled'], isFalse); - channel.onPresenceSync((payload) {}); + channel.onPresenceSync.listen((payload) {}); expect(channel.parameters['config']['presence']['enabled'], isTrue); }, @@ -1044,7 +1041,7 @@ void main() { channel.joinPush.trigger('ok', {}); final initialPayload = Map.from(channel.parameters); - channel.onPresenceSync((payload) {}); + channel.onPresenceSync.listen((payload) {}); expect(channel.parameters['config']['presence']['enabled'], isTrue); expect(channel.parameters, equals(initialPayload)); @@ -1064,13 +1061,13 @@ void main() { channel.joinPush.trigger('ok', {}); expect(channel.parameters['config']['presence']['enabled'], isFalse); - channel.onPresenceSync((payload) {}); + channel.onPresenceSync.listen((payload) {}); expect(channel.parameters['config']['presence']['enabled'], isTrue); final payloadAfterFirst = Map.from(channel.parameters); - channel.onPresenceJoin((payload) {}); - channel.onPresenceLeave((payload) {}); + channel.onPresenceJoin.listen((payload) {}); + channel.onPresenceLeave.listen((payload) {}); expect(channel.parameters, equals(payloadAfterFirst)); }, @@ -1088,7 +1085,7 @@ void main() { expect(channel.joinedOnce, isFalse); - channel.onPresenceSync((payload) {}); + channel.onPresenceSync.listen((payload) {}); expect(channel.parameters['config']['presence']['enabled'], isFalse); }, @@ -1097,7 +1094,7 @@ void main() { test( 'should receive presence events after resubscription triggered by adding ' 'callback', - () { + () async { channel = RealtimeChannel( 'topic', socket, @@ -1108,11 +1105,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.delayed(Duration.zero); expect(syncCalled, isTrue); }, @@ -1129,7 +1127,7 @@ void main() { channel.joinPush.trigger('ok', {}); expect(channel.parameters['config']['presence']['enabled'], isFalse); - channel.onPresenceJoin((payload) {}); + channel.onPresenceJoin.listen((payload) {}); expect(channel.parameters['config']['presence']['enabled'], isTrue); }); @@ -1145,7 +1143,7 @@ void main() { channel.joinPush.trigger('ok', {}); expect(channel.parameters['config']['presence']['enabled'], isFalse); - channel.onPresenceLeave((payload) {}); + channel.onPresenceLeave.listen((payload) {}); expect(channel.parameters['config']['presence']['enabled'], isTrue); }); diff --git a/packages/supabase_realtime/test/mock_test.dart b/packages/supabase_realtime/test/mock_test.dart index 90157ea7b..675f3846b 100644 --- a/packages/supabase_realtime/test/mock_test.dart +++ b/packages/supabase_realtime/test/mock_test.dart @@ -187,22 +187,16 @@ void main() { }); test('.on()', () { - final streamController = StreamController(); - - 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', @@ -236,27 +230,21 @@ void main() { }); test('.on() with filter', () { - final streamController = StreamController(); - - 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', @@ -281,34 +269,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()); - error as RealtimeCloseEvent; + if (change.status == RealtimeSubscribeStatus.channelError) { + expect(change.error, isA()); + 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( @@ -453,22 +439,16 @@ void main() { }); test('.on()', () { - final streamController = StreamController(); - - 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', @@ -502,27 +482,21 @@ void main() { }); test('.on() with filter', () { - final streamController = StreamController(); - - 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/supabase_realtime/test/realtime_integration_test.dart b/packages/supabase_realtime/test/realtime_integration_test.dart index d74a98652..dd61b1042 100644 --- a/packages/supabase_realtime/test/realtime_integration_test.dart +++ b/packages/supabase_realtime/test/realtime_integration_test.dart @@ -36,7 +36,7 @@ void main() { test('connects and reports the open state', () async { final opened = Completer(); - 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>(); - 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>(); 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(); - channel.onPresenceSync((_) { + channel.onPresenceSync.listen((_) { if (channel.presenceState().isNotEmpty && !synced.isCompleted) { synced.complete(); } }); final joined = Completer(); - channel.onPresenceJoin((payload) { + channel.onPresenceJoin.listen((payload) { if (!joined.isCompleted) joined.complete(payload); }); @@ -192,7 +186,7 @@ void main() { ); final left = Completer(); - channel.onPresenceLeave((payload) { + channel.onPresenceLeave.listen((payload) { if (!left.isCompleted) left.complete(payload); }); @@ -227,23 +221,24 @@ void main() { final deletes = Completer(); final channel = client.channel('db-changes-${version.wireVersion}'); - 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.delayed(const Duration(seconds: 2)); @@ -282,19 +277,20 @@ void main() { final matched = Completer(); final channel = client.channel('db-filter-${version.wireVersion}'); - 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.delayed(const Duration(seconds: 2)); @@ -330,18 +326,21 @@ void main() { /// Subscribes to [channel] and resolves with the terminal subscribe status. Future _subscribe(RealtimeChannel channel) { final completer = Completer(); - 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) { 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/supabase_realtime/test/socket_test.dart b/packages/supabase_realtime/test/socket_test.dart index ab56f3ae8..3fa2670e6 100644 --- a/packages/supabase_realtime/test/socket_test.dart +++ b/packages/supabase_realtime/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.heartbeatInterval, @@ -124,12 +118,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.heartbeatInterval, const Duration(seconds: 60)); expect( @@ -214,17 +202,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; }); @@ -242,16 +230,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()); + expect(await errorFuture, isA()); }); test('is idempotent', () { @@ -521,8 +506,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); @@ -535,9 +520,11 @@ void main() { expect(mockedSocket.connectionState, SocketState.open); await mockedSocket.disconnect(); + // Wait for the async stream delivery of the close event. + await Future.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(); @@ -972,15 +959,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? 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'); @@ -997,6 +983,8 @@ void main() { ]); socket.onConnectionMessage(frame); + // Wait for the async stream delivery of the broadcast event. + await Future.delayed(Duration.zero); expect(received, { 'type': 'broadcast', @@ -1007,7 +995,7 @@ void main() { test( 'decodes a legacy object frame and dispatches it when version is v1', - () { + () async { final socket = RealtimeClient( socketEndpoint, version: RealtimeProtocolVersion.v1, @@ -1015,10 +1003,9 @@ void main() { final channel = socket.channel('room'); Map? received; - channel.onBroadcast( - event: 'cursor', - callback: (payload) => received = payload, - ); + channel.onBroadcast(event: 'cursor').listen((payload) { + received = payload; + }); socket.onConnectionMessage( json.encode({ @@ -1032,6 +1019,8 @@ void main() { 'ref': null, }), ); + // Wait for the async stream delivery of the broadcast event. + await Future.delayed(Duration.zero); expect(received, { 'type': 'broadcast', @@ -1207,7 +1196,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); @@ -1224,6 +1213,8 @@ void main() { verify(() => erroredChannel.rejoin()).called(1); verifyNever(() => healthyChannel.rejoin()); + // Wait for the async stream delivery of the open event. + await Future.delayed(Duration.zero); expect(opens, 1); expect(socket.connectionState, SocketState.open); diff --git a/packages/supabase_realtime/test/utils/realtime_test_utils.dart b/packages/supabase_realtime/test/utils/realtime_test_utils.dart index b049e1fe2..6671e243a 100644 --- a/packages/supabase_realtime/test/utils/realtime_test_utils.dart +++ b/packages/supabase_realtime/test/utils/realtime_test_utils.dart @@ -116,28 +116,30 @@ Future primePostgresChanges({ final received = Completer(); 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(); - 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) { 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 +184,22 @@ Future waitForRealtimeServer({ httpReachable = await _isRealtimeHttpReachable(); final client = createRealtimeClient(RealtimeProtocolVersion.v1); - client.onError((error) => lastError = error); + client.onError.listen((error) => lastError = error); final completer = Completer(); 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) { completer.complete(false); } }); + channel.subscribe(); var ready = false; try { diff --git a/sdk-compliance.yaml b/sdk-compliance.yaml index d800bedfc..2bec7a119 100644 --- a/sdk-compliance.yaml +++ b/sdk-compliance.yaml @@ -1625,7 +1625,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: From 23bd7ddbcb12722242e07ac228c252c141f98f0f Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Thu, 13 Aug 2026 15:19:25 +0200 Subject: [PATCH 2/8] style: fix DCM lint findings --- packages/supabase_realtime/example/main.dart | 4 +++- packages/supabase_realtime/lib/src/realtime_channel.dart | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/packages/supabase_realtime/example/main.dart b/packages/supabase_realtime/example/main.dart index 35d2ccc7a..dede02898 100644 --- a/packages/supabase_realtime/example/main.dart +++ b/packages/supabase_realtime/example/main.dart @@ -43,7 +43,9 @@ Future main() async { // on connect and subscribe await socket.connect(); - channel.onStatusChange.listen((change) => print('STATUS ${change.status}')); + channel.onStatusChange.listen( + (change) => print('STATUS ${change.status.name}'), + ); channel.subscribe(); // delay 20s to receive events from server diff --git a/packages/supabase_realtime/lib/src/realtime_channel.dart b/packages/supabase_realtime/lib/src/realtime_channel.dart index 5aed8428e..8b98466ca 100644 --- a/packages/supabase_realtime/lib/src/realtime_channel.dart +++ b/packages/supabase_realtime/lib/src/realtime_channel.dart @@ -578,7 +578,7 @@ class RealtimeChannel { 'system', ChannelFilter(), (payload) => payload is Map - ? RealtimeSystemPayload.fromJson(Map.from(payload)) + ? RealtimeSystemPayload.fromJson(Map.from(payload)) : null, ); } From b6f7679ab934ca0b1447dff98055cdbc2fc7a8d2 Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Thu, 13 Aug 2026 15:41:31 +0200 Subject: [PATCH 3/8] refactor(realtime): address review comments Reuse one binding and stream for repeated onPostgresChanges and onBroadcast calls with the same arguments, type _onConnectionError as Object, treat a pre-subscription closed status as a failed subscription in the tests and the room example, and drop an unused variable from the README presence snippet. --- examples/realtime_room/lib/room_channel.dart | 5 ++ packages/supabase_flutter/README.md | 3 +- .../lib/src/realtime_channel.dart | 46 ++++++++++----- .../lib/src/realtime_client.dart | 4 +- .../supabase_realtime/test/channel_test.dart | 57 +++++++++++++++++++ .../test/realtime_integration_test.dart | 3 +- .../test/utils/realtime_test_utils.dart | 6 +- 7 files changed, 103 insertions(+), 21 deletions(-) diff --git a/examples/realtime_room/lib/room_channel.dart b/examples/realtime_room/lib/room_channel.dart index 6e13dfd25..347777fd6 100644 --- a/examples/realtime_room/lib/room_channel.dart +++ b/examples/realtime_room/lib/room_channel.dart @@ -123,6 +123,11 @@ class RoomChannel { ); } 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'), + ); } }); diff --git a/packages/supabase_flutter/README.md b/packages/supabase_flutter/README.md index 48cb7c6d5..558ca00e3 100644 --- a/packages/supabase_flutter/README.md +++ b/packages/supabase_flutter/README.md @@ -432,8 +432,7 @@ myChannel.onPresenceLeave.listen((payload) { 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()}); } }); diff --git a/packages/supabase_realtime/lib/src/realtime_channel.dart b/packages/supabase_realtime/lib/src/realtime_channel.dart index 8b98466ca..ebd404d32 100644 --- a/packages/supabase_realtime/lib/src/realtime_channel.dart +++ b/packages/supabase_realtime/lib/src/realtime_channel.dart @@ -50,6 +50,14 @@ class RealtimeChannel { Stream? _presenceLeaveStream; Stream? _systemEventsStream; + /// Streams handed out by [onPostgresChanges], keyed by the encoded filter, + /// so that repeated calls with the same arguments reuse one binding. + final Map> _postgresChangeStreams = {}; + + /// Streams handed out by [onBroadcast], keyed by the event name, so that + /// repeated calls with the same event reuse one binding. + final Map>> _broadcastStreams = {}; + RealtimeChannel( this.topic, this.socket, { @@ -419,6 +427,7 @@ class RealtimeChannel { /// 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 /// final channel = supabase.channel('my_channel'); @@ -457,16 +466,21 @@ class RealtimeChannel { ]; final filterString = allFilters.isEmpty ? null : allFilters.join(','); - return _eventStream( - '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) => PostgresChangePayload.fromPayload(payload), ); } @@ -474,7 +488,8 @@ class RealtimeChannel { /// /// [event] is the broadcast event name to which you want to listen. /// - /// Returns a broadcast stream of the matching messages. + /// Returns a broadcast stream of the matching messages. Repeated calls with + /// the same [event] return the same stream. /// /// ```dart /// final channel = supabase.channel('my_channel'); @@ -484,10 +499,13 @@ class RealtimeChannel { /// channel.subscribe(); /// ``` Stream> onBroadcast({required String event}) { - return _eventStream( - 'broadcast', - ChannelFilter(event: event), - (payload) => Map.from(payload), + return _broadcastStreams.putIfAbsent( + event, + () => _eventStream( + 'broadcast', + ChannelFilter(event: event), + (payload) => Map.from(payload), + ), ); } diff --git a/packages/supabase_realtime/lib/src/realtime_client.dart b/packages/supabase_realtime/lib/src/realtime_client.dart index 22503175d..ebcf75b6b 100644 --- a/packages/supabase_realtime/lib/src/realtime_client.dart +++ b/packages/supabase_realtime/lib/src/realtime_client.dart @@ -699,10 +699,10 @@ class RealtimeClient { _closeController.add(event); } - void _onConnectionError(dynamic error) { + void _onConnectionError(Object error) { log('transport', error.toString()); _triggerChanError(error); - _errorController.add(error as Object); + _errorController.add(error); } void _triggerChanError([dynamic error]) { diff --git a/packages/supabase_realtime/test/channel_test.dart b/packages/supabase_realtime/test/channel_test.dart index f570cd8ea..dfc0bfe6c 100644 --- a/packages/supabase_realtime/test/channel_test.dart +++ b/packages/supabase_realtime/test/channel_test.dart @@ -498,6 +498,63 @@ 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('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.delayed(Duration.zero); + + expect(firstEvents, 1); + expect(secondEvents, 1); + }); + }); + group('blocking listeners after subscribe', () { setUp(() { socket = RealtimeClient('wss://example.com/socket'); diff --git a/packages/supabase_realtime/test/realtime_integration_test.dart b/packages/supabase_realtime/test/realtime_integration_test.dart index dd61b1042..b198aebd4 100644 --- a/packages/supabase_realtime/test/realtime_integration_test.dart +++ b/packages/supabase_realtime/test/realtime_integration_test.dart @@ -331,7 +331,8 @@ Future _subscribe(RealtimeChannel channel) { if (change.status == RealtimeSubscribeStatus.subscribed) { completer.complete(change.status); } else if (change.status == RealtimeSubscribeStatus.channelError || - change.status == RealtimeSubscribeStatus.timedOut) { + change.status == RealtimeSubscribeStatus.timedOut || + change.status == RealtimeSubscribeStatus.closed) { completer.completeError( StateError( 'Failed to subscribe: ${change.status.name} (${change.error})', diff --git a/packages/supabase_realtime/test/utils/realtime_test_utils.dart b/packages/supabase_realtime/test/utils/realtime_test_utils.dart index 6671e243a..8d5848065 100644 --- a/packages/supabase_realtime/test/utils/realtime_test_utils.dart +++ b/packages/supabase_realtime/test/utils/realtime_test_utils.dart @@ -132,7 +132,8 @@ Future primePostgresChanges({ if (change.status == RealtimeSubscribeStatus.subscribed) { subscribed.complete(); } else if (change.status == RealtimeSubscribeStatus.channelError || - change.status == RealtimeSubscribeStatus.timedOut) { + change.status == RealtimeSubscribeStatus.timedOut || + change.status == RealtimeSubscribeStatus.closed) { subscribed.completeError( StateError('warmup subscribe failed: ${change.status.name}'), StackTrace.current, @@ -195,7 +196,8 @@ Future waitForRealtimeServer({ if (change.status == RealtimeSubscribeStatus.subscribed) { completer.complete(true); } else if (change.status == RealtimeSubscribeStatus.channelError || - change.status == RealtimeSubscribeStatus.timedOut) { + change.status == RealtimeSubscribeStatus.timedOut || + change.status == RealtimeSubscribeStatus.closed) { completer.complete(false); } }); From 205eaafc73cea92c6e0d7292284e94639f03528a Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Thu, 13 Aug 2026 15:44:14 +0200 Subject: [PATCH 4/8] refactor(realtime): complete late-created channel streams and document the migration A stream created after a subscribed channel has closed can never receive events or complete, so it is now handed out already closed. Adds the v2 to v3 migration entries for the callback-to-stream changes. --- MIGRATION.md | 109 ++++++++++++++++++ .../lib/src/realtime_channel.dart | 9 ++ .../supabase_realtime/test/channel_test.dart | 14 +++ 3 files changed, 132 insertions(+) diff --git a/MIGRATION.md b/MIGRATION.md index 55e5b0e74..f06c2f6dd 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -308,6 +308,115 @@ 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.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. + ### Plural enum names singularized A Dart enum type names one value rather than the set, so its name should be singular. Five enums diff --git a/packages/supabase_realtime/lib/src/realtime_channel.dart b/packages/supabase_realtime/lib/src/realtime_channel.dart index ebd404d32..cfc5461f0 100644 --- a/packages/supabase_realtime/lib/src/realtime_channel.dart +++ b/packages/supabase_realtime/lib/src/realtime_channel.dart @@ -613,6 +613,15 @@ class RealtimeChannel { T? Function(dynamic payload) transform, ) { final controller = StreamController.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) { diff --git a/packages/supabase_realtime/test/channel_test.dart b/packages/supabase_realtime/test/channel_test.dart index dfc0bfe6c..050658681 100644 --- a/packages/supabase_realtime/test/channel_test.dart +++ b/packages/supabase_realtime/test/channel_test.dart @@ -537,6 +537,20 @@ void main() { 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.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; From 3edffae3468d092104749fc9988135392d55cb6c Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 14 Aug 2026 11:52:21 +0200 Subject: [PATCH 5/8] refactor(realtime)!: internalize RealtimePresence (#1707) Marks `RealtimePresence`, its helper types (`PresenceOptions`, `PresenceEvents`, `PresenceChooser`, `PresenceOnJoinCallback`, `PresenceOnLeaveCallback`), and the `RealtimeChannel.presence` field as `@internal`, and stops exporting them from the barrel (`realtime_presence.dart` now only exports the `Presence` payload class, which stays public). `RealtimePresence` is presence bookkeeping that leaked into the public API, and it hides a footgun: `onJoin` / `onLeave` / `onSync` are single-slot callback setters, and the constructor installs the forwarders that feed the channel presence streams through those same slots. A user calling `channel.presence.onJoin(...)` therefore silently disabled the channel's `onPresenceJoin` / `onPresenceLeave` / `onPresenceSync` events. Everything the class offered is available on the channel: the presence streams for events and `presenceState()` for the current state. A migration entry documents the before/after. - Stacked on #1706 (the callback-to-stream conversion) since it points users at the stream API; based on `feat/realtime-streams-v3`. - Removes the internalized symbols from `sdk-compliance.yaml`; local symbol and drift checks pass. All realtime unit tests (195) and the integration suite (both protocol versions, run locally against a real Realtime server) pass, plus `supabase` (134) and analyzer/DCM across the workspace. Resolves SDK-1477 --- MIGRATION.md | 26 +++++++++++++++++++ .../lib/src/realtime_channel.dart | 2 ++ .../lib/src/realtime_presence.dart | 17 ++++++++++++ .../lib/supabase_realtime.dart | 2 +- sdk-compliance.yaml | 25 ------------------ 5 files changed, 46 insertions(+), 26 deletions(-) diff --git a/MIGRATION.md b/MIGRATION.md index f06c2f6dd..3153419aa 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -417,6 +417,32 @@ 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. +### `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/packages/supabase_realtime/lib/src/realtime_channel.dart b/packages/supabase_realtime/lib/src/realtime_channel.dart index cfc5461f0..049195a8f 100644 --- a/packages/supabase_realtime/lib/src/realtime_channel.dart +++ b/packages/supabase_realtime/lib/src/realtime_channel.dart @@ -8,6 +8,7 @@ import 'package:meta/meta.dart'; import 'package:supabase_realtime/supabase_realtime.dart'; import 'package:supabase_realtime/src/constants.dart'; import 'package:supabase_realtime/src/push.dart'; +import 'package:supabase_realtime/src/realtime_presence.dart'; import 'package:supabase_realtime/src/retry_timer.dart'; import 'package:supabase_realtime/src/transformers.dart'; import 'package:supabase_realtime/src/types.dart'; @@ -22,6 +23,7 @@ class RealtimeChannel { late Push joinPush; late RetryTimer _rejoinTimer; List _pushBuffer = []; + @internal late RealtimePresence presence; @internal late final String broadcastEndpointUrl; diff --git a/packages/supabase_realtime/lib/src/realtime_presence.dart b/packages/supabase_realtime/lib/src/realtime_presence.dart index cd1575893..0b1c3d7a7 100644 --- a/packages/supabase_realtime/lib/src/realtime_presence.dart +++ b/packages/supabase_realtime/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:supabase_realtime/supabase_realtime.dart'; import 'package:supabase_realtime/src/types.dart'; @@ -37,20 +38,25 @@ class Presence { 'Presence(presenceReference: $presenceReference, payload: $payload)'; } +@internal typedef PresenceChooser = 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); +@internal class PresenceOptions { final PresenceEvents events; const PresenceOptions({required this.events}); } +@internal class PresenceEvents { final String state; final String diff; @@ -58,6 +64,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> state = >{}; List> pendingDiffs = []; diff --git a/packages/supabase_realtime/lib/supabase_realtime.dart b/packages/supabase_realtime/lib/supabase_realtime.dart index 7d2f323fb..7c2a249b7 100644 --- a/packages/supabase_realtime/lib/supabase_realtime.dart +++ b/packages/supabase_realtime/lib/supabase_realtime.dart @@ -7,6 +7,6 @@ export 'src/constants.dart' export 'src/realtime_channel.dart'; export 'src/realtime_client.dart'; export 'src/realtime_constants.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; diff --git a/sdk-compliance.yaml b/sdk-compliance.yaml index 2bec7a119..5d989dc8e 100644 --- a/sdk-compliance.yaml +++ b/sdk-compliance.yaml @@ -1773,17 +1773,7 @@ features: - RealtimeChannel.onPresenceJoin - RealtimeChannel.onPresenceLeave supporting_symbols: - - PresenceChooser - PresenceEvent - - PresenceEvents - - PresenceEvents.PresenceEvents - - PresenceEvents.diff - - PresenceEvents.state - - PresenceOnJoinCallback - - PresenceOnLeaveCallback - - PresenceOptions - - PresenceOptions.PresenceOptions - - PresenceOptions.events - RealtimePresenceJoinPayload - RealtimePresenceJoinPayload.RealtimePresenceJoinPayload - RealtimePresenceJoinPayload.currentPresences @@ -1851,21 +1841,6 @@ features: - Presence.payload - Presence.presenceReference - 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 From 183db0f4a2de206f82c7b6e74f7e9d86724627e0 Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Fri, 14 Aug 2026 12:59:14 +0200 Subject: [PATCH 6/8] refactor(realtime)!: hide the Binding registration primitives (#1708) Marks `Binding` and `BindingCallback` as `@internal` and removes them from the barrel export of `realtime_client`. They are the raw registration primitives underneath the channel listeners, and their only consumers, `RealtimeChannel.onEvents` and `RealtimeChannel.off`, have always been `@internal`. The raw-callback escape hatch underneath the v3 stream API should not be public. This also removed a stray import of the package barrel from `lib/src/message.dart`. The ticket also covered `RealtimeChannel.joinPush` leaking the internal `Push` type, but that field is already annotated `@internal`, so no change was needed there. - Stacked on #1706; based on `feat/realtime-streams-v3`. Independent of - Deregisters the `Binding` symbols from `sdk-compliance.yaml`; local symbol and drift checks pass. - Adds a migration entry. All realtime unit tests (195) and the integration suite (both protocol versions, run locally against a real Realtime server) pass, plus `supabase` (134) and analyzer/DCM across the workspace. Resolves SDK-1478 --- MIGRATION.md | 8 ++++++++ packages/supabase_realtime/lib/src/message.dart | 4 ++-- packages/supabase_realtime/lib/src/types.dart | 2 ++ packages/supabase_realtime/lib/supabase_realtime.dart | 3 ++- sdk-compliance.yaml | 8 -------- 5 files changed, 14 insertions(+), 11 deletions(-) diff --git a/MIGRATION.md b/MIGRATION.md index 3153419aa..ddcca9e58 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -417,6 +417,14 @@ 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`, diff --git a/packages/supabase_realtime/lib/src/message.dart b/packages/supabase_realtime/lib/src/message.dart index 6f32fde2d..ad8933824 100644 --- a/packages/supabase_realtime/lib/src/message.dart +++ b/packages/supabase_realtime/lib/src/message.dart @@ -1,6 +1,6 @@ -import 'package:supabase_realtime/supabase_realtime.dart'; -import 'package:supabase_realtime/src/constants.dart'; import 'package:meta/meta.dart'; +import 'package:supabase_realtime/src/constants.dart'; +import 'package:supabase_realtime/src/types.dart'; @internal class Message { diff --git a/packages/supabase_realtime/lib/src/types.dart b/packages/supabase_realtime/lib/src/types.dart index 670e08a06..07a7d5d6c 100644 --- a/packages/supabase_realtime/lib/src/types.dart +++ b/packages/supabase_realtime/lib/src/types.dart @@ -3,8 +3,10 @@ import 'package:collection/collection.dart'; import 'package:meta/meta.dart'; import 'package:supabase_realtime/supabase_realtime.dart'; +@internal typedef BindingCallback = void Function(dynamic payload, [dynamic ref]); +@internal class Binding { String type; Map filter; diff --git a/packages/supabase_realtime/lib/supabase_realtime.dart b/packages/supabase_realtime/lib/supabase_realtime.dart index 7c2a249b7..2321e38a5 100644 --- a/packages/supabase_realtime/lib/supabase_realtime.dart +++ b/packages/supabase_realtime/lib/supabase_realtime.dart @@ -9,4 +9,5 @@ export 'src/realtime_client.dart'; export 'src/realtime_constants.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/sdk-compliance.yaml b/sdk-compliance.yaml index 5d989dc8e..9d54faf8a 100644 --- a/sdk-compliance.yaml +++ b/sdk-compliance.yaml @@ -2103,14 +2103,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 From 66d1fe209131b8e0da9422446cfbae2d59620db2 Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Mon, 17 Aug 2026 10:24:59 +0200 Subject: [PATCH 7/8] refactor(realtime)!: merge the connection streams into onStatusChange The socket exposed onOpen, onClose, onError and onMessage while a channel reported the same open, closed and error events on a single onStatusChange stream. Give the connection the same shape: onStatusChange for the lifecycle, with errors delivered as stream errors, and onMessage for the decoded frames. Fewer streams also mean fewer subscriptions to cancel, which streams need and the callbacks did not. --- MIGRATION.md | 21 ++++-- .../lib/src/realtime_client.dart | 71 ++++++++++++++----- .../test/realtime_integration_test.dart | 7 +- .../supabase_realtime/test/socket_test.dart | 63 ++++++++++++---- .../test/utils/realtime_test_utils.dart | 5 +- sdk-compliance.yaml | 11 ++- 6 files changed, 134 insertions(+), 44 deletions(-) diff --git a/MIGRATION.md b/MIGRATION.md index ddcca9e58..357ae35b8 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -315,8 +315,10 @@ following the shape `RealtimeClient.onHeartbeat` already had. Streams compose (` `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: +On `RealtimeClient`, the four connection callbacks are replaced by two broadcast streams: +`onStatusChange` for the connection lifecycle and `onMessage` for every decoded frame. Connection +errors are emitted as stream errors on `onStatusChange`, so they arrive through the `onError` +handler of `listen`: ```dart // Before @@ -326,12 +328,21 @@ 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.onStatusChange.listen( + (change) => switch (change.status) { + RealtimeConnectionStatus.open => print('open'), + RealtimeConnectionStatus.closed => print('closed: ${change.closeEvent}'), + }, + onError: (error) => print('error: $error'), +); client.onMessage.listen((message) => print('message: $message')); ``` +Four separate streams for one connection was a different shape than the channel, where all of +open, closed and error already arrive on a single `RealtimeChannel.onStatusChange`. Since a stream +needs `listen` and a cancelled subscription to clean up, one status stream is also less +bookkeeping than three. + 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 diff --git a/packages/supabase_realtime/lib/src/realtime_client.dart b/packages/supabase_realtime/lib/src/realtime_client.dart index ebcf75b6b..e31b101b2 100644 --- a/packages/supabase_realtime/lib/src/realtime_client.dart +++ b/packages/supabase_realtime/lib/src/realtime_client.dart @@ -62,6 +62,35 @@ enum RealtimeHeartbeatStatus { timeout, } +/// The status of the WebSocket connection reported by +/// [RealtimeClient.onStatusChange]. +enum RealtimeConnectionStatus { + /// The connection is open and messages can be sent and received. + open, + + /// The connection is closed, either because [RealtimeClient.disconnect] was + /// called or because it dropped, in which case the client reconnects with + /// backoff. [RealtimeClient.connectionState] tells the two apart. + closed, +} + +/// A connection status change emitted by [RealtimeClient.onStatusChange]. +class RealtimeConnectionStatusChange { + /// The new status of the WebSocket connection. + final RealtimeConnectionStatus status; + + /// The close code and reason sent by the server, `null` for + /// [RealtimeConnectionStatus.open] and for a close without one. + final RealtimeCloseEvent? closeEvent; + + const RealtimeConnectionStatusChange(this.status, [this.closeEvent]); + + @override + String toString() => + 'RealtimeConnectionStatusChange(status: ${status.name}, ' + 'closeEvent: $closeEvent)'; +} + /// Manages a persistent WebSocket connection to the Supabase Realtime server. /// /// [RealtimeClient] is the central hub for all real-time communication. It owns @@ -149,9 +178,8 @@ class RealtimeClient { @internal List sendBuffer = []; - final _openController = StreamController.broadcast(); - final _closeController = StreamController.broadcast(); - final _errorController = StreamController.broadcast(); + final _statusController = + StreamController.broadcast(); final _messageController = StreamController>.broadcast(); final _heartbeatController = @@ -430,23 +458,24 @@ class RealtimeClient { logger?.call(kind, message, data); } - /// Emits whenever the WebSocket connection is opened. + /// Emits whenever the WebSocket connection opens or closes. + /// + /// Connection errors are emitted as stream errors, so they are observed with + /// the `onError` handler of [Stream.listen] rather than as status changes: /// /// ```dart - /// final subscription = client.onOpen.listen((_) { - /// print('Socket opened.'); - /// }); + /// final subscription = client.onStatusChange.listen( + /// (change) => print('Socket ${change.status.name}.'), + /// onError: (error) => print('Socket error: $error'), + /// ); /// ``` - Stream get onOpen => _openController.stream; - - /// Emits whenever the WebSocket connection is closed. /// - /// The emitted [RealtimeCloseEvent] carries the close code and reason sent - /// by the server, or `null` when the connection closed without one. - Stream get onClose => _closeController.stream; - - /// Emits whenever the WebSocket connection reports an error. - Stream get onError => _errorController.stream; + /// The connection level statuses and errors are informational: a dropped + /// connection and its cause also reach every channel through + /// [RealtimeChannel.onStatusChange], which is what a subscription should + /// react to. + Stream get onStatusChange => + _statusController.stream; /// Emits every decoded message received over the WebSocket. Stream> get onMessage => _messageController.stream; @@ -674,7 +703,9 @@ class RealtimeClient { log('transport', 'error while rejoining channels', error, Level.WARNING); } - _openController.add(null); + _statusController.add( + const RealtimeConnectionStatusChange(RealtimeConnectionStatus.open), + ); } /// communication has been closed @@ -696,13 +727,15 @@ class RealtimeClient { reconnectTimer.scheduleTimeout(); } if (heartbeatTimer != null) heartbeatTimer!.cancel(); - _closeController.add(event); + _statusController.add( + RealtimeConnectionStatusChange(RealtimeConnectionStatus.closed, event), + ); } void _onConnectionError(Object error) { log('transport', error.toString()); _triggerChanError(error); - _errorController.add(error); + _statusController.addError(error); } void _triggerChanError([dynamic error]) { diff --git a/packages/supabase_realtime/test/realtime_integration_test.dart b/packages/supabase_realtime/test/realtime_integration_test.dart index b198aebd4..d6a808e96 100644 --- a/packages/supabase_realtime/test/realtime_integration_test.dart +++ b/packages/supabase_realtime/test/realtime_integration_test.dart @@ -36,8 +36,11 @@ void main() { test('connects and reports the open state', () async { final opened = Completer(); - client.onOpen.listen((_) { - if (!opened.isCompleted) opened.complete(); + client.onStatusChange.listen((change) { + if (change.status == RealtimeConnectionStatus.open && + !opened.isCompleted) { + opened.complete(); + } }); await client.connect(); await opened.future.timeout(const Duration(seconds: 10)); diff --git a/packages/supabase_realtime/test/socket_test.dart b/packages/supabase_realtime/test/socket_test.dart index 3fa2670e6..83901343b 100644 --- a/packages/supabase_realtime/test/socket_test.dart +++ b/packages/supabase_realtime/test/socket_test.dart @@ -203,13 +203,9 @@ void main() { }); test('emits connection state events on the streams', () async { - int opens = 0; - socket.onOpen.listen((_) { - opens += 1; - }); - int closes = 0; - socket.onClose.listen((_) { - closes += 1; + final statuses = []; + socket.onStatusChange.listen((change) { + statuses.add(change.status); }); late dynamic lastMessage; socket.onMessage.listen((message) { @@ -218,7 +214,7 @@ void main() { await socket.connect(); await Future.delayed(const Duration(milliseconds: 200)); - expect(opens, 1); + expect(statuses, [RealtimeConnectionStatus.open]); await socket.sendHeartbeat(); // need to wait for event to trigger @@ -227,16 +223,51 @@ void main() { await socket.disconnect(); await Future.delayed(const Duration(seconds: 1)); - expect(closes, 1); + expect(statuses, [ + RealtimeConnectionStatus.open, + RealtimeConnectionStatus.closed, + ]); }); - test('emits errors on the onError stream', () async { + test('emits errors on the onStatusChange stream', () async { final RealtimeClient erroneousSocket = RealtimeClient('badurl'); - final errorFuture = erroneousSocket.onError.first; + final errorFuture = erroneousSocket.onStatusChange.first; unawaited(erroneousSocket.connect()); - expect(await errorFuture, isA()); + await expectLater(errorFuture, throwsA(isA())); + }); + + test('reports the close code and reason with the closed status', () async { + final mockedSocketChannel = MockIOWebSocketChannel(); + final mockedSink = MockWebSocketSink(); + final streamController = StreamController(); + final mockedSocket = RealtimeClient( + socketEndpoint, + reconnectAfter: (tries) => const Duration(seconds: 100), + transport: (url, headers) => mockedSocketChannel, + ); + when(() => mockedSocketChannel.ready).thenAnswer((_) => Future.value()); + when(() => mockedSocketChannel.sink).thenReturn(mockedSink); + when( + () => mockedSocketChannel.stream, + ).thenAnswer((_) => streamController.stream); + when(() => mockedSocketChannel.closeCode).thenReturn(1011); + when(() => mockedSocketChannel.closeReason).thenReturn('server error'); + when(() => mockedSink.close()).thenAnswer((_) => Future.value()); + + final closed = mockedSocket.onStatusChange.firstWhere( + (change) => change.status == RealtimeConnectionStatus.closed, + ); + + await mockedSocket.connect(); + await streamController.close(); + + final change = await closed; + expect(change.closeEvent?.code, 1011); + expect(change.closeEvent?.reason, 'server error'); + + await mockedSocket.disconnect(); }); test('is idempotent', () { @@ -507,7 +538,9 @@ void main() { transport: (url, headers) => mockedSocketChannel, ); var closeEvents = 0; - mockedSocket.onClose.listen((_) => closeEvents += 1); + mockedSocket.onStatusChange + .where((change) => change.status == RealtimeConnectionStatus.closed) + .listen((_) => closeEvents += 1); when(() => mockedSocketChannel.ready).thenAnswer((_) => Future.value()); when(() => mockedSocketChannel.sink).thenReturn(mockedSink); @@ -1196,7 +1229,9 @@ void main() { transport: (url, headers) => mockedSocketChannel, ); var opens = 0; - socket.onOpen.listen((_) => opens += 1); + socket.onStatusChange + .where((change) => change.status == RealtimeConnectionStatus.open) + .listen((_) => opens += 1); when(() => mockedSocketChannel.ready).thenAnswer((_) => Future.value()); when(() => mockedSocketChannel.sink).thenReturn(mockedSink); diff --git a/packages/supabase_realtime/test/utils/realtime_test_utils.dart b/packages/supabase_realtime/test/utils/realtime_test_utils.dart index 8d5848065..0b59b3174 100644 --- a/packages/supabase_realtime/test/utils/realtime_test_utils.dart +++ b/packages/supabase_realtime/test/utils/realtime_test_utils.dart @@ -185,7 +185,10 @@ Future waitForRealtimeServer({ httpReachable = await _isRealtimeHttpReachable(); final client = createRealtimeClient(RealtimeProtocolVersion.v1); - client.onError.listen((error) => lastError = error); + client.onStatusChange.listen( + (_) {}, + onError: (Object error) => lastError = error, + ); final completer = Completer(); final channel = client.channel('readiness-check'); diff --git a/sdk-compliance.yaml b/sdk-compliance.yaml index 9d54faf8a..468b58cdd 100644 --- a/sdk-compliance.yaml +++ b/sdk-compliance.yaml @@ -1659,9 +1659,14 @@ features: - RealtimeClient.connection - RealtimeClient.onConnectionMessage supporting_symbols: - - RealtimeClient.onError - RealtimeClient.onMessage - - RealtimeClient.onOpen + - RealtimeClient.onStatusChange + - RealtimeConnectionStatus + - RealtimeConnectionStatusChange + - RealtimeConnectionStatusChange.RealtimeConnectionStatusChange + - RealtimeConnectionStatusChange.closeEvent + - RealtimeConnectionStatusChange.status + - RealtimeConnectionStatusChange.toString realtime.client.disconnect: status: implemented symbols: @@ -1670,7 +1675,7 @@ features: - RealtimeClientOptions.connectionCloseTimeout - RealtimeConstants.defaultConnectionCloseTimeout supporting_symbols: - - RealtimeClient.onClose + - RealtimeConstants.webSocketCloseNormal - RealtimeCloseEvent - RealtimeCloseEvent.RealtimeCloseEvent - RealtimeCloseEvent.code From 04cd83e0ac0463f23bcd250c3e88d82906f5cf7e Mon Sep 17 00:00:00 2001 From: Lukas Klingsbo Date: Mon, 17 Aug 2026 10:57:31 +0200 Subject: [PATCH 8/8] docs: correct the presence state migration `presence.state` was a map and `presenceState()` returns a list of `SinglePresenceState`, so the two are not interchangeable. Name the type by its v2 name `PresenceOpts` as well, since `PresenceOptions` only exists after the rename documented further down. --- MIGRATION.md | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/MIGRATION.md b/MIGRATION.md index 357ae35b8..1902da397 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -438,7 +438,7 @@ channel streams (`onPostgresChanges`, `onBroadcast`, `onPresenceSync`, `onPresen ### `RealtimePresence` is internal -`RealtimePresence` and its helper types (`PresenceOptions`, `PresenceEvents`, `PresenceChooser`, +`RealtimePresence` and its helper types (`PresenceOpts`, `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 @@ -451,13 +451,24 @@ Everything the class offered is available on the channel: channel.presence.onJoin((key, current, joined) { /* ... */ }); channel.presence.onLeave((key, current, left) { /* ... */ }); channel.presence.onSync(() { /* ... */ }); -final state = channel.presence.state; +final Map> state = channel.presence.state; // After channel.onPresenceJoin.listen((payload) { /* ... */ }); channel.onPresenceLeave.listen((payload) { /* ... */ }); channel.onPresenceSync.listen((payload) { /* ... */ }); -final state = channel.presenceState(); +final List state = channel.presenceState(); +``` + +`presenceState()` is not a drop-in replacement for `presence.state`: it returns a +`List` rather than a map, so a presence key is read from +`SinglePresenceState.key` and its payloads from `SinglePresenceState.presences`. When code depended +on the map, rebuild it from the list: + +```dart +final byKey = { + for (final state in channel.presenceState()) state.key: state.presences, +}; ``` The `Presence` payload class is unchanged and stays public.