initRSocketCallStubs method

void initRSocketCallStubs()

Implementation

void initRSocketCallStubs() {
  //RSocket requestResponse
  requestResponse = (payload) {
    var completer = Completer<Payload>();
    var streamId = streamIdSupplier.nextStreamId(senders)!;
    connection
        .write(FrameCodec.encodeRequestResponseFrame(streamId, payload!));
    senders[streamId] = CompleterSubscriber(completer);
    return completer.future;
  };
  //RSocket fireAndForget
  fireAndForget = (payload) {
    var streamId = streamIdSupplier.nextStreamId(senders)!;
    connection.write(FrameCodec.encodeFireAndForgetFrame(streamId, payload!));
    return Future.value(() {});
  };
  //RSocket requestStream
  requestStream = (payload) {
    var streamId = streamIdSupplier.nextStreamId(senders)!;
    connection.write(FrameCodec.encodeRequestStreamFrame(
        streamId, MAX_REQUEST_N_SIZE, payload!));
    var streamSubscriber = StreamSubscriber(onCancel: () {
      connection.write(FrameCodec.encodeCancelFrame(streamId));
      senders.remove(streamId);
    });
    senders[streamId] = streamSubscriber;
    return streamSubscriber.payloadStream();
  };
  //RSocket metadataPush
  metadataPush = (payload) {
    connection.write(FrameCodec.encodeMetadataFrame(0, payload!));
    return Future.value(() {});
  };
  //Rsocket Channel
  /*requestChannel = (payloads) {
    var streamId = streamIdSupplier.nextStreamId(senders);
    connection.write(FrameCodec.encodeChannelFrame(streamId, MAX_REQUEST_N_SIZE, payload));
    var streamSubscriber = StreamSubscriber();
    senders[streamId] = streamSubscriber;
    return streamSubscriber.payloadStream();
  };*/
}