Added clinetID=empty then reject connect rule

broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java

@@ -20,13 +20,13 @@ import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 import org.fusesource.mqtt.client.BlockingConnection;
 import org.fusesource.mqtt.client.MQTT;
+import org.fusesource.mqtt.client.Message;
 import org.fusesource.mqtt.client.QoS;
 import org.fusesource.mqtt.client.Topic;
-import org.fusesource.mqtt.client.Message;
 import org.junit.After;
+import static org.junit.Assert.*;
 import org.junit.Before;
 import org.junit.Test;
-import static org.junit.Assert.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java

@@ -89,7 +89,7 @@ public class ServerIntegrationPahoTest {
 
         assertEquals("/topic", m_callback.getTopic());
     }
-
+    
 
     @Test
     public void testCleanSession_maintainClientSubscriptions() throws Exception {

broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java

@@ -17,8 +17,11 @@ package org.dna.mqtt.moquette.server;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
+import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
 import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
 import org.dna.mqtt.moquette.testclient.Client;
 import org.fusesource.mqtt.client.BlockingConnection;
@@ -27,9 +30,9 @@ import org.fusesource.mqtt.client.Message;
 import org.fusesource.mqtt.client.QoS;
 import org.fusesource.mqtt.client.Topic;
 import org.junit.After;
+import static org.junit.Assert.*;
 import org.junit.Before;
 import org.junit.Test;
-import static org.junit.Assert.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -121,4 +124,37 @@ public class ServerLowlevelMessagesIntegrationTests {
         assertEquals(willTestamentMsg, new String(msg.getPayload()));
         willSubscriber.disconnect();
     }
+    
+    AbstractMessage receivedMsg;
+    
+    @Test
+    public void checkRejectConnectWithEmptyClientID() throws InterruptedException {
+        LOG.info("*** checkRejectConnectWithEmptyClientID ***");
+        final CountDownLatch latch = new CountDownLatch(1);
+        m_client.setCallback(new Client.ICallback() {
+
+            public void call(AbstractMessage msg) {
+                receivedMsg = msg;
+                latch.countDown();
+            }
+        });
+        
+        int keepAlive = 2; //secs
+        ConnectMessage connectMessage = new ConnectMessage();
+        connectMessage.setProcotolVersion((byte) 4);
+        connectMessage.setClientID("");
+        connectMessage.setKeepAlive(keepAlive);
+        connectMessage.setWillFlag(false);
+        connectMessage.setWillQos((byte) QOSType.MOST_ONE.ordinal());
+        
+        //Execute
+        m_client.sendMessage(connectMessage);
+        
+        latch.await(200, TimeUnit.MILLISECONDS);
+        
+        assertTrue(receivedMsg instanceof ConnAckMessage);
+        ConnAckMessage connAck = (ConnAckMessage) receivedMsg;
+        assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, connAck.getReturnCode());
+    }
+    
 }

broker/src/test/java/org/dna/mqtt/moquette/testclient/Client.java

@@ -24,6 +24,7 @@ import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
+import java.util.concurrent.Callable;
 import org.dna.mqtt.commons.Constants;
 import org.dna.mqtt.moquette.parser.netty.MQTTDecoder;
 import org.dna.mqtt.moquette.parser.netty.MQTTEncoder;
@@ -39,6 +40,10 @@ import org.slf4j.LoggerFactory;
  */
 public class Client {
     
+    public static interface ICallback {
+        void call(AbstractMessage msg);
+    }
+    
     private static final Logger LOG = LoggerFactory.getLogger(Client.class);
     final ClientNettyMQTTHandler handler = new ClientNettyMQTTHandler();
     EventLoopGroup workerGroup;
@@ -78,12 +83,20 @@ public class Client {
         }
     }
     
+    ICallback callback;
+    public void setCallback(ICallback callback) {
+        this.callback = callback;
+    }
+    
     public void sendMessage(AbstractMessage msg) {
         m_channel.writeAndFlush(msg);
     }
 
     void messageReceived(AbstractMessage msg) {
         LOG.info("Received message " + msg);
+        if (this.callback != null) {
+            this.callback.call(msg);
+        }
     }
     
     void setConnectionLost(boolean status) {