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

Added rough implementation for QoS 2 publish

Browse Source
This commit is contained in:
selva.andre
2012-09-10 23:30:40 +02:00
parent 87a51cf4c2
commit c4f186b3c8
5 changed files with 140 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java
+6
View File
@@ -39,4 +39,10 @@ public interface IStorageService {
List<Subscription> retrieveAllSubscriptions();
void close();
void persistQoS2Message(String publishKey, PublishEvent evt);
void removeQoS2Message(String publishKey);
PublishEvent retrieveQoS2Message(String publishKey);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+24 -1
View File
@@ -52,6 +52,8 @@ public class HawtDBStorageService implements IStorageService {
private SortedIndex<String, StoredMessage> m_retainedStore;
//bind clientID+MsgID -> evt message published
private SortedIndex<String, PublishEvent> m_inflightStore;
//bind clientID+MsgID -> evt message published
private SortedIndex<String, PublishEvent> m_qos2Store;
//persistent Map of clientID, set of Subscriptions
private SortedIndex<String, Set<Subscription>> m_persistentSubscriptions;
@@ -81,6 +83,7 @@ public class HawtDBStorageService implements IStorageService {
initPersistentMessageStore();
initInflightMessageStore();
initPersistentSubscriptions();
initPersistentQoS2MessageStore();
}
private void initRetainedStore() {
@@ -88,7 +91,6 @@ public class HawtDBStorageService implements IStorageService {
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_retainedStore = (SortedIndex<String, StoredMessage>) m_multiIndexFactory.openOrCreate("retained", indexFactory);
}
@@ -117,6 +119,13 @@ public class HawtDBStorageService implements IStorageService {
m_inflightStore = (SortedIndex<String, PublishEvent>) m_multiIndexFactory.openOrCreate("inflight", indexFactory);
}
private void initPersistentQoS2MessageStore() {
BTreeIndexFactory<String, PublishEvent> indexFactory = new BTreeIndexFactory<String, PublishEvent>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_qos2Store = (SortedIndex<String, PublishEvent>) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory);
}
public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) {
if (message.length == 0) {
//clean the message from topic
@@ -202,4 +211,18 @@ public class HawtDBStorageService implements IStorageService {
LOG.error(null, ex);
}
}
/*-------- 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);
}
public void removeQoS2Message(String publishKey) {
m_qos2Store.remove(publishKey);
}
public PublishEvent retrieveQoS2Message(String publishKey) {
return m_qos2Store.get(publishKey);
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+74 -17
View File
@@ -19,6 +19,7 @@ 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;
@@ -178,6 +179,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
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 {
throw new RuntimeException("Illegal message received " + message);
}
@@ -274,6 +279,30 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
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();
@@ -286,24 +315,14 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
//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());
}
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(), evt.getMessageID(), null);
m_storageService.storePublishForFuture(newPublishEvt);
} else {
notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false));
}
}
}
publish2Subscribers(topic, qos, message, retain, evt.getMessageID());
if (qos == QOSType.LEAST_ONE) {
assert publishKey != null;
@@ -316,6 +335,28 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
}
}
/**
* 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 {
notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false));
}
}
}
}
private void subscribeSingleTopic(Subscription newSubscription, final String topic) {
subscriptions.add(newSubscription);
@@ -455,4 +496,20 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
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);
}
}
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java
+12
View File
@@ -60,4 +60,16 @@ public class DummyStorageService implements IStorageService {
public void close() {
//To change body of implemented methods use File | Settings | File Templates.
}
public void persistQoS2Message(String publishKey, PublishEvent evt) {
//To change body of implemented methods use File | Settings | File Templates.
}
public void removeQoS2Message(String publishKey) {
//To change body of implemented methods use File | Settings | File Templates.
}
public PublishEvent retrieveQoS2Message(String publishKey) {
return null; //To change body of implemented methods use File | Settings | File Templates.
}
}
parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java
+24
View File
@@ -121,4 +121,28 @@ public class ServerIntegrationFuseTest {
futurePublish.await();
connectionPub.disconnect().await();
}
@Test
public void testPublishWithQoS2() throws Exception {
m_mqtt.setClientId("TestClient");
m_connection = m_mqtt.futureConnection();
m_connection.connect().await();
//subscribe to a QoS 2 topic
Topic[] topics = {new Topic("/topic", QoS.EXACTLY_ONCE)};
Future<byte[]> subscribeFuture = m_connection.subscribe(topics);
byte[] qoses = subscribeFuture.await();
Future<Message> receive = m_connection.receive();
//publish a QoS 2 message
Future<Void> f3 = m_connection.publish("/topic", "Hello MQTT".getBytes(), QoS.EXACTLY_ONCE, false);
//verify the reception
Message message = receive.await();
message.ack();
assertEquals("Hello MQTT", new String(message.getPayload()));
}
}