Fixed issue #65

ChangeLog.txt

@@ -1,5 +1,6 @@
 Version 0.8:
     Introduced authenticator and authorizator implementations.
+    Fixed bug, doesnt' close the channel when connect doesn't contains a valid proto name, issue #65
 
 Version 0.7:
     Fix issue #38, added aligned to MQTT 3.1.1

broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java

@@ -20,6 +20,8 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import java.util.HashMap;
 import java.util.Map;
+
+import io.netty.handler.codec.CorruptedFrameException;
 import org.eclipse.moquette.spi.IMessaging;
 import org.eclipse.moquette.proto.Utils;
 import org.eclipse.moquette.proto.messages.AbstractMessage;
@@ -37,8 +39,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
     
     private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class);
     private IMessaging m_messaging;
-//    private final Map<ChannelHandlerContext, NettyChannel> m_channelMapper = new HashMap<>();
-    
+
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object message) {
         AbstractMessage msg = (AbstractMessage) message;
@@ -54,14 +55,6 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
                 case PUBREL:
                 case DISCONNECT:
                 case PUBACK:    
-//                    NettyChannel channel;
-//                    synchronized(m_channelMapper) {
-//                        if (!m_channelMapper.containsKey(ctx)) {
-//                            m_channelMapper.put(ctx, new NettyChannel(ctx));
-//                        }
-//                        channel = m_channelMapper.get(ctx);
-//                    }
-                    
                     m_messaging.handleProtocolMessage(new NettyChannel(ctx), msg);
                     break;
                 case PINGREQ:
@@ -76,16 +69,22 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
     
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-//        NettyChannel channel = m_channelMapper.get(ctx);
-//        String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
         String clientID = (String) NettyUtils.getAttribute(ctx, NettyChannel.ATTR_KEY_CLIENTID);
         m_messaging.lostConnection(clientID);
         ctx.close(/*false*/);
-//        synchronized(m_channelMapper) {
-//            m_channelMapper.remove(ctx);
-//        }
     }
-    
+
+    @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();
+    }
+
     public void setMessaging(IMessaging messaging) {
         m_messaging = messaging;
     }

broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java

@@ -23,6 +23,8 @@ import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
+import io.netty.util.internal.StringUtil;
+import org.apache.commons.codec.binary.StringUtils;
 import org.eclipse.moquette.server.netty.NettyChannel;
 import org.eclipse.moquette.spi.IMatchingCondition;
 import org.eclipse.moquette.spi.IMessagesStore;
@@ -144,7 +146,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         // Get the ring buffer from the Disruptor to be used for publishing.
         m_ringBuffer = disruptor.getRingBuffer();
     }
-    
+
     @MQTTMessage(message = ConnectMessage.class)
     void processConnect(ServerChannel session, ConnectMessage msg) {
         LOG.debug("CONNECT for client <{}>", msg.getClientID());

broker/src/test/java/org/eclipse/moquette/server/netty/NettyMQTTHandlerTest.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright (c) 2012-2014 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 org.eclipse.moquette.server.netty;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.CorruptedFrameException;
+import org.eclipse.moquette.parser.netty.ConnectDecoder;
+import org.eclipse.moquette.spi.IMessaging;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+import java.util.List;
+
+import static org.mockito.Mockito.mock;
+
+/**
+ *
+ * @author andrea
+ */
+public class NettyMQTTHandlerTest {
+
+//    class ConnectDecoderAdapter extends ByteToMessageDecoder {
+//
+//        final ConnectDecoder decoder;
+//
+//        ConnectDecoderAdapter(ConnectDecoder decoder) {
+//            this.decoder = decoder;
+//        }
+//
+//        @Override
+//        protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List<Object> list) throws Exception {
+//            decoder.decode(ctx, byteBuf, list);
+//        }
+//    }
+
+    class ThrowingExceptionInboundHandler extends ChannelInboundHandlerAdapter {
+        @Override
+        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+            throw new CorruptedFrameException("Invalid protoName: SUPERFAKE");
+        }
+    }
+
+    /**
+     * Identified by issue #65
+     * */
+    @Test
+    public void testCatchingCorruptedFrameException() {
+        IMessaging mockMsg = mock(IMessaging.class);
+        NettyMQTTHandler sut = new NettyMQTTHandler();
+        sut.setMessaging(mockMsg);
+        EmbeddedChannel channel = new EmbeddedChannel(new ThrowingExceptionInboundHandler(), sut);
+
+        //prepare a packet with bad protocol name
+        ByteBuf input = Unpooled.buffer();
+        input.writeByte(0x01);
+
+        //Exercise
+        assertTrue(channel.isOpen());
+        channel.writeInbound(input.readBytes(1));
+
+        //Verify the handler close the channel
+        assertFalse(channel.isOpen());
+    }
+}

broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java

@@ -185,7 +185,7 @@ public class ProtocolProcessorTest {
         //Verify
         assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode);
     }
-    
+
     @Test
     public void testConnect_badClientID() {
         connMsg.setClientID("extremely_long_clientID_greater_than_23");

netty_parser/src/test/java/org/eclipse/moquette/parser/netty/ConnectDecoderTest.java

@@ -66,7 +66,7 @@ public class ConnectDecoderTest {
     public void testBaseHeader_311() throws UnsupportedEncodingException {
         m_buff = Unpooled.buffer(12);
         initBaseHeader311(m_buff);
-        List<Object> results = new ArrayList<Object>();
+        List<Object> results = new ArrayList<>();
         
         //Excercise
         m_msgdec.decode(this.attrMap, m_buff, results);