diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java index 7937a845..12f557a5 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java @@ -1,12 +1,32 @@ package org.dna.mqtt.moquette.client; +import java.util.logging.Logger; import org.apache.mina.core.service.IoHandlerAdapter; +import org.apache.mina.core.session.IoSession; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; /** * * @author andrea */ public class ClientMQTTHandler extends IoHandlerAdapter { -// public void sessionOpened(IoSession session) {} -// public void messageReceived(IoSession session, Object message) { + + private static final Logger LOG = Logger.getLogger(ClientMQTTHandler.class.getName()); + + @Override + public void messageReceived(IoSession session, Object message) throws Exception { + AbstractMessage msg = (AbstractMessage) message; + LOG.fine("Received a message of type " + msg.getMessageType()); +// switch (msg.getMessageType()) { +// case CONNECT: +// handleConnect(session, (ConnectMessage) msg); +// break; +// case SUBSCRIBE: +// handleSubscribe(session, (SubscribeMessage) msg); +// break; +// case PUBLISH: +// handlePublish(session, (PublishMessage) msg); +// break; +// } + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java index 69f7803e..a8f7e443 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java @@ -5,6 +5,7 @@ import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import java.util.HashMap; import java.util.Map; +import java.util.logging.Level; import java.util.logging.Logger; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.apache.mina.core.service.IoHandlerAdapter; @@ -34,13 +35,20 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { public void messageReceived(IoSession session, Object message) throws Exception { AbstractMessage msg = (AbstractMessage) message; LOG.fine("Received a message of type " + msg.getMessageType()); - switch (msg.getMessageType()) { - case CONNECT: - handleConnect(session, (ConnectMessage) msg); - case SUBSCRIBE: - handleSubscribe(session, (SubscribeMessage) msg); - case PUBLISH: - handlePublish(session, (PublishMessage) msg); + try { + switch (msg.getMessageType()) { + case CONNECT: + handleConnect(session, (ConnectMessage) msg); + break; + case SUBSCRIBE: + handleSubscribe(session, (SubscribeMessage) msg); + break; + case PUBLISH: + handlePublish(session, (PublishMessage) msg); + break; + } + } catch (Exception ex) { + LOG.log(Level.SEVERE, "Bad error in processing the message", ex); } } @@ -53,7 +61,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { return; } - if (msg.getClientID().length() > 23) { + if (msg.getClientID() == null || msg.getClientID().length() > 23) { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); session.write(okResp); diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java index 69a636f5..370e34d7 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -25,6 +25,11 @@ public class Server { public static final int DEFAULT_CONNECT_TIMEOUT = 10; public static void main(String[] args) throws IOException { + new Server().startServer(); + + } + + protected void startServer() throws IOException { DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); decoder.addMessageDecoder(new ConnectDecoder());