使用Netty在WebSockets上进行MQTT？

我想在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() { 
                 @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吗？

干杯!

让我假设您MQTTDecoder消耗ByteBufs并生成一些MQTT消息对象,并MQTTEncoder执行相反的操作,通常就是这种情况.

然后,ByteBuf您的编解码器使用的s不是Web Socket消息.它们需要成为Web Socket帧的有效载荷.我会将以下处理程序插入管道:

阿MessageToMessageDecoder是一种把WebSocket的文本(或二进制)帧分割为ByteBuf使得MQTTDecoder能够使用它.转换应该非常简单 - 只需获取Web Socket框架的内容即可.

A MessageToMessageEncoder将一个ByteBuf转换为Web Socket文本(或二进制)框架,以便Netty WebSocketFrameEncoder可以使用它.转换也应该非常简单 - 只需用Web Socket框架对象包装ByteBuf编码MQTTEncoder.

生成的管道将如下所示:

HttpResponseEncoder

HttpRequestDecoder

HttpObjectAggregator(65536)

WebSocketServerProtocolHandler("/your-websocket-endpoint-path")

WebSocketFrameToByteBufDecoder 扩展 MessageToMessageDecoder

ByteBufToWebSocketFrameEncoder 扩展 MessageToMessageEncoder

MQTTEncoder

MQTTDecoder

MessageMetricsHandler

handler

WebSocketServerProtocolHandler将与您的Web套接字客户端进行必要的握手,WebSocketFrameEncoder并WebSocketFrameDecoder在之前插入WebSocketFrameToByteBufDecoder.成功握手后生成的管道将如下所示:

WebSocketFrameEncoder

WebSocketFrameDecoder

WebSocketFrameToByteBufDecoder 扩展 MessageToMessageDecoder

ByteBufToWebSocketFrameEncoder 扩展 MessageToMessageEncoder

MQTTEncoder

MQTTDecoder

MessageMetricsHandler

handler