Netty, ProtoBuf, WebSocket; Как преобразовать BinaryWebSocketFrame в тип protobuf?

Question

Netty, ProtoBuf, WebSocket; Как преобразовать BinaryWebSocketFrame в тип protobuf?

У меня есть сервер с использованием protobuf. Обработчик определяется как -

public class ServerHandler extends SimpleChannelInboundHandler<MyProtoBufType> {

Трубопровод выглядит следующим образом -

pipeline.addLast(new ProtobufVarint32FrameDecoder());
pipeline.addLast(new ProtobufDecoder(MyProtoBufType.getDefaultInstance()));
pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
pipeline.addLast(new ProtobufEncoder());
pipeline.addLast("ServerHandler", new ServerHandler());

С этим и клиентом Java все работает отлично.

Теперь, чтобы иметь возможность подключаться с помощью веб-сокета Javascript, я изменил конвейер сервера следующим образом:

pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new HttpObjectAggregator(65536));
pipeline.addLast(new WebSocketServerCompressionHandler());
pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
pipeline.addLast(new MessageToMessageDecoder<WebSocketFrame>() {
    @Override
    protected void decode(ChannelHandlerContext ctx, WebSocketFrame frame, List<Object> objs) throws Exception {
        ByteBuf buf = ((BinaryWebSocketFrame) frame).content();
        objs.add(buf);
        buf.retain();
    }
});
pipeline.addLast(new ProtobufVarint32FrameDecoder());
pipeline.addLast(new ProtobufDecoder(MyProtoBufType.getDefaultInstance()));
pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
pipeline.addLast(new ProtobufEncoder());
pipeline.addLast("ServerHandler", new ServerHandler());

Ниже приведен код клиента Javascript, основанный на библиотеке protobufjs.

var ProtoBuf = dcodeIO.ProtoBuf,
    builder = ProtoBuf.loadProtoFile("./example.proto"),
    pkg = builder.build("com.example");
    socket = new WebSocket("ws://localhost:8484");

socket.binaryType = "arraybuffer"; 

socket.onopen = function() {
    var login = new pkg.Login({"userName": "hello", "password": "pass"});
    var wrapper = new pkg.MyProtoBufType();
    wrapper.set("login", login);
    socket.send(wrapper.toArrayBuffer());
};

Когда приведенный выше код выполняется, на сервере происходит сбой в декодере protobuf. Следующее - это трассировка стека -

WARN [nioEventLoopGroup-3-6] (DefaultChannelPipeline.java:1144) - An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
io.netty.handler.codec.DecoderException: com.google.protobuf.InvalidProtocolBufferException: While parsing a protocol message, the input ended unexpectedly in the middle of a field.  This could mean either than the input has been truncated or that an embedded message misreported its own length.
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:98)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:335)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:288)
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:275)
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:391)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:243)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:335)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:102)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:335)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:102)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:335)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
    at io.netty.handler.codec.http.websocketx.Utf8FrameValidator.channelRead(Utf8FrameValidator.java:77)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:335)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    at io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler$1.channelRead(WebSocketServerProtocolHandler.java:159)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:335)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:102)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:335)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:288)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:262)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:335)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    at io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:233)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:335)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:327)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1336)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:335)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:911)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:123)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:544)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:485)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:399)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:371)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:742)
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:145)
    at java.lang.Thread.run(Unknown Source)
Caused by: com.google.protobuf.InvalidProtocolBufferException: While parsing a protocol message, the input ended unexpectedly in the middle of a field.  This could mean either than the input has been truncated or that an embedded message misreported its own length.
    at com.google.protobuf.InvalidProtocolBufferException.truncatedMessage(InvalidProtocolBufferException.java:70)
    at com.google.protobuf.CodedInputStream.readRawBytesSlowPath(CodedInputStream.java:1187)
    at com.google.protobuf.CodedInputStream.readBytes(CodedInputStream.java:517)
    at com.google.protobuf.UnknownFieldSet$Builder.mergeFieldFrom(UnknownFieldSet.java:501)
    at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:478)
    at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:596)
    at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:281)
    at com.google.protobuf.CodedInputStream.readGroup(CodedInputStream.java:424)
    at com.google.protobuf.UnknownFieldSet$Builder.mergeFieldFrom(UnknownFieldSet.java:505)
    at com.google.protobuf.GeneratedMessage.parseUnknownField(GeneratedMessage.java:205)
    at com.exmaple.MyProtoBufType.<init>(CallBreakProtocol.java:15290)
    at com.exmaple.MyProtoBufType.<init>(CallBreakProtocol.java:15273)
    at com.exmaple.MyProtoBufType$1.parsePartialFrom(CallBreakProtocol.java:15638)
    at com.exmaple.MyProtoBufType$1.parsePartialFrom(CallBreakProtocol.java:1)
    at com.google.protobuf.AbstractParser.parsePartialFrom(AbstractParser.java:137)
    at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:168)
    at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:174)
    at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:49)
    at io.netty.handler.codec.protobuf.ProtobufDecoder.decode(ProtobufDecoder.java:121)
    at io.netty.handler.codec.protobuf.ProtobufDecoder.decode(ProtobufDecoder.java:64)
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:88)
    ... 52 more

Кто-нибудь использовал Netty, ProtoBuf и WebSocket вместе? Как этого достичь?

2

javascript java protocol-buffers netty protobufjs

Источник

user738746 13 июл '16 в 23:11

3 ответа

Решение

я считаю, что ProtobufVarint32FrameDecoder — это оптимизация самой сети, что означает, что клиенту также необходимо использовать ProtobufVarint32LengthFieldPrepender в соответствии с этим https://issues.redhat.com/browse/NETTY-203?page=com.atlassian.jira.plugin.system.issuetabpanels%3Aall-tabpanel

0

Источник

user3507175 24 янв '22 в 11:57

Вы можете отправить varint, который ProtobufVarint32FrameDecoder ищет из javascript, с помощью этой функции:

      function sendSeiralizedMsgWithVarInt(sock, data) {
    
    let dataByteLen = data.byteLength;
    
    let dynamicArray = [];

    let done = false;
    while (!done) {
        if ((dataByteLen & ~0x7F) === 0) {
            dynamicArray.push(dataByteLen);
            done = true;
        } else {
            out.writeByte((dataByteLen & 0x7F) | 0x80);
            dataByteLen >>>= 7;
        }
    }
    
    const uint8Array = new Uint8Array(dynamicArray);
    sock.send(uint8Array);
    sock.send(data);
}

0

Источник

user3228402 13 сен '23 в 20:01

Другие вопросы по тегам javascript java protocol-buffers netty protobufjs

user738746 14 июл '16 в 00:28 2016-07-14 00:28 · Accepted Answer · 2016-07-14 00:28

Мне нужно было только удалить следующее -

pipeline.addLast(new ProtobufVarint32FrameDecoder());

и сейчас работает нормально.

1

Источник

user738746 14 июл '16 в 00:28