From dd61be59b37f6cb815881b02185f6dab81f80e3b Mon Sep 17 00:00:00 2001 From: andsel Date: Fri, 30 Dec 2016 00:32:53 +0100 Subject: [PATCH] Switched from http encoder and decoder to codec, and fixed logger to check message type before log --- .../main/java/io/moquette/server/netty/NettyAcceptor.java | 5 +++-- .../io/moquette/server/netty/metrics/MQTTMessageLogger.java | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java index 1612085c..84097945 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java @@ -36,7 +36,9 @@ import io.netty.handler.codec.MessageToMessageEncoder; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpRequestDecoder; import io.netty.handler.codec.http.HttpResponseEncoder; +import io.netty.handler.codec.http.HttpServerCodec; import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; +import io.netty.handler.codec.http.websocketx.Utf8FrameValidator; import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; import io.netty.handler.ssl.SslHandler; import io.netty.handler.timeout.IdleStateHandler; @@ -184,8 +186,7 @@ public class NettyAcceptor implements ServerAcceptor { initFactory(host, port, new PipelineInitializer() { @Override void init(ChannelPipeline pipeline) { - pipeline.addLast("httpEncoder", new HttpResponseEncoder()); - pipeline.addLast("httpDecoder", new HttpRequestDecoder()); + pipeline.addLast(new HttpServerCodec()); pipeline.addLast("aggregator", new HttpObjectAggregator(65536)); pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST)); pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java index 4367e688..a062b5b8 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java @@ -97,7 +97,7 @@ public class MQTTMessageLogger extends ChannelDuplexHandler { @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { - logMQTTMessage(ctx, (AbstractMessage) msg, "C<-B"); + logMQTTMessage(ctx, msg, "C<-B"); ctx.write(msg, promise); } //