How to use vert.x-rx to create reactive client-server TCP communication

user_7125243 0
Viewed 1522

I'm currently working on a project which requires TCP communication between an external system and an application which I will write (in Java). As we all know, this can easily be achieved using regular NIO. However, as part of this new project I'm working on, I have to use Vert.x to provide the TCP communication. Please refer to the image below:

enter image description here

On the right, I have my application which runs as a TCP server waiting for a connection from the external system, on the left. I've read that to create a TCP and listen for connections you simple do something like this:

NetServer server = vertx.createNetServer();
server.listen(1234, "localhost", res -> {
  if (res.succeeded()) {
    System.out.println("Server is now listening!");
  } else {
    System.out.println("Failed to bind!");
  }
});

However, the bit I can't figure out is how to handle when the external system connects to my application and sends EchoRequestMessages via TCP. My application has to take the received Buffer of bytes, decode it into a EchoRequestMessage POJO and then encode the EchoResponseMessage into a Buffer of bytes to send back to the external system.

How do i use vert.x-rx to perform reactive programming of the receipt of the EchoRequestMessage, its decoding, the encoding of the EchoResponseMessage, and then sending that back to the external system, all in one builder pattern type setup. I've read about Observables and subscribing, but i can't figure out what to observe or what to subscribe to. Any help would be greatly appreciated.

vert.x reactive reactive-programming java
2 Answers

I find rather complicated to use the RecordParser.fixedSizeMode() method for aggregating incoming packets to one message.

I use RecordParser.newDelimited. If you can manipulate the data sent to your TCP receiver, then instead of prefixing the 4 byte message length to every message sent, you can suffix a unique delimiter to the end of it.

code sample: (I'm not using the RxJava syntax)

sending:

byte[] serialized = SerializationUtils.serialize(o);
senderSocket.write(Buffer.buffer(serialized).appendString(MSG_DELIMITER));
System.out.println("I sent some bytes: " + serialized.length);

reading:

protected static String MSG_DELIMITER = "_I_Am_so0o_UNIQUEQE_";
protected Vertx vertx;
protected final RecordParser parser = RecordParser.newDelimited(MSG_DELIMITER, new Output());

public static main(String[] args) {
    this.vertx = Vertx.vertx();
    this.vertx.deployVerticle(new Receiver());
}

public class Receiver extends AbstractVerticle {
    @Override
    public void start() {

        NetServer receiver = this.vertx.createNetServer();

        receiver.connectHandler(socket ->

                socket.handler(parser::handle));

        receiver.listen(port, host, res -> {
            if (res.succeeded()) {
                System.out.println("Receiver is now listening!");
            } else {
                System.out.println("Failed to bind!");
            }
        });
    }

    @Override
    public void stop() {
        System.out.println("Receiver stopped");
    }
}

public class Output implements Handler<Buffer> {

    @Override
    public void handle(Buffer event) {
        try {
            E deserialized = (E) SerializationUtils.deserialize(event.getBytes());
            queue.put(deserialized);
            System.out.println("I received some bytes: " + event.length());

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
user_31420170
Related