How to convert custom data to JSON from a TcpClientTransport in Java

Hi all! I am new to rsocket and need to connect to a Tcp provider which emits several packets of data. After connecting, I need to convert the custom data format separated with pipes into JSON. My sample below successfully connects to the provider but, I don’t know how to handle the data conversion and apply backpressure so 1000 or so packets are processed then continues in that manner.
I’ll REALLY appreciate any help!!!

    public static void main(String[] args) {
        final int port = 8282;
        final String host = "localhost";
        RSocket socket = RSocketFactory.connect()
                                       .frameDecoder(PayloadDecoder.ZERO_COPY)
                                       .transport(TcpClientTransport.create(host, port))
                                       .start()
                                       .block();

        System.out.printf("Tcp Client initialized, connecting to port %d%n", port);

        socket.requestStream(DefaultPayload.create("request-stream-message"))
              .map(Payload::getData)
              .map(ByteBuffer::asReadOnlyBuffer)
              .doOnNext(System.out::println)
              //.take(10)
              .doFinally(signalType -> socket.dispose())
              .then()
              // Block until all the bytes are received
              .onTerminateDetach();
    }
}

Hi,

Take a look at this example- It uses limitRate to to control back-pressure:

In your example onTerminateDetach probably subscribes with a request n of Long.MAX disabling back-pressure. If you use limitRate a smaller request n will be propagated over the network.

The take operator can cause issues too - it subscribers with long.max so you could use limitRequest instead and it will only request n for what you ask for.

Thanks,
Robert