subscribe<T> function

XRPCResponse<Subscription<T>> subscribe<T>(NSID methodId, { Protocol protocol = Protocol.https, String? service, Map<String, dynamic>? parameters, ResponseDataBuilder<T>? to, ResponseDataAdaptor? adaptor, WebSocketChannelFactory? channelFactory, })

Subscribes endpoints associated with methodId in WebSocket.

Error Handling and Stream Lifecycle

• When the WebSocket connection fails or reports an error, the error is added to Subscription.stream and the stream is closed. Listeners always receive a done event afterwards.
• When the server closes the connection, Subscription.stream is closed and listeners receive a done event.
• When an individual event cannot be converted with adaptor or to, the conversion error is added to Subscription.stream and the subscription continues with the next event.
• Subscription.close cleans up the WebSocket connection, the internal subscription, and the stream, and always delivers a done event.

Protocol

The WebSocket scheme is derived from protocol: wss for Protocol.https (default) and ws for Protocol.http.

Mocking Channel

When testing, pass channelFactory to inject a mocked WebSocketChannel instead of establishing a real connection.

Note that the returned XRPCResponse is constructed synchronously before the connection is established; connection failures are reported through Subscription.stream as described above.

Implementation

XRPCResponse<Subscription<T>> subscribe<T>(
  final nsid.NSID methodId, {
  final Protocol protocol = Protocol.https,
  final String? service,
  final Map<String, dynamic>? parameters,
  final type.ResponseDataBuilder<T>? to,
  final type.ResponseDataAdaptor? adaptor,
  final type.WebSocketChannelFactory? channelFactory,
}) {
  final uri = _buildWsUri(methodId, service, parameters, protocol);
  final channel = (channelFactory ?? WebSocketChannel.connect).call(uri);

  //! Connection failures are also delivered through `channel.stream`,
  //! so just prevent an unhandled async error here.
  unawaited(channel.ready.then((_) {}, onError: (_) {}));

  //! Assigned synchronously below, before any of the controller callbacks
  //! (which only fire once a listener interacts with the stream) can run.
  late final StreamSubscription<dynamic> subscription;

  //! Tears down the underlying channel subscription and socket. Safe to
  //! call multiple times: cancelling an already-cancelled subscription and
  //! closing an already-closed sink are both no-ops.
  Future<void> teardownChannel() async {
    await subscription.cancel();
    await Future.sync(channel.sink.close).then((_) {}, onError: (_) {});
  }

  final controller = StreamController<T>(
    //! Nothing is pulled from the socket until a consumer actually listens:
    //! the underlying channel subscription is created paused (see below) and
    //! only resumed here on the first listen. Without this, `channel.stream`
    //! would start draining the socket the moment `subscribe()` is called and
    //! buffer every frame inside this controller while no listener drains it —
    //! an unbounded firehose OOM / socket-leak path for a consumer that delays
    //! or never listens. Backpressure (`onPause`) can only engage once a
    //! listener exists, so deferring the pull to `onListen` is what actually
    //! bounds the pre-listen buffer.
    onListen: () => subscription.resume(),
    //! Propagate backpressure: when the consumer pauses, stop pulling from
    //! the socket so events are not buffered unboundedly (firehose OOM
    //! path); resume pulling when the consumer resumes.
    onPause: () => subscription.pause(),
    onResume: () => subscription.resume(),
    //! A consumer that cancels its subscription (without calling
    //! [Subscription.close]) must not leak the socket: tear it down here.
    onCancel: teardownChannel,
  );

  void closeController() {
    if (!controller.isClosed) {
      //! Don't await: the future returned by [StreamController.close]
      //! completes only after a listener has received the done event,
      //! and would therefore hang if the stream is never listened to.
      unawaited(controller.close());
    }
  }

  //! The subscription is owned by the returned [Subscription] and is
  //! cancelled by [Subscription.close].
  // ignore: cancel_subscriptions
  subscription = channel.stream.listen(
    (event) {
      if (controller.isClosed) return;

      try {
        final data = adaptor != null ? adaptor.call(event) : event;

        controller.add(to != null ? to.call(data) : data as T);
      } catch (error, stackTrace) {
        //! A single malformed event must not kill the subscription:
        //! report the conversion failure and keep listening.
        controller.addError(error, stackTrace);
      }
    },
    onError: (Object error, StackTrace stackTrace) {
      if (!controller.isClosed) {
        controller.addError(error, stackTrace);
      }

      closeController();
      unawaited(Future.sync(channel.sink.close).then((_) {}, onError: (_) {}));
    },
    onDone: () {
      closeController();
      unawaited(Future.sync(channel.sink.close).then((_) {}, onError: (_) {}));
    },
  );

  //! Start paused so the socket is not drained before a consumer listens on
  //! [Subscription.stream]; [StreamController.onListen] resumes it on the
  //! first listen. This pause is balanced by that single `onListen` resume,
  //! leaving the normal immediate-listen path unaffected.
  subscription.pause();

  return XRPCResponse<Subscription<T>>(
    headers: const {},
    status: HttpStatus.ok,
    request: XRPCRequest(method: HttpMethod.get, url: uri),
    rateLimit: RateLimit.unlimited(),
    data: Subscription(
      channel: channel,
      controller: controller,
      subscription: subscription,
    ),
  );
}