问题：

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

钱繁

2023-03-14

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

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

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

问题2
一旦添加了问题1中的行，onNext将以块的形式打印它收到的所有数据。我想我明白了，它在接收时转发数据。
输出

5bd34970-da5a-44e1-b167-0d32d2e70bb2
成功
？[它是一个不可打印的字节4]
12
34

我的最终目标是只获取没有长度标题的消息。任何指针都将受到赞赏。

密码

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

    }

}

p. s.
我从SO响应中定制了如何使用vert. x-rx创建反应式客户端-服务器TCP通信的LengthPrefix edStream，行为没有差异。

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

共有3个答案

荣轶

2023-03-14

我发现使用RecordParser相当复杂。fixedSizeMode（）方法，用于将传入数据包聚合到一条消息。

对于任何寻求更简单解决方案的人，我建议使用RecordParser。新分隔符。

如果您可以操作发送到TCP接收器的数据，那么您可以在每条发送的消息末尾附加一个唯一分隔符，而不是在消息长度的前几个第一个字节。

代码示例：（我没有使用RxJava语法）

发送：

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

阅读：

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

微生智刚

2023-03-14

我无法运行您粘贴的代码。如果您的目标是只获取没有长度头的消息，那么以下内容如何：（我没有使用RecordParser）

import io.reactivex.Flowable;
import io.reactivex.Single;
import io.vertx.core.net.NetServerOptions;
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.NetClient;
import io.vertx.reactivex.core.net.NetServer;
import io.vertx.reactivex.core.net.NetSocket;
import io.vertx.reactivex.core.parsetools.RecordParser;

public class NewTCPRxServer extends AbstractVerticle {

    Single<NetSocket>        observableNetSocket;
    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 -> {
                    socket.handler(buffer -> onNewframe(buffer));
                          })
                          .rxListen(8888);

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

    private void onNewframe(Buffer buffer) {
        int size = buffer.getInt(0);
        Buffer content = buffer.getBuffer(4, size + 4);
        vertx.eventBus().rxSend("MESSAGE", content.toString()).subscribe();
    }

    public static void main(String[] args) {
        NewTCPRxServer server = new NewTCPRxServer();
        Vertx vertxx = Vertx.vertx();
        Single<String> deployment = io.vertx.reactivex.core.RxHelper.deployVerticle(vertxx, server);

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

        // a test client
        io.vertx.reactivex.core.RxHelper.deployVerticle(vertxx, new TestTCPClient()).subscribe();
    }

    static class TestTCPClient extends AbstractVerticle {

        @Override
        public void start() throws Exception {

            vertx.eventBus().consumer("MESSAGE", message -> {
                Object body = message.body();
                // we receive 'hello world' here means the TCP server processed
                // it successfully!
                System.out.println("[Message] - " + body);
            });

            NetClient netclient = vertx.createNetClient();
            netclient.connect(8888, "localhost", result -> {

                if (result.failed()) {
                    System.err.println("Client unable to connect");
                    return;
                }
                NetSocket socket = result.result();

                // sending "hello world" every second
                vertx.periodicStream(1000).handler(tick -> {
                    String msg = "Hello world";
                    Buffer frame = Buffer.buffer(msg.length() + 4);
                    frame.appendInt(msg.length());
                    frame.appendString(msg);
                    socket.write(frame);
                });

            });
        }

    }

}

我在最后创建了一个测试客户端，该客户端部署在同一个vertx实例上。

姬振

2023-03-14

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

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

            }
        });

    }

}

类似资料：

BigInteger没有最大长度吗？如果它有，我如何找到BigInteger DataType的最大长度？

当我尝试对BigInteger使用MAX_VALUE时，它给出>>ERROR>>找不到符号变量MAX_VALUE
发电机输出的长度

问题内容： Python提供了一种很好的方法来获取渴望的迭代的长度。但是对于以生成器理解和函数表示的惰性可迭代对象，我找不到类似的东西。当然，编写类似以下内容并不难：但是我无法摆脱自己重新设计自行车的感觉。（当我键入函数时，我想到了一个想法：也许确实没有这样的函数，因为它“破坏了”其参数。不过，对于我的情况而言，这不是问题）。 PS：关于第一个答案-是的，类似的方法也可以工作，但是会大大增加内
列出所有具有最大长度和最大长度的SQL列

问题内容：我试图从表中获取所有列的列表，这些列表包含它们的数据类型，数据长度和该列中最长值的长度。我使用此SQL来获取列及其数据类型和长度：我有此SQL，用于获取值的最大长度：但是我不知道如何将它们结合起来。我正在使用SQL Server 2008。问题答案：感谢您的建议。我想出了以下解决方案。它为我获取了我需要的数据，但是希望了解它是否可以提高效率。
responseEntity-头中没有内容-长度

使用简单的按预期设置所有标头。什么样的问题可能导致这种行为？
密钥长度对输出长度的影响

考虑以下java代码：此代码输出“”。所以是64字节，而SHA-256是32字节哈希。我知道我指定了512位（64字节）作为密钥长度然而，我希望生成的密钥（PBKDF2）将由SHA-256进行散列，这样无论我使用的密钥大小如何，输出都应始终为32字节。我错过了什么（或者为什么我的期望是错误的）？
JMS消息没有使用spring JMSTemplate重新出现？

我最近两天都在处理简单的任务，我已经在JBOSS6.x中配置了jms消息队列。我在spring中使用applicationContext.xml文件中的jmstemplate配置了jndi、jms MQ, 下面是邮件发送者

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

共有3个答案

相关问答

相关文章

相关阅读

相关工具

相关文档