stream<I, T> method

Future<Stream<T>> stream<I, T>(ServerStreamFunctionRef<I, T> function, I data, { Future<void>? cancelled, })

Calls a streaming server function.

Completing cancelled stops request and response streaming with RpcCancelledException.

Implementation

Future<Stream<T>> stream<I, T>(
  ServerStreamFunctionRef<I, T> function,
  I data, {
  Future<void>? cancelled,
}) async {
  final response = await _send(
    function.id,
    function.method,
    data,
    encodeInput: function.encodeInput,
    cancelled: cancelled,
  );
  final contentType = response.headers.value('content-type') ?? '';
  if (!contentType.startsWith('application/x-ndjson')) {
    _decodeFrame<Object?>(
      await _readFrame(response, cancelled),
      response.status,
    );
    throw const RpcProtocolException(
      'The server returned one value for a streaming function.',
    );
  }
  if (!_isSuccessful(response.status)) {
    await _cancelBody(response.body);
    _invalidResponse(
      response.status,
      'The server returned a stream for a failed stream request.',
    );
  }
  return _decodeResponseFrames(
    stopOnRpcCancellation(response.body, cancelled),
    response.status,
    function.decodeOutput,
  );
}