How to use the rsocket-flowable.Flowable function in rsocket-flowable

connectionStatus(): Flowable {
    return new Flowable(subscriber => {
      subscriber.onSubscribe({
        cancel: () => {
          this._statusSubscribers.delete(subscriber);
        },
        request: () => {
          this._statusSubscribers.add(subscriber);
          subscriber.onNext(this._status);
        },
      });
    });
  }

receive(): Flowable {
    return new Flowable(subject => {
      let added = false;
      subject.onSubscribe({
        cancel: () => {
          this._receivers.delete(subject);
        },
        request: () => {
          if (!added) {
            added = true;
            this._receivers.add(subject);
          }
        },
      });
    });
  }

RecordsServiceServer.prototype.requestStream = function requestStream(payload) {
    try {
      if (payload.metadata == null) {
        return rsocket_flowable.Flowable.error(new Error('metadata is empty'));
      }
      var method = rsocket_rpc_frames.getMethod(payload.metadata);
      var spanContext = rsocket_rpc_tracing.deserializeTraceData(this._tracer, payload.metadata);
      switch (method) {
        case 'records':
          return this.recordsMetrics(
            this.recordsTrace(spanContext)(new rsocket_flowable.Flowable(subscriber => {
              var binary = !payload.data || payload.data.constructor === Buffer || payload.data.constructor === Uint8Array ? payload.data : new Uint8Array(payload.data);
              return this._service
                .records(service_pb.RecordsRequest.deserializeBinary(binary), payload.metadata)
                .map(function (message) {
                  return {
                    data: Buffer.from(message.serializeBinary()),
                    metadata: Buffer.alloc(0)
                  }
                }).subscribe(subscriber);
              }
            )
          )
        );
        default:
          return rsocket_flowable.Flowable.error(new Error('unknown method'));
      }

connectionStatus(): Flowable {
    return new Flowable(subscriber => {
      subscriber.onSubscribe({
        cancel: () => {
          this.connStatusSubscribers.delete(subscriber);
        },
        request: _ => {
          this.connStatusSubscribers.add(subscriber);
          subscriber.onNext(this.connStatus);
        },
      });
    });
  }

requestStream(payload: Payload): Flowable> {
    const leaseError = this._useLeaseOrError(this._requesterLeaseHandler);
    if (leaseError) {
      return Flowable.error(new Error(leaseError));
    }

    const streamId = this._getNextStreamId(this._receivers);

    return new Flowable(
      subscriber => {
        this._receivers.set(streamId, subscriber);
        let initialized = false;

        subscriber.onSubscribe({
          cancel: () => {
            this._receivers.delete(streamId);
            if (!initialized) {
              return;
            }
            const cancelFrame = {
              flags: 0,
              streamId,
              type: FRAME_TYPES.CANCEL,
            };
            this._connection.sendOne(cancelFrame);

start(): Flowable {
    return new Flowable(subscriber => {
      let server: ws.Server;
      const onClose = () => {
        if (server) {
          server.stop();
        }
        subscriber.onComplete();
      };
      const onError = error => subscriber.onError(error);
      const onConnection = socket => {
        subscriber.onNext(new WSDuplexConnection(socket, this._encoders));
      };
      subscriber.onSubscribe({
        cancel: () => {
          if (!server) {
            return;
          }

receive(): Flowable {
    return new Flowable(subject => {
      subject.onSubscribe({
        cancel: () => {
          this._receivers.delete(subject);
        },
        request: () => {
          this._receivers.add(subject);
        },
      });
    });
  }

map(fn: (data: T) => R): IPublisher {
    return new Flowable(subscriber => this.subscribe(subscriber)).map(fn);
  }
}

requestChannel(payloads: Flowable>) {
    return new Flowable(subscriber => payloads.subscribe(subscriber))
    .lift(sub => 
      new SwitchTransformOperator(
        sub,
        (firstPayload, restOfPayloads) => {
          const method: string = this._getMethod(firstPayload);
          const spanContext = deserializeTraceData(
            this._tracer,
            firstPayload.metadata,
          );
          const unmarshalledData = restOfPayloads
            .map(this._marshaller.unmarshall)
            .map(payload => payload.data);
          return this._getMetricsWrapper(true, method)(
            this._getTracingWrapper(true, method)(spanContext)(
              this._handler[method](
                unmarshalledData,

_handleRequestChannel(streamId: number, frame: RequestChannelFrame): void {
    const existingSubscription = this._subscriptions.get(streamId);
    if (existingSubscription) {
      //Likely a duplicate REQUEST_CHANNEL frame, ignore per spec
      return;
    }

    const payloads = new Flowable(
      subscriber => {
        let firstRequest = true;

        subscriber.onSubscribe({
          cancel: () => {
            this._receivers.delete(streamId);
            const cancelFrame = {
              flags: 0,
              streamId,
              type: FRAME_TYPES.CANCEL,
            };
            this._connection.sendOne(cancelFrame);
          },
          request: n => {
            if (n > MAX_REQUEST_N) {
              n = MAX_REQUEST_N;