subscribe method

RealtimeChannel subscribe([ void callback(RealtimeSubscribeStatus status, Object? error)?, Duration? timeout ])

Subscribes to receive real-time changes

Pass a callback to react to different status changes.

timeout parameter can be used to override the default timeout set on RealtimeClient.

Implementation

RealtimeChannel subscribe([
  void Function(RealtimeSubscribeStatus status, Object? error)? callback,
  Duration? timeout,
]) {
  if (!socket.isConnected) {
    unawaited(socket.connect());
  }
  if (joinedOnce == true) {
    throw "tried to subscribe multiple times. 'subscribe' can only be called a single time per channel instance";
  }
  final broadcast = params['config']['broadcast'];
  final presenceConfig = params['config']['presence'];
  final isPrivate = params['config']['private'];

  _onError((e) {
    if (callback != null) callback(RealtimeSubscribeStatus.channelError, e);
  });
  _onClose(() {
    if (callback != null) callback(RealtimeSubscribeStatus.closed, null);
  });

  // A `postgres_changes` subscription's replication setup happens
  // asynchronously on the server, AFTER the phoenix join succeeds. The join
  // reply optimistically echoes the requested config (with server-assigned
  // binding ids), so the join is reported as `subscribed` before the
  // replication is actually live. If that setup then fails (e.g. RLS denies
  // 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
  // `subscribed` yet delivers nothing. Forward it as `channelError` so
  // consumers can detect and recover from the failed subscription.
  onSystemEvents((payload) {
    if (payload is Map && payload['status'] == 'error') {
      if (callback != null) {
        callback(
          RealtimeSubscribeStatus.channelError,
          Exception(
            payload['message']?.toString() ??
                'postgres_changes subscription failed',
          ),
        );
      }
    }
  });

  final presenceEnabled = _shouldEnablePresence();

  final accessTokenPayload = <String, String>{};
  final config = {
    'broadcast': broadcast,
    'presence': {...presenceConfig, 'enabled': presenceEnabled},
    'postgres_changes':
        _bindings['postgres_changes']?.map((r) => r.filter).toList() ?? [],
    'private': isPrivate == true,
  };

  if (socket.accessToken != null) {
    accessTokenPayload['access_token'] = socket.accessToken!;
  }

  updateJoinPayload({'config': config, ...accessTokenPayload});

  joinedOnce = true;
  rejoin(timeout ?? _timeout);

  joinPush
      .receive(
        'ok',
        (response) => unawaited(
          _handleJoinOk(response as Map<String, dynamic>, callback),
        ),
      )
      .receive('error', (error) {
        if (callback != null) {
          callback(
            RealtimeSubscribeStatus.channelError,
            Exception(
              jsonEncode(
                (error as Map<String, dynamic>).isNotEmpty
                    ? (error).values.join(', ')
                    : 'error',
              ),
            ),
          );
        }
      })
      .receive('timeout', (_) {
        if (callback != null) {
          callback(RealtimeSubscribeStatus.timedOut, null);
        }
      });
  return this;
}