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

Merge branch 'implements_support_for_qos1'

Browse Source
This commit is contained in:
selva.andre
2012-09-01 12:33:27 +02:00
parent c889a4ef45 e5fbf5c655
commit 42c04b074c
24 changed files with 480 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/client/Client.java
+40 -24
View File
@@ -24,17 +24,7 @@ import org.dna.mqtt.moquette.ConnectionException;
import org.dna.mqtt.moquette.MQTTException;
import org.dna.mqtt.moquette.PublishException;
import org.dna.mqtt.moquette.SubscribeException;
import org.dna.mqtt.moquette.proto.ConnAckDecoder;
import org.dna.mqtt.moquette.proto.ConnectEncoder;
import org.dna.mqtt.moquette.proto.DisconnectEncoder;
import org.dna.mqtt.moquette.proto.PingReqEncoder;
import org.dna.mqtt.moquette.proto.PingRespDecoder;
import org.dna.mqtt.moquette.proto.PublishDecoder;
import org.dna.mqtt.moquette.proto.PublishEncoder;
import org.dna.mqtt.moquette.proto.SubAckDecoder;
import org.dna.mqtt.moquette.proto.SubscribeEncoder;
import org.dna.mqtt.moquette.proto.UnsubAckDecoder;
import org.dna.mqtt.moquette.proto.UnsubscribeEncoder;
import org.dna.mqtt.moquette.proto.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
@@ -120,6 +110,7 @@ public final class Client {
decoder.addMessageDecoder(new SubAckDecoder());
decoder.addMessageDecoder(new UnsubAckDecoder());
decoder.addMessageDecoder(new PublishDecoder());
decoder.addMessageDecoder(new PubAckDecoder());
decoder.addMessageDecoder(new PingRespDecoder());
DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
@@ -230,27 +221,46 @@ public final class Client {
public void publish(String topic, byte[] payload) throws PublishException {
publish(topic, payload, false);
}
/**
* Publish by default with QoS 0
* */
public void publish(String topic, byte[] payload, boolean retain) throws PublishException {
publish(topic, payload, AbstractMessage.QOSType.MOST_ONE, retain);
}
public void publish(String topic, byte[] payload, AbstractMessage.QOSType qos, boolean retain) throws PublishException {
PublishMessage msg = new PublishMessage();
msg.setRetainFlag(retain);
msg.setTopicName(topic);
msg.setPayload(payload);
//Untill the server could handle all the Qos levels
msg.setQos(AbstractMessage.QOSType.MOST_ONE);
//Untill the server could handle all the Qos 2 level
if (qos != AbstractMessage.QOSType.MOST_ONE) {
msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
int messageID = m_messageIDGenerator.next();
msg.setMessageID(messageID);
WriteFuture wf = m_session.write(msg);
try {
wf.await();
} catch (InterruptedException ex) {
LOG.debug(null, ex);
throw new PublishException(ex);
}
try {
manageSendQoS1(msg);
} catch (Throwable ex) {
throw new MQTTException(ex);
}
} else {
//QoS 0 case
msg.setQos(AbstractMessage.QOSType.MOST_ONE);
WriteFuture wf = m_session.write(msg);
try {
wf.await();
} catch (InterruptedException ex) {
LOG.debug(null, ex);
throw new PublishException(ex);
}
Throwable ex = wf.getException();
if (ex != null) {
throw new PublishException(ex);
Throwable ex = wf.getException();
if (ex != null) {
throw new PublishException(ex);
}
}
updatePinger();
@@ -368,6 +378,12 @@ public final class Client {
m_receivedSubAckMessageID = messageID;
}
void publishAckCallback(Integer messageID) {
LOG.info("publishAckCallback invoked");
m_subscribeBarrier.countDown();
m_receivedSubAckMessageID = messageID;
}
protected void publishCallback(String topic, byte[] payload) {
IPublishCallback callback = m_subscribersList.get(topic);
if (callback == null) {
parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java
+10 -5
View File
@@ -2,12 +2,10 @@ package org.dna.mqtt.moquette.client;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.*;
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,11 +43,18 @@ public class ClientMQTTHandler extends IoHandlerAdapter {
case PUBLISH:
handlePublish(session, (PublishMessage) msg);
break;
case PUBACK:
handlePublishAck(session, (PubAckMessage) msg);
break;
case PINGRESP:
break;
}
}
private void handlePublishAck(IoSession session, PubAckMessage msg) {
m_callback.publishAckCallback(msg.getMessageID());
}
private void handleConnectAck(IoSession session, ConnAckMessage connAckMessage) {
m_callback.connectionAckCallback(connAckMessage.getReturnCode());
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+13 -2
View File
@@ -1,6 +1,8 @@
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.messaging.spi.impl.events.PubAckEvent;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
/**
@@ -16,12 +18,17 @@ 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, IoSession session);
/**
* Used to publish messages with QoS > 0
* */
void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID, IoSession session);
/**
* Subscribe a client to a specified topic with a defined level
*/
void subscribe(String clientId, String topic, QOSType qos);
void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession);
/**
* Remove all subscription to any topic the client (identified by clientID)
@@ -38,4 +45,8 @@ public interface IMessaging {
void unsubscribe(String topic, String clientID);
void disconnect(IoSession session);
void refill(MessagingEvent evt) throws InterruptedException;
void republishStored(String clientID);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java
+7 -2
View File
@@ -1,7 +1,10 @@
package org.dna.mqtt.moquette.messaging.spi;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
/**
*
@@ -9,8 +12,10 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
*/
public interface INotifier {
public void notify(String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained);
public void notify(NotifyEvent evt/*String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained*/);
public void disconnect(IoSession session);
void sendPubAck(PubAckEvent evt);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+122 -27
View File
@@ -4,6 +4,7 @@ import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
@@ -12,15 +13,9 @@ 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.Constants;
import org.dna.mqtt.moquette.server.Server;
import org.fusesource.hawtbuf.codec.StringCodec;
import org.fusesource.hawtdb.api.BTreeIndexFactory;
@@ -66,6 +61,10 @@ 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;
//maps clientID to the list of pending messages stored
private SortedIndex<String, List<PublishEvent>> m_persistentMessageStore;
public SimpleMessaging() {
String storeFile = Server.STORAGE_FILE_PATH;
@@ -86,6 +85,9 @@ public class SimpleMessaging implements IMessaging, Runnable {
subscriptions.init(m_multiIndexFactory);
initRetainedMessageStore();
initInflightMessageStore();
//init the message store for QoS 1/2 messages in clean sessions
initPersistentMessageStore();
}
private void initRetainedMessageStore() {
@@ -96,24 +98,47 @@ 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 in flight.
*/
private void initInflightMessageStore() {
BTreeIndexFactory<String, PublishEvent> indexFactory = new BTreeIndexFactory<String, PublishEvent>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_inflightStore = (SortedIndex<String, PublishEvent>) m_multiIndexFactory.openOrCreate("inflight", indexFactory);
}
private void initPersistentMessageStore() {
BTreeIndexFactory<String, List<PublishEvent>> indexFactory = new BTreeIndexFactory<String, List<PublishEvent>>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_persistentMessageStore = (SortedIndex<String, List<PublishEvent>>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory);
}
public void run() {
eventLoop();
}
public void publish(String topic, byte[] message, QOSType qos, boolean retain) {
//TODO for this moment no qos > 0 trivial implementation
QOSType defQos = QOSType.MOST_ONE;
public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, IoSession session) {
try {
m_inboundQueue.put(new PublishEvent(topic, defQos, message, retain));
m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, session));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
public void subscribe(String clientId, String topic, QOSType qos) {
Subscription newSubscription = new Subscription(clientId, topic, qos);
public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID, IoSession session) {
try {
m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, messageID, session));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession) {
Subscription newSubscription = new Subscription(clientId, topic, qos, cleanSession);
try {
LOG.debug("subscribe invoked for topic: " + topic);
m_inboundQueue.put(new SubscribeEvent(newSubscription));
@@ -139,8 +164,21 @@ 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);
}
public void republishStored(String clientID) {
//create the event to push
try {
m_inboundQueue.put(new RepublishEvent(clientID));
} catch (InterruptedException iex) {
LOG.error(null, iex);
}
}
/**
* Verify if the 2 topics matching respecting the rules of MQTT Appendix A
*/
@@ -198,8 +236,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;
@@ -217,7 +255,12 @@ public class SimpleMessaging implements IMessaging, Runnable {
} else if (evt instanceof CloseEvent) {
processClose();
} else if (evt instanceof DisconnectEvent) {
m_outboundQueue.put(evt);
processDisconnect((DisconnectEvent) evt);
} else if (evt instanceof CleanInFlightEvent) {
//remove the message from inflight storage
m_inflightStore.remove(((CleanInFlightEvent)evt).getMsgId());
} else if (evt instanceof RepublishEvent) {
processRepublish((RepublishEvent) evt);
}
} catch (InterruptedException ex) {
processClose();
@@ -225,21 +268,50 @@ 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 = String.format("%s%d", evt.getClientID(), evt.getMessageID());
m_inflightStore.put(publishKey, evt);
cleanEvt = new CleanInFlightEvent(publishKey);
}
for (final Subscription sub : subscriptions.matches(topic)) {
try {
if (qos == QOSType.MOST_ONE) {
//QoS 0
m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false));
} catch (InterruptedException ex) {
LOG.error(null, ex);
} else {
//QoS 1 or 2
//if the target subscription is not clean session and is not connected => store it
if (!sub.isCleanSession() && !sub.isActive()) {
List<PublishEvent> storedEvents;
String clientID = evt.getClientID();
if (!m_persistentMessageStore.containsKey(clientID)) {
storedEvents = new ArrayList<PublishEvent>();
} else {
storedEvents = m_persistentMessageStore.get(clientID);
}
storedEvents.add(evt);
m_persistentMessageStore.put(clientID, storedEvents);
}
m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false, evt.getMessageID()));
}
}
if (cleanEvt != null) {
m_outboundQueue.put(cleanEvt);
m_outboundQueue.put(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
}
if (retain) {
if (message.length == 0) {
//clean the message from topic
@@ -250,8 +322,8 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
}
}
protected void processSubscribe(SubscribeEvent evt)/* throws InterruptedException*/ {
LOG.debug("processSubscribe invoked");
Subscription newSubscription = evt.getSubscription();
@@ -284,6 +356,17 @@ public class SimpleMessaging implements IMessaging, Runnable {
protected void processRemoveAllSubscriptions(RemoveAllSubscriptionsEvent evt) {
LOG.debug("processRemoveAllSubscriptions invoked");
subscriptions.removeForClient(evt.getClientID());
//remove also the messages stored of type QoS1/2
m_persistentMessageStore.remove(evt.getClientID());
}
private void processDisconnect(DisconnectEvent evt) throws InterruptedException {
m_outboundQueue.put(evt);
//de-activate the subscriptions for this ClientID
String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
subscriptions.disconnect(clientID);
}
private void processClose() {
@@ -294,4 +377,16 @@ public class SimpleMessaging implements IMessaging, Runnable {
LOG.error(null, ex);
}
}
private void processRepublish(RepublishEvent evt) throws InterruptedException {
List<PublishEvent> publishedEvents = m_persistentMessageStore.get(evt.getClientID());
if (publishedEvents == null) {
return;
}
for (PublishEvent pubEvt : publishedEvents) {
m_outboundQueue.put(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID()));
}
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java
+16 -1
View File
@@ -15,11 +15,14 @@ public class Subscription implements Serializable {
QOSType requestedQos;
String clientId;
String topic;
boolean cleanSession;
boolean active = true;
public Subscription(String clientId, String topic, QOSType requestedQos) {
public Subscription(String clientId, String topic, QOSType requestedQos, boolean cleanSession) {
this.requestedQos = requestedQos;
this.clientId = clientId;
this.topic = topic;
this.cleanSession = cleanSession;
}
public String getClientId() {
@@ -34,6 +37,18 @@ public class Subscription implements Serializable {
return topic;
}
public boolean isCleanSession() {
return this.cleanSession;
}
public boolean isActive() {
return active;
}
public void setActive(boolean active) {
this.active = active;
}
@Override
public boolean equals(Object obj) {
if (obj == null) {
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java
+19 -2
View File
@@ -134,6 +134,19 @@ public class SubscriptionsStore {
child.removeClientSubscriptions(clientID);
}
}
void disconnect(String clientID) {
for (Subscription s : m_subscriptions) {
if (s.clientId.equals(clientID)) {
s.setActive(false);
}
}
//go deep
for (TreeNode child : m_children) {
child.disconnect(clientID);
}
}
}
protected static class Token {
@@ -192,7 +205,7 @@ public class SubscriptionsStore {
}
private TreeNode subscriptions = new TreeNode(null);
private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
//pesistent Map of clientID, list of Subscriptions
//persistent Map of clientID, list of Subscriptions
private SortedIndex<String, List<Subscription>> m_persistent;
/**
@@ -207,7 +220,7 @@ public class SubscriptionsStore {
m_persistent = (SortedIndex<String, List<Subscription>>) multiFactory.openOrCreate("subscriptions", indexFactory);
//reaload any subscriptions persisted
//reload any subscriptions persisted
LOG.debug("Reloading all stored subscriptions...");
for (Map.Entry<String, List<Subscription>> entry : m_persistent) {
for (Subscription subscription : entry.getValue()) {
@@ -293,6 +306,10 @@ public class SubscriptionsStore {
m_persistent.remove(clientID);
}
public void disconnect(String clientID) {
subscriptions.disconnect(clientID);
}
public List<Subscription> matches(String topic) {
List<Token> tokens = new ArrayList<Token>();
try {
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/NotifyEvent.java
+13
View File
@@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
/**
* TODO it's identical to PublishEvent
*
* @author andrea
*/
@@ -13,6 +14,9 @@ public class NotifyEvent extends MessagingEvent {
private QOSType m_qos;
private byte[] m_message;
private boolean m_retaned;
//Optional attribute, available only fo QoS 1 and 2
int m_msgID;
public NotifyEvent(String clientId, String topic, QOSType qos, byte[] message, boolean retained) {
@@ -23,6 +27,11 @@ public class NotifyEvent extends MessagingEvent {
m_retaned = retained;
}
public NotifyEvent(String clientId, String topic, QOSType qos, byte[] message, boolean retained, int msgID) {
this(clientId, topic, qos, message, retained);
m_msgID = msgID;
}
public String getClientId() {
return m_clientId;
}
@@ -42,5 +51,9 @@ public class NotifyEvent extends MessagingEvent {
public boolean isRetained() {
return m_retaned;
}
public int getMessageID() {
return m_msgID;
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.java
+24
View File
@@ -0,0 +1,24 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
/**
* Used to send the ack message back to the client after a publish
*/
public class PubAckEvent extends MessagingEvent {
int m_messageId;
String m_clientID;
public PubAckEvent(int messageID, String clientID) {
m_messageId = messageID ;
m_clientID = clientID;
}
public int getMessageId() {
return m_messageId;
}
public String getClientID() {
return m_clientID;
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
+31 -2
View File
@@ -1,22 +1,39 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import java.io.Serializable;
/**
*
* @author andrea
*/
public class PublishEvent extends MessagingEvent {
public class PublishEvent extends MessagingEvent 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;
transient IoSession m_session;
public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain) {
public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain,
String clientID, IoSession session) {
m_topic = topic;
m_qos = qos;
m_message = message;
m_retain = retain;
m_clientID = clientID;
m_session = session;
}
public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain,
String clientID, int msgID, IoSession session) {
this(topic, qos, message, retain, clientID, session);
m_msgID = msgID;
}
public String getTopic() {
@@ -34,4 +51,16 @@ public class PublishEvent extends MessagingEvent {
public boolean isRetain() {
return m_retain;
}
public String getClientID() {
return m_clientID;
}
public int getMessageID() {
return m_msgID;
}
public IoSession getSession() {
return m_session;
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RepublishEvent.java
+16
View File
@@ -0,0 +1,16 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
/**
* Used to re-fire stored QoS1 QoS2 events once a client reconnects.
*/
public class RepublishEvent extends MessagingEvent {
private String m_clientID;
public RepublishEvent(String clientID) {
this.m_clientID = clientID;
}
public String getClientID() {
return m_clientID;
}
}
parser/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java
+4 -4
View File
@@ -9,14 +9,14 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public abstract class MessageIDMessage extends AbstractMessage {
int m_messageID;
private Integer m_messageID; //could be null if Qos is == 0
public int getMessageID() {
public Integer getMessageID() {
return m_messageID;
}
public void setMessageID(int messageID) {
public void setMessageID(Integer messageID) {
this.m_messageID = messageID;
}
}
parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java
+4 -4
View File
@@ -4,19 +4,19 @@ package org.dna.mqtt.moquette.proto.messages;
*
* @author andrea
*/
public class PublishMessage extends AbstractMessage {
public class PublishMessage extends MessageIDMessage {
private String m_topicName;
private Integer m_messageID; //could be null if Qos is == 0
// private Integer m_messageID; //could be null if Qos is == 0
private byte[] m_payload;
public Integer getMessageID() {
/*public Integer getMessageID() {
return m_messageID;
}
public void setMessageID(Integer messageID) {
this.m_messageID = messageID;
}
}*/
public String getTopicName() {
return m_topicName;
parser/src/main/java/org/dna/mqtt/moquette/server/Constants.java
+9
View File
@@ -0,0 +1,9 @@
package org.dna.mqtt.moquette.server;
/**
* Server constants keeper
*/
public class Constants {
public static final String ATTR_CLIENTID = "ClientID";
public static final String CLEAN_SESSION = "cleanSession";
}
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+63 -31
View File
@@ -9,18 +9,12 @@ import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.INotifier;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent;
import org.dna.mqtt.moquette.proto.messages.*;
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.dna.mqtt.moquette.proto.messages.DisconnectMessage;
import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -30,7 +24,6 @@ import org.slf4j.LoggerFactory;
*/
public class MQTTHandler extends IoHandlerAdapter implements INotifier {
protected static final String ATTR_CLIENTID = "ClientID";
private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class);
/**
* Maps CLIENT_ID to the IoSession that represents the connection
@@ -93,7 +86,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
if (m_clientIDs.containsKey(msg.getClientID())) {
//clean the subscriptions if the old used a cleanSession = true
IoSession oldSession = m_clientIDs.get(msg.getClientID()).getSession();
boolean cleanSession = (Boolean) oldSession.getAttribute("cleanSession");
boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
m_messaging.removeSubscriptions(msg.getClientID());
@@ -110,9 +103,9 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
int keepAlive = msg.getKeepAlive();
session.setAttribute("keepAlive", keepAlive);
session.setAttribute("cleanSession", msg.isCleanSession());
session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
//used to track the client in the subscription and publishing phases.
session.setAttribute(ATTR_CLIENTID, msg.getClientID());
session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID());
session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f));
@@ -120,7 +113,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(), session);
}
//handle user authentication
@@ -137,13 +130,14 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
}
}
//TODO for QoS1 and QoS remember to store all the messages arrived to
//the dormat subscriptionsonce the QoS1 and QoS2
//handle clean session flag
if (msg.isCleanSession()) {
//remove all prev subscriptions
//cleanup topic subscriptions
m_messaging.removeSubscriptions(msg.getClientID());
} else {
//force the republish of stored QoS1 and QoS2
m_messaging.republishStored(msg.getClientID());
}
ConnAckMessage okResp = new ConnAckMessage();
@@ -154,8 +148,10 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
protected void handleSubscribe(IoSession session, SubscribeMessage msg) {
LOG.debug("handleSubscribe, registering the subscriptions");
for (SubscribeMessage.Couple req : msg.subscriptions()) {
m_messaging.subscribe((String) session.getAttribute(ATTR_CLIENTID),
req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()]);
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
m_messaging.subscribe(clientID, req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()],
cleanSession);
}
//ack the client
@@ -173,23 +169,30 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
private void handleUnsubscribe(IoSession session, UnsubscribeMessage msg) {
LOG.info("unregistering the subscriptions");
for (String topic : msg.topics()) {
m_messaging.unsubscribe(topic, (String) session.getAttribute(ATTR_CLIENTID));
m_messaging.unsubscribe(topic, (String) session.getAttribute(Constants.ATTR_CLIENTID));
}
//ack the client
UnsubAckMessage ackMessage = new UnsubAckMessage();
ackMessage.setMessageID(msg.getMessageID());
LOG.info("replying with UnsubAct to MSG ID {0}", msg.getMessageID());
LOG.info("replying with UnsubAck to MSG ID {0}", msg.getMessageID());
session.write(ackMessage);
}
protected void handlePublish(IoSession session, PublishMessage message) {
m_messaging.publish(message.getTopicName(), message.getPayload(),
message.getQos(), message.isRetainFlag());
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
if (message.getQos() == QOSType.MOST_ONE) {
m_messaging.publish(message.getTopicName(), message.getPayload(),
message.getQos(), message.isRetainFlag(), clientID, session);
} else {
m_messaging.publish(message.getTopicName(), message.getPayload(),
message.getQos(), message.isRetainFlag(), clientID, message.getMessageID(), session);
}
}
protected void handleDisconnect(IoSession session, DisconnectMessage disconnectMessage) {
String clientID = (String) session.getAttribute(ATTR_CLIENTID);
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
//remove from clientIDs
// m_clientIDsLock.lock();
// try {
@@ -197,7 +200,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
// } finally {
// m_clientIDsLock.unlock();
// }
boolean cleanSession = (Boolean) session.getAttribute("cleanSession");
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
m_messaging.removeSubscriptions(clientID);
@@ -221,12 +224,19 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
m_authenticator = authenticator;
}
public void notify(String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained) {
public void notify(NotifyEvent evt) {
LOG.debug("notify invoked with event " + evt);
String clientId = evt.getClientId();
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(retained);
pubMessage.setTopicName(topic);
pubMessage.setQos(qOSType);
pubMessage.setPayload(payload);
pubMessage.setRetainFlag(evt.isRetained());
pubMessage.setTopicName(evt.getTopic());
pubMessage.setQos(evt.getQos());
pubMessage.setPayload(evt.getMessage());
if (pubMessage.getQos() != QOSType.MOST_ONE) {
pubMessage.setMessageID(evt.getMessageID());
}
LOG.debug("notify invoked");
m_clientIDsLock.lock();
@@ -244,7 +254,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
}
public void disconnect(IoSession session) {
String clientID = (String) session.getAttribute(ATTR_CLIENTID);
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
m_clientIDsLock.lock();
try {
m_clientIDs.remove(clientID);
@@ -253,4 +263,26 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
}
session.close(true);
}
public void sendPubAck(PubAckEvent evt) {
LOG.debug("sendPubAck invoked");
String clientId = evt.getClientID();
PubAckMessage pubAckMessage = new PubAckMessage();
pubAckMessage.setMessageID(evt.getMessageId());
m_clientIDsLock.lock();
try {
assert m_clientIDs != null;
LOG.debug("clientIDs are " + m_clientIDs);
assert m_clientIDs.get(clientId) != null;
LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession());
m_clientIDs.get(clientId).getSession().write(pubAckMessage);
}catch(Throwable t) {
LOG.error(null, t);
} finally {
m_clientIDsLock.unlock();
}
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java
+11 -5
View File
@@ -1,10 +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.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.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -18,10 +18,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() {
@@ -33,10 +35,14 @@ class Notifier implements Runnable {
LOG.debug("Notifing event " + evt);
if (evt instanceof NotifyEvent) {
NotifyEvent evtN = (NotifyEvent) evt;
m_notifier.notify(evtN.getClientId(), evtN.getTopic(), evtN.getQos(), evtN.getMessage(), evtN.isRetained());
m_notifier.notify(evtN);
} else if (evt instanceof DisconnectEvent) {
DisconnectEvent evtD = (DisconnectEvent) evt;
m_notifier.disconnect(evtD.getSession());
} else if (evt instanceof CleanInFlightEvent) {
m_messaging.refill(evt);
} else if (evt instanceof PubAckEvent) {
m_notifier.sendPubAck((PubAckEvent) evt);
}
} catch (InterruptedException ex) {
interrupted = true;
parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
+5 -18
View File
@@ -15,23 +15,8 @@ import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.proto.ConnAckEncoder;
import org.dna.mqtt.moquette.proto.ConnectDecoder;
import org.dna.mqtt.moquette.proto.DisconnectDecoder;
import org.dna.mqtt.moquette.proto.MQTTLoggingFilter;
import org.dna.mqtt.moquette.proto.PingReqDecoder;
import org.dna.mqtt.moquette.proto.PingRespEncoder;
import org.dna.mqtt.moquette.proto.PublishDecoder;
import org.dna.mqtt.moquette.proto.PublishEncoder;
import org.dna.mqtt.moquette.proto.SubAckEncoder;
import org.dna.mqtt.moquette.proto.SubscribeDecoder;
import org.dna.mqtt.moquette.proto.UnsubAckEncoder;
import org.dna.mqtt.moquette.proto.UnsubscribeDecoder;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
import org.dna.mqtt.moquette.proto.*;
import org.dna.mqtt.moquette.proto.messages.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
@@ -73,6 +58,7 @@ public class Server {
encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder());
encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder());
encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder());
encoder.addMessageEncoder(PubAckMessage.class, new PubAckEncoder());
encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder());
encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder());
@@ -94,7 +80,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);
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
+6 -6
View File
@@ -28,36 +28,36 @@ public class SimpleMessagingTest {
@Test
public void testSubscribe() {
//Exercise
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE));
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
messaging.processSubscribe(evt);
//Verify
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE);
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false);
assertTrue(messaging.getSubscriptions().contains(expectedSubscription));
}
@Test
public void testDoubleSubscribe() {
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE));
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
messaging.processSubscribe(evt);
//Exercise
messaging.processSubscribe(evt);
//Verify
Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE);
Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false);
assertEquals(1, messaging.getSubscriptions().size());
}
@Test
public void testPublish() throws InterruptedException {
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE));
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
messaging.processSubscribe(evt);
// INotifier notifier = mock(INotifier.class);
// messaging.setNotifier(notifier);
//Exercise
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false);
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null);
messaging.processPublish(pubEvt);
//Verify
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java
+18 -18
View File
@@ -105,11 +105,11 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimple() {
Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", AbstractMessage.QOSType.MOST_ONE);
Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", AbstractMessage.QOSType.MOST_ONE, false);
store.add(slashSub);
assertTrue(store.matches("finance").isEmpty());
Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", AbstractMessage.QOSType.MOST_ONE);
Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", AbstractMessage.QOSType.MOST_ONE, false);
store.add(slashFinanceSub);
assertTrue(store.matches("finance").isEmpty());
@@ -119,19 +119,19 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleMulti() {
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false);
store.add(anySub);
assertTrue(store.matches("finance").contains(anySub));
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false);
store.add(financeAnySub);
assertTrue(store.matches("finance").containsAll(Arrays.asList(financeAnySub, anySub)));
}
@Test
public void testMatchingDeepMulti_one_layer() {
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false);
store.add(anySub);
store.add(financeAnySub);
@@ -143,7 +143,7 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchingDeepMulti_two_layer() {
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", AbstractMessage.QOSType.MOST_ONE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", AbstractMessage.QOSType.MOST_ONE, false);
store.add(financeAnySub);
//Verify
@@ -152,18 +152,18 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleSingle() {
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(anySub);
assertTrue(store.matches("finance").contains(anySub));
Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", AbstractMessage.QOSType.MOST_ONE);
Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(financeOne);
assertTrue(store.matches("finance/stock").contains(financeOne));
}
@Test
public void testMatchManySingle() {
Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", AbstractMessage.QOSType.MOST_ONE);
Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(manySub);
//verify
@@ -173,9 +173,9 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSlashSingle() {
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", AbstractMessage.QOSType.MOST_ONE);
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(slashPlusSub);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(anySub);
//Verify
@@ -187,10 +187,10 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchManyDeepSingle() {
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE);
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE, false);
store.add(slashPlusSub);
Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_1", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE);
Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_1", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(slashPlusDeepSub);
//Verify
@@ -199,7 +199,7 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleMulti_allTheTree() {
store.add(new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE));
store.add(new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false));
assertFalse(store.matches("finance").isEmpty());
assertFalse(store.matches("finance/ibm").isEmpty());
}
@@ -207,7 +207,7 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleMulti_zeroLevel() {
//check MULTI in case of zero level match
store.add(new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE));
store.add(new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false));
assertFalse(store.matches("finance").isEmpty());
}
@@ -215,7 +215,7 @@ public class SubscriptionsStoreTest {
@Test
public void testRemoveClientSubscriptions_existingClientID() {
String cliendID = "FAKE_CLID_1";
store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE));
store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false));
//Exercise
store.removeForClient(cliendID);
@@ -227,7 +227,7 @@ public class SubscriptionsStoreTest {
@Test
public void testRemoveClientSubscriptions_notexistingClientID() {
String cliendID = "FAKE_CLID_1";
store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE));
store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false));
//Exercise
store.removeForClient("FAKE_CLID_2");
parser/src/test/java/org/dna/mqtt/moquette/proto/PubAckDecoderTest.java
+1 -1
View File
@@ -38,7 +38,7 @@ public class PubAckDecoderTest {
assertNotNull(m_mockProtoDecoder.getMessage());
assertEquals(MessageDecoder.OK, res);
assertEquals(messageId, m_mockProtoDecoder.getMessage().getMessageID());
assertEquals(messageId, m_mockProtoDecoder.getMessage().getMessageID().intValue());
assertEquals(AbstractMessage.PUBACK, m_mockProtoDecoder.getMessage().getMessageType());
}
parser/src/test/java/org/dna/mqtt/moquette/proto/SubAckDecoderTest.java
+2 -2
View File
@@ -39,7 +39,7 @@ public class SubAckDecoderTest {
//Verify
assertEquals(MessageDecoderResult.OK, res);
assertEquals(0xAABB, m_mockProtoDecoder.getMessage().getMessageID());
assertEquals(0xAABB, m_mockProtoDecoder.getMessage().getMessageID().intValue());
List<QOSType> qoses = m_mockProtoDecoder.getMessage().types();
assertEquals(3, qoses.size());
assertEquals(QOSType.LEAST_ONE, qoses.get(0));
@@ -64,7 +64,7 @@ public class SubAckDecoderTest {
assertNotNull(m_mockProtoDecoder.getMessage());
assertEquals(MessageDecoder.OK, res);
SubAckMessage message = (SubAckMessage) m_mockProtoDecoder.getMessage();
assertEquals(0x0A, message.getMessageID());
assertEquals(0x0A, message.getMessageID().intValue());
assertEquals(1, message.types().size());
assertEquals(AbstractMessage.QOSType.LEAST_ONE, message.types().get(0));
}
parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java
+4 -4
View File
@@ -86,7 +86,7 @@ public class MQTTHandlerTest {
//Verify
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode);
verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
any(AbstractMessage.QOSType.class), anyBoolean());
any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));
}
@@ -99,8 +99,8 @@ public class MQTTHandlerTest {
(byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName));
IMessaging mockedMessaging = mock(IMessaging.class);
m_session.setAttribute(MQTTHandler.ATTR_CLIENTID, "fakeID");
m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID");
m_session.setAttribute(Constants.CLEAN_SESSION, false);
//Exercise
m_handler.setMessaging(mockedMessaging);
@@ -108,7 +108,7 @@ public class MQTTHandlerTest {
//Verify
ArgumentCaptor<AbstractMessage.QOSType> argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class);
verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture());
verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture(), eq(false));
assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE, argument.getValue());
//TODO verify the message ACK and corresponding QoS!!
}
parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java
+25
View File
@@ -6,6 +6,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.dna.mqtt.moquette.client.Client;
import org.dna.mqtt.moquette.client.IPublishCallback;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.junit.After;
import static org.junit.Assert.*;
import org.junit.Before;
@@ -313,5 +314,29 @@ public class ServerIntegrationTest {
client.close();
client.shutdown();
}
/*-------------- QoS 1 tests ----*/
@Test
public void testPublishWithQoS1() throws IOException, InterruptedException {
// startServer();
Client client = new Client("localhost", Server.PORT);
// Client client = new Client("test.mosquitto.org", 1883);
client.connect();
client.subscribe("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
received = true;
}
});
client.publish("/topic", "Test my payload".getBytes(), AbstractMessage.QOSType.LEAST_ONE, false);
client.close();
client.shutdown();
assertTrue(received);
}
}