GoogleCodeExporter
commented
9 years ago I've managed to successfully connect through WebSockets to moquette broker.
I've changed NettyAcceptor method:
public void initialize(IMessaging messaging, Properties props) throws
IOException {
m_bossGroup = new NioEventLoopGroup();
m_workerGroup = new NioEventLoopGroup();
final NettyMQTTHandler handler = new NettyMQTTHandler();
handler.setMessaging(messaging);
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
{
boolean handleCloseFrames = false;
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("codec-http", new HttpServerCodec());
pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
pipeline.addLast("handler", new WebSocketServerProtocolHandler("/mqtt", "mqttv3.1"));
pipeline.addLast("messageToMessage", new MessageToMessageCodec<WebSocketFrame, AbstractMessage>()
{
@Override
protected void encode(ChannelHandlerContext ctx, AbstractMessage msg, List<Object> out) throws Exception
{
BinaryWebSocketFrame result = new BinaryWebSocketFrame();
MQTTEncoder encoder = new MQTTEncoder();
encoder.encode(ctx, msg, result.content());
}
@Override
protected void decode(ChannelHandlerContext ctx, WebSocketFrame msg, List<Object> out) throws Exception
{
if(msg instanceof BinaryWebSocketFrame)
{
ByteBuf bb = msg.content();
MQTTDecoder mqttDecoder = new MQTTDecoder();
mqttDecoder.decode(ctx, bb, out);
}
}
}) ;
;
//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);
try {
// Bind and start to accept incoming connections.
// ChannelFuture f = b.bind(Constants.PORT);
ChannelFuture f = b.bind(props.getProperty("host"), Integer.parseInt("1884"));
f.sync();
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
Unfortunately just after connection it is lost:
3503 [pool-1-thread-1] INFO ProtocolProcessor - Connected client ID
<clientId-AVKc68oWHt> with clean session true
3505 [pool-1-thread-1] INFO ProtocolProcessor - Lost connection with client
<clientId-AVKc68oWHt>
Do you have any idea how to make it work?Original comment by radoslaw...@gmail.com on 16 Mar 2014 at 4:39
Original issue reported on code.google.com by
radoslaw...@gmail.comon 16 Mar 2014 at 9:15