diff --git a/broker/pom.xml b/broker/pom.xml index 0bb73b19..1b0ee69d 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -41,7 +41,7 @@ org.slf4j slf4j-log4j12 - 1.6.4 + 1.7.5 diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java index 16b346af..500a8a88 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi; +import java.nio.ByteBuffer; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; @@ -18,7 +19,7 @@ public interface IStorageService { * */ void initStore(); - void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos); + void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos); Collection searchMatching(IMatchingCondition condition); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index 86a5f8f3..f0349be8 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -1,9 +1,15 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +import java.io.File; +import java.io.IOException; +import java.io.Serializable; +import java.nio.ByteBuffer; +import java.util.*; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.storage.StoredPublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.server.Server; @@ -12,11 +18,6 @@ import org.fusesource.hawtdb.api.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.File; -import java.io.IOException; -import java.io.Serializable; -import java.util.*; - /** * Implementation of IStorageService backed by HawtDB @@ -38,8 +39,8 @@ public class HawtDBStorageService implements IStorageService { return m_qos; } - byte[] getPayload() { - return m_payload; + ByteBuffer getPayload() { + return (ByteBuffer) ByteBuffer.allocate(m_payload.length).put(m_payload).flip(); } String getTopic() { @@ -54,12 +55,12 @@ public class HawtDBStorageService implements IStorageService { private PageFileFactory pageFactory; //maps clientID to the list of pending messages stored - private SortedIndex> m_persistentMessageStore; + private SortedIndex> m_persistentMessageStore; private SortedIndex m_retainedStore; //bind clientID+MsgID -> evt message published - private SortedIndex m_inflightStore; + private SortedIndex m_inflightStore; //bind clientID+MsgID -> evt message published - private SortedIndex m_qos2Store; + private SortedIndex m_qos2Store; //persistent Map of clientID, set of Subscriptions private SortedIndex> m_persistentSubscriptions; @@ -101,10 +102,10 @@ public class HawtDBStorageService implements IStorageService { private void initPersistentMessageStore() { - BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); + BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); indexFactory.setKeyCodec(StringCodec.INSTANCE); - m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory); + m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory); } private void initPersistentSubscriptions() { @@ -119,31 +120,33 @@ public class HawtDBStorageService implements IStorageService { * messages in flight. */ private void initInflightMessageStore() { - BTreeIndexFactory indexFactory = new BTreeIndexFactory(); + BTreeIndexFactory indexFactory = new BTreeIndexFactory(); indexFactory.setKeyCodec(StringCodec.INSTANCE); - m_inflightStore = (SortedIndex) m_multiIndexFactory.openOrCreate("inflight", indexFactory); + m_inflightStore = (SortedIndex) m_multiIndexFactory.openOrCreate("inflight", indexFactory); } private void initPersistentQoS2MessageStore() { - BTreeIndexFactory indexFactory = new BTreeIndexFactory(); + BTreeIndexFactory indexFactory = new BTreeIndexFactory(); indexFactory.setKeyCodec(StringCodec.INSTANCE); - m_qos2Store = (SortedIndex) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory); + m_qos2Store = (SortedIndex) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory); } - public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) { - if (message.length == 0) { + public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) { + if (!message.hasRemaining()) { //clean the message from topic m_retainedStore.remove(topic); } else { //store the message to the topic - m_retainedStore.put(topic, new StoredMessage(message, qos, topic)); + byte[] raw = new byte[message.remaining()]; + message.get(raw); + m_retainedStore.put(topic, new StoredMessage(raw, qos, topic)); } } public Collection searchMatching(IMatchingCondition condition) { - LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size()); + LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); List results = new ArrayList(); @@ -158,21 +161,29 @@ public class HawtDBStorageService implements IStorageService { } public void storePublishForFuture(PublishEvent evt) { - List storedEvents; + List storedEvents; String clientID = evt.getClientID(); if (!m_persistentMessageStore.containsKey(clientID)) { - storedEvents = new ArrayList(); + storedEvents = new ArrayList(); } else { storedEvents = m_persistentMessageStore.get(clientID); } - storedEvents.add(evt); + storedEvents.add(convertToStored(evt)); m_persistentMessageStore.put(clientID, storedEvents); - LOG.debug(String.format("Stored published message for client <%s> on topic <%s>, content %s", - clientID, evt.getTopic(), new String(evt.getMessage()))); + //NB rewind the evt message content + LOG.debug("Stored published message for client <{}> on topic <{}>", clientID, evt.getTopic()); } public List retrivePersistedPublishes(String clientID) { - return m_persistentMessageStore.get(clientID); + List storedEvts = m_persistentMessageStore.get(clientID); + if (storedEvts == null) { + return null; + } + List liveEvts = new ArrayList(); + for (StoredPublishEvent storedEvt : storedEvts) { + liveEvts.add(convertFromStored(storedEvt)); + } + return liveEvts; } public void cleanPersistedPublishes(String clientID) { @@ -184,7 +195,8 @@ public class HawtDBStorageService implements IStorageService { } public void addInFlight(PublishEvent evt, String publishKey) { - m_inflightStore.put(publishKey, evt); + StoredPublishEvent storedEvt = convertToStored(evt); + m_inflightStore.put(publishKey, storedEvt); } public void addNewSubscription(Subscription newSubscription, String clientID) { @@ -221,8 +233,8 @@ public class HawtDBStorageService implements IStorageService { /*-------- QoS 2 storage management --------------*/ public void persistQoS2Message(String publishKey, PublishEvent evt) { - LOG.debug(String.format("persistQoS2Message store pubKey %s, evt %s", publishKey, evt)); - m_qos2Store.put(publishKey, evt); + LOG.debug("persistQoS2Message store pubKey {}, evt {}", publishKey, evt); + m_qos2Store.put(publishKey, convertToStored(evt)); } public void removeQoS2Message(String publishKey) { @@ -230,6 +242,21 @@ public class HawtDBStorageService implements IStorageService { } public PublishEvent retrieveQoS2Message(String publishKey) { - return m_qos2Store.get(publishKey); + StoredPublishEvent storedEvt = m_qos2Store.get(publishKey); + return convertFromStored(storedEvt); + } + + private StoredPublishEvent convertToStored(PublishEvent evt) { + StoredPublishEvent storedEvt = new StoredPublishEvent(evt); + return storedEvt; + } + + private PublishEvent convertFromStored(StoredPublishEvent evt) { + byte[] message = evt.getMessage(); + ByteBuffer bbmessage = ByteBuffer.wrap(message); + //bbmessage.flip(); + PublishEvent liveEvt = new PublishEvent(evt.getTopic(), evt.getQos(), + bbmessage, evt.isRetain(), evt.getClientID(), evt.getMessageID(), null); + return liveEvt; } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java index e29157a1..7570832e 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +import java.nio.ByteBuffer; import java.util.ArrayList; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IStorageService; @@ -32,18 +33,20 @@ public class MemoryStorageService implements IStorageService { //To change body of implemented methods use File | Settings | File Templates. } - public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) { - if (message.length == 0) { + public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) { + if (!message.hasRemaining()) { //clean the message from topic m_retainedStore.remove(topic); } else { //store the message to the topic - m_retainedStore.put(topic, new HawtDBStorageService.StoredMessage(message, qos, topic)); + byte[] raw = new byte[message.remaining()]; + message.get(raw); + m_retainedStore.put(topic, new HawtDBStorageService.StoredMessage(raw, qos, topic)); } } public Collection searchMatching(IMatchingCondition condition) { - LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size()); + LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); List results = new ArrayList(); @@ -58,7 +61,7 @@ public class MemoryStorageService implements IStorageService { } public void storePublishForFuture(PublishEvent evt) { - LOG.debug("storePublishForFuture store evt " + evt); + LOG.debug("storePublishForFuture store evt {}", evt); List storedEvents; String clientID = evt.getClientID(); if (!m_persistentMessageStore.containsKey(clientID)) { @@ -115,7 +118,7 @@ public class MemoryStorageService implements IStorageService { } public void persistQoS2Message(String publishKey, PublishEvent evt) { - LOG.debug(String.format("persistQoS2Message store pubKey %s, evt %s", publishKey, evt)); + LOG.debug("persistQoS2Message store pubKey {}, evt {}", publishKey, evt); m_qos2Store.put(publishKey, evt); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 524f5573..ae1eb3ad 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -4,6 +4,7 @@ import com.lmax.disruptor.BatchEventProcessor; import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.SequenceBarrier; +import java.nio.ByteBuffer; import java.util.Collection; import java.util.HashMap; import java.util.List; @@ -78,7 +79,7 @@ class ProtocolProcessor implements EventHandler { } void processConnect(ServerChannel session, ConnectMessage msg) { - LOG.debug("processConnect for client " + msg.getClientID()); + LOG.debug("processConnect for client {}", msg.getClientID()); if (msg.getProcotolVersion() != 0x03) { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); @@ -112,7 +113,7 @@ class ProtocolProcessor implements EventHandler { m_clientIDs.put(msg.getClientID(), connDescr); int keepAlive = msg.getKeepAlive(); - LOG.debug(String.format("Connect with keepAlive %d s", keepAlive)); + LOG.debug("Connect with keepAlive {} s", keepAlive); session.setAttribute(Constants.KEEP_ALIVE, keepAlive); session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession()); //used to track the client in the subscription and publishing phases. @@ -123,8 +124,10 @@ class ProtocolProcessor implements EventHandler { //Handle will flag if (msg.isWillFlag()) { AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()]; - PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, msg.getWillMessage().getBytes(), - msg.isWillRetain(), msg.getClientID(), session); + byte[] willPayload = msg.getWillMessage().getBytes(); + ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); + PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, + bb, msg.isWillRetain(), msg.getClientID(), session); processPublish(pubEvt); } @@ -155,7 +158,7 @@ class ProtocolProcessor implements EventHandler { okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); LOG.debug("processConnect sent OK ConnAck"); session.write(okResp); - LOG.info(String.format("Connected client ID <%s> with clean session %s", msg.getClientID(), msg.isCleanSession())); + LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession()); if (!msg.isCleanSession()) { //force the republish of stored QoS1 and QoS2 @@ -167,11 +170,11 @@ class ProtocolProcessor implements EventHandler { LOG.debug("republishStored invoked"); List publishedEvents = m_storageService.retrivePersistedPublishes(clientID); if (publishedEvents == null) { - LOG.info(String.format("No stored messages for client <%s>", clientID)); + LOG.info("No stored messages for client <{}>", clientID); return; } - LOG.info(String.format("republishing stored messages to client <%s>", clientID)); + LOG.info("republishing stored messages to client <{}>", clientID); for (PublishEvent pubEvt : publishedEvents) { sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage(), false, pubEvt.getMessageID()); @@ -179,7 +182,7 @@ class ProtocolProcessor implements EventHandler { } private void processRemoveAllSubscriptions(String clientID) { - LOG.info(String.format("cleaning old saved subscriptions for client <%s>", clientID)); + LOG.info("cleaning old saved subscriptions for client <{}>", clientID); subscriptions.removeForClient(clientID); //remove also the messages stored of type QoS1/2 @@ -190,11 +193,11 @@ class ProtocolProcessor implements EventHandler { LOG.debug("processPublish invoked with " + evt); final String topic = evt.getTopic(); final AbstractMessage.QOSType qos = evt.getQos(); - final byte[] message = evt.getMessage(); + final ByteBuffer message = evt.getMessage(); boolean retain = evt.isRetain(); - LOG.info(String.format("Publish recieved from clientID <%s> on topic <%s> with QoS %s", - evt.getClientID(), evt.getTopic(), evt.getQos())); + LOG.info("Publish recieved from clientID <{}> on topic <{}> with QoS {}", + evt.getClientID(), evt.getTopic(), evt.getQos()); String publishKey = null; if (qos == AbstractMessage.QOSType.LEAST_ONE) { @@ -216,7 +219,7 @@ class ProtocolProcessor implements EventHandler { } m_storageService.cleanInFlight(publishKey); sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID())); - LOG.debug("replying with PubAck to MSG ID " + evt.getMessageID()); + LOG.debug("replying with PubAck to MSG ID {}", evt.getMessageID()); } if (retain) { @@ -227,11 +230,11 @@ class ProtocolProcessor implements EventHandler { /** * 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, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) { - LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic " + topic); + private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retain, Integer messageID) { + LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic); for (final Subscription sub : subscriptions.matches(topic)) { - LOG.debug(String.format("Broker publishing to client <%s> topic <%s> qos <%s>", - sub.getClientId(), sub.getTopic(), qos)); + LOG.debug("Broker publishing to client <{}> topic <{}> qos <{}>", + sub.getClientId(), sub.getTopic(), qos); if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS 0 sendPublish(sub.getClientId(), topic, qos, message, false); @@ -256,19 +259,19 @@ class ProtocolProcessor implements EventHandler { } } - private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, byte[] message, boolean retained) { + private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) { sendPublish(clientId, topic, qos, message, retained, 0); } - private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, byte[] message, boolean retained, int messageID) { + private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, int messageID) { LOG.debug("notify invoked with event "); PublishMessage pubMessage = new PublishMessage(); pubMessage.setRetainFlag(retained); pubMessage.setTopicName(topic); pubMessage.setQos(qos); pubMessage.setPayload(message); - LOG.info(String.format("send publish message to <%s> on topic <%s>, message is [%s]", - clientId, topic, new String(message))); + + LOG.info("send publish message to <{}> on topic <{}>", clientId, topic); if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) { pubMessage.setMessageID(messageID); } @@ -277,11 +280,11 @@ class ProtocolProcessor implements EventHandler { if (m_clientIDs == null) { throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!"); } - LOG.debug("clientIDs are " + m_clientIDs); + LOG.debug("clientIDs are {}", m_clientIDs); if (m_clientIDs.get(clientId) == null) { throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs)); } - LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); + LOG.debug("Session for clientId {} is {}", clientId, m_clientIDs.get(clientId).getSession()); // m_clientIDs.get(clientId).getSession().write(pubMessage); disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubMessage)); }catch(Throwable t) { @@ -290,7 +293,7 @@ class ProtocolProcessor implements EventHandler { } private void sendPubRec(String clientID, int messageID) { - LOG.debug(String.format("sendPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); + LOG.debug("sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); PubRecMessage pubRecMessage = new PubRecMessage(); pubRecMessage.setMessageID(messageID); @@ -310,11 +313,11 @@ class ProtocolProcessor implements EventHandler { if (m_clientIDs == null) { throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!"); } - LOG.debug("clientIDs are " + m_clientIDs); - if (m_clientIDs.get(clientId) == null) { + LOG.debug("clientIDs are {}", m_clientIDs); + if (m_clientIDs.get(clientId) == null) { throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs)); } - LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); +// LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); // m_clientIDs.get(clientId).getSession().write(pubAckMessage); disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubAckMessage)); }catch(Throwable t) { @@ -332,22 +335,20 @@ class ProtocolProcessor implements EventHandler { final String topic = evt.getTopic(); final AbstractMessage.QOSType qos = evt.getQos(); - final byte[] message = evt.getMessage(); - boolean retain = evt.isRetain(); - publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); + publish2Subscribers(topic, qos, evt.getMessage(), evt.isRetain(), evt.getMessageID()); m_storageService.removeQoS2Message(publishKey); - if (retain) { - m_storageService.storeRetained(topic, message, qos); + if (evt.isRetain()) { + m_storageService.storeRetained(topic, evt.getMessage(), qos); } sendPubComp(clientID, messageID); } private void sendPubComp(String clientID, int messageID) { - LOG.debug(String.format("sendPubComp invoked for clientID %s ad messageID %d", clientID, messageID)); + LOG.debug("sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID); PubCompMessage pubCompMessage = new PubCompMessage(); pubCompMessage.setMessageID(messageID); @@ -357,7 +358,7 @@ class ProtocolProcessor implements EventHandler { 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)); + LOG.debug("processPubRec invoked for clientID {} ad messageID {}", clientID, messageID); PubRelMessage pubRelMessage = new PubRelMessage(); pubRelMessage.setMessageID(messageID); @@ -383,7 +384,7 @@ class ProtocolProcessor implements EventHandler { //de-activate the subscriptions for this ClientID // String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); subscriptions.deactivate(clientID); - LOG.info(String.format("Disconnected client <%s> with clean session %s", clientID, cleanSession)); + LOG.info("Disconnected client <{}> with clean session {}", clientID, cleanSession); } void proccessConnectionLost(String clientID) { @@ -392,7 +393,7 @@ class ProtocolProcessor implements EventHandler { //de-activate the subscriptions for this ClientID subscriptions.deactivate(clientID); - LOG.info(String.format("Lost connection with client <%s>", clientID)); + LOG.info("Lost connection with client <{}>", clientID); } } @@ -410,13 +411,13 @@ class ProtocolProcessor implements EventHandler { UnsubAckMessage ackMessage = new UnsubAckMessage(); ackMessage.setMessageID(messageID); - LOG.info(String.format("replying with UnsubAck to MSG ID %s", messageID)); + LOG.info("replying with UnsubAck to MSG ID {}", messageID); session.write(ackMessage); } void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) { - LOG.debug(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID())); + LOG.debug("processSubscribe invoked from client {} with msgID {}", clientID, msg.getMessageID()); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; @@ -432,14 +433,14 @@ class ProtocolProcessor implements EventHandler { for (int i = 0; i < msg.subscriptions().size(); i++) { ackMessage.addType(AbstractMessage.QOSType.MOST_ONE); } - LOG.debug("replying with SubAck to MSG ID " + msg.getMessageID()); + LOG.debug("replying with SubAck to MSG ID {}", msg.getMessageID()); session.write(ackMessage); } private void subscribeSingleTopic(Subscription newSubscription, final String topic) { - LOG.info(String.format("<%s> subscribed to topic <%s> with QoS %s", + LOG.info("<{}> subscribed to topic <{}> with QoS {}", newSubscription.getClientId(), topic, - AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()))); + AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos())); subscriptions.add(newSubscription); //scans retained messages to be published to the new subscription @@ -451,13 +452,13 @@ class ProtocolProcessor implements EventHandler { for (HawtDBStorageService.StoredMessage storedMsg : messages) { //fire the as retained the message - LOG.debug("send publish message for topic " + topic); + LOG.debug("send publish message for topic {}", topic); sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true); } } private void disruptorPublish(OutputMessagingEvent msgEvent) { - LOG.debug("disruptorPublish publishing event on output " + msgEvent); + LOG.debug("disruptorPublish publishing event on output {}", msgEvent); long sequence = m_ringBuffer.next(); ValueEvent event = m_ringBuffer.get(sequence); 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 c8d6b608..681c7044 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 @@ -69,7 +69,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { private void disruptorPublish(MessagingEvent msgEvent) { - LOG.debug("disruptorPublish publishing event " + msgEvent); + LOG.debug("disruptorPublish publishing event {}", msgEvent); long sequence = m_ringBuffer.next(); ValueEvent event = m_ringBuffer.get(sequence); @@ -97,7 +97,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { public void onEvent(ValueEvent t, long l, boolean bln) throws Exception { MessagingEvent evt = t.getEvent(); - LOG.info("onEvent processing messaging event from input ringbuffer" + evt); + LOG.info("onEvent processing messaging event from input ringbuffer {}", evt); if (evt instanceof PublishEvent) { m_processor.processPublish((PublishEvent) evt); } else if (evt instanceof StopEvent) { @@ -112,17 +112,15 @@ public class SimpleMessaging implements IMessaging, EventHandler { if (message instanceof ConnectMessage) { m_processor.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); - } + pubEvt = new PublishEvent((PublishMessage) message, clientID, session); +// 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); +// } m_processor.processPublish(pubEvt); } else if (message instanceof DisconnectMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java index 7e2c2b5b..0666ca3b 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java @@ -2,25 +2,39 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import java.io.Serializable; +import java.nio.ByteBuffer; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.server.ServerChannel; /** * * @author andrea */ -public class PublishEvent extends MessagingEvent implements Serializable { +public class PublishEvent extends MessagingEvent { String m_topic; QOSType m_qos; - byte[] m_message; + //byte[] m_message; + ByteBuffer m_message; boolean m_retain; String m_clientID; //Optional attribute, available only fo QoS 1 and 2 int m_msgID; transient ServerChannel m_session; + + public PublishEvent(PublishMessage pubMsg, String clientID, ServerChannel session) { + m_topic = pubMsg.getTopicName(); + m_qos = pubMsg.getQos(); + m_message = pubMsg.getPayload(); + m_retain = pubMsg.isRetainFlag(); + m_clientID = clientID; + m_session = session; + if (pubMsg.getQos() != QOSType.MOST_ONE) { + m_msgID = pubMsg.getMessageID(); + } + } - public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, + public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain, String clientID, ServerChannel session) { m_topic = topic; m_qos = qos; @@ -30,7 +44,7 @@ public class PublishEvent extends MessagingEvent implements Serializable { m_session = session; } - public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, + public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain, String clientID, int msgID, ServerChannel session) { this(topic, qos, message, retain, clientID, session); m_msgID = msgID; @@ -44,7 +58,7 @@ public class PublishEvent extends MessagingEvent implements Serializable { return m_qos; } - public byte[] getMessage() { + public ByteBuffer getMessage() { return m_message; } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/storage/StoredPublishEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/storage/StoredPublishEvent.java new file mode 100644 index 00000000..70de8856 --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/storage/StoredPublishEvent.java @@ -0,0 +1,58 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.storage; + +import java.io.Serializable; +import java.nio.ByteBuffer; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; + +/** + * Publish event serialized to the DB. + * + * @author andrea + */ +public class StoredPublishEvent implements Serializable { + String m_topic; + QOSType m_qos; + byte[] m_message; + boolean m_retain; + String m_clientID; + //Optional attribute, available only fo QoS 1 and 2 + int m_msgID; + + public StoredPublishEvent(PublishEvent wrapped) { + m_topic = wrapped.getTopic(); + m_qos = wrapped.getQos(); + m_retain = wrapped.isRetain(); + m_clientID = wrapped.getClientID(); + m_msgID = wrapped.getMessageID(); + + ByteBuffer buffer = wrapped.getMessage(); + m_message = new byte[buffer.remaining()]; + buffer.get(m_message); + buffer.rewind(); + } + + public String getTopic() { + return m_topic; + } + + public QOSType getQos() { + return m_qos; + } + + public byte[] getMessage() { + return m_message; + } + + public boolean isRetain() { + return m_retain; + } + + public String getClientID() { + return m_clientID; + } + + public int getMessageID() { + return m_msgID; + } +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java index 3871d843..9708460e 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java @@ -72,7 +72,7 @@ public class SubscriptionsStore { //reload any subscriptions persisted LOG.debug("Reloading all stored subscriptions..."); for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) { - LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic()); + LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopic()); addDirect(subscription); } LOG.debug("Finished loading"); @@ -166,7 +166,7 @@ public class SubscriptionsStore { } public void activate(String clientID) { - LOG.debug(String.format("Activating subscriptions for clientID <%s>", clientID)); + LOG.debug("Activating subscriptions for clientID <{}>", clientID); subscriptions.activate(clientID); } @@ -176,7 +176,7 @@ public class SubscriptionsStore { * listeners subscriptions, and not topic publishing. */ public List matches(String topic) { - List tokens = new ArrayList(); + List tokens; try { tokens = splitTopic(topic); } catch (ParseException ex) { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java index 22c65a2c..da0cf1cd 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java @@ -85,7 +85,7 @@ public class NettyAcceptor implements ServerAcceptor { MessageMetrics metrics = m_metricsCollector.computeMetrics(); //LOG.info(String.format("Bytes read: %d, bytes wrote: %d", metrics.readBytes(), metrics.wroteBytes())); - LOG.info(String.format("Msg read: %d, msg wrote: %d", metrics.messagesRead(), metrics.messagesWrote())); + LOG.info("Msg read: {}, msg wrote: {}", metrics.messagesRead(), metrics.messagesWrote()); } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java index 88b8dff5..4e6d1aa0 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java @@ -28,7 +28,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object message) { AbstractMessage msg = (AbstractMessage) message; - LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType()))); + LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType())); try { switch (msg.getMessageType()) { case CONNECT: diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index f13b8af1..16968927 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +import java.nio.ByteBuffer; import java.util.Arrays; import java.util.HashMap; import java.util.List; @@ -168,7 +169,8 @@ public class ProtocolProcessorTest { //Exercise - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null); + ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, false, "FakeCLI", null); m_processor.processPublish(pubEvt); //Verify @@ -254,7 +256,8 @@ public class ProtocolProcessorTest { connectMessage.setProcotolVersion((byte)3); connectMessage.setCleanSession(subscription.isCleanSession()); m_processor.processConnect(m_session, connectMessage); - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); + ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, true, "FakeCLI", null); m_processor.processPublish(pubEvt); //Exercise diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java index 151fb330..6b21a34c 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java @@ -306,6 +306,7 @@ public class ServerIntegrationPahoTest { anotherClient.disconnect(); } + //BUG @Test public void testPublishWithQoS2() throws Exception { LOG.info("*** testPublishWithQoS2 ***"); @@ -324,6 +325,7 @@ public class ServerIntegrationPahoTest { assertEquals(2, m_callback.getMessage().getQos()); } + //BUG @Test public void testPublishReceiveWithQoS2() throws Exception { LOG.info("*** testPublishReceiveWithQoS2 ***"); diff --git a/netty_parser/pom.xml b/netty_parser/pom.xml index 3e1bee64..27276ef7 100644 --- a/netty_parser/pom.xml +++ b/netty_parser/pom.xml @@ -23,7 +23,7 @@ org.dna.mqtt moquette-parser-commons - 0.2 + ${project.version} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java index 00e869f8..bdace02b 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java @@ -1,6 +1,7 @@ package org.dna.mqtt.moquette.parser.netty; import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import java.util.List; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; @@ -18,14 +19,14 @@ class PublishDecoder extends DemuxDecoder { @Override void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { - LOG.info("decode invoked with buffer " + in); + LOG.info("decode invoked with buffer {}", in); in.resetReaderIndex(); int startPos = in.readerIndex(); //Common decoding part PublishMessage message = new PublishMessage(); if (!decodeCommonHeader(message, in)) { - LOG.info("decode ask for more data after " + in); + LOG.info("decode ask for more data after {}", in); in.resetReaderIndex(); return; } @@ -51,9 +52,10 @@ class PublishDecoder extends DemuxDecoder { in.resetReaderIndex(); return; } - byte[] b = new byte[payloadSize]; - in.readBytes(b); - message.setPayload(b); +// byte[] b = new byte[payloadSize]; + ByteBuf bb = Unpooled.buffer(payloadSize); + in.readBytes(bb); + message.setPayload(bb.nioBuffer()); out.add(message); } diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java index 4cebae39..b84df9f7 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java @@ -2,6 +2,8 @@ package org.dna.mqtt.moquette.parser.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; +import java.nio.Buffer; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; @@ -65,7 +67,8 @@ public class PublishDecoderTest { public void testHeaderWithMessageID_Payload() throws Exception { m_buff = Unpooled.buffer(14); int messageID = 123; - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; +// byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + ByteBuffer payload = ByteBuffer.allocate(3).put(new byte[]{0x0A, 0x0B, 0x0C}); initHeaderWithMessageID_Payload(m_buff, messageID, payload); //Exercise @@ -76,7 +79,8 @@ public class PublishDecoderTest { assertNotNull(message); assertEquals("Fake Topic", message.getTopicName()); assertEquals(messageID, (int) message.getMessageID()); - TestUtils.verifyEquals(payload, message.getPayload()); +// TestUtils.verifyEquals(payload, message.getPayload()); + assertEquals(payload, message.getPayload()); } @Test @@ -141,7 +145,9 @@ public class PublishDecoderTest { PublishMessage message = (PublishMessage)m_results.get(0); assertNotNull(message); assertEquals("/topic", message.getTopicName()); - assertEquals("Test my payload", new String(message.getPayload())); +// assertEquals("Test my payload", new String(message.getPayload())); + Buffer expectedPayload = ByteBuffer.allocate(15).put("Test my payload".getBytes()).flip(); + assertEquals(expectedPayload, message.getPayload()); } private void initHeader(ByteBuf buff) throws IllegalAccessException { @@ -170,6 +176,16 @@ public class PublishDecoderTest { buff.writeBytes(tmp); } + private void initHeaderWithMessageID_Payload(ByteBuf buff, int messageID, ByteBuffer payload) throws IllegalAccessException { + ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic")); + tmp.writeShort(messageID); + tmp.writeBytes(payload); + buff.clear().writeByte(AbstractMessage.PUBLISH << 4 | 0x02) //set Qos to 1 + .writeBytes(Utils.encodeRemainingLength(tmp.readableBytes())); + //topic name + buff.writeBytes(tmp); + } + private ByteBuf generatePublishQoS0(ByteBuf payload) throws IllegalAccessException { int size = payload.capacity(); ByteBuf messageBody = Unpooled.buffer(size); diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java index a7216af1..ec0e9261 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java @@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.parser.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; +import java.nio.ByteBuffer; import org.junit.Before; import org.junit.Test; import static org.junit.Assert.*; @@ -35,7 +36,8 @@ public class PublishEncoderTest { msg.setTopicName(topic); //variable part - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + byte[] bpayload = new byte[]{0x0A, 0x0B, 0x0C}; + ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(3).put(bpayload).flip(); msg.setPayload(payload); //Exercise @@ -48,7 +50,8 @@ public class PublishEncoderTest { //Variable part verifyString(topic, m_out); - verifyBuff(payload.length, payload, m_out); + payload.rewind(); + verifyBuff(bpayload.length, payload, m_out); } @@ -61,7 +64,8 @@ public class PublishEncoderTest { msg.setTopicName(topic); //variable part - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + byte[] bpayload = new byte[]{0x0A, 0x0B, 0x0C}; + ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(3).put(bpayload).flip(); msg.setPayload(payload); //Exercise @@ -76,7 +80,8 @@ public class PublishEncoderTest { verifyString(topic, m_out); assertEquals(0, m_out.readByte()); //MessageID MSB assertEquals(1, m_out.readByte()); //MessageID LSB - verifyBuff(payload.length, payload, m_out); + payload.rewind(); + verifyBuff(bpayload.length, payload, m_out); } @Test(expected = IllegalArgumentException.class) @@ -120,7 +125,8 @@ public class PublishEncoderTest { msg.setTopicName(null); //variable part - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + byte[] bpayload = new byte[]{0x0A, 0x0B, 0x0C}; + ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(3).put(bpayload).flip(); msg.setPayload(payload); ByteBuf out = Unpooled.buffer(); @@ -136,7 +142,8 @@ public class PublishEncoderTest { msg.setTopicName(topic); //variable part - byte[] payload = "Test my payload".getBytes(); + byte[] bpayload = "Test my payload".getBytes(); + ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(bpayload.length).put(bpayload).flip(); msg.setPayload(payload); //Exercise @@ -145,16 +152,17 @@ public class PublishEncoderTest { //Verify //2 byte header + (2+6) topic name + [2 message ID if QoQ <> 0]+ payload int size = m_out.readableBytes(); - assertEquals(10+payload.length, size); + assertEquals(10 + bpayload.length, size); assertEquals(0x30, m_out.readByte()); //1 byte assertEquals(23, m_out.readByte()); //1 byte the length } @Test public void testEncodeBigMessage() throws Exception { - byte[] payload = new byte[256]; + byte[] bpayload = new byte[256]; - TestUtils.generateRandomPayload(256).readBytes(payload); + TestUtils.generateRandomPayload(256).readBytes(bpayload); + ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(bpayload.length).put(bpayload).flip(); String topic = "/topic"; PublishMessage msg = new PublishMessage(); @@ -168,7 +176,7 @@ public class PublishEncoderTest { m_encoder.encode(m_mockedContext, msg, m_out); int size = m_out.readableBytes(); - assertEquals(11+payload.length, size); + assertEquals(11 + bpayload.length, size); assertEquals(0x30, m_out.readByte()); //1 byte } diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java index 29a93411..250dc109 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java @@ -6,6 +6,7 @@ import io.netty.buffer.Unpooled; import io.netty.buffer.UnpooledByteBufAllocator; import io.netty.channel.ChannelHandlerContext; import java.io.UnsupportedEncodingException; +import java.nio.ByteBuffer; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.*; @@ -31,6 +32,11 @@ public class TestUtils { } } + static void verifyBuff(int numBytes, ByteBuffer bytes, ByteBuf buff) { + assertTrue(numBytes <= buff.readableBytes()); + assertEquals(bytes, buff.nioBuffer()); + } + /** * Verify the presence of the given string starting from the current position * inside the buffer. diff --git a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java index 67a5da3f..24818db9 100644 --- a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java +++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java @@ -1,5 +1,7 @@ package org.dna.mqtt.moquette.proto.messages; +import java.nio.ByteBuffer; + /** * * @author andrea @@ -8,7 +10,7 @@ public class PublishMessage extends MessageIDMessage { private String m_topicName; // private Integer m_messageID; //could be null if Qos is == 0 - private byte[] m_payload; + private ByteBuffer m_payload; /*public Integer getMessageID() { return m_messageID; @@ -30,11 +32,11 @@ public class PublishMessage extends MessageIDMessage { this.m_topicName = topicName; } - public byte[] getPayload() { + public ByteBuffer getPayload() { return m_payload; } - public void setPayload(byte[] payload) { + public void setPayload(ByteBuffer payload) { this.m_payload = payload; } } diff --git a/tools_scripts/fuse_publisher.groovy b/tools_scripts/fuse_publisher.groovy new file mode 100644 index 00000000..7d53ee29 --- /dev/null +++ b/tools_scripts/fuse_publisher.groovy @@ -0,0 +1,27 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') + +import java.net.URISyntaxException + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.QoS + +String host = args[0] + +MQTT mqtt = new MQTT() +//mqtt.setHost("test.mosquitto.org", 1883); +mqtt.setHost(host, 1883) +BlockingConnection connection = mqtt.blockingConnection() +connection.connect() + +print "publishing.." +long startTime = System.currentTimeMillis() +(1..100000).each { + connection.publish("/topic", 'Hello world!!'.bytes, QoS.AT_MOST_ONCE, false) +} +connection.publish('/exit', 'Exit'.bytes, QoS.AT_MOST_ONCE, false) +long stopTime = System.currentTimeMillis() +long spentTime = stopTime -startTime +println "published in ${spentTime} ms" +connection.disconnect() +println "disconnected" \ No newline at end of file diff --git a/tools_scripts/fuse_subscriber.groovy b/tools_scripts/fuse_subscriber.groovy new file mode 100644 index 00000000..2fbc055d --- /dev/null +++ b/tools_scripts/fuse_subscriber.groovy @@ -0,0 +1,51 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') + +import java.net.URISyntaxException + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.Message +import org.fusesource.mqtt.client.QoS +import org.fusesource.mqtt.client.Topic +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit + +String host = args[0] +MQTT mqtt = new MQTT() +//mqtt.setHost("test.mosquitto.org", 1883); +mqtt.setHost(host, 1883) +mqtt.setCleanSession(true) +mqtt.setHost(host, 1883) + +mqtt.setClientId("SubscriberClient") +BlockingConnection connection = mqtt.blockingConnection() +connection.connect() +println "Connected to ${host}" + +Topic[] topics = [new Topic("/topic", QoS.AT_MOST_ONCE), new Topic("/exit", QoS.AT_MOST_ONCE)] +byte[] qoses = connection.subscribe(topics) +println "subscribed to /topic qos: ${qoses[0]}" +println "subscribed to /exit qos: ${qoses[1]}" + +int numReceived = 0 +long startTime = System.currentTimeMillis() +boolean exit = false + +while(!exit) { + Message message = connection.receive() + if (message.topic == "/exit") { + exit = true + } else { + numReceived++ +// if ((numReceived % 10000) == 0) { +// print '.' +// } + } + message.ack() +} +connection.disconnect() +long stopTime = System.currentTimeMillis() +long spentTime = stopTime - startTime +println "/exit received" +println "subscriber disconnected, received ${numReceived} messages in ${spentTime} ms" +