anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed the switching from byte[] to ByteBuffer on server storage

Browse Source
This commit is contained in:
andrea
2013-11-26 16:21:01 +01:00
parent d9d2d13d0d
commit f592db516f
6 changed files with 121 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+52 -26
View File
@@ -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<String, List<PublishEvent>> m_persistentMessageStore;
private SortedIndex<String, List<StoredPublishEvent>> m_persistentMessageStore;
private SortedIndex<String, StoredMessage> m_retainedStore;
//bind clientID+MsgID -> evt message published
private SortedIndex<String, PublishEvent> m_inflightStore;
private SortedIndex<String, StoredPublishEvent> m_inflightStore;
//bind clientID+MsgID -> evt message published
private SortedIndex<String, PublishEvent> m_qos2Store;
private SortedIndex<String, StoredPublishEvent> m_qos2Store;
//persistent Map of clientID, set of Subscriptions
private SortedIndex<String, Set<Subscription>> m_persistentSubscriptions;
@@ -102,10 +102,10 @@ public class HawtDBStorageService implements IStorageService {
private void initPersistentMessageStore() {
BTreeIndexFactory<String, List<PublishEvent>> indexFactory = new BTreeIndexFactory<String, List<PublishEvent>>();
BTreeIndexFactory<String, List<StoredPublishEvent>> indexFactory = new BTreeIndexFactory<String, List<StoredPublishEvent>>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_persistentMessageStore = (SortedIndex<String, List<PublishEvent>>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory);
m_persistentMessageStore = (SortedIndex<String, List<StoredPublishEvent>>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory);
}
private void initPersistentSubscriptions() {
@@ -120,26 +120,28 @@ public class HawtDBStorageService implements IStorageService {
* messages in flight.
*/
private void initInflightMessageStore() {
BTreeIndexFactory<String, PublishEvent> indexFactory = new BTreeIndexFactory<String, PublishEvent>();
BTreeIndexFactory<String, StoredPublishEvent> indexFactory = new BTreeIndexFactory<String, StoredPublishEvent>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_inflightStore = (SortedIndex<String, PublishEvent>) m_multiIndexFactory.openOrCreate("inflight", indexFactory);
m_inflightStore = (SortedIndex<String, StoredPublishEvent>) m_multiIndexFactory.openOrCreate("inflight", indexFactory);
}
private void initPersistentQoS2MessageStore() {
BTreeIndexFactory<String, PublishEvent> indexFactory = new BTreeIndexFactory<String, PublishEvent>();
BTreeIndexFactory<String, StoredPublishEvent> indexFactory = new BTreeIndexFactory<String, StoredPublishEvent>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_qos2Store = (SortedIndex<String, PublishEvent>) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory);
m_qos2Store = (SortedIndex<String, StoredPublishEvent>) 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<PublishEvent> storedEvents;
List<StoredPublishEvent> storedEvents;
String clientID = evt.getClientID();
if (!m_persistentMessageStore.containsKey(clientID)) {
storedEvents = new ArrayList<PublishEvent>();
storedEvents = new ArrayList<StoredPublishEvent>();
} 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<PublishEvent> retrivePersistedPublishes(String clientID) {
return m_persistentMessageStore.get(clientID);
List<StoredPublishEvent> storedEvts = m_persistentMessageStore.get(clientID);
if (storedEvts == null) {
return null;
}
List<PublishEvent> liveEvts = new ArrayList<PublishEvent>();
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;
}
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java
+6 -3
View File
@@ -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));
}
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+2 -2
View File
@@ -270,8 +270,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
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);
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
+1 -2
View File
@@ -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;
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/storage/StoredPublishEvent.java
+58
View File
@@ -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;
}
}
broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java
+2
View File
@@ -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 ***");