diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 001c05ea..6f901cc3 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -4,6 +4,7 @@ import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; /** * Interface to the underling messaging system used to publish, subscribe. @@ -47,4 +48,6 @@ public interface IMessaging { void disconnect(IoSession session); void republishStored(String clientID); + + void connect(IoSession session, ConnectMessage msg); } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 2b2032c8..22c92176 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IMessaging; @@ -8,14 +9,22 @@ import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; 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.proto.messages.PubAckMessage; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.dna.mqtt.moquette.server.ConnectionDescriptor; import org.dna.mqtt.moquette.server.Constants; +import org.dna.mqtt.moquette.server.IAuthenticator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.Serializable; import java.text.ParseException; import java.util.Collection; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; @@ -53,6 +62,10 @@ public class SimpleMessaging implements IMessaging, Runnable { private INotifier m_notifier; private IStorageService m_storageService; + + Map m_clientIDs = new HashMap(); + + private IAuthenticator m_authenticator; public SimpleMessaging() { m_storageService = new HawtDBStorageService(); @@ -129,6 +142,14 @@ public class SimpleMessaging implements IMessaging, Runnable { } } + public void connect(IoSession session, ConnectMessage msg) { + try { + m_inboundQueue.put(new ConnectEvent(session, msg)); + } catch (InterruptedException iex) { + LOG.error(null, iex); + } + } + /** * Verify if the 2 topics matching respecting the rules of MQTT Appendix A */ @@ -207,6 +228,8 @@ public class SimpleMessaging implements IMessaging, Runnable { m_storageService.cleanInFlight(((CleanInFlightEvent) evt).getMsgId()); } else if (evt instanceof RepublishEvent) { processRepublish((RepublishEvent) evt); + } else if (evt instanceof ConnectEvent) { + processConnect((ConnectEvent) evt); } } catch (InterruptedException ex) { processClose(); @@ -215,6 +238,85 @@ public class SimpleMessaging implements IMessaging, Runnable { } } + private void processConnect(ConnectEvent evt) { + ConnectMessage msg = evt.getMessage(); + IoSession session = evt.getSession(); + + if (msg.getProcotolVersion() != 0x03) { + ConnAckMessage badProto = new ConnAckMessage(); + badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); + session.write(badProto); + session.close(false); + return; + } + + if (msg.getClientID() == null || msg.getClientID().length() > 23) { + ConnAckMessage okResp = new ConnAckMessage(); + okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); + session.write(okResp); + return; + } + + //if an old client with the same ID already exists close its session. + if (m_clientIDs.containsKey(msg.getClientID())) { + //clean the subscriptions if the old used a cleanSession = true + IoSession oldSession = m_clientIDs.get(msg.getClientID()).getSession(); + boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION); + if (cleanSession) { + //cleanup topic subscriptions + removeSubscriptions(msg.getClientID()); + } + + m_clientIDs.get(msg.getClientID()).getSession().close(false); + } + + ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession()); + m_clientIDs.put(msg.getClientID(), connDescr); + + int keepAlive = msg.getKeepAlive(); + session.setAttribute("keepAlive", keepAlive); + session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession()); + //used to track the client in the subscription and publishing phases. + session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID()); + + session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f)); + + //Handle will flag + if (msg.isWillFlag()) { + QOSType willQos = QOSType.values()[msg.getWillQos()]; + publish(msg.getWillTopic(), msg.getWillMessage().getBytes(), + willQos, msg.isWillRetain(), msg.getClientID(), session); + } + + //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; + } + } + + //handle clean session flag + if (msg.isCleanSession()) { + //remove all prev subscriptions + //cleanup topic subscriptions + removeSubscriptions(msg.getClientID()); + } else { + //force the republish of stored QoS1 and QoS2 + republishStored(msg.getClientID()); + } + + ConnAckMessage okResp = new ConnAckMessage(); + okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); + session.write(okResp); + } + protected void processPublish(PublishEvent evt) throws InterruptedException { LOG.debug("processPublish invoked"); final String topic = evt.getTopic(); @@ -234,20 +336,20 @@ public class SimpleMessaging implements IMessaging, Runnable { for (final Subscription sub : subscriptions.matches(topic)) { if (qos == QOSType.MOST_ONE) { //QoS 0 - m_notifier.notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); + notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); } else { //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it if (!sub.isCleanSession() && !sub.isActive()) { m_storageService.storePublishForFuture(evt); } - m_notifier.notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); + notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); } } if (cleanEvt != null) { refill(cleanEvt); - m_notifier.sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID())); + sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID())); } if (retain) { @@ -292,10 +394,14 @@ public class SimpleMessaging implements IMessaging, Runnable { } private void processDisconnect(DisconnectEvent evt) throws InterruptedException { - m_notifier.disconnect(evt.getSession()); +// m_notifier.disconnect(evt.getSession()); + IoSession session = evt.getSession(); + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + m_clientIDs.remove(clientID); + session.close(true); //de-activate the subscriptions for this ClientID - String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); +// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); subscriptions.disconnect(clientID); } @@ -315,4 +421,47 @@ public class SimpleMessaging implements IMessaging, Runnable { pubEvt.getMessage(), false, pubEvt.getMessageID())); } } + + private void notify(NotifyEvent evt) { + LOG.debug("notify invoked with event " + evt); + String clientId = evt.getClientId(); + PublishMessage pubMessage = new PublishMessage(); + pubMessage.setRetainFlag(evt.isRetained()); + pubMessage.setTopicName(evt.getTopic()); + pubMessage.setQos(evt.getQos()); + pubMessage.setPayload(evt.getMessage()); + if (pubMessage.getQos() != QOSType.MOST_ONE) { + pubMessage.setMessageID(evt.getMessageID()); + } + + LOG.debug("notify invoked"); + try { + assert m_clientIDs != null; + LOG.debug("clientIDs are " + m_clientIDs); + assert m_clientIDs.get(clientId) != null; + LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); + m_clientIDs.get(clientId).getSession().write(pubMessage); + }catch(Throwable t) { + LOG.error(null, t); + } + } + + private void sendPubAck(PubAckEvent evt) { + LOG.debug("sendPubAck invoked"); + + String clientId = evt.getClientID(); + + PubAckMessage pubAckMessage = new PubAckMessage(); + pubAckMessage.setMessageID(evt.getMessageId()); + + try { + assert m_clientIDs != null; + LOG.debug("clientIDs are " + m_clientIDs); + assert m_clientIDs.get(clientId) != null; + LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); + m_clientIDs.get(clientId).getSession().write(pubAckMessage); + }catch(Throwable t) { + LOG.error(null, t); + } + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ConnectEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ConnectEvent.java new file mode 100644 index 00000000..7e2f6bea --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ConnectEvent.java @@ -0,0 +1,25 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +import org.apache.mina.core.session.IoSession; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; + +/** + */ +public class ConnectEvent extends MessagingEvent { + + IoSession m_session; + ConnectMessage message; + + public ConnectEvent(IoSession session, ConnectMessage message) { + this.m_session = session; + this.message = message; + } + + public IoSession getSession() { + return m_session; + } + + public ConnectMessage getMessage() { + return message; + } +} 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 1d64d78c..b86b90e5 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 @@ -22,16 +22,16 @@ import org.slf4j.LoggerFactory; * * @author andrea */ -public class MQTTHandler extends IoHandlerAdapter implements INotifier { +public class MQTTHandler extends IoHandlerAdapter/* implements INotifier*/ { private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class); /** * Maps CLIENT_ID to the IoSession that represents the connection */ Lock m_clientIDsLock = new ReentrantLock(); - Map m_clientIDs = new HashMap(); +// Map m_clientIDs = new HashMap(); private IMessaging m_messaging; - private IAuthenticator m_authenticator; +// private IAuthenticator m_authenticator; @Override public void messageReceived(IoSession session, Object message) throws Exception { @@ -65,84 +65,8 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { protected void handleConnect(IoSession session, ConnectMessage msg) { LOG.info("handleConnect invoked"); - if (msg.getProcotolVersion() != 0x03) { - ConnAckMessage badProto = new ConnAckMessage(); - badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); - session.write(badProto); - session.close(false); - return; - } - if (msg.getClientID() == null || msg.getClientID().length() > 23) { - ConnAckMessage okResp = new ConnAckMessage(); - okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); - session.write(okResp); - return; - } - - m_clientIDsLock.lock(); - try { - //if an old client with the same ID already exists close its session. - if (m_clientIDs.containsKey(msg.getClientID())) { - //clean the subscriptions if the old used a cleanSession = true - IoSession oldSession = m_clientIDs.get(msg.getClientID()).getSession(); - boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION); - if (cleanSession) { - //cleanup topic subscriptions - m_messaging.removeSubscriptions(msg.getClientID()); - } - - m_clientIDs.get(msg.getClientID()).getSession().close(false); - } - - ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession()); - m_clientIDs.put(msg.getClientID(), connDescr); - } finally { - m_clientIDsLock.unlock(); - } - - int keepAlive = msg.getKeepAlive(); - session.setAttribute("keepAlive", keepAlive); - session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession()); - //used to track the client in the subscription and publishing phases. - session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID()); - - session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f)); - - //Handle will flag - if (msg.isWillFlag()) { - QOSType willQos = QOSType.values()[msg.getWillQos()]; - m_messaging.publish(msg.getWillTopic(), msg.getWillMessage().getBytes(), - willQos, msg.isWillRetain(), msg.getClientID(), session); - } - - //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; - } - } - - //handle clean session flag - if (msg.isCleanSession()) { - //remove all prev subscriptions - //cleanup topic subscriptions - m_messaging.removeSubscriptions(msg.getClientID()); - } else { - //force the republish of stored QoS1 and QoS2 - m_messaging.republishStored(msg.getClientID()); - } - - ConnAckMessage okResp = new ConnAckMessage(); - okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); - session.write(okResp); + m_messaging.connect(session, msg); } protected void handleSubscribe(IoSession session, SubscribeMessage msg) { @@ -220,69 +144,69 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { m_messaging = messaging; } - public void setAuthenticator(IAuthenticator authenticator) { - m_authenticator = authenticator; - } +// public void setAuthenticator(IAuthenticator authenticator) { +// m_authenticator = authenticator; +// } - public void notify(NotifyEvent evt) { - LOG.debug("notify invoked with event " + evt); - String clientId = evt.getClientId(); - PublishMessage pubMessage = new PublishMessage(); - pubMessage.setRetainFlag(evt.isRetained()); - pubMessage.setTopicName(evt.getTopic()); - pubMessage.setQos(evt.getQos()); - pubMessage.setPayload(evt.getMessage()); - if (pubMessage.getQos() != QOSType.MOST_ONE) { - pubMessage.setMessageID(evt.getMessageID()); - } - - LOG.debug("notify invoked"); - m_clientIDsLock.lock(); - try { - assert m_clientIDs != null; - LOG.debug("clientIDs are " + m_clientIDs); - assert m_clientIDs.get(clientId) != null; - LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); - m_clientIDs.get(clientId).getSession().write(pubMessage); - }catch(Throwable t) { - LOG.error(null, t); - } finally { - m_clientIDsLock.unlock(); - } - } - - public void disconnect(IoSession session) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - m_clientIDsLock.lock(); - try { - m_clientIDs.remove(clientID); - } finally { - m_clientIDsLock.unlock(); - } - session.close(true); - } - - public void sendPubAck(PubAckEvent evt) { - LOG.debug("sendPubAck invoked"); - - String clientId = evt.getClientID(); - - PubAckMessage pubAckMessage = new PubAckMessage(); - pubAckMessage.setMessageID(evt.getMessageId()); - - m_clientIDsLock.lock(); - try { - assert m_clientIDs != null; - LOG.debug("clientIDs are " + m_clientIDs); - assert m_clientIDs.get(clientId) != null; - LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); - m_clientIDs.get(clientId).getSession().write(pubAckMessage); - }catch(Throwable t) { - LOG.error(null, t); - } finally { - m_clientIDsLock.unlock(); - } - } +// public void notify(NotifyEvent evt) { +// LOG.debug("notify invoked with event " + evt); +// String clientId = evt.getClientId(); +// PublishMessage pubMessage = new PublishMessage(); +// pubMessage.setRetainFlag(evt.isRetained()); +// pubMessage.setTopicName(evt.getTopic()); +// pubMessage.setQos(evt.getQos()); +// pubMessage.setPayload(evt.getMessage()); +// if (pubMessage.getQos() != QOSType.MOST_ONE) { +// pubMessage.setMessageID(evt.getMessageID()); +// } +// +// LOG.debug("notify invoked"); +// m_clientIDsLock.lock(); +// try { +// assert m_clientIDs != null; +// LOG.debug("clientIDs are " + m_clientIDs); +// assert m_clientIDs.get(clientId) != null; +// LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); +// m_clientIDs.get(clientId).getSession().write(pubMessage); +// }catch(Throwable t) { +// LOG.error(null, t); +// } finally { +// m_clientIDsLock.unlock(); +// } +// } +// +// public void disconnect(IoSession session) { +// String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); +// m_clientIDsLock.lock(); +// try { +// m_clientIDs.remove(clientID); +// } finally { +// m_clientIDsLock.unlock(); +// } +// session.close(true); +// } +// +// public void sendPubAck(PubAckEvent evt) { +// LOG.debug("sendPubAck invoked"); +// +// String clientId = evt.getClientID(); +// +// PubAckMessage pubAckMessage = new PubAckMessage(); +// pubAckMessage.setMessageID(evt.getMessageId()); +// +// m_clientIDsLock.lock(); +// try { +// assert m_clientIDs != null; +// LOG.debug("clientIDs are " + m_clientIDs); +// assert m_clientIDs.get(clientId) != null; +// LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); +// m_clientIDs.get(clientId).getSession().write(pubAckMessage); +// }catch(Throwable t) { +// LOG.error(null, t); +// } finally { +// m_clientIDsLock.unlock(); +// } +// } } 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 7392d2f9..e7c61039 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 @@ -71,7 +71,7 @@ public class Server { messaging = new SimpleMessaging(); //TODO fix this hugly wiring handler.setMessaging(messaging); - messaging.setNotifier(handler); +// messaging.setNotifier(handler); messagingEventLoop = new Thread(messaging); messagingEventLoop.setName("Event Loop" + System.currentTimeMillis()); messagingEventLoop.start();