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 0e766eb5..27b19220 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java @@ -38,6 +38,7 @@ import io.netty.handler.codec.http.HttpRequestDecoder; import io.netty.handler.codec.http.HttpResponseEncoder; import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; +import io.netty.handler.logging.LoggingHandler; import io.netty.handler.ssl.SslHandler; import io.netty.handler.timeout.IdleStateHandler; import io.netty.util.concurrent.Future; @@ -158,11 +159,12 @@ public class NettyAcceptor implements ServerAcceptor { void init(ChannelPipeline pipeline) { pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); - //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); +// pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); pipeline.addLast("decoder", new MQTTDecoder()); pipeline.addLast("encoder", new MQTTEncoder()); pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); + pipeline.addLast("messageLogger", new MQTTMessageLogger()); pipeline.addLast("handler", handler); } }); 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 new file mode 100644 index 00000000..318647e8 --- /dev/null +++ b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java @@ -0,0 +1,120 @@ +/* + * Copyright (c) 2012-2015 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package io.moquette.server.netty.metrics; + +import io.moquette.parser.proto.messages.*; +import io.moquette.server.netty.NettyUtils; +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandler.Sharable; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static io.moquette.parser.proto.messages.AbstractMessage.*; + +/** + * + * @author andrea + */ +@Sharable +public class MQTTMessageLogger extends ChannelDuplexHandler { + + private static final Logger LOG = LoggerFactory.getLogger("messageLogger"); + + @Override + public void channelRead(ChannelHandlerContext ctx, Object message) { + logMQTTMessage(ctx, (AbstractMessage) message, "C->B"); + ctx.fireChannelRead(message); + } + + private void logMQTTMessage(ChannelHandlerContext ctx, AbstractMessage message, String direction) { + AbstractMessage msg = message; + String clientID = NettyUtils.clientID(ctx.channel()); + switch (msg.getMessageType()) { + case CONNECT: + ConnectMessage connect = (ConnectMessage) msg; + LOG.info("{} CONNECT client <{}>", direction, connect.getClientID()); + break; + case SUBSCRIBE: + SubscribeMessage subscribe = (SubscribeMessage) msg; + LOG.info("{} SUBSCRIBE <{}> to topics {}", direction, clientID, subscribe.subscriptions()); + break; + case UNSUBSCRIBE: + UnsubscribeMessage unsubscribe = (UnsubscribeMessage) msg; + LOG.info("{} UNSUBSCRIBE <{}> to topics <{}>", direction, clientID, unsubscribe.topicFilters()); + break; + case PUBLISH: + PublishMessage publish = (PublishMessage) msg; + LOG.info("{} PUBLISH <{}> to topics <{}>", direction, clientID, publish.getTopicName()); + break; + case PUBREC: + PubRecMessage pubrec = (PubRecMessage) msg; + LOG.info("{} PUBREC <{}> packetID <{}>", direction, clientID, pubrec.getMessageID()); + break; + case PUBCOMP: + PubCompMessage pubCompleted = (PubCompMessage) msg; + LOG.info("{} PUBCOMP <{}> packetID <{}>", direction, clientID, pubCompleted.getMessageID()); + break; + case PUBREL: + PubRelMessage pubRelease = (PubRelMessage) msg; + LOG.info("{} PUBREL <{}> packetID <{}>", direction, clientID, pubRelease.getMessageID()); + break; + case DISCONNECT: + LOG.info("{} DISCONNECT <{}>", direction, clientID); + break; + case PUBACK: + PubAckMessage pubAck = (PubAckMessage) msg; + LOG.info("{} PUBACK <{}> packetID <{}>", direction, clientID, pubAck.getMessageID()); + break; + } + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + String clientID = NettyUtils.clientID(ctx.channel()); + if (clientID != null && !clientID.isEmpty()) { + LOG.info("Channel closed <{}>", clientID); + } + ctx.fireChannelActive(); + } + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + logMQTTMessage(ctx, (AbstractMessage) msg, "C<-B"); + ctx.write(msg, promise); + } +// +// @Override +// public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { +// if (cause instanceof CorruptedFrameException) { +// //something goes bad with decoding +// LOG.warn("Error decoding a packet, probably a bad formatted packet, message: " + cause.getMessage()); +// } else { +// LOG.error("Ugly error on networking", cause); +// } +// ctx.close(); +// } +// +// @Override +// public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { +// if (ctx.channel().isWritable()) { +// m_processor.notifyChannelWritable(ctx.channel()); +// } +// ctx.fireChannelWritabilityChanged(); +// } + +} diff --git a/broker/src/test/resources/log4j.properties b/broker/src/test/resources/log4j.properties index 4957aeb8..7913dd43 100644 --- a/broker/src/test/resources/log4j.properties +++ b/broker/src/test/resources/log4j.properties @@ -20,8 +20,8 @@ log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=WARN log4j.logger.io.moquette.spi.persistence.MapDBSessionsStore=WARN #HazelCast Handler -log4j.logger.io.moquette.interception.HazelcastInterceptHandler=INFO -log4j.logger.io.moquette.server.HazelcastListener=INFO +log4j.logger.io.moquette.interception.HazelcastInterceptHandler=WARN +log4j.logger.io.moquette.server.HazelcastListener=WARN # stdout appender is set to be a ConsoleAppender. log4j.appender.stdout=org.apache.log4j.ConsoleAppender @@ -40,3 +40,17 @@ log4j.appender.file.MaxBackupIndex=1 log4j.appender.file.layout=org.apache.log4j.PatternLayout #log4j.appender.file.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n log4j.appender.file.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1} %L %x - %m%n + +#################################### +# Message Logger Configuration # +##################################### +log4j.appender.messagelog=org.apache.log4j.RollingFileAppender +log4j.appender.messagelog.Threshold=INFO +log4j.appender.messagelog.File=moquette_messages.log +log4j.appender.messagelog.MaxFileSize=100MB +log4j.appender.messagelog.MaxBackupIndex=1 +log4j.appender.messagelog.layout=org.apache.log4j.PatternLayout +log4j.appender.messagelog.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1} %L %x - %m%n + +log4j.category.messageLogger=INFO, messagelog +log4j.additivity.messageLogger=false diff --git a/netty_parser/src/main/java/io/moquette/parser/proto/messages/SubscribeMessage.java b/netty_parser/src/main/java/io/moquette/parser/proto/messages/SubscribeMessage.java index d2974f22..022cfa37 100644 --- a/netty_parser/src/main/java/io/moquette/parser/proto/messages/SubscribeMessage.java +++ b/netty_parser/src/main/java/io/moquette/parser/proto/messages/SubscribeMessage.java @@ -33,7 +33,13 @@ public class SubscribeMessage extends MessageIDMessage { this.qos = qos; this.topicFilter = topic; } - + + @Override + public String toString() { + return "[qos=" + qos + + ", topicFilter='" + topicFilter + '\'' + + ']'; + } } private List m_subscriptions = new ArrayList<>();