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 18c2c107..af7f7a1d 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,12 +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.nio.ByteBuffer; -import java.util.*; - /** * Implementation of IStorageService backed by HawtDB @@ -55,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; @@ -102,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() { @@ -120,26 +120,28 @@ 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, ByteBuffer message, AbstractMessage.QOSType qos) { - if (message.length == 0) { + 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)); } } @@ -159,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(String.format("Stored published message for client <%s> on topic <%s>", 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) { @@ -185,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) { @@ -223,7 +234,7 @@ 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); + m_qos2Store.put(publishKey, convertToStored(evt)); } public void removeQoS2Message(String publishKey) { @@ -231,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..c4625669 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,13 +33,15 @@ 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)); } } 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 1c1d988e..027db096 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 @@ -270,8 +270,8 @@ class ProtocolProcessor implements EventHandler { 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(String.format("send publish message to <%s> on topic <%s>", clientId, topic)); if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) { pubMessage.setMessageID(messageID); } 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 750cb374..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,7 +2,6 @@ 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; @@ -11,7 +10,7 @@ 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; 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/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 ***");