diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index f8ea68c0..c90d3f21 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -1,544 +1,544 @@ -package org.dna.mqtt.moquette.messaging.spi.impl; - -import com.lmax.disruptor.BatchEventProcessor; -import com.lmax.disruptor.EventHandler; -import com.lmax.disruptor.RingBuffer; -import com.lmax.disruptor.SequenceBarrier; - -import java.util.Collection; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -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; -import org.dna.mqtt.moquette.messaging.spi.IStorageService; -import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; -import org.dna.mqtt.moquette.messaging.spi.impl.events.*; -import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; -import org.dna.mqtt.moquette.proto.PubCompMessage; -import org.dna.mqtt.moquette.proto.messages.*; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -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 org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService.StoredMessage; - -/** - * - * Singleton class that orchestrate the execution of the protocol. - * - * Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion. - * - * @author andrea - */ -public class SimpleMessaging implements IMessaging, EventHandler { - - private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class); - - private SubscriptionsStore subscriptions = new SubscriptionsStore(); - - private RingBuffer m_ringBuffer; - -// private INotifier m_notifier; - - private IStorageService m_storageService; - - Map m_clientIDs = new HashMap(); - - private IAuthenticator m_authenticator; - - private ExecutorService m_executor; - BatchEventProcessor m_eventProcessor; - - private static SimpleMessaging INSTANCE; - - private SimpleMessaging() { - } - - public static SimpleMessaging getInstance() { - if (INSTANCE == null) { - INSTANCE = new SimpleMessaging(); - } - return INSTANCE; - } - - public void init() { - m_executor = Executors.newFixedThreadPool(1); - - m_ringBuffer = new RingBuffer(ValueEvent.EVENT_FACTORY, 1024 * 32); - - SequenceBarrier barrier = m_ringBuffer.newBarrier(); - m_eventProcessor = new BatchEventProcessor(m_ringBuffer, barrier, this); - m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence()); - m_executor.submit(m_eventProcessor); - - disruptorPublish(new InitEvent()); - } - - -/* public void setNotifier(INotifier notifier) { - m_notifier= notifier; - } */ - - - private void disruptorPublish(MessagingEvent msgEvent) { - long sequence = m_ringBuffer.next(); - ValueEvent event = m_ringBuffer.get(sequence); - - event.setEvent(msgEvent); - - m_ringBuffer.publish(sequence); - } - - - public void disconnect(IoSession session) { - disruptorPublish(new DisconnectEvent(session)); - } - - //method used by hte Notifier to re-put an event on the inbound queue - private void refill(MessagingEvent evt) { - disruptorPublish(evt); - } - - public void republishStored(String clientID) { - //create the event to push - LOG.debug("republishStored invoked to publish soterd messages for clientID " + clientID); - disruptorPublish(new RepublishEvent(clientID)); - } - - public void handleProtocolMessage(IoSession session, AbstractMessage msg) { - disruptorPublish(new ProtocolEvent(session, msg)); - } - - - /** - * NOT SAFE Method, to be removed because used only in tests - */ - protected SubscriptionsStore getSubscriptions() { - return subscriptions; - } - - - public void stop() { - disruptorPublish(new StopEvent()); - } - - public void onEvent(ValueEvent t, long l, boolean bln) throws Exception { - MessagingEvent evt = t.getEvent(); - LOG.debug("onEvent processing messaging event " + evt); - if (evt instanceof PublishEvent) { - processPublish((PublishEvent) evt); - } else if (evt instanceof StopEvent) { - processStop(); - } else if (evt instanceof DisconnectEvent) { - DisconnectEvent disEvt = (DisconnectEvent) evt; - String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID); - processDisconnect(disEvt.getSession(), clientID); - } else if (evt instanceof RepublishEvent) { - processRepublish((RepublishEvent) evt); - } else if (evt instanceof ProtocolEvent) { - IoSession session = ((ProtocolEvent) evt).getSession(); - AbstractMessage message = ((ProtocolEvent) evt).getMessage(); - if (message instanceof ConnectMessage) { - processConnect(session, (ConnectMessage) message); - } else if (message instanceof PublishMessage) { - PublishMessage pubMsg = (PublishMessage) message; - PublishEvent pubEvt; - - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - - if (message.getQos() == QOSType.MOST_ONE) { - pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, session); - - } else { - pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session); - } - processPublish(pubEvt); - } else if (message instanceof DisconnectMessage) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); - if (cleanSession) { - //cleanup topic subscriptions - processRemoveAllSubscriptions(clientID); - } - - //close the TCP connection - //session.close(true); - processDisconnect(session, clientID); - } else if (message instanceof UnsubscribeMessage) { - UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message; - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID()); - } else if (message instanceof SubscribeMessage) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); - processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession); - } else if (message instanceof PubRelMessage) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - int messageID = ((PubRelMessage) message).getMessageID(); - processPubRel(clientID, messageID); - } else if (message instanceof PubRecMessage) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - int messageID = ((PubRecMessage) message).getMessageID(); - processPubRec(clientID, messageID); - } else if (message instanceof PubCompMessage) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - int messageID = ((PubCompMessage) message).getMessageID(); - processPubComp(clientID, messageID); - } else { - throw new RuntimeException("Illegal message received " + message); - } - - } else if (evt instanceof InitEvent) { - processInit(); - } - } - - private void processInit() { - m_storageService = new HawtDBStorageService(); - m_storageService.initStore(); - - subscriptions.init(m_storageService); - } - - - protected void processConnect(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 (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 - processRemoveAllSubscriptions(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()]; - PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, msg.getWillMessage().getBytes(), - msg.isWillRetain(), msg.getClientID(), session); - processPublish(pubEvt); - } - - //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; - } - } - - subscriptions.activate(msg.getClientID()); - - //handle clean session flag - if (msg.isCleanSession()) { - //remove all prev subscriptions - //cleanup topic subscriptions - processRemoveAllSubscriptions(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); - } - - /** - * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish - * to all interested subscribers. - * */ - protected void processPubRel(String clientID, int messageID) { - String publishKey = String.format("%s%d", clientID, messageID); - PublishEvent evt = m_storageService.retrieveQoS2Message(publishKey); - - final String topic = evt.getTopic(); - final QOSType qos = evt.getQos(); - final byte[] message = evt.getMessage(); - boolean retain = evt.isRetain(); - - publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); - - m_storageService.removeQoS2Message(publishKey); - - if (retain) { - m_storageService.storeRetained(topic, message, qos); - } - - sendPubComp(clientID, messageID); - } - - protected void processPublish(PublishEvent evt) { - LOG.debug("processPublish invoked with " + evt); - final String topic = evt.getTopic(); - final QOSType qos = evt.getQos(); - final byte[] message = evt.getMessage(); - boolean retain = evt.isRetain(); - - String publishKey = null; - if (qos == QOSType.LEAST_ONE) { - //store the temporary message - publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); - m_storageService.addInFlight(evt, publishKey); - } else if (qos == QOSType.EXACTLY_ONCE) { - publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); - //store the message in temp store - m_storageService.persistQoS2Message(publishKey, evt); - sendPubRec(evt.getClientID(), evt.getMessageID()); - } - - publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); - - if (qos == QOSType.LEAST_ONE) { - assert publishKey != null; - m_storageService.cleanInFlight(publishKey); - sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID())); - } - - if (retain) { - m_storageService.storeRetained(topic, message, qos); - } - } - - /** - * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2 - * */ - private void publish2Subscribers(String topic, QOSType qos, byte[] message, boolean retain, Integer messageID) { - for (final Subscription sub : subscriptions.matches(topic)) { - if (qos == QOSType.MOST_ONE) { - //QoS 0 - notify(new NotifyEvent(sub.getClientId(), 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()) { - //clone the event with matching clientID - PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null); - m_storageService.storePublishForFuture(newPublishEvt); - } else { - //if QoS 2 then store it in temp memory - if (qos ==QOSType.EXACTLY_ONCE) { - String publishKey = String.format("%s%d", sub.getClientId(), messageID); - PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null); - m_storageService.addInFlight(newPublishEvt, publishKey); - } - //publish - notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); - } - } - } - } - - private void processPubRec(String clientID, int messageID) { - //once received a PUBREC reply with a PUBREL(messageID) - LOG.debug(String.format("processPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); - PubRelMessage pubRelMessage = new PubRelMessage(); - pubRelMessage.setMessageID(messageID); - - m_clientIDs.get(clientID).getSession().write(pubRelMessage); - } - - private void processPubComp(String clientID, int messageID) { - //once received the PUBCOMP then remove the message from the temp memory - String publishKey = String.format("%s%d", clientID, messageID); - m_storageService.cleanInFlight(publishKey); - } - - private void subscribeSingleTopic(Subscription newSubscription, final String topic) { - subscriptions.add(newSubscription); - - //scans retained messages to be published to the new subscription - Collection messages = m_storageService.searchMatching(new IMatchingCondition() { - public boolean match(String key) { - return SubscriptionsStore.matchTopics(key, topic); - } - }); - - for (StoredMessage storedMsg : messages) { - //fire the as retained the message - LOG.debug("Inserting NotifyEvent into outbound for topic " + topic); - notify(new NotifyEvent(newSubscription.getClientId(), topic, storedMsg.getQos(), storedMsg.getPayload(), true)); - } - } - - protected void processSubscribe(IoSession session, SubscribeMessage msg, String clientID, boolean cleanSession) { - LOG.debug("processSubscribe invoked"); - - for (SubscribeMessage.Couple req : msg.subscriptions()) { - QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; - Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); - subscribeSingleTopic(newSubscription, req.getTopic()); - } - - //ack the client - SubAckMessage ackMessage = new SubAckMessage(); - ackMessage.setMessageID(msg.getMessageID()); - - //TODO by now it handles only QoS 0 messages - for (int i = 0; i < msg.subscriptions().size(); i++) { - ackMessage.addType(QOSType.MOST_ONE); - } - LOG.info("replying with SubAct to MSG ID " + msg.getMessageID()); - session.write(ackMessage); - - } - - /** - * Remove the clientID from topic subscription, if not previously subscribed, - * doesn't reply any error - */ - protected void processUnsubscribe(IoSession session, String clientID, List topics, int messageID) { - LOG.debug("processSubscribe invoked"); - - for (String topic : topics) { - subscriptions.removeSubscription(topic, clientID); - } - //ack the client - UnsubAckMessage ackMessage = new UnsubAckMessage(); - ackMessage.setMessageID(messageID); - - LOG.info("replying with UnsubAck to MSG ID {0}", messageID); - session.write(ackMessage); - } - - protected void processRemoveAllSubscriptions(String clientID) { - LOG.debug("processRemoveAllSubscriptions invoked"); - subscriptions.removeForClient(clientID); - - //remove also the messages stored of type QoS1/2 - m_storageService.cleanPersistedPublishes(clientID); - } - - private void processDisconnect(IoSession session, String clientID) throws InterruptedException { -// m_notifier.disconnect(evt.getSession()); - m_clientIDs.remove(clientID); - session.close(true); - - //de-activate the subscriptions for this ClientID -// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); - subscriptions.deactivate(clientID); - } - - private void processStop() { - LOG.debug("processStop invoked"); - m_storageService.close(); - -// m_eventProcessor.halt(); - m_executor.shutdown(); - } - - private void processRepublish(RepublishEvent evt) throws InterruptedException { - LOG.debug("processRepublish invoked"); - List publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID()); - if (publishedEvents == null) { - LOG.debug("processRepublish, no stored publish events"); - return; - } - - for (PublishEvent pubEvt : publishedEvents) { - notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), - 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()); - } - - 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); - } - } - - private void sendPubRec(String clientID, int messageID) { - LOG.debug(String.format("sendPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); - PubRecMessage pubRecMessage = new PubRecMessage(); - pubRecMessage.setMessageID(messageID); - - m_clientIDs.get(clientID).getSession().write(pubRecMessage); - } - - private void sendPubComp(String clientID, int messageID) { - LOG.debug(String.format("sendPubComp invoked for clientID %s ad messageID %d", clientID, messageID)); - PubCompMessage pubCompMessage = new PubCompMessage(); - pubCompMessage.setMessageID(messageID); - - m_clientIDs.get(clientID).getSession().write(pubCompMessage); - } -} +package org.dna.mqtt.moquette.messaging.spi.impl; + +import com.lmax.disruptor.BatchEventProcessor; +import com.lmax.disruptor.EventHandler; +import com.lmax.disruptor.RingBuffer; +import com.lmax.disruptor.SequenceBarrier; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +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; +import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService.StoredMessage; +import org.dna.mqtt.moquette.messaging.spi.impl.events.*; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; +import org.dna.mqtt.moquette.proto.PubCompMessage; +import org.dna.mqtt.moquette.proto.messages.*; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +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; + +/** + * + * Singleton class that orchestrate the execution of the protocol. + * + * Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion. + * + * @author andrea + */ +public class SimpleMessaging implements IMessaging, EventHandler { + + private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class); + + private SubscriptionsStore subscriptions = new SubscriptionsStore(); + + private RingBuffer m_ringBuffer; + +// private INotifier m_notifier; + + private IStorageService m_storageService; + + Map m_clientIDs = new HashMap(); + + private IAuthenticator m_authenticator; + + private ExecutorService m_executor; + BatchEventProcessor m_eventProcessor; + + private static SimpleMessaging INSTANCE; + + private SimpleMessaging() { + } + + public static SimpleMessaging getInstance() { + if (INSTANCE == null) { + INSTANCE = new SimpleMessaging(); + } + return INSTANCE; + } + + public void init() { + m_executor = Executors.newFixedThreadPool(1); + + m_ringBuffer = new RingBuffer(ValueEvent.EVENT_FACTORY, 1024 * 32); + + SequenceBarrier barrier = m_ringBuffer.newBarrier(); + m_eventProcessor = new BatchEventProcessor(m_ringBuffer, barrier, this); + m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence()); + m_executor.submit(m_eventProcessor); + + disruptorPublish(new InitEvent()); + } + + +/* public void setNotifier(INotifier notifier) { + m_notifier= notifier; + } */ + + + private void disruptorPublish(MessagingEvent msgEvent) { + long sequence = m_ringBuffer.next(); + ValueEvent event = m_ringBuffer.get(sequence); + + event.setEvent(msgEvent); + + m_ringBuffer.publish(sequence); + } + + + public void disconnect(IoSession session) { + disruptorPublish(new DisconnectEvent(session)); + } + + //method used by hte Notifier to re-put an event on the inbound queue + private void refill(MessagingEvent evt) { + disruptorPublish(evt); + } + + public void republishStored(String clientID) { + //create the event to push + LOG.debug("republishStored invoked to publish soterd messages for clientID " + clientID); + disruptorPublish(new RepublishEvent(clientID)); + } + + public void handleProtocolMessage(IoSession session, AbstractMessage msg) { + disruptorPublish(new ProtocolEvent(session, msg)); + } + + + /** + * NOT SAFE Method, to be removed because used only in tests + */ + protected SubscriptionsStore getSubscriptions() { + return subscriptions; + } + + + public void stop() { + disruptorPublish(new StopEvent()); + } + + public void onEvent(ValueEvent t, long l, boolean bln) throws Exception { + MessagingEvent evt = t.getEvent(); + LOG.debug("onEvent processing messaging event " + evt); + if (evt instanceof PublishEvent) { + processPublish((PublishEvent) evt); + } else if (evt instanceof StopEvent) { + processStop(); + } else if (evt instanceof DisconnectEvent) { + DisconnectEvent disEvt = (DisconnectEvent) evt; + String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID); + processDisconnect(disEvt.getSession(), clientID); + } else if (evt instanceof RepublishEvent) { + processRepublish((RepublishEvent) evt); + } else if (evt instanceof ProtocolEvent) { + IoSession session = ((ProtocolEvent) evt).getSession(); + AbstractMessage message = ((ProtocolEvent) evt).getMessage(); + if (message instanceof ConnectMessage) { + processConnect(session, (ConnectMessage) message); + } else if (message instanceof PublishMessage) { + PublishMessage pubMsg = (PublishMessage) message; + PublishEvent pubEvt; + + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + + if (message.getQos() == QOSType.MOST_ONE) { + pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, session); + + } else { + pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session); + } + processPublish(pubEvt); + } else if (message instanceof DisconnectMessage) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); + if (cleanSession) { + //cleanup topic subscriptions + processRemoveAllSubscriptions(clientID); + } + + //close the TCP connection + //session.close(true); + processDisconnect(session, clientID); + } else if (message instanceof UnsubscribeMessage) { + UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message; + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID()); + } else if (message instanceof SubscribeMessage) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); + processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession); + } else if (message instanceof PubRelMessage) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + int messageID = ((PubRelMessage) message).getMessageID(); + processPubRel(clientID, messageID); + } else if (message instanceof PubRecMessage) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + int messageID = ((PubRecMessage) message).getMessageID(); + processPubRec(clientID, messageID); + } else if (message instanceof PubCompMessage) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + int messageID = ((PubCompMessage) message).getMessageID(); + processPubComp(clientID, messageID); + } else { + throw new RuntimeException("Illegal message received " + message); + } + + } else if (evt instanceof InitEvent) { + processInit(); + } + } + + private void processInit() { + m_storageService = new HawtDBStorageService(); + m_storageService.initStore(); + + subscriptions.init(m_storageService); + } + + + protected void processConnect(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 (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 + processRemoveAllSubscriptions(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(); + LOG.debug(String.format("Connect with keepAlive %d s", keepAlive)); + 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()]; + PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, msg.getWillMessage().getBytes(), + msg.isWillRetain(), msg.getClientID(), session); + processPublish(pubEvt); + } + + //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; + } + } + + subscriptions.activate(msg.getClientID()); + + //handle clean session flag + if (msg.isCleanSession()) { + //remove all prev subscriptions + //cleanup topic subscriptions + processRemoveAllSubscriptions(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); + } + + /** + * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish + * to all interested subscribers. + * */ + protected void processPubRel(String clientID, int messageID) { + String publishKey = String.format("%s%d", clientID, messageID); + PublishEvent evt = m_storageService.retrieveQoS2Message(publishKey); + + final String topic = evt.getTopic(); + final QOSType qos = evt.getQos(); + final byte[] message = evt.getMessage(); + boolean retain = evt.isRetain(); + + publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); + + m_storageService.removeQoS2Message(publishKey); + + if (retain) { + m_storageService.storeRetained(topic, message, qos); + } + + sendPubComp(clientID, messageID); + } + + protected void processPublish(PublishEvent evt) { + LOG.debug("processPublish invoked with " + evt); + final String topic = evt.getTopic(); + final QOSType qos = evt.getQos(); + final byte[] message = evt.getMessage(); + boolean retain = evt.isRetain(); + + String publishKey = null; + if (qos == QOSType.LEAST_ONE) { + //store the temporary message + publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); + m_storageService.addInFlight(evt, publishKey); + } else if (qos == QOSType.EXACTLY_ONCE) { + publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); + //store the message in temp store + m_storageService.persistQoS2Message(publishKey, evt); + sendPubRec(evt.getClientID(), evt.getMessageID()); + } + + publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); + + if (qos == QOSType.LEAST_ONE) { + assert publishKey != null; + m_storageService.cleanInFlight(publishKey); + sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID())); + } + + if (retain) { + m_storageService.storeRetained(topic, message, qos); + } + } + + /** + * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2 + * */ + private void publish2Subscribers(String topic, QOSType qos, byte[] message, boolean retain, Integer messageID) { + for (final Subscription sub : subscriptions.matches(topic)) { + if (qos == QOSType.MOST_ONE) { + //QoS 0 + notify(new NotifyEvent(sub.getClientId(), 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()) { + //clone the event with matching clientID + PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null); + m_storageService.storePublishForFuture(newPublishEvt); + } else { + //if QoS 2 then store it in temp memory + if (qos ==QOSType.EXACTLY_ONCE) { + String publishKey = String.format("%s%d", sub.getClientId(), messageID); + PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null); + m_storageService.addInFlight(newPublishEvt, publishKey); + } + //publish + notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); + } + } + } + } + + private void processPubRec(String clientID, int messageID) { + //once received a PUBREC reply with a PUBREL(messageID) + LOG.debug(String.format("processPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); + PubRelMessage pubRelMessage = new PubRelMessage(); + pubRelMessage.setMessageID(messageID); + + m_clientIDs.get(clientID).getSession().write(pubRelMessage); + } + + private void processPubComp(String clientID, int messageID) { + //once received the PUBCOMP then remove the message from the temp memory + String publishKey = String.format("%s%d", clientID, messageID); + m_storageService.cleanInFlight(publishKey); + } + + private void subscribeSingleTopic(Subscription newSubscription, final String topic) { + subscriptions.add(newSubscription); + + //scans retained messages to be published to the new subscription + Collection messages = m_storageService.searchMatching(new IMatchingCondition() { + public boolean match(String key) { + return SubscriptionsStore.matchTopics(key, topic); + } + }); + + for (StoredMessage storedMsg : messages) { + //fire the as retained the message + LOG.debug("Inserting NotifyEvent into outbound for topic " + topic); + notify(new NotifyEvent(newSubscription.getClientId(), topic, storedMsg.getQos(), storedMsg.getPayload(), true)); + } + } + + protected void processSubscribe(IoSession session, SubscribeMessage msg, String clientID, boolean cleanSession) { + LOG.debug("processSubscribe invoked"); + + for (SubscribeMessage.Couple req : msg.subscriptions()) { + QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; + Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); + subscribeSingleTopic(newSubscription, req.getTopic()); + } + + //ack the client + SubAckMessage ackMessage = new SubAckMessage(); + ackMessage.setMessageID(msg.getMessageID()); + + //TODO by now it handles only QoS 0 messages + for (int i = 0; i < msg.subscriptions().size(); i++) { + ackMessage.addType(QOSType.MOST_ONE); + } + LOG.info("replying with SubAct to MSG ID " + msg.getMessageID()); + session.write(ackMessage); + + } + + /** + * Remove the clientID from topic subscription, if not previously subscribed, + * doesn't reply any error + */ + protected void processUnsubscribe(IoSession session, String clientID, List topics, int messageID) { + LOG.debug("processSubscribe invoked"); + + for (String topic : topics) { + subscriptions.removeSubscription(topic, clientID); + } + //ack the client + UnsubAckMessage ackMessage = new UnsubAckMessage(); + ackMessage.setMessageID(messageID); + + LOG.info("replying with UnsubAck to MSG ID {0}", messageID); + session.write(ackMessage); + } + + protected void processRemoveAllSubscriptions(String clientID) { + LOG.debug("processRemoveAllSubscriptions invoked"); + subscriptions.removeForClient(clientID); + + //remove also the messages stored of type QoS1/2 + m_storageService.cleanPersistedPublishes(clientID); + } + + private void processDisconnect(IoSession session, String clientID) throws InterruptedException { +// m_notifier.disconnect(evt.getSession()); + m_clientIDs.remove(clientID); + session.close(true); + + //de-activate the subscriptions for this ClientID +// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); + subscriptions.deactivate(clientID); + } + + private void processStop() { + LOG.debug("processStop invoked"); + m_storageService.close(); + +// m_eventProcessor.halt(); + m_executor.shutdown(); + } + + private void processRepublish(RepublishEvent evt) throws InterruptedException { + LOG.debug("processRepublish invoked"); + List publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID()); + if (publishedEvents == null) { + LOG.debug("processRepublish, no stored publish events"); + return; + } + + for (PublishEvent pubEvt : publishedEvents) { + notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), + 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()); + } + + 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); + } + } + + private void sendPubRec(String clientID, int messageID) { + LOG.debug(String.format("sendPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); + PubRecMessage pubRecMessage = new PubRecMessage(); + pubRecMessage.setMessageID(messageID); + + m_clientIDs.get(clientID).getSession().write(pubRecMessage); + } + + private void sendPubComp(String clientID, int messageID) { + LOG.debug(String.format("sendPubComp invoked for clientID %s ad messageID %d", clientID, messageID)); + PubCompMessage pubCompMessage = new PubCompMessage(); + pubCompMessage.setMessageID(messageID); + + m_clientIDs.get(clientID).getSession().write(pubCompMessage); + } +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java index 7bdc07fe..507d4e06 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java @@ -1,56 +1,62 @@ -package org.dna.mqtt.moquette.server; - -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.messaging.spi.IMessaging; -import org.dna.mqtt.moquette.proto.Utils; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; - -/** - * MINA MQTT Handler used to route messages to protocol logic - * - * @author andrea - */ -public class MQTTHandler extends IoHandlerAdapter { - - private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class); - private IMessaging m_messaging; - - @Override - public void messageReceived(IoSession session, Object message) throws Exception { - AbstractMessage msg = (AbstractMessage) message; - LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType()))); - try { - switch (msg.getMessageType()) { - case CONNECT: - case SUBSCRIBE: - case UNSUBSCRIBE: - case PUBLISH: - case PUBREC: - case PUBCOMP: - case PUBREL: - case PINGREQ: - case DISCONNECT: - m_messaging.handleProtocolMessage(session, msg); - break; - } - } catch (Exception ex) { - LOG.error("Bad error in processing the message", ex); - } - } - - @Override - public void sessionIdle(IoSession session, IdleStatus status) { - session.close(false); - } - - public void setMessaging(IMessaging messaging) { - m_messaging = messaging; - } - -} +package org.dna.mqtt.moquette.server; + +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.messaging.spi.IMessaging; +import org.dna.mqtt.moquette.proto.Utils; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; +import org.dna.mqtt.moquette.proto.messages.PingRespMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * MINA MQTT Handler used to route messages to protocol logic + * + * @author andrea + */ +public class MQTTHandler extends IoHandlerAdapter { + + private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class); + private IMessaging m_messaging; + + @Override + public void messageReceived(IoSession session, Object message) throws Exception { + AbstractMessage msg = (AbstractMessage) message; + LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType()))); + try { + switch (msg.getMessageType()) { + case CONNECT: + case SUBSCRIBE: + case UNSUBSCRIBE: + case PUBLISH: + case PUBREC: + case PUBCOMP: + case PUBREL: + case DISCONNECT: + m_messaging.handleProtocolMessage(session, msg); + break; + case PINGREQ: + PingRespMessage pingResp = new PingRespMessage(); + session.write(pingResp); + break; + } + } catch (Exception ex) { + LOG.error("Bad error in processing the message", ex); + } + } + + @Override + public void sessionIdle(IoSession session, IdleStatus status) { + if (status == IdleStatus.READER_IDLE) { + session.close(false); + //TODO send a notification to messaging part to remove the bining clientID-ConnConfig + } + } + + public void setMessaging(IMessaging messaging) { + m_messaging = messaging; + } + +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java index 9f8624e6..3a94ca55 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -1,137 +1,136 @@ -package org.dna.mqtt.moquette.server; - -import java.io.File; -import java.io.IOException; -import java.net.InetSocketAddress; - -import org.apache.mina.core.service.IoAcceptor; -import org.apache.mina.core.service.IoServiceStatistics; -import org.apache.mina.core.session.IdleStatus; -import org.apache.mina.core.session.IoSession; -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.transport.socket.nio.NioSocketAcceptor; -import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; -import org.dna.mqtt.moquette.proto.*; -import org.dna.mqtt.moquette.proto.messages.*; -import org.dna.mqtt.commons.Constants; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -/** - * Launch a configured version of the server. - * @author andrea - */ -public class Server { - - private static final Logger LOG = LoggerFactory.getLogger(Server.class); - -// public static final int PORT = 1883; -// public static final int DEFAULT_CONNECT_TIMEOUT = 10; - public static final String STORAGE_FILE_PATH = System.getProperty("user.home") + - File.separator + "moquette_store.hawtdb"; - - //TODO one notifier per outbound queue, else no serialization of message flow is granted!! - //private static final int NOTIFIER_POOL_SIZE = 1; - private IoAcceptor m_acceptor; - SimpleMessaging messaging; -// Thread messagingEventLoop; - //private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/; - - public static void main(String[] args) throws IOException { - new Server().startServer(); - - } - - protected void startServer() throws IOException { - DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); - decoder.addMessageDecoder(new ConnectDecoder()); - decoder.addMessageDecoder(new PublishDecoder()); - decoder.addMessageDecoder(new PubAckDecoder()); - decoder.addMessageDecoder(new PubRelDecoder()); - decoder.addMessageDecoder(new PubRecDecoder()); - decoder.addMessageDecoder(new PubCompDecoder()); - decoder.addMessageDecoder(new SubscribeDecoder()); - decoder.addMessageDecoder(new UnsubscribeDecoder()); - decoder.addMessageDecoder(new DisconnectDecoder()); - decoder.addMessageDecoder(new PingReqDecoder()); - - DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); -// encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); - encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder()); - encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder()); - encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder()); - encoder.addMessageEncoder(PubAckMessage.class, new PubAckEncoder()); - encoder.addMessageEncoder(PubRecMessage.class, new PubRecEncoder()); - encoder.addMessageEncoder(PubCompMessage.class, new PubCompEncoder()); - encoder.addMessageEncoder(PubRelMessage.class, new PubRelEncoder()); - encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); - encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder()); - - m_acceptor = new NioSocketAcceptor(); - - m_acceptor.getFilterChain().addLast( "logger", new MQTTLoggingFilter("SERVER LOG") ); - m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder)); - - MQTTHandler handler = new MQTTHandler(); - messaging = SimpleMessaging.getInstance(); - messaging.init(); - //TODO fix this hugly wiring - handler.setMessaging(messaging); -// messaging.setNotifier(handler); -// messagingEventLoop = new Thread(messaging); -// messagingEventLoop.setName("Event Loop" + System.currentTimeMillis()); -// messagingEventLoop.start(); - - m_acceptor.setHandler(handler); - ((NioSocketAcceptor)m_acceptor).setReuseAddress(true); - ((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true); - m_acceptor.getSessionConfig().setReadBufferSize( 2048 ); - m_acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, Constants.DEFAULT_CONNECT_TIMEOUT ); - m_acceptor.getStatistics().setThroughputCalculationInterval(10); - m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis()); - m_acceptor.bind( new InetSocketAddress(Constants.PORT) ); - LOG.info("Server binded"); - - //Bind a shutdown hook - Runtime.getRuntime().addShutdownHook(new Thread() { - @Override - public void run() { - stopServer(); - } - }); - } - - protected void stopServer() { - LOG.info("Server stopping..."); - - messaging.stop(); - //messaging.stop(); -// messagingEventLoop.interrupt(); -// LOG.info("shutting down evet loop"); -// try { -// messagingEventLoop.join(); -// } catch (InterruptedException ex) { -// LOG.error(null, ex); -// } - - /*m_notifierPool.shutdown();*/ - - //log statistics - IoServiceStatistics statistics = m_acceptor.getStatistics(); - statistics.updateThroughput(System.currentTimeMillis()); - System.out.println(String.format("Total read bytes: %d, read throughtput: %f (b/s)", statistics.getReadBytes(), statistics.getReadBytesThroughput())); - System.out.println(String.format("Total read msgs: %d, read msg throughtput: %f (msg/s)", statistics.getReadMessages(), statistics.getReadMessagesThroughput())); - - for(IoSession session: m_acceptor.getManagedSessions().values()) { - if(session.isConnected() && !session.isClosing()){ - session.close(false); - } - } - - m_acceptor.unbind(); - m_acceptor.dispose(); - LOG.info("Server stopped"); - } - -} +package org.dna.mqtt.moquette.server; + +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; +import org.apache.mina.core.service.IoAcceptor; +import org.apache.mina.core.service.IoServiceStatistics; +import org.apache.mina.core.session.IdleStatus; +import org.apache.mina.core.session.IoSession; +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.transport.socket.nio.NioSocketAcceptor; +import org.dna.mqtt.commons.Constants; +import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; +import org.dna.mqtt.moquette.proto.*; +import org.dna.mqtt.moquette.proto.messages.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +/** + * Launch a configured version of the server. + * @author andrea + */ +public class Server { + + private static final Logger LOG = LoggerFactory.getLogger(Server.class); + +// public static final int PORT = 1883; +// public static final int DEFAULT_CONNECT_TIMEOUT = 10; + public static final String STORAGE_FILE_PATH = System.getProperty("user.home") + + File.separator + "moquette_store.hawtdb"; + + //TODO one notifier per outbound queue, else no serialization of message flow is granted!! + //private static final int NOTIFIER_POOL_SIZE = 1; + private IoAcceptor m_acceptor; + SimpleMessaging messaging; +// Thread messagingEventLoop; + //private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/; + + public static void main(String[] args) throws IOException { + new Server().startServer(); + + } + + protected void startServer() throws IOException { + DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); + decoder.addMessageDecoder(new ConnectDecoder()); + decoder.addMessageDecoder(new PublishDecoder()); + decoder.addMessageDecoder(new PubAckDecoder()); + decoder.addMessageDecoder(new PubRelDecoder()); + decoder.addMessageDecoder(new PubRecDecoder()); + decoder.addMessageDecoder(new PubCompDecoder()); + decoder.addMessageDecoder(new SubscribeDecoder()); + decoder.addMessageDecoder(new UnsubscribeDecoder()); + decoder.addMessageDecoder(new DisconnectDecoder()); + decoder.addMessageDecoder(new PingReqDecoder()); + + DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); +// encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); + encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder()); + encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder()); + encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder()); + encoder.addMessageEncoder(PubAckMessage.class, new PubAckEncoder()); + encoder.addMessageEncoder(PubRecMessage.class, new PubRecEncoder()); + encoder.addMessageEncoder(PubCompMessage.class, new PubCompEncoder()); + encoder.addMessageEncoder(PubRelMessage.class, new PubRelEncoder()); + encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); + encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder()); + + m_acceptor = new NioSocketAcceptor(); + + m_acceptor.getFilterChain().addLast( "logger", new MQTTLoggingFilter("SERVER LOG") ); + m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder)); + + MQTTHandler handler = new MQTTHandler(); + messaging = SimpleMessaging.getInstance(); + messaging.init(); + //TODO fix this hugly wiring + handler.setMessaging(messaging); +// messaging.setNotifier(handler); +// messagingEventLoop = new Thread(messaging); +// messagingEventLoop.setName("Event Loop" + System.currentTimeMillis()); +// messagingEventLoop.start(); + + m_acceptor.setHandler(handler); + ((NioSocketAcceptor)m_acceptor).setReuseAddress(true); + ((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true); + m_acceptor.getSessionConfig().setReadBufferSize( 2048 ); + m_acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, Constants.DEFAULT_CONNECT_TIMEOUT ); + m_acceptor.getStatistics().setThroughputCalculationInterval(10); + m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis()); + m_acceptor.bind( new InetSocketAddress(Constants.PORT) ); + LOG.info("Server binded"); + + //Bind a shutdown hook + Runtime.getRuntime().addShutdownHook(new Thread() { + @Override + public void run() { + stopServer(); + } + }); + } + + protected void stopServer() { + LOG.info("Server stopping..."); + + messaging.stop(); + //messaging.stop(); +// messagingEventLoop.interrupt(); +// LOG.info("shutting down evet loop"); +// try { +// messagingEventLoop.join(); +// } catch (InterruptedException ex) { +// LOG.error(null, ex); +// } + + /*m_notifierPool.shutdown();*/ + + //log statistics + IoServiceStatistics statistics = m_acceptor.getStatistics(); + statistics.updateThroughput(System.currentTimeMillis()); + System.out.println(String.format("Total read bytes: %d, read throughtput: %f (b/s)", statistics.getReadBytes(), statistics.getReadBytesThroughput())); + System.out.println(String.format("Total read msgs: %d, read msg throughtput: %f (msg/s)", statistics.getReadMessages(), statistics.getReadMessagesThroughput())); + + for(IoSession session: m_acceptor.getManagedSessions().values()) { + if(session.isConnected() && !session.isClosing()){ + session.close(false); + } + } + + m_acceptor.unbind(); + m_acceptor.dispose(); + LOG.info("Server stopped"); + } + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java b/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java index f98aa81d..e879e09f 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java @@ -1,57 +1,63 @@ -package org.dna.mqtt.moquette.proto; - -import org.apache.mina.core.buffer.IoBuffer; -import org.apache.mina.core.session.IoSession; -import org.apache.mina.filter.codec.ProtocolDecoderOutput; -import org.apache.mina.filter.codec.demux.MessageDecoderResult; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import org.dna.mqtt.moquette.proto.messages.PublishMessage; - -/** - * - * @author andrea - */ -public class PublishDecoder extends MqttDecoder { - - public MessageDecoderResult decodable(IoSession session, IoBuffer in) { - return Utils.checkDecodable(AbstractMessage.PUBLISH, in); - } - - public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { - int startPos = in.position(); - - //Common decoding part - PublishMessage message = new PublishMessage(); - if (decodeCommonHeader(message, in) == NEED_DATA) { - return NEED_DATA; - } - int remainingLength = message.getRemainingLength(); - - //Topic name - String topic = Utils.decodeString(in); - if (topic == null) { - return NEED_DATA; - } - message.setTopicName(topic); - - if (message.getQos() == QOSType.LEAST_ONE || - message.getQos() == QOSType.EXACTLY_ONCE) { - message.setMessageID(Utils.readWord(in)); - } - int stopPos = in.position(); - - //read the payload - int payloadSize = remainingLength - (stopPos - startPos - 2); - byte[] b = new byte[payloadSize]; - if (in.remaining() < payloadSize) { - return NEED_DATA; - } - in.get(b); - message.setPayload(b); - - out.write(message); - return OK; - } - -} +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +public class PublishDecoder extends MqttDecoder { + + private static Logger LOG = LoggerFactory.getLogger(PublishDecoder.class); + + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.PUBLISH, in); + } + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + LOG.debug("decode invoked with buffer " + in); + int startPos = in.position(); + + //Common decoding part + PublishMessage message = new PublishMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + LOG.info("decode ask for more data after " + in); + return NEED_DATA; + } + int remainingLength = message.getRemainingLength(); + + //Topic name + String topic = Utils.decodeString(in); + if (topic == null) { + return NEED_DATA; + } + message.setTopicName(topic); + + if (message.getQos() == QOSType.LEAST_ONE || + message.getQos() == QOSType.EXACTLY_ONCE) { + message.setMessageID(Utils.readWord(in)); + } + int stopPos = in.position(); + + //read the payload + int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); + if (in.remaining() < payloadSize) { + return NEED_DATA; + } + byte[] b = new byte[payloadSize]; + in.get(b); + message.setPayload(b); + + out.write(message); + return OK; + } + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java b/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java index 9ac028b8..b44dafa0 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java @@ -1,48 +1,48 @@ -package org.dna.mqtt.moquette.proto; - -import org.apache.mina.core.buffer.IoBuffer; -import org.apache.mina.core.session.IoSession; -import org.apache.mina.filter.codec.ProtocolEncoderOutput; -import org.apache.mina.filter.codec.demux.MessageEncoder; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import org.dna.mqtt.moquette.proto.messages.PublishMessage; - -/** - * - * @author andrea - */ -public class PublishEncoder implements MessageEncoder { - - public void encode(IoSession session, PublishMessage message, ProtocolEncoderOutput out) throws Exception { - if (message.getQos() == QOSType.RESERVED) { - throw new IllegalArgumentException("Found a message with RESERVED Qos"); - } - if (message.getTopicName() == null || message.getTopicName().isEmpty()) { - throw new IllegalArgumentException("Found a message with empty or null topic name"); - } - - IoBuffer variableHeaderBuff = IoBuffer.allocate(2).setAutoExpand(true); - variableHeaderBuff.put(Utils.encodeString(message.getTopicName())); - if (message.getQos() == QOSType.LEAST_ONE || - message.getQos() == QOSType.EXACTLY_ONCE ) { - if (message.getMessageID() == null) { - throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted"); - } - Utils.writeWord(variableHeaderBuff, message.getMessageID()); - } - variableHeaderBuff.put(message.getPayload()); - variableHeaderBuff.flip(); - int variableHeaderSize = variableHeaderBuff.remaining(); - - byte flags = Utils.encodeFlags(message); - - IoBuffer buff = IoBuffer.allocate(2 + variableHeaderSize); - buff.put((byte) (AbstractMessage.PUBLISH << 4 | flags)); - buff.put(Utils.encodeRemainingLength(variableHeaderSize)); - buff.put(variableHeaderBuff).flip(); - - out.write(buff); - } - -} +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; + +/** + * + * @author andrea + */ +public class PublishEncoder implements MessageEncoder { + + public void encode(IoSession session, PublishMessage message, ProtocolEncoderOutput out) throws Exception { + if (message.getQos() == QOSType.RESERVED) { + throw new IllegalArgumentException("Found a message with RESERVED Qos"); + } + if (message.getTopicName() == null || message.getTopicName().isEmpty()) { + throw new IllegalArgumentException("Found a message with empty or null topic name"); + } + + IoBuffer variableHeaderBuff = IoBuffer.allocate(2).setAutoExpand(true); + variableHeaderBuff.put(Utils.encodeString(message.getTopicName())); + if (message.getQos() == QOSType.LEAST_ONE || + message.getQos() == QOSType.EXACTLY_ONCE ) { + if (message.getMessageID() == null) { + throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted"); + } + Utils.writeWord(variableHeaderBuff, message.getMessageID()); + } + variableHeaderBuff.put(message.getPayload()); + variableHeaderBuff.flip(); + int variableHeaderSize = variableHeaderBuff.remaining(); + + byte flags = Utils.encodeFlags(message); + + IoBuffer buff = IoBuffer.allocate(2 + variableHeaderSize).setAutoExpand(true);; + buff.put((byte) (AbstractMessage.PUBLISH << 4 | flags)); + buff.put(Utils.encodeRemainingLength(variableHeaderSize)); + buff.put(variableHeaderBuff).flip(); + + out.write(buff); + } + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/Utils.java b/parser/src/main/java/org/dna/mqtt/moquette/proto/Utils.java index b5dd63d8..899b5e7e 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/proto/Utils.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/proto/Utils.java @@ -1,180 +1,191 @@ -package org.dna.mqtt.moquette.proto; - -import java.io.UnsupportedEncodingException; -import org.apache.mina.core.buffer.IoBuffer; -import org.apache.mina.filter.codec.demux.MessageDecoderResult; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.slf4j.LoggerFactory; - -/** - * Common utils methodd used in codecs. - * - * @author andrea - */ -public class Utils { - - public static final int MAX_LENGTH_LIMIT = 268435455; - - /** - * Read 2 bytes from in buffer first MSB, and then LSB returning as int. - */ - static int readWord(IoBuffer in) { - int msb = in.get() & 0x00FF; //remove sign extension due to casting - int lsb = in.get() & 0x00FF; - msb = (msb << 8) | lsb ; - return msb; - } - - /** - * Writes as 2 bytes the int value into buffer first MSB, and then LSB. - */ - static void writeWord(IoBuffer out, int value) { - out.put((byte) ((value & 0xFF00) >> 8)); //msb - out.put((byte) (value & 0x00FF)); //lsb - } - - /** - * Decode the variable remaining lenght as defined in MQTT v3.1 specification - * (section 2.1). - * - * @return the decoded length or -1 if needed more data to decode the length field. - */ - static int decodeRemainingLenght(IoBuffer in) { - int multiplier = 1; - int value = 0; - byte digit; - do { - if (in.remaining() < 1) { - return -1; - } - digit = in.get(); - value += (digit & 0x7F) * multiplier; - multiplier *= 128; - } while ((digit & 0x80) != 0); - return value; - } - - /** - * Encode the value in the format defined in specification as variable length - * array. - * - * @throws IllegalArgumentException if the value is not in the specification bounds - * [0..268435455]. - */ - static IoBuffer encodeRemainingLength(int value) throws IllegalAccessException { - if (value > MAX_LENGTH_LIMIT || value < 0) { - throw new IllegalAccessException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value); - } - - IoBuffer encoded = IoBuffer.allocate(4); - byte digit; - do { - digit = (byte) (value % 128); - value = value / 128; - // if there are more digits to encode, set the top bit of this digit - if (value > 0) { - digit = (byte) (digit | 0x80); - } - encoded.put(digit); - } while (value > 0); - encoded.flip(); - return encoded; - } - - static MessageDecoderResult checkDecodable(byte type, IoBuffer in) { - if (in.remaining() < 1) { - return MessageDecoderResult.NEED_DATA; - } - byte h1 = in.get(); - byte messageType = (byte) ((h1 & 0x00F0) >> 4); - - int remainingLength = Utils.decodeRemainingLenght(in); - if (remainingLength == -1) { - return MessageDecoderResult.NEED_DATA; - } - - //check remaining length - if (in.remaining() < remainingLength) { - return MessageDecoderResult.NEED_DATA; - } - - return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK; - } - - /** - * Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded - * string content. - */ - static IoBuffer encodeString(String str) { - IoBuffer out = IoBuffer.allocate(2).setAutoExpand(true); - byte[] raw; - try { - raw = str.getBytes("UTF-8"); - //NB every Java platform has got UTF-8 encoding by default, so this - //exception are never raised. - } catch (UnsupportedEncodingException ex) { - LoggerFactory.getLogger(ConnectEncoder.class).error(null, ex); - return null; - } - Utils.writeWord(out, raw.length); - out.put(raw).flip(); - return out; - } - - - /** - * Load a string from the given buffer, reading first the two bytes of len - * and then the UTF-8 bytes of the string. - * - * @return the decoded string or null if NEED_DATA - */ - static String decodeString(IoBuffer in) throws UnsupportedEncodingException { - if (in.remaining() < 2) { - return null; - } - int strLen = Utils.readWord(in); - if (in.remaining() < strLen) { - return null; - } - byte[] strRaw = new byte[strLen]; - in.get(strRaw); - - return new String(strRaw, "UTF-8"); - } - - static byte encodeFlags(AbstractMessage message) { - byte flags = 0; - if (message.isDupFlag()) { - flags |= 0x08; - } - if (message.isRetainFlag()) { - flags |= 0x01; - } - - flags |= ((message.getQos().ordinal() & 0x03) << 1); - return flags; - } - - /** - * Converts MQTT message type to a textual description. - * */ - public static String msgType2String(int type) { - switch (type) { - case AbstractMessage.CONNECT: return "CONNECT"; - case AbstractMessage.CONNACK: return "CONNACK"; - case AbstractMessage.PUBLISH: return "PUBLISH"; - case AbstractMessage.PUBACK: return "PUBACK"; - case AbstractMessage.PUBREC: return "PUBREC"; - case AbstractMessage.PUBREL: return "PUBREL"; - case AbstractMessage.PUBCOMP: return "PUBCOMP"; - case AbstractMessage.SUBSCRIBE: return "SUBSCRIBE"; - case AbstractMessage.SUBACK: return "SUBACK"; - case AbstractMessage.UNSUBSCRIBE: return "UNSUBSCRIBE"; - case AbstractMessage.UNSUBACK: return "UNSUBACK"; - case AbstractMessage.PINGREQ: return "PINGREQ"; - case AbstractMessage.PINGRESP: return "PINGRESP"; - case AbstractMessage.DISCONNECT: return "DISCONNECT"; - default: throw new RuntimeException("Can't decode message type " + type); - } - } -} +package org.dna.mqtt.moquette.proto; + +import java.io.UnsupportedEncodingException; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.slf4j.LoggerFactory; + +/** + * Common utils methodd used in codecs. + * + * @author andrea + */ +public class Utils { + + public static final int MAX_LENGTH_LIMIT = 268435455; + + /** + * Read 2 bytes from in buffer first MSB, and then LSB returning as int. + */ + static int readWord(IoBuffer in) { + int msb = in.get() & 0x00FF; //remove sign extension due to casting + int lsb = in.get() & 0x00FF; + msb = (msb << 8) | lsb ; + return msb; + } + + /** + * Writes as 2 bytes the int value into buffer first MSB, and then LSB. + */ + static void writeWord(IoBuffer out, int value) { + out.put((byte) ((value & 0xFF00) >> 8)); //msb + out.put((byte) (value & 0x00FF)); //lsb + } + + /** + * Decode the variable remaining lenght as defined in MQTT v3.1 specification + * (section 2.1). + * + * @return the decoded length or -1 if needed more data to decode the length field. + */ + static int decodeRemainingLenght(IoBuffer in) { + int multiplier = 1; + int value = 0; + byte digit; + do { + if (in.remaining() < 1) { + return -1; + } + digit = in.get(); + value += (digit & 0x7F) * multiplier; + multiplier *= 128; + } while ((digit & 0x80) != 0); + return value; + } + + /** + * Return the number of bytes to encode the gicen remaining length value + */ + static int numBytesToEncode(int len) { + if (0 <= len && len <= 127) return 1; + if (128 <= len && len <= 16383) return 2; + if (16384 <= len && len <= 2097151) return 3; + if (2097152 <= len && len <= 268435455) return 4; + throw new IllegalArgumentException("value shoul be in the range [0..268435455]"); + } + + /** + * Encode the value in the format defined in specification as variable length + * array. + * + * @throws IllegalArgumentException if the value is not in the specification bounds + * [0..268435455]. + */ + static IoBuffer encodeRemainingLength(int value) throws IllegalAccessException { + if (value > MAX_LENGTH_LIMIT || value < 0) { + throw new IllegalAccessException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value); + } + + IoBuffer encoded = IoBuffer.allocate(4); + byte digit; + do { + digit = (byte) (value % 128); + value = value / 128; + // if there are more digits to encode, set the top bit of this digit + if (value > 0) { + digit = (byte) (digit | 0x80); + } + encoded.put(digit); + } while (value > 0); + encoded.flip(); + return encoded; + } + + static MessageDecoderResult checkDecodable(byte type, IoBuffer in) { + if (in.remaining() < 1) { + return MessageDecoderResult.NEED_DATA; + } + byte h1 = in.get(); + byte messageType = (byte) ((h1 & 0x00F0) >> 4); + + int remainingLength = Utils.decodeRemainingLenght(in); + if (remainingLength == -1) { + return MessageDecoderResult.NEED_DATA; + } + + //check remaining length + if (in.remaining() < remainingLength) { + return MessageDecoderResult.NEED_DATA; + } + + return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK; + } + + /** + * Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded + * string content. + */ + static IoBuffer encodeString(String str) { + IoBuffer out = IoBuffer.allocate(2).setAutoExpand(true); + byte[] raw; + try { + raw = str.getBytes("UTF-8"); + //NB every Java platform has got UTF-8 encoding by default, so this + //exception are never raised. + } catch (UnsupportedEncodingException ex) { + LoggerFactory.getLogger(ConnectEncoder.class).error(null, ex); + return null; + } + Utils.writeWord(out, raw.length); + out.put(raw).flip(); + return out; + } + + + /** + * Load a string from the given buffer, reading first the two bytes of len + * and then the UTF-8 bytes of the string. + * + * @return the decoded string or null if NEED_DATA + */ + static String decodeString(IoBuffer in) throws UnsupportedEncodingException { + if (in.remaining() < 2) { + return null; + } + int strLen = Utils.readWord(in); + if (in.remaining() < strLen) { + return null; + } + byte[] strRaw = new byte[strLen]; + in.get(strRaw); + + return new String(strRaw, "UTF-8"); + } + + static byte encodeFlags(AbstractMessage message) { + byte flags = 0; + if (message.isDupFlag()) { + flags |= 0x08; + } + if (message.isRetainFlag()) { + flags |= 0x01; + } + + flags |= ((message.getQos().ordinal() & 0x03) << 1); + return flags; + } + + /** + * Converts MQTT message type to a textual description. + * */ + public static String msgType2String(int type) { + switch (type) { + case AbstractMessage.CONNECT: return "CONNECT"; + case AbstractMessage.CONNACK: return "CONNACK"; + case AbstractMessage.PUBLISH: return "PUBLISH"; + case AbstractMessage.PUBACK: return "PUBACK"; + case AbstractMessage.PUBREC: return "PUBREC"; + case AbstractMessage.PUBREL: return "PUBREL"; + case AbstractMessage.PUBCOMP: return "PUBCOMP"; + case AbstractMessage.SUBSCRIBE: return "SUBSCRIBE"; + case AbstractMessage.SUBACK: return "SUBACK"; + case AbstractMessage.UNSUBSCRIBE: return "UNSUBSCRIBE"; + case AbstractMessage.UNSUBACK: return "UNSUBACK"; + case AbstractMessage.PINGREQ: return "PINGREQ"; + case AbstractMessage.PINGRESP: return "PINGRESP"; + case AbstractMessage.DISCONNECT: return "DISCONNECT"; + default: throw new RuntimeException("Can't decode message type " + type); + } + } +} diff --git a/parser/src/main/resources/log4j.properties b/parser/src/main/resources/log4j.properties index 86ef418d..2bf76ccf 100644 --- a/parser/src/main/resources/log4j.properties +++ b/parser/src/main/resources/log4j.properties @@ -1,17 +1,19 @@ -# Set root logger level to DEBUG and its only appender to A1. -log4j.rootLogger=ERROR, A1 -#log4j.rootLogger=DEBUG, A1 -#log4j.rootLogger=WARN, A1 -#log4j.logger.org.dna.mqtt.moquette.client=WARN -#log4j.logger.org.mina=WARN -#log4j.logger.org.dna.mqtt.moquette.server=DEBUG -#log4j.logger.org.dna.mqtt.moquette.messaging=DEBUG - -log4j.logger.org.dna.mqtt.bechnmark=INFO - -# A1 is set to be a ConsoleAppender. -log4j.appender.A1=org.apache.log4j.ConsoleAppender - -# A1 uses PatternLayout. -log4j.appender.A1.layout=org.apache.log4j.PatternLayout +# Set root logger level to DEBUG and its only appender to A1. +#log4j.rootLogger=ERROR, A1 +log4j.rootLogger=INFO, A1 +#log4j.rootLogger=WARN, A1 +#log4j.logger.org.dna.mqtt.moquette.client=WARN +#log4j.logger.org.mina=WARN +#log4j.logger.org.dna.mqtt.moquette.server=DEBUG +#log4j.logger.org.dna.mqtt.moquette.messaging=DEBUG +#log4j.logger.org.dna.mqtt.moquette.proto.MQTTLoggingFilter=DEBUG +#log4j.logger.org.dna.mqtt.moquette.proto.PublishDecoder=DEBUG + +log4j.logger.org.dna.mqtt.bechnmark=INFO + +# A1 is set to be a ConsoleAppender. +log4j.appender.A1=org.apache.log4j.ConsoleAppender + +# A1 uses PatternLayout. +log4j.appender.A1.layout=org.apache.log4j.PatternLayout log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n \ No newline at end of file diff --git a/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java b/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java index 7fcf0be5..d4e0510f 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java @@ -1,132 +1,180 @@ -package org.dna.mqtt.moquette.proto; - -import org.apache.mina.core.buffer.IoBuffer; -import org.apache.mina.filter.codec.demux.MessageDecoder; -import org.apache.mina.filter.codec.demux.MessageDecoderResult; -import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.PublishMessage; -import static org.junit.Assert.*; -import org.junit.Before; -import org.junit.Test; - -/** - * - * @author andrea - */ -public class PublishDecoderTest { - - IoBuffer m_buff; - PublishDecoder m_msgdec; - MockProtocolDecoderOutput m_mockProtoDecoder; - - @Before - public void setUp() { - m_buff = IoBuffer.allocate(4); - m_msgdec = new PublishDecoder(); - m_mockProtoDecoder = new MockProtocolDecoderOutput(); - } - - @Test - public void testDecodable_OK() { - m_buff.put((byte) (AbstractMessage.PUBLISH << 4)) - .put((byte)0) //0 length - .flip(); - assertEquals(MessageDecoder.OK, m_msgdec.decodable(null, m_buff)); - } - - @Test - public void testHeader() throws Exception { - m_buff = IoBuffer.allocate(14); - initHeader(m_buff); - m_buff.flip(); - - //Exercise - MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); - - assertNotNull(m_mockProtoDecoder.getMessage()); - assertEquals(MessageDecoder.OK, res); - assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); - assertNull(m_mockProtoDecoder.getMessage().getMessageID()); - assertEquals(AbstractMessage.PUBLISH, m_mockProtoDecoder.getMessage().getMessageType()); - } - - @Test - public void testHeaderWithMessageID() throws Exception { - m_buff = IoBuffer.allocate(14).setAutoExpand(true); - int messageID = 123; - initHeaderWithMessageID(m_buff, messageID); - m_buff.flip(); - - //Exercise - MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); - - assertNotNull(m_mockProtoDecoder.getMessage()); - assertEquals(MessageDecoder.OK, res); - assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); - assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID()); - } - - @Test - public void testHeaderWithMessageID_Payload() throws Exception { - m_buff = IoBuffer.allocate(14).setAutoExpand(true); - int messageID = 123; - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; - initHeaderWithMessageID_Payload(m_buff, messageID, payload); - m_buff.flip(); - - //Exercise - MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); - - assertNotNull(m_mockProtoDecoder.getMessage()); - assertEquals(MessageDecoder.OK, res); - assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); - assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID()); - TestUtils.verifyEquals(payload, m_mockProtoDecoder.getMessage().getPayload()); - } - - @Test - public void testBugOnRealCase() throws Exception { - byte[] overallMessage = new byte[] {0x30, 0x17, //fixed header, 25 byte lenght - 0x00, 0x06, 0x2f, 0x74, 0x6f, 0x70, 0x69, 0x63, //[/topic] string 2 len + 6 content - 0x54, 0x65, 0x73, 0x74, 0x20, 0x6d, 0x79, // [Test my payload] encoding - 0x20, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64}; - m_buff = IoBuffer.allocate(overallMessage.length).setAutoExpand(true); - m_buff.put(overallMessage); - m_buff.flip(); - - //Exercise - MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); - - assertNotNull(m_mockProtoDecoder.getMessage()); - assertEquals(MessageDecoder.OK, res); - } - - private void initHeader(IoBuffer buff) throws IllegalAccessException { - IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")).flip(); - buff.clear().put((byte) (AbstractMessage.PUBLISH << 4)).put(Utils.encodeRemainingLength(tmp.remaining())); - //topic name - buff.put(tmp); - } - - private void initHeaderWithMessageID(IoBuffer buff, int messageID) throws IllegalAccessException { - IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")); - Utils.writeWord(tmp, messageID); - tmp.flip(); - buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1 - .put(Utils.encodeRemainingLength(tmp.remaining())); - //topic name - buff.put(tmp); - } - - private void initHeaderWithMessageID_Payload(IoBuffer buff, int messageID, byte[] payload) throws IllegalAccessException { - IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")); - Utils.writeWord(tmp, messageID); - tmp.put(payload); - tmp.flip(); - buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1 - .put(Utils.encodeRemainingLength(tmp.remaining())); - //topic name - buff.put(tmp); - } -} +package org.dna.mqtt.moquette.proto; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.filter.codec.demux.MessageDecoder; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import static org.junit.Assert.*; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +/** + * + * @author andrea + */ +public class PublishDecoderTest { + + IoBuffer m_buff; + PublishDecoder m_msgdec; + MockProtocolDecoderOutput m_mockProtoDecoder; + + @Before + public void setUp() { + m_buff = IoBuffer.allocate(4); + m_msgdec = new PublishDecoder(); + m_mockProtoDecoder = new MockProtocolDecoderOutput(); + } + + @Test + public void testDecodable_OK() { + m_buff.put((byte) (AbstractMessage.PUBLISH << 4)) + .put((byte) 0) //0 length + .flip(); + assertEquals(MessageDecoder.OK, m_msgdec.decodable(null, m_buff)); + } + + @Test + public void testHeader() throws Exception { + m_buff = IoBuffer.allocate(14); + initHeader(m_buff); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); + assertNull(m_mockProtoDecoder.getMessage().getMessageID()); + assertEquals(AbstractMessage.PUBLISH, m_mockProtoDecoder.getMessage().getMessageType()); + } + + @Test + public void testHeaderWithMessageID() throws Exception { + m_buff = IoBuffer.allocate(14).setAutoExpand(true); + int messageID = 123; + initHeaderWithMessageID(m_buff, messageID); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); + assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID()); + } + + @Test + public void testHeaderWithMessageID_Payload() throws Exception { + m_buff = IoBuffer.allocate(14).setAutoExpand(true); + int messageID = 123; + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + initHeaderWithMessageID_Payload(m_buff, messageID, payload); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); + assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID()); + TestUtils.verifyEquals(payload, m_mockProtoDecoder.getMessage().getPayload()); + } + + @Test + public void testBugOnRealCase() throws Exception { + byte[] overallMessage = new byte[]{0x30, 0x17, //fixed header, 25 byte lenght + 0x00, 0x06, 0x2f, 0x74, 0x6f, 0x70, 0x69, 0x63, //[/topic] string 2 len + 6 content + 0x54, 0x65, 0x73, 0x74, 0x20, 0x6d, 0x79, // [Test my payload] encoding + 0x20, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64}; + m_buff = IoBuffer.allocate(overallMessage.length).setAutoExpand(true); + m_buff.put(overallMessage); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + } + + @Test + public void testDecodeBigContent() throws Exception { + int size = 129; + IoBuffer payload = TestUtils.generateRandomPayload(size); + + IoBuffer firstPublish = generatePublishQoS0(payload); + IoBuffer secondPublish = generatePublishQoS0(TestUtils.generateRandomPayload(size)); + + IoBuffer doubleMessageBuf = IoBuffer.allocate(size * 2).setAutoExpand(true); + doubleMessageBuf.put(firstPublish).put(secondPublish).flip(); + + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, doubleMessageBuf, m_mockProtoDecoder); + + assertEquals(MessageDecoder.OK, res); + PublishMessage pubMsg = m_mockProtoDecoder.getMessage(); + assertNotNull(pubMsg); + + res = m_msgdec.decode(null, doubleMessageBuf, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + + m_buff.flip(); + } + + private void initHeader(IoBuffer buff) throws IllegalAccessException { + IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")).flip(); + buff.clear().put((byte) (AbstractMessage.PUBLISH << 4)).put(Utils.encodeRemainingLength(tmp.remaining())); + //topic name + buff.put(tmp); + } + + private void initHeaderWithMessageID(IoBuffer buff, int messageID) throws IllegalAccessException { + IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")); + Utils.writeWord(tmp, messageID); + tmp.flip(); + buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1 + .put(Utils.encodeRemainingLength(tmp.remaining())); + //topic name + buff.put(tmp); + } + + private void initHeaderWithMessageID_Payload(IoBuffer buff, int messageID, byte[] payload) throws IllegalAccessException { + IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")); + Utils.writeWord(tmp, messageID); + tmp.put(payload); + tmp.flip(); + buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1 + .put(Utils.encodeRemainingLength(tmp.remaining())); + //topic name + buff.put(tmp); + } + + private IoBuffer generatePublishQoS0(IoBuffer payload) throws IllegalAccessException { + int size = payload.capacity(); + IoBuffer messageBody = IoBuffer.allocate(size).setAutoExpand(true); + messageBody.put(Utils.encodeString("/topic")); + + //ONLY for QoS > 1 Utils.writeWord(messageBody, messageID); + messageBody.put(payload).flip(); + + IoBuffer completeMsg = IoBuffer.allocate(size).setAutoExpand(true); + completeMsg.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x00)) //set Qos to 0 + .put(Utils.encodeRemainingLength(messageBody.remaining())); + completeMsg.put(messageBody).flip(); + + return completeMsg; + } +} diff --git a/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java b/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java index 6aeac635..c3dd9f15 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java @@ -1,145 +1,167 @@ -package org.dna.mqtt.moquette.proto; - -import org.dna.mqtt.moquette.proto.PublishEncoder; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import org.dna.mqtt.moquette.proto.messages.PublishMessage; -import org.junit.Before; -import org.junit.Test; -import static org.junit.Assert.*; -import static org.dna.mqtt.moquette.proto.TestUtils.*; - -/** - * - * @author andrea - */ -public class PublishEncoderTest { - - PublishEncoder m_encoder; - TestUtils.MockProtocolEncoderOutput m_mockProtoEncoder; - - @Before - public void setUp() { - m_encoder = new PublishEncoder(); - m_mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput(); - } - - @Test - public void testEncodeWithQos_0_noMessageID() throws Exception { - String topic = "/photos"; - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.MOST_ONE); - msg.setTopicName(topic); - - //variable part - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; - msg.setPayload(payload); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - - //Verify - assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte - //(2+7) topic + 3 payload - assertEquals(12, m_mockProtoEncoder.getBuffer().get()); //remaining length - - //Variable part - verifyString(topic, m_mockProtoEncoder.getBuffer()); - verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer()); - } - - @Test - public void testEncodeWithQos_1_MessageID() throws Exception { - String topic = "/photos"; - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.LEAST_ONE); - msg.setMessageID(1); - msg.setTopicName(topic); - - //variable part - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; - msg.setPayload(payload); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - - //Verify - assertEquals(0x32, m_mockProtoEncoder.getBuffer().get()); //1 byte - //(2+7) topic + 2 messageID + 3 payload - assertEquals(14, m_mockProtoEncoder.getBuffer().get()); //remaining length - - //Variable part - verifyString(topic, m_mockProtoEncoder.getBuffer()); - assertEquals(0, m_mockProtoEncoder.getBuffer().get()); //MessageID MSB - assertEquals(1, m_mockProtoEncoder.getBuffer().get()); //MessageID LSB - verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer()); - } - - @Test(expected = IllegalArgumentException.class) - public void testEncodeWithQos_1_noMessageID_fail() throws Exception { - String topic = "pictures/photos"; - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.LEAST_ONE); - msg.setTopicName(topic); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - } - - @Test(expected = IllegalArgumentException.class) - public void testEncode_empty_topic() throws Exception { - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.LEAST_ONE); - msg.setMessageID(1); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - } - - @Test(expected = IllegalArgumentException.class) - public void testEncode_null_topic() throws Exception { - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.LEAST_ONE); - msg.setMessageID(1); - msg.setTopicName(null); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - } - - @Test(expected = IllegalArgumentException.class) - public void testEncode_qos_reserved() throws Exception { - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.RESERVED); - msg.setMessageID(1); - msg.setTopicName(null); - - //variable part - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; - msg.setPayload(payload); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - } - - @Test - public void testBugMissedMessageIDEncoding() throws Exception{ - String topic = "/topic"; - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.MOST_ONE); - msg.setTopicName(topic); - - //variable part - byte[] payload = "Test my payload".getBytes(); - msg.setPayload(payload); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - - //Verify - //2 byte header + (2+6) topic name + [2 message ID if QoQ <> 0]+ payload - int size = m_mockProtoEncoder.getBuffer().remaining(); - assertEquals(10+payload.length, size); - assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte - assertEquals(23, m_mockProtoEncoder.getBuffer().get()); //1 byte the length - } -} +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; +import static org.dna.mqtt.moquette.proto.TestUtils.*; + +/** + * + * @author andrea + */ +public class PublishEncoderTest { + + PublishEncoder m_encoder; + TestUtils.MockProtocolEncoderOutput m_mockProtoEncoder; + + @Before + public void setUp() { + m_encoder = new PublishEncoder(); + m_mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput(); + } + + @Test + public void testEncodeWithQos_0_noMessageID() throws Exception { + String topic = "/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.MOST_ONE); + msg.setTopicName(topic); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte + //(2+7) topic + 3 payload + assertEquals(12, m_mockProtoEncoder.getBuffer().get()); //remaining length + + //Variable part + verifyString(topic, m_mockProtoEncoder.getBuffer()); + verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer()); + } + + @Test + public void testEncodeWithQos_1_MessageID() throws Exception { + String topic = "/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + msg.setTopicName(topic); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + assertEquals(0x32, m_mockProtoEncoder.getBuffer().get()); //1 byte + //(2+7) topic + 2 messageID + 3 payload + assertEquals(14, m_mockProtoEncoder.getBuffer().get()); //remaining length + + //Variable part + verifyString(topic, m_mockProtoEncoder.getBuffer()); + assertEquals(0, m_mockProtoEncoder.getBuffer().get()); //MessageID MSB + assertEquals(1, m_mockProtoEncoder.getBuffer().get()); //MessageID LSB + verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer()); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncodeWithQos_1_noMessageID_fail() throws Exception { + String topic = "pictures/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setTopicName(topic); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_empty_topic() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_null_topic() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + msg.setTopicName(null); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_qos_reserved() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.RESERVED); + msg.setMessageID(1); + msg.setTopicName(null); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test + public void testBugMissedMessageIDEncoding() throws Exception{ + String topic = "/topic"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.MOST_ONE); + msg.setTopicName(topic); + + //variable part + byte[] payload = "Test my payload".getBytes(); + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + //2 byte header + (2+6) topic name + [2 message ID if QoQ <> 0]+ payload + int size = m_mockProtoEncoder.getBuffer().remaining(); + assertEquals(10+payload.length, size); + assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(23, m_mockProtoEncoder.getBuffer().get()); //1 byte the length + } + + @Test + public void testEncodeBigMessage() throws Exception { + byte[] payload = new byte[256]; + + TestUtils.generateRandomPayload(256).get(payload); + + String topic = "/topic"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.MOST_ONE); + msg.setTopicName(topic); + + //variable part + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + int size = m_mockProtoEncoder.getBuffer().remaining(); + assertEquals(11+payload.length, size); + assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte + } +} diff --git a/parser/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java b/parser/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java index 71d0d17e..fb4bb602 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java @@ -1,97 +1,124 @@ -package org.dna.mqtt.moquette.proto; - -import java.io.UnsupportedEncodingException; -import org.apache.mina.core.buffer.IoBuffer; -import org.apache.mina.core.filterchain.IoFilter.NextFilter; -import org.apache.mina.core.future.WriteFuture; -import org.apache.mina.core.session.IoSession; -import org.apache.mina.filter.codec.ProtocolDecoderOutput; -import org.apache.mina.filter.codec.ProtocolEncoderOutput; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import static org.junit.Assert.*; - -/** - * - * @author andrea - */ -public class TestUtils { - - static final class MockProtocolDecoderOutput implements ProtocolDecoderOutput { - - protected T m_connMessage; - - public T getMessage() { - return m_connMessage; - } - - public void write(Object message) { - m_connMessage = (T) message; - } - - public void flush(NextFilter nextFilter, IoSession session) { - throw new UnsupportedOperationException("Not supported yet."); - } - } - - - static final class MockProtocolEncoderOutput implements ProtocolEncoderOutput { - - protected IoBuffer m_buffer; - - public IoBuffer getBuffer() { - return m_buffer; - } - - public void write(Object buffer) { - m_buffer = (IoBuffer) buffer; - } - - public void mergeAll() { - throw new UnsupportedOperationException("Not supported yet."); - } - - public WriteFuture flush() { - throw new UnsupportedOperationException("Not supported yet."); - } - } - - /** - * Verify the presence of the given string starting from the current position - * inside the buffer. - */ - static void verifyString(String str, IoBuffer buff) throws UnsupportedEncodingException { - IoBuffer tmpBuff = IoBuffer.allocate(2).setAutoExpand(true); - byte[] raw = str.getBytes("UTF-8"); - Utils.writeWord(tmpBuff, raw.length); - tmpBuff.put(raw).flip(); - - verifyIoBuffer(tmpBuff, buff); - } - - static void verifyIoBuffer(IoBuffer expected, IoBuffer found) { - while(expected.hasRemaining()) { - assertEquals(expected.get(), found.get()); - } - } - - /** - * Verify that the given bytes buffer og the given numBytes length is present - * in the buff starting from the current position. - */ - static void verifyBuff(int numBytes, byte[] bytes, IoBuffer buff) { - assertTrue(numBytes <= buff.remaining()); - byte[] toCheck = new byte[numBytes]; - buff.get(toCheck); - - for (int i = 0; i < numBytes; i++) { - assertEquals(bytes[i], toCheck[i]); - } - } - - static void verifyEquals(byte[] expected, byte[] found) { - assertEquals(expected.length, found.length); - for (int i = 0; i < expected.length; i++) { - assertEquals(expected[i], found[i]); - } - } -} +package org.dna.mqtt.moquette.proto; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.filterchain.IoFilter.NextFilter; +import org.apache.mina.core.future.WriteFuture; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class TestUtils { + + static final class MockProtocolDecoderOutput implements ProtocolDecoderOutput { + + protected T m_connMessage; + + public T getMessage() { + return m_connMessage; + } + + public void write(Object message) { + m_connMessage = (T) message; + } + + public void flush(NextFilter nextFilter, IoSession session) { + throw new UnsupportedOperationException("Not supported yet."); + } + } + + + static final class MockProtocolEncoderOutput implements ProtocolEncoderOutput { + + protected IoBuffer m_buffer; + + public IoBuffer getBuffer() { + return m_buffer; + } + + public void write(Object buffer) { + m_buffer = (IoBuffer) buffer; + } + + public void mergeAll() { + throw new UnsupportedOperationException("Not supported yet."); + } + + public WriteFuture flush() { + throw new UnsupportedOperationException("Not supported yet."); + } + } + + /** + * Verify the presence of the given string starting from the current position + * inside the buffer. + */ + static void verifyString(String str, IoBuffer buff) throws UnsupportedEncodingException { + IoBuffer tmpBuff = IoBuffer.allocate(2).setAutoExpand(true); + byte[] raw = str.getBytes("UTF-8"); + Utils.writeWord(tmpBuff, raw.length); + tmpBuff.put(raw).flip(); + + verifyIoBuffer(tmpBuff, buff); + } + + static void verifyIoBuffer(IoBuffer expected, IoBuffer found) { + while(expected.hasRemaining()) { + assertEquals(expected.get(), found.get()); + } + } + + /** + * Verify that the given bytes buffer og the given numBytes length is present + * in the buff starting from the current position. + */ + static void verifyBuff(int numBytes, byte[] bytes, IoBuffer buff) { + assertTrue(numBytes <= buff.remaining()); + byte[] toCheck = new byte[numBytes]; + buff.get(toCheck); + + for (int i = 0; i < numBytes; i++) { + assertEquals(bytes[i], toCheck[i]); + } + } + + static void verifyEquals(byte[] expected, byte[] found) { + assertEquals(expected.length, found.length); + for (int i = 0; i < expected.length; i++) { + assertEquals(expected[i], found[i]); + } + } + + + public static IoBuffer generateRandomPayload(int size) { + IoBuffer payloadBuffer = IoBuffer.allocate(size); + for (int i = 0; i < size; i++) { + payloadBuffer.put((byte) (Math.random() * 255)); + } + payloadBuffer.flip(); + return payloadBuffer; + } + + public static IoBuffer loadFile(String path) throws FileNotFoundException, IOException { + File f = new File(path); + long size = f.length(); + byte[] payload = new byte[(int) size]; + + FileInputStream fr = new FileInputStream(f); + fr.read(payload); + + IoBuffer payloadBuffer = IoBuffer.allocate((int) size); + payloadBuffer.put(payload).flip(); + return payloadBuffer; + } +} diff --git a/parser/src/test/java/org/dna/mqtt/moquette/proto/UtilsTests.java b/parser/src/test/java/org/dna/mqtt/moquette/proto/UtilsTests.java index e32df123..b9b77850 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/proto/UtilsTests.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/proto/UtilsTests.java @@ -1,26 +1,25 @@ -package org.dna.mqtt.moquette.proto; - -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; -import org.junit.Before; -import org.junit.Test; -import static org.junit.Assert.*; - -/** - * - * @author andrea - */ -public class UtilsTests { - - @Test - public void testEncodeFlags() { - UnsubscribeMessage msg = new UnsubscribeMessage(); - msg.setRetainFlag(true); - msg.setQos(AbstractMessage.QOSType.MOST_ONE); - - //Exercise - assertEquals(1, Utils.encodeFlags(msg)); - } - -} - +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; +import static org.junit.Assert.*; +import org.junit.Test; + +/** + * + * @author andrea + */ +public class UtilsTests { + + @Test + public void testEncodeFlags() { + UnsubscribeMessage msg = new UnsubscribeMessage(); + msg.setRetainFlag(true); + msg.setQos(AbstractMessage.QOSType.MOST_ONE); + + //Exercise + assertEquals(1, Utils.encodeFlags(msg)); + } + +} +