How to use vert.x-rx to create reactive client-ser

问题:

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:

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.

回答1:

To read data from the socket you can use a RecordParser. On a socket connection, data is often separated by newline characters:

RecordParser parser = RecordParser.newDelimited("\n", sock);

A RecordParser is a Vert.x ReadStream so it can be transformed into a Flowable:

FlowableHelper.toFlowable(parser)

Now if an EchoRequestMessage can be created from a Buffer:

public class EchoRequestMessage {
  private String message;

  public static EchoRequestMessage fromBuffer(Buffer buffer) {
    // Deserialize
  }

  public String getMessage() {
    return message;
   }
 }

And an EchoResponseMessage converted to a Buffer:

public class EchoResponseMessage {
  private final String message;

  public EchoResponseMessage(String message) {
    this.message = message;
  }

  public Buffer toBuffer() {
    // Serialize;
  }
}

You can use RxJava operators to implement an echo server flow:

vertx.createNetServer().connectHandler(sock -> {

  RecordParser parser = RecordParser.newDelimited("\n", sock);

  FlowableHelper.toFlowable(parser)
    .map(EchoRequestMessage::fromBuffer)
    .map(echoRequestMessage -> {
      return new EchoResponseMessage(echoRequestMessage.getMessage());
    })
    .subscribe(echoResponseMessage -> sock.write(echoResponseMessage.toBuffer()).write("\n"), throwable -> {
      throwable.printStackTrace();
      sock.close();
    }, sock::close);

}).listen(1234);

[EDIT] If in your protocol messages are not line separated but length-prefixed, then you can create a custom ReadStream:

class LengthPrefixedStream implements ReadStream<Buffer> {
  final RecordParser recordParser;
  boolean prefix = false;

  private LengthPrefixedStream(ReadStream<Buffer> stream) {
    recordParser = RecordParser.newFixed(4, stream);
  }

  @Override
  public ReadStream<Buffer> exceptionHandler(Handler<Throwable> handler) {
    recordParser.exceptionHandler(handler);
    return this;
  }

  @Override
  public ReadStream<Buffer> handler(Handler<Buffer> handler) {
    if (handler == null) {
      recordParser.handler(null);
      return this;
    }
    recordParser.handler(buffer -> {
      if (prefix) {
        prefix = false;
        recordParser.fixedSizeMode(buffer.getInt(0));
      } else {
        prefix = true;
        recordParser.fixedSizeMode(4);
        handler.handle(buffer);
      }
    });
    return this;
  }

  @Override
  public ReadStream<Buffer> pause() {
    recordParser.pause();
    return this;
  }

  @Override
  public ReadStream<Buffer> resume() {
    recordParser.resume();
    return this;
  }

  @Override
  public ReadStream<Buffer> endHandler(Handler<Void> endHandler) {
    recordParser.endHandler(endHandler);
    return this;
  }
}

And transform it to a Flowable:

FlowableHelper.toFlowable(new LengthPrefixedStream(sock))

回答2:

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();
        }
    }
}