java - 如何使 Vertx Reactive RecordParser 发出没有长度的消息？

Question

我有一个响应式(Reactive) vertx TCPserver，它使用 RecordParser 来处理具有 2 字节长度 header 的消息。 RecordParser 的工作是理解长度部分和消息部分。

问题1
recordparser 正确解释 recordparser 内的数据和系统输出。但是当我添加以下行时，系统输出不再打印。它可能与它在不同的线程上执行有关？我不确定。

frameParser.toFlowable()
         .subscribe(onNext -> System.out.println(onNext));  

问题2
添加问题 1 中的行后，onNext 会分块打印其接收到的所有数据。我想我明白，它在收到时转发数据。
输出

5bd34970-da5a-44e1-b167-0d32d2e70bb2
OnSuccess
? [its a non-printable byte 4]
12
34

我的最终目标是只获取没有长度 header 的消息。任何指针将不胜感激。

---

代码

// data that is sent by client byte[] b = { 0, 4, 0x31, 0x32, 0x33, 0x34 };

import io.reactivex.Flowable;
import io.reactivex.Single;
import io.vertx.core.net.NetServerOptions;
import io.vertx.reactivex.RxHelper;
import io.vertx.reactivex.core.AbstractVerticle;
import io.vertx.reactivex.core.Vertx;
import io.vertx.reactivex.core.buffer.Buffer;
import io.vertx.reactivex.core.net.NetServer;
import io.vertx.reactivex.core.net.NetSocket;
import io.vertx.reactivex.core.parsetools.RecordParser;

public class TCPRxServer extends AbstractVerticle {

    Single<NetSocket>        observableNetSocket;
    private FrameToken       expectedToken = FrameToken.SIZE;
    RecordParser             parser, frameParser = null;
    static Flowable<Buffer>  y;
    static Single<NetServer> observable;

    @Override
    public void start() throws Exception {
        NetServerOptions opt = new NetServerOptions();
        opt.setIdleTimeout(500); // large to assist in debugging for now
        observable = vertx.createNetServer(opt)
                          .connectHandler(socket -> {
                              frameParser = get2byteFrameParser();
//                              frameParser.toFlowable()
//                                         .subscribe(onNext -> System.out.println(onNext));

                              socket.handler(frameParser);

                          })
                          .rxListen(8888);

        observable.subscribe(onSuccess -> System.out.println("OnSuccess"), onError -> onError.printStackTrace());
    }

    protected RecordParser get2byteFrameParser() {
        return RecordParser.newFixed(2, buffer -> {

            switch (expectedToken) {

                case SIZE:

                    byte[] b = buffer.getString(0, 2)
                                     .getBytes();
                    int frameSize = (((b[0]) & 0xFF) << 8) | ((b[1]) & 0xFF);
                    System.out.println("size: " + frameSize);
                    frameParser.fixedSizeMode(frameSize);
                    expectedToken = FrameToken.PAYLOAD;
                    break;

                case PAYLOAD:
                    System.out.println("Payload: |" + buffer + "|");
                    frameParser.fixedSizeMode(2);
                    expectedToken = FrameToken.SIZE;
                vertx.eventBus() // this solves the sending messsge in the non rx api. When I turn the recordparser to a flowable I still cant get the parser logic to parse.
                     .rxSend("MESSAGE", buffer.toString())
                     .subscribe();

                    break;
            }

        });
    }

    public static void main(String[] args) throws InterruptedException {

        TCPRxServer server = new TCPRxServer();
        Single<String> deployment = io.vertx.reactivex.core.RxHelper.deployVerticle(Vertx.vertx(), server);

        deployment.subscribe(id -> {
            // Deployed
            System.out.println(id);
            //
        }, err -> {
            // Could not deploy
            err.printStackTrace();
        });

    }

}  

附注
我已经根据 How to use vert.x-rx to create reactive client-server TCP communication 的 SO 响应定制了 LengthPrefixedStream行为上没有差异。

    observable = vertx.createNetServer(opt)
                      .connectHandler(socket -> {
                          new LengthPrefixedStream(socket).toFlowable()
                                                          .subscribe(onNext -> System.out.println(onNext));
                      })
                      .rxListen(8888);

Answer 1

来自 vertx google 用户组的 @tsegismont 提供了适合我的代码。

LengthPrefixedStream.java

import io.vertx.core.Handler;
import io.vertx.reactivex.core.buffer.Buffer;
import io.vertx.reactivex.core.parsetools.RecordParser;
import io.vertx.reactivex.core.streams.ReadStream;

public class LengthPrefixedStream implements io.vertx.core.streams.ReadStream<Buffer> {
    private final RecordParser recordParser;

    private FrameToken         expectedToken = FrameToken.SIZE;

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

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

    @Override
    public io.vertx.core.streams.ReadStream<Buffer> handler(Handler<Buffer> handler) {

        if (handler == null) {
            recordParser.handler(null);
            return this;
        }
        recordParser.handler(buffer -> {
            switch (expectedToken) {

                case SIZE:

                    byte[] b = buffer.getString(0, 2)
                                     .getBytes();
                    int frameSize = (((b[0]) & 0xFF) << 8) | ((b[1]) & 0xFF);
                    System.out.println("size: " + frameSize);
                    recordParser.fixedSizeMode(frameSize);
                    expectedToken = FrameToken.PAYLOAD;
                    break;

                case PAYLOAD:
                    System.out.println("Payload: |" + buffer + "|");
                    recordParser.fixedSizeMode(2);
                    expectedToken = FrameToken.SIZE;
                    handler.handle(buffer);

                    break;
            }
        });
        return this;
    }

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

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

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

TCPRxServer.java

import io.reactivex.Single;
import io.reactivex.disposables.Disposable;
import io.vertx.core.VertxOptions;
import io.vertx.core.net.NetServerOptions;
import io.vertx.reactivex.FlowableHelper;
import io.vertx.reactivex.core.AbstractVerticle;
import io.vertx.reactivex.core.RxHelper;
import io.vertx.reactivex.core.Vertx;
import io.vertx.reactivex.core.net.NetServer;
import io.vertx.reactivex.core.parsetools.RecordParser;

public class TCPRxServer extends AbstractVerticle {

    RecordParser       frameParser = null;
    private NetServer  server;
    private Disposable disposable;

    @Override
    public void start() throws Exception {

        NetServerOptions opt = new NetServerOptions();
        opt.setIdleTimeout(500); // large to assist in debugging for now
        server = vertx.createNetServer(opt);// Create a TCP server

        Single<NetServer> rxSingleServer = server.connectHandler(socket -> {
            LengthPrefixedStream frameParser = new LengthPrefixedStream(socket);
            FlowableHelper.toFlowable(frameParser)
                          .subscribe(onNext -> System.out.println(onNext));

        })
                                                 .rxListen(8888, "localhost");
        disposable = rxSingleServer.subscribe(s -> System.out.println("Server is listening"),
                                              onError -> System.out.println(onError));

    }

    @Override
    public void stop() throws Exception {

        disposable.dispose();
        server.close(completionHandler -> {
            if (completionHandler.succeeded()) {
                System.out.println("TCPRxServer stopped");
            }
            else {
                System.out.println(completionHandler.cause());
            }
        });
    }

    public static void main(String[] args) {

        VertxOptions ops = new VertxOptions();

        // We are using clustered as we have verticles in different VMs and we
        // want to make sure when something is placed on the eventbus, a
        // verticle on a different VM receives it.
        ops.setClustered(true);
        Vertx.clusteredVertx(ops, result -> {
            if (result.succeeded()) {

                Single<String> deployment = RxHelper.deployVerticle(result.result(), new TCPRxServer());

                deployment.subscribe(id -> {
                    System.out.println(id);
                    // Deployed
                }, err -> {
                    err.printStackTrace();
                });

            }
        });

    }

}