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

First handling of QoS 1 level publish

Browse Source
This commit is contained in:
selva.andre
2012-08-31 16:56:46 +02:00
parent ecaf50a1a8
commit 90259d4148
7 changed files with 100 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+4 -1
View File
@@ -1,6 +1,7 @@
package org.dna.mqtt.moquette.messaging.spi;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
/**
@@ -16,7 +17,7 @@ public interface IMessaging {
* notification also after all the current subscriber to the given topic has
* been notified.
*/
void publish(String topic, byte[] message, QOSType qos, boolean retain);
void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID);
/**
* Subscribe a client to a specified topic with a defined level
@@ -38,4 +39,6 @@ public interface IMessaging {
void unsubscribe(String topic, String clientID);
void disconnect(IoSession session);
void refill(MessagingEvent evt) throws InterruptedException;
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+46 -21
View File
@@ -12,14 +12,7 @@ import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.MQTTException;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token;
import org.dna.mqtt.moquette.messaging.spi.impl.events.CloseEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.DisconnectEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.RemoveAllSubscriptionsEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.SubscribeEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.UnsubscribeEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.server.Server;
import org.fusesource.hawtbuf.codec.StringCodec;
@@ -66,6 +59,8 @@ public class SimpleMessaging implements IMessaging, Runnable {
private MultiIndexFactory m_multiIndexFactory;
private PageFileFactory pageFactory;
private SortedIndex<String, StoredMessage> m_retainedStore;
//bind clientID+MsgID -> evt message published
private SortedIndex<String, PublishEvent> m_inflightStore;
public SimpleMessaging() {
String storeFile = Server.STORAGE_FILE_PATH;
@@ -86,6 +81,7 @@ public class SimpleMessaging implements IMessaging, Runnable {
subscriptions.init(m_multiIndexFactory);
initRetainedMessageStore();
initPersistentMessageStore();
}
private void initRetainedMessageStore() {
@@ -96,17 +92,28 @@ public class SimpleMessaging implements IMessaging, Runnable {
m_retainedStore = (SortedIndex<String, StoredMessage>) m_multiIndexFactory.openOrCreate("retained", indexFactory);
}
/**
* Initialize the message store used to handle the temporary storage of QoS 1,2
* messages.
*/
private void initPersistentMessageStore() {
BTreeIndexFactory<String, PublishEvent> indexFactory = new BTreeIndexFactory<String, PublishEvent>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_inflightStore = (SortedIndex<String, PublishEvent>) m_multiIndexFactory.openOrCreate("inflight", indexFactory);
}
public void run() {
eventLoop();
}
public void publish(String topic, byte[] message, QOSType qos, boolean retain) {
public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID) {
//TODO for this moment no qos > 0 trivial implementation
QOSType defQos = QOSType.MOST_ONE;
// QOSType defQos = QOSType.MOST_ONE;
try {
m_inboundQueue.put(new PublishEvent(topic, defQos, message, retain));
m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
@@ -139,6 +146,11 @@ public class SimpleMessaging implements IMessaging, Runnable {
LOG.error(null, ex);
}
}
//method used by hte Notifier to re-put an event on the inbound queue
public void refill(MessagingEvent evt) throws InterruptedException {
m_inboundQueue.put(evt);
}
/**
@@ -198,8 +210,8 @@ public class SimpleMessaging implements IMessaging, Runnable {
public BlockingQueue<MessagingEvent> getNotifyEventQueue() {
return m_outboundQueue;
}
protected void eventLoop() {
LOG.debug("Started event loop");
boolean interrupted = false;
@@ -218,6 +230,9 @@ public class SimpleMessaging implements IMessaging, Runnable {
processClose();
} else if (evt instanceof DisconnectEvent) {
m_outboundQueue.put(evt);
} else if (evt instanceof CleanInFlightEvent) {
//remove the message from inflight storage
m_inflightStore.remove(((CleanInFlightEvent)evt).getMsgId());
}
} catch (InterruptedException ex) {
processClose();
@@ -226,20 +241,30 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
}
protected void processPublish(PublishEvent evt) {
protected void processPublish(PublishEvent evt) throws InterruptedException {
LOG.debug("processPublish invoked");
final String topic = evt.getTopic();
final QOSType qos = evt.getQos();
final byte[] message = evt.getMessage();
boolean retain = evt.isRetain();
CleanInFlightEvent cleanEvt = null;
if (qos == QOSType.LEAST_ONE) {
//store the temporary message
String publishKey = evt.getClientID() + evt.getMessageID();
m_inflightStore.put(publishKey, evt);
cleanEvt = new CleanInFlightEvent(publishKey);
}
for (final Subscription sub : subscriptions.matches(topic)) {
try {
m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false));
}
if (cleanEvt != null) {
m_outboundQueue.put(cleanEvt);
}
if (retain) {
if (message.length == 0) {
//clean the message from topic
@@ -250,8 +275,8 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
}
}
protected void processSubscribe(SubscribeEvent evt)/* throws InterruptedException*/ {
LOG.debug("processSubscribe invoked");
Subscription newSubscription = evt.getSubscription();
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CleanInFlightEvent.java
+17
View File
@@ -0,0 +1,17 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
/**
* Message to remove the message of type QoS 1 or 2 from in flight store
*/
public class CleanInFlightEvent extends MessagingEvent {
private String m_msgId;
public CleanInFlightEvent(String msgId) {
m_msgId = msgId;
}
public String getMsgId() {
return m_msgId;
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
+20 -1
View File
@@ -11,12 +11,23 @@ public class PublishEvent extends MessagingEvent {
QOSType m_qos;
byte[] m_message;
boolean m_retain;
String m_clientID;
//Optional attribute, available only fo QoS 1 and 2
String m_msgID;
public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain) {
public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain,
String clientID) {
m_topic = topic;
m_qos = qos;
m_message = message;
m_retain = retain;
m_clientID = clientID;
}
public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain,
String clientID, String msgID) {
this(topic, qos, message, retain, clientID);
m_msgID = msgID;
}
public String getTopic() {
@@ -34,4 +45,12 @@ public class PublishEvent extends MessagingEvent {
public boolean isRetain() {
return m_retain;
}
public String getClientID() {
return m_clientID;
}
public String getMessageID() {
return m_msgID;
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+3 -2
View File
@@ -120,7 +120,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
if (msg.isWillFlag()) {
QOSType willQos = QOSType.values()[msg.getWillQos()];
m_messaging.publish(msg.getWillTopic(), msg.getWillMessage().getBytes(),
willQos, msg.isWillRetain());
willQos, msg.isWillRetain(), msg.getClientID());
}
//handle user authentication
@@ -184,8 +184,9 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
}
protected void handlePublish(IoSession session, PublishMessage message) {
String clientID = (String) session.getAttribute(ATTR_CLIENTID);
m_messaging.publish(message.getTopicName(), message.getPayload(),
message.getQos(), message.isRetainFlag());
message.getQos(), message.isRetainFlag(), clientID);
}
protected void handleDisconnect(IoSession session, DisconnectMessage disconnectMessage) {
parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java
+8 -1
View File
@@ -1,7 +1,10 @@
package org.dna.mqtt.moquette.server;
import java.util.concurrent.BlockingQueue;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.INotifier;
import org.dna.mqtt.moquette.messaging.spi.impl.events.CleanInFlightEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.DisconnectEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent;
@@ -18,10 +21,12 @@ class Notifier implements Runnable {
BlockingQueue<MessagingEvent> m_queue;
INotifier m_notifier;
IMessaging m_messaging;
Notifier(BlockingQueue<MessagingEvent> queue, INotifier notifier) {
Notifier(BlockingQueue<MessagingEvent> queue, INotifier notifier, IMessaging messaging) {
m_queue = queue;
m_notifier = notifier;
m_messaging = messaging;
}
public void run() {
@@ -37,6 +42,8 @@ class Notifier implements Runnable {
} else if (evt instanceof DisconnectEvent) {
DisconnectEvent evtD = (DisconnectEvent) evt;
m_notifier.disconnect(evtD.getSession());
} else if (evt instanceof CleanInFlightEvent) {
m_messaging.refill(evt);
}
} catch (InterruptedException ex) {
interrupted = true;
parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
+2 -1
View File
@@ -94,7 +94,8 @@ public class Server {
BlockingQueue<MessagingEvent> notifyQueue = messaging.getNotifyEventQueue();
for (int i = 0; i < NOTIFIER_POOL_SIZE; i++) {
m_notifierPool.submit(new Notifier(notifyQueue, handler));
Notifier notifier = new Notifier(notifyQueue, handler, messaging);
m_notifierPool.submit(notifier);
}
m_acceptor.setHandler(handler);