java - 使用 Netty 的 WebSockets 上的 MQTT？

Question

我想通过 Websockets 使用 MQTT。在 Netty 中使用 Websockets 非常简单:

ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("codec-http", new HttpServerCodec());
pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
pipeline.addLast("handler", new WebSocketServerHandler());

我找到了基于 Netty 的 MQTT 代理 (moquette)。

 NettyMQTTHandler handler = new NettyMQTTHandler();
 ServerBootstrap b = new ServerBootstrap();
            b.group(m_bossGroup, m_workerGroup)
             .channel(NioServerSocketChannel.class) 
             .childHandler(new ChannelInitializer<SocketChannel>() { 
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    //pipeline.addFirst("metrics", new BytesMetricsHandler(m_metricsCollector));
                    pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
                    pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
                    //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
                    pipeline.addLast("decoder", new MQTTDecoder());
                    pipeline.addLast("encoder", new MQTTEncoder());
                    pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
                    pipeline.addLast("handler", handler);
                 }
             })
             .option(ChannelOption.SO_BACKLOG, 128)
             .option(ChannelOption.SO_REUSEADDR, true)
             .childOption(ChannelOption.SO_KEEPALIVE, true); 

所以理论上我应该能够通过 Websocket 发送 MQTT，但我不知道是否可以使用 Netty？有没有任何线索或想法如何做到这一点？我应该使用 MessageToMessageCodec 和 BinaryWebSocketFrame 吗？

干杯!

Answer 1

让我假设您的 MQTTDecoder 使用 ByteBuf 并生成一些 MQTT 消息对象，而 MQTTEncoder 做相反的事情，这通常是这种情况.

然后，您的编解码器使用的 ByteBuf 不是 Web Socket 消息。它们需要成为 Web Socket 帧的负载。我会将以下处理程序插入到管道中:

• MessageToMessageDecoder 将 WebSocket 文本(或二进制)帧转换为 ByteBuf，以便 MQTTDecoder 可以使用它。转换应该非常简单 - 只需获取 Web Socket 框架的内容。
• MessageToMessageEncoder 将 ByteBuf 转换为 Web Socket 文本(或二进制)帧，以便 Netty 的 WebSocketFrameEncoder 可以使用它。转换也应该非常简单——只需用 Web Socket 框架对象包装由 MQTTEncoder 编码的 ByteBuf。

生成的管道将如下所示:

1. HttpResponseEncoder
2. HttpRequestDecoder
3. HttpObjectAggregator(65536)
4. WebSocketServerProtocolHandler("/your-websocket-endpoint-path")
5. WebSocketFrameToByteBufDecoder 扩展 MessageToMessageDecoder
6. ByteBufToWebSocketFrameEncoder 扩展 MessageToMessageEncoder
7. MQTTEncoder
8. MQTT解码器
9. MessageMetricsHandler
10. 处理程序

WebSocketServerProtocolHandler 将与您的网络套接字客户端执行必要的握手，并在 WebSocketFrameToByteBufDecoder 之前插入 WebSocketFrameEncoder 和 WebSocketFrameDecoder。成功握手后生成的管道如下所示:

1. WebSocketFrameEncoder
2. WebSocketFrameDecoder
3. WebSocketFrameToByteBufDecoder 扩展 MessageToMessageDecoder
4. ByteBufToWebSocketFrameEncoder 扩展 MessageToMessageEncoder
5. MQTTEncoder
6. MQTT解码器
7. MessageMetricsHandler
8. 处理程序