RpcFrameMultiplexedChannel constructor

RpcFrameMultiplexedChannel({
  1. required IRpcChannel channel,
  2. required bool isClient,
  3. RpcSecurityPolicy policy = const RpcSecurityPolicy(),
  4. bool? closeOnOversizedFrame,
  5. Duration? prefaceTimeout,
})

Creates a multiplexed channel that encodes/decodes frames over channel.

Each end writes RpcChannelFrame.preface once, before its first frame; see isClient for the order and ready for how a mismatch is reported. On a mismatch incoming fails with the same status and the channel closes.

prefaceTimeout bounds the wait for the peer's preface; past it the channel closes. Null waits for ever, which suits a peer that may take long to start, such as a wasm guest. A server facing the network sets it, so a peer that connects and stays silent costs a connection for that long and no more.

policy bounds the RECEIVE path: a declared frame payload larger than _maxFramePayloadBytes is rejected from the header without buffering, and the reassembly buffer is capped at _maxBufferedFrameBytes.

Implementation

RpcFrameMultiplexedChannel({
  required IRpcChannel channel,
  required this.isClient,
  RpcSecurityPolicy policy = const RpcSecurityPolicy(),
  bool? closeOnOversizedFrame,
  Duration? prefaceTimeout,
}) : _channel = channel,
     _policy = policy,
     closeOnOversizedFrame = closeOnOversizedFrame ?? !isClient {
  // Marked handled, or a failed handshake nobody awaited is an unhandled
  // async error. Whoever awaits [ready] still sees it fail.
  unawaited(_ready.future.catchError((Object _) {}));
  if (prefaceTimeout != null) {
    _prefaceTimer = Timer(
      prefaceTimeout,
      () => _failPreface(
        RpcStatusException(
          RpcStatus.unavailable,
          'No rpc_dart preface from the peer within $prefaceTimeout',
        ),
      ),
    );
  }
  if (isClient) {
    // Before any frame: frames go out through the same channel, in call
    // order.
    unawaited(
      _channel.send(RpcChannelFrame.preface).catchError((Object _) {}),
    );
    _released.complete();
  } else {
    _held = [];
  }
  _channelSub = _channel.incoming.listen(
    _onData,
    onError: (Object e) {
      // A transport failure before the preface is why the handshake failed.
      if (!_ready.isCompleted && e is RpcStatusException) {
        _ready.completeError(e);
      }
      if (!_incomingCtl.isClosed) _incomingCtl.addError(e);
    },
    onDone: () {
      if (!_closed) close();
    },
  );
}