Introduced MQTT logging filter to display more clear message's headers

parser/src/main/java/org/dna/mqtt/moquette/proto/MQTTLoggingFilter.java

@@ -0,0 +1,155 @@
+package org.dna.mqtt.moquette.proto;
+
+import org.apache.mina.core.buffer.AbstractIoBuffer;
+import org.apache.mina.core.buffer.IoBuffer;
+import org.apache.mina.core.session.IoSession;
+import org.apache.mina.core.write.WriteRequest;
+import org.apache.mina.filter.logging.LogLevel;
+import org.apache.mina.filter.logging.LoggingFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
+
+/**
+ * Apache MINA logging filter that pretty prints the header of messages
+ *
+ * @author andrea
+ */
+public class MQTTLoggingFilter extends LoggingFilter {
+
+    private final Logger log;
+
+    protected static String computeLoggerName(String name) {
+        return name == null ? MQTTLoggingFilter.class.getName() : name;
+    }
+
+    /**
+     * Default Constructor.
+     */
+    public MQTTLoggingFilter() {
+        this(LoggingFilter.class.getName());
+    }
+
+    /**
+     * Create a new NoopFilter using a class name
+     *
+     * @param clazz the class which name will be used to create the logger
+     */
+    public MQTTLoggingFilter(Class<?> clazz) {
+        this(clazz.getName());
+    }
+
+    /**
+     * Create a new NoopFilter using a name
+     *
+     * @param name the name used to create the logger. If null, will default to
+     * "NoopFilter"
+     */
+    public MQTTLoggingFilter(String name) {
+        super(computeLoggerName(name));
+        log = LoggerFactory.getLogger(getName());
+    }
+
+    protected String decodeCommonHeader(IoBuffer in) {
+        StringBuilder sb = new StringBuilder();
+
+        byte h1 = in.get();
+        byte messageType = (byte) ((h1 & 0x00F0) >> 4);
+        boolean dupFlag = ((byte) ((h1 & 0x0008) >> 3) == 1);
+        byte qosLevel = (byte) ((h1 & 0x0006) >> 1);
+        boolean retainFlag = ((byte) (h1 & 0x0001) == 1);
+        int remainingLength = Utils.decodeRemainingLenght(in);
+
+        sb.append("type: ").append(decodeMessageType(messageType)).append(", ");
+        sb.append("dup: ").append(Boolean.toString(dupFlag)).append(", ");
+        sb.append("QoS: ").append(qosLevel).append(", ");
+        sb.append("retain: ").append(Boolean.toString(retainFlag));
+        
+        if (remainingLength != -1) {
+            sb.append(", remainingLen: ").append(remainingLength);
+        }
+
+        in.rewind();
+        return sb.toString();
+    }
+
+    protected String decodeMessageType(byte type) {
+        switch (type) {
+            case CONNECT:
+                return "CONNECT";
+            case CONNACK:
+                return "CONNACK";
+            case PUBLISH:
+                return "PUBLISH";
+            case PUBACK:
+                return "PUBACK";
+            case PUBREC:
+                return "PUBREC";
+            case PUBREL:
+                return "PUBREL";
+            case PUBCOMP:
+                return "PUBCOMP";
+            case SUBSCRIBE:
+                return "SUBSCRIBE";
+            case SUBACK:
+                return "SUBACK";
+            case UNSUBSCRIBE:
+                return "UNSUBSCRIBE";
+            case UNSUBACK:
+                return "UNSUBACK";
+            case PINGREQ:
+                return "PINGREQ";
+            case PINGRESP:
+                return "PINGRESP";
+            case DISCONNECT:
+                return "DISCONNECT";
+            default:
+                return "undefined";
+        }
+    }
+
+    @Override
+    public void messageReceived(NextFilter nextFilter, IoSession session,
+            Object message) throws Exception {
+
+        if (message instanceof AbstractIoBuffer) {
+            IoBuffer buff = (AbstractIoBuffer) message;
+            log(getMessageReceivedLogLevel(), "RECEIVED: {}", decodeCommonHeader(buff));
+        } else {
+            log(getMessageReceivedLogLevel(), "RECEIVED: {}", message);
+        }
+        nextFilter.messageReceived(session, message);
+    }
+
+    @Override
+    public void messageSent(NextFilter nextFilter, IoSession session,
+            WriteRequest writeRequest) throws Exception {
+        log(getMessageSentLogLevel(), "SENT: {}", writeRequest.getMessage());
+        nextFilter.messageSent(session, writeRequest);
+    }
+
+    /**
+     * Copied by parent
+     */
+    private void log(LogLevel eventLevel, String message, Object param) {
+        switch (eventLevel) {
+            case TRACE:
+                log.trace(message, param);
+                return;
+            case DEBUG:
+                log.debug(message, param);
+                return;
+            case INFO:
+                log.info(message, param);
+                return;
+            case WARN:
+                log.warn(message, param);
+                return;
+            case ERROR:
+                log.error(message, param);
+                return;
+            default:
+                return;
+        }
+    }
+}

parser/src/main/java/org/dna/mqtt/moquette/server/Server.java

@@ -13,6 +13,7 @@ import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging;
 import org.dna.mqtt.moquette.proto.ConnAckEncoder;
 import org.dna.mqtt.moquette.proto.ConnectDecoder;
 import org.dna.mqtt.moquette.proto.DisconnectDecoder;
+import org.dna.mqtt.moquette.proto.MQTTLoggingFilter;
 import org.dna.mqtt.moquette.proto.PublishDecoder;
 import org.dna.mqtt.moquette.proto.PublishEncoder;
 import org.dna.mqtt.moquette.proto.SubAckEncoder;
@@ -54,7 +55,7 @@ public class Server {
         
         m_acceptor = new NioSocketAcceptor();
 
-        m_acceptor.getFilterChain().addLast( "logger", new LoggingFilter("SERVER LOG") );
+        m_acceptor.getFilterChain().addLast( "logger", new MQTTLoggingFilter("SERVER LOG") );
         m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
 
         MQTTHandler handler = new MQTTHandler();