streamBidi method

@experimental BidiActionStream<Chunk, Output, Input> streamBidi({ Stream<Input>? inputStream, StreamingCallback<Chunk>? onChunk, Map<String, dynamic>? context, Init? init, CancellationToken? cancel, })

Starts a bidirectional session with this action: send inputs with BidiActionStream.send while consuming output chunks.

Experimental: bidirectional streaming is not covered by semver and may change in any minor release.

Implementation

@experimental
BidiActionStream<Chunk, Output, Input> streamBidi({
  Stream<Input>? inputStream,
  StreamingCallback<Chunk>? onChunk,
  Map<String, dynamic>? context,
  Init? init,
  CancellationToken? cancel,
}) {
  StreamController<Input>? internalInputController;
  if (inputStream == null) {
    internalInputController = StreamController<Input>();
    inputStream = internalInputController.stream;
  }

  final streamController = StreamController<Chunk>();
  final bidiStream = BidiActionStream<Chunk, Output, Input>(
    streamController.stream,
    internalInputController?.sink,
  );

  run(
        null, // Pass null for unary input
        onChunk: (chunk) {
          if (!streamController.isClosed) {
            streamController.add(chunk);
          }
          if (onChunk != null) {
            onChunk(chunk);
          }
        },
        context: context,
        inputStream: inputStream,
        init: init,
        cancel: cancel,
      )
      .then((result) {
        bidiStream.setResult(result.result);
        if (!streamController.isClosed) {
          streamController.close();
        }
      })
      .catchError((Object e, StackTrace s) {
        bidiStream.setError(e, s);
        if (!streamController.isClosed) {
          streamController.addError(e, s);
          streamController.close();
        }
      });

  return bidiStream;
}