diff --git a/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java new file mode 100644 index 00000000..5a6d2ef1 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -0,0 +1,20 @@ +package org.dna.mqtt.moquette.messaging.spi; + +/** + * Interface to the underling messaging system used to publish, subscribe. + * + * @author andrea + */ +public class IMessaging { + + /** + * Publish a message on the messaging system. The qos is the level of quality + * the retain flag is true iff the message should be persisted for future + * notification also after all the current subscriber to the given topic has + * been notified. + */ + public void publish(String topic, Object message, byte qos, boolean retain) { + throw new UnsupportedOperationException("Not yet implemented"); + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/server/IAuthenticator.java b/src/main/java/org/dna/mqtt/moquette/server/IAuthenticator.java new file mode 100644 index 00000000..b3b76ace --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/server/IAuthenticator.java @@ -0,0 +1,11 @@ +package org.dna.mqtt.moquette.server; + +/** + * username and password checker + * + * @author andrea + */ +public interface IAuthenticator { + + boolean checkValid(String username, String password); +} diff --git a/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java index 30cb6921..2fb90a22 100644 --- a/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java @@ -1,9 +1,12 @@ package org.dna.mqtt.moquette.server; +import org.dna.mqtt.moquette.messaging.spi.IMessaging; import java.util.HashMap; import java.util.Map; +import java.util.logging.Logger; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.apache.mina.core.service.IoHandlerAdapter; +import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; @@ -14,41 +17,86 @@ import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; * @author andrea */ public class MQTTHandler extends IoHandlerAdapter { - + + private static final Logger LOG = Logger.getLogger(MQTTHandler.class.getName()); Map m_clientIDs = new HashMap(); + private IMessaging m_messaging; + private IAuthenticator m_authenticator; @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); } } - protected void handleConnect(IoSession session, ConnectMessage connectMessage) { - if (connectMessage.getProcotolVersion() != 0x03) { + protected void handleConnect(IoSession session, ConnectMessage msg) { + if (msg.getProcotolVersion() != 0x03) { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); session.write(badProto); session.close(false); return; } - - //if an old client with the same ID already exists close its session.s - if (m_clientIDs.containsKey(connectMessage.getClientID())) { - m_clientIDs.get(connectMessage.getClientID()).close(false); + + if (msg.getClientID().length() > 23) { + ConnAckMessage okResp = new ConnAckMessage(); + okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); + session.write(okResp); + return; } - m_clientIDs.put(connectMessage.getClientID(), session); - - session.setAttribute("keepAlive", connectMessage.getKeepAlive()); - session.setAttribute("cleanSession", connectMessage.isCleanSession()); - - //TODO Handle will flag - //TODO handle user authentication - + + //if an old client with the same ID already exists close its session. + if (m_clientIDs.containsKey(msg.getClientID())) { + m_clientIDs.get(msg.getClientID()).close(false); + } + m_clientIDs.put(msg.getClientID(), session); + + int keepAlive = msg.getKeepAlive(); + session.setAttribute("keepAlive", keepAlive); + session.setAttribute("cleanSession", msg.isCleanSession()); + + session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f)); + + //Handle will flag + if (msg.isWillFlag()) { + m_messaging.publish(msg.getWillTopic(), msg.getWillMessage(), msg.getWillQos(), msg.isWillRetain()); + } + + //handle user authentication + if (msg.isUserFlag()) { + String pwd = null; + if (msg.isPasswordFlag()) { + pwd = msg.getPassword(); + } + if (!m_authenticator.checkValid(msg.getUsername(), pwd)) { + ConnAckMessage okResp = new ConnAckMessage(); + okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); + session.write(okResp); + return; + } + } + + //TODO handle clean session flag + ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); session.write(okResp); } + + @Override + public void sessionIdle(IoSession session, IdleStatus status) { + session.close(false); + } + + public void setMessaging(IMessaging messaging) { + m_messaging = messaging; + } + + public void setAuthenticator(IAuthenticator authenticator) { + m_authenticator = authenticator; + } } diff --git a/src/main/java/org/dna/mqtt/moquette/server/Server.java b/src/main/java/org/dna/mqtt/moquette/server/Server.java new file mode 100644 index 00000000..f08d01a0 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -0,0 +1,47 @@ +package org.dna.mqtt.moquette.server; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.logging.Logger; +import org.apache.mina.core.service.IoAcceptor; +import org.apache.mina.core.session.IdleStatus; +import org.apache.mina.filter.codec.ProtocolCodecFilter; +import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder; +import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder; +import org.apache.mina.filter.logging.LoggingFilter; +import org.apache.mina.transport.socket.nio.NioSocketAcceptor; +import org.dna.mqtt.moquette.proto.ConnAckEncoder; +import org.dna.mqtt.moquette.proto.ConnectDecoder; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +/** + * Launch a configured version of the server. + * @author andrea + */ +public class Server { + + private static final Logger LOG = Logger.getLogger(Server.class.getName()); + + public static final int PORT = 9191; + public static final int DEFAULT_CONNECT_TIMEOUT = 10; + + public static void main(String[] args) throws IOException { + DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); + decoder.addMessageDecoder(new ConnectDecoder()); + + DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); +// encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); + encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder()); + + IoAcceptor acceptor = new NioSocketAcceptor(); + + acceptor.getFilterChain().addLast( "logger", new LoggingFilter() ); + acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder)); + + acceptor.setHandler( new MQTTHandler() ); + acceptor.getSessionConfig().setReadBufferSize( 2048 ); + acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, DEFAULT_CONNECT_TIMEOUT ); + acceptor.bind( new InetSocketAddress(PORT) ); + LOG.info("Server binded"); + } + +} diff --git a/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java b/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java index 73a96998..da4d4a7a 100644 --- a/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java +++ b/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java @@ -31,7 +31,7 @@ public class MQTTHandlerTest { ConnAckMessage buf = (ConnAckMessage) writeRequest.getMessage(); assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, buf.getReturnCode()); } catch (Exception ex) { - throw new AssertionError("Wrong return code"); + fail("Wrong return code"); } } }); @@ -39,4 +39,34 @@ public class MQTTHandlerTest { //Exercise h.handleConnect(session, connMsg); } -} \ No newline at end of file + + + @Test + public void testConnect_badClientID() { + MQTTHandler h = new MQTTHandler(); + ConnectMessage connMsg = new ConnectMessage(); + connMsg.setProcotolVersion((byte) 0x03); + connMsg.setClientID("extremely_long_clientID_grtaer_than_23"); + + IoSession session = new DummySession(); + session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { + + @Override + public void filterWrite(NextFilter nextFilter, IoSession session, + WriteRequest writeRequest) throws Exception { + try { + ConnAckMessage buf = (ConnAckMessage) writeRequest.getMessage(); + assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, buf.getReturnCode()); + } catch (Exception ex) { + fail("Wrong return code"); + } + } + }); + + //Exercise + h.handleConnect(session, connMsg); + } + + + +}