Merge branch 'protocol_processor_extraction'
Splitted lmax messsaging part from protocol processing part
This commit is contained in:
selva.andre
@@ -1,11 +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.messaging.spi.impl.events.PubAckEvent;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
|
||||
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
|
||||
|
||||
/**
|
||||
* Interface to the underling messaging system used to publish, subscribe.
|
||||
@@ -18,7 +14,5 @@ public interface IMessaging {
|
||||
|
||||
void disconnect(IoSession session);
|
||||
|
||||
void republishStored(String clientID);
|
||||
|
||||
void handleProtocolMessage(IoSession session, AbstractMessage msg);
|
||||
}
|
||||
|
||||
@@ -1,21 +0,0 @@
|
||||
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;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public interface INotifier {
|
||||
|
||||
public void notify(NotifyEvent evt/*String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained*/);
|
||||
|
||||
public void disconnect(IoSession session);
|
||||
|
||||
void sendPubAck(PubAckEvent evt);
|
||||
|
||||
}
|
||||
+129
@@ -0,0 +1,129 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl;
|
||||
|
||||
import java.util.ArrayList;
|
||||
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.subscriptions.Subscription;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
*/
|
||||
public class MemoryStorageService implements IStorageService {
|
||||
|
||||
private Map<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<String, Set<Subscription>>();
|
||||
private Map<String, HawtDBStorageService.StoredMessage> m_retainedStore = new HashMap<String, HawtDBStorageService.StoredMessage>();
|
||||
private Map<String, List<PublishEvent>> m_persistentMessageStore = new HashMap<String, List<PublishEvent>>();
|
||||
private Map<String, PublishEvent> m_inflightStore = new HashMap<String, PublishEvent>();
|
||||
private Map<String, PublishEvent> m_qos2Store = new HashMap<String, PublishEvent>();
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class);
|
||||
|
||||
public void initStore() {
|
||||
//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) {
|
||||
//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));
|
||||
}
|
||||
}
|
||||
|
||||
public Collection<HawtDBStorageService.StoredMessage> searchMatching(IMatchingCondition condition) {
|
||||
LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size());
|
||||
|
||||
List<HawtDBStorageService.StoredMessage> results = new ArrayList<HawtDBStorageService.StoredMessage>();
|
||||
|
||||
for (Map.Entry<String, HawtDBStorageService.StoredMessage> entry : m_retainedStore.entrySet()) {
|
||||
HawtDBStorageService.StoredMessage storedMsg = entry.getValue();
|
||||
if (condition.match(entry.getKey())) {
|
||||
results.add(storedMsg);
|
||||
}
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
public void storePublishForFuture(PublishEvent evt) {
|
||||
LOG.debug("storePublishForFuture store evt " + evt);
|
||||
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);
|
||||
}
|
||||
|
||||
public List<PublishEvent> retrivePersistedPublishes(String clientID) {
|
||||
return m_persistentMessageStore.get(clientID);
|
||||
}
|
||||
|
||||
public void cleanPersistedPublishes(String clientID) {
|
||||
m_persistentMessageStore.remove(clientID);
|
||||
}
|
||||
|
||||
public void cleanInFlight(String msgID) {
|
||||
m_inflightStore.remove(msgID);
|
||||
}
|
||||
|
||||
public void addInFlight(PublishEvent evt, String publishKey) {
|
||||
m_inflightStore.put(publishKey, evt);
|
||||
}
|
||||
|
||||
public void addNewSubscription(Subscription newSubscription, String clientID) {
|
||||
if (!m_persistentSubscriptions.containsKey(clientID)) {
|
||||
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
|
||||
}
|
||||
|
||||
Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
|
||||
if (!subs.contains(newSubscription)) {
|
||||
subs.add(newSubscription);
|
||||
m_persistentSubscriptions.put(clientID, subs);
|
||||
}
|
||||
}
|
||||
|
||||
public void removeAllSubscriptions(String clientID) {
|
||||
m_persistentSubscriptions.remove(clientID);
|
||||
}
|
||||
|
||||
public List<Subscription> retrieveAllSubscriptions() {
|
||||
List<Subscription> allSubscriptions = new ArrayList<Subscription>();
|
||||
for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
|
||||
allSubscriptions.addAll(entry.getValue());
|
||||
}
|
||||
return allSubscriptions;
|
||||
}
|
||||
|
||||
public void close() {
|
||||
//To change body of implemented methods use File | Settings | File Templates.
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,406 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import org.apache.mina.core.session.IdleStatus;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
|
||||
import org.dna.mqtt.moquette.proto.PubCompMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.PubRecMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.PubRelMessage;
|
||||
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.server.ConnectionDescriptor;
|
||||
import org.dna.mqtt.moquette.server.Constants;
|
||||
import org.dna.mqtt.moquette.server.IAuthenticator;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* Class responsible to handle the logic of MQTT protocol it's the director of
|
||||
* the protocol execution.
|
||||
*
|
||||
* Used by the front facing class SimpleMessaging.
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
class ProtocolProcessor {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
|
||||
|
||||
private Map<String, ConnectionDescriptor> m_clientIDs;
|
||||
private SubscriptionsStore subscriptions;
|
||||
private IStorageService m_storageService;
|
||||
private IAuthenticator m_authenticator;
|
||||
|
||||
ProtocolProcessor() {
|
||||
|
||||
}
|
||||
|
||||
void init(Map<String, ConnectionDescriptor> clientIDs, SubscriptionsStore subscriptions,
|
||||
IStorageService storageService) {
|
||||
m_clientIDs = clientIDs;
|
||||
this.subscriptions = subscriptions;
|
||||
m_storageService = storageService;
|
||||
}
|
||||
|
||||
void processConnect(IoSession session, ConnectMessage msg) {
|
||||
if (msg.getProcotolVersion() != 0x03) {
|
||||
ConnAckMessage badProto = new ConnAckMessage();
|
||||
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
|
||||
session.write(badProto);
|
||||
session.close(false);
|
||||
return;
|
||||
}
|
||||
|
||||
if (msg.getClientID() == null || msg.getClientID().length() > 23) {
|
||||
ConnAckMessage okResp = new ConnAckMessage();
|
||||
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
|
||||
session.write(okResp);
|
||||
return;
|
||||
}
|
||||
|
||||
//if an old client with the same ID already exists close its session.
|
||||
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(Constants.CLEAN_SESSION);
|
||||
if (cleanSession) {
|
||||
//cleanup topic subscriptions
|
||||
processRemoveAllSubscriptions(msg.getClientID());
|
||||
}
|
||||
|
||||
m_clientIDs.get(msg.getClientID()).getSession().close(false);
|
||||
}
|
||||
|
||||
ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession());
|
||||
m_clientIDs.put(msg.getClientID(), connDescr);
|
||||
|
||||
int keepAlive = msg.getKeepAlive();
|
||||
LOG.debug(String.format("Connect with keepAlive %d s", keepAlive));
|
||||
session.setAttribute("keepAlive", keepAlive);
|
||||
session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
|
||||
//used to track the client in the subscription and publishing phases.
|
||||
session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID());
|
||||
|
||||
session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f));
|
||||
|
||||
//Handle will flag
|
||||
if (msg.isWillFlag()) {
|
||||
AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()];
|
||||
PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, msg.getWillMessage().getBytes(),
|
||||
msg.isWillRetain(), msg.getClientID(), session);
|
||||
processPublish(pubEvt);
|
||||
}
|
||||
|
||||
//handle user authentication
|
||||
if (msg.isUserFlag()) {
|
||||
String pwd = null;
|
||||
if (msg.isPasswordFlag()) {
|
||||
pwd = msg.getPassword();
|
||||
}
|
||||
if (!m_authenticator.checkValid(msg.getUsername(), pwd)) {
|
||||
ConnAckMessage okResp = new ConnAckMessage();
|
||||
okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
|
||||
session.write(okResp);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
subscriptions.activate(msg.getClientID());
|
||||
|
||||
//handle clean session flag
|
||||
if (msg.isCleanSession()) {
|
||||
//remove all prev subscriptions
|
||||
//cleanup topic subscriptions
|
||||
processRemoveAllSubscriptions(msg.getClientID());
|
||||
}
|
||||
|
||||
ConnAckMessage okResp = new ConnAckMessage();
|
||||
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
|
||||
session.write(okResp);
|
||||
|
||||
if (!msg.isCleanSession()) {
|
||||
//force the republish of stored QoS1 and QoS2
|
||||
republishStored(msg.getClientID());
|
||||
}
|
||||
}
|
||||
|
||||
private void republishStored(String clientID) {
|
||||
LOG.debug("republishStored invoked");
|
||||
List<PublishEvent> publishedEvents = m_storageService.retrivePersistedPublishes(clientID);
|
||||
if (publishedEvents == null) {
|
||||
LOG.debug("republishStored, no stored publish events");
|
||||
return;
|
||||
}
|
||||
|
||||
for (PublishEvent pubEvt : publishedEvents) {
|
||||
sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
|
||||
pubEvt.getMessage(), false, pubEvt.getMessageID());
|
||||
}
|
||||
}
|
||||
|
||||
private void processRemoveAllSubscriptions(String clientID) {
|
||||
LOG.debug("processRemoveAllSubscriptions invoked");
|
||||
subscriptions.removeForClient(clientID);
|
||||
|
||||
//remove also the messages stored of type QoS1/2
|
||||
m_storageService.cleanPersistedPublishes(clientID);
|
||||
}
|
||||
|
||||
protected void processPublish(PublishEvent evt) {
|
||||
LOG.debug("processPublish invoked with " + evt);
|
||||
final String topic = evt.getTopic();
|
||||
final AbstractMessage.QOSType qos = evt.getQos();
|
||||
final byte[] message = evt.getMessage();
|
||||
boolean retain = evt.isRetain();
|
||||
|
||||
String publishKey = null;
|
||||
if (qos == AbstractMessage.QOSType.LEAST_ONE) {
|
||||
//store the temporary message
|
||||
publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID());
|
||||
m_storageService.addInFlight(evt, publishKey);
|
||||
} else if (qos == AbstractMessage.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());
|
||||
}
|
||||
|
||||
publish2Subscribers(topic, qos, message, retain, evt.getMessageID());
|
||||
|
||||
if (qos == AbstractMessage.QOSType.LEAST_ONE) {
|
||||
if (publishKey == null) {
|
||||
throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + evt);
|
||||
}
|
||||
m_storageService.cleanInFlight(publishKey);
|
||||
sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
|
||||
}
|
||||
|
||||
if (retain) {
|
||||
m_storageService.storeRetained(topic, message, qos);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) {
|
||||
for (final Subscription sub : subscriptions.matches(topic)) {
|
||||
if (qos == AbstractMessage.QOSType.MOST_ONE) {
|
||||
//QoS 0
|
||||
sendPublish(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 {
|
||||
//if QoS 2 then store it in temp memory
|
||||
if (qos ==AbstractMessage.QOSType.EXACTLY_ONCE) {
|
||||
String publishKey = String.format("%s%d", sub.getClientId(), messageID);
|
||||
PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null);
|
||||
m_storageService.addInFlight(newPublishEvt, publishKey);
|
||||
}
|
||||
//publish
|
||||
sendPublish(sub.getClientId(), topic, qos, message, false);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, byte[] message, boolean retained) {
|
||||
sendPublish(clientId, topic, qos, message, retained, 0);
|
||||
}
|
||||
|
||||
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, byte[] message, boolean retained, int messageID) {
|
||||
LOG.debug("notify invoked with event ");
|
||||
PublishMessage pubMessage = new PublishMessage();
|
||||
pubMessage.setRetainFlag(retained);
|
||||
pubMessage.setTopicName(topic);
|
||||
pubMessage.setQos(qos);
|
||||
pubMessage.setPayload(message);
|
||||
if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) {
|
||||
pubMessage.setMessageID(messageID);
|
||||
}
|
||||
|
||||
try {
|
||||
if (m_clientIDs == null) {
|
||||
throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!");
|
||||
}
|
||||
LOG.debug("clientIDs are " + m_clientIDs);
|
||||
if (m_clientIDs.get(clientId) == null) {
|
||||
throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs));
|
||||
}
|
||||
LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession());
|
||||
m_clientIDs.get(clientId).getSession().write(pubMessage);
|
||||
}catch(Throwable t) {
|
||||
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 sendPubAck(PubAckEvent evt) {
|
||||
LOG.debug("sendPubAck invoked");
|
||||
|
||||
String clientId = evt.getClientID();
|
||||
|
||||
PubAckMessage pubAckMessage = new PubAckMessage();
|
||||
pubAckMessage.setMessageID(evt.getMessageId());
|
||||
|
||||
try {
|
||||
if (m_clientIDs == null) {
|
||||
throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!");
|
||||
}
|
||||
LOG.debug("clientIDs are " + m_clientIDs);
|
||||
if (m_clientIDs.get(clientId) == null) {
|
||||
throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs));
|
||||
}
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish
|
||||
* to all interested subscribers.
|
||||
* */
|
||||
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 AbstractMessage.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);
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
void processPubRec(String clientID, int messageID) {
|
||||
//once received a PUBREC reply with a PUBREL(messageID)
|
||||
LOG.debug(String.format("processPubRec invoked for clientID %s ad messageID %d", clientID, messageID));
|
||||
PubRelMessage pubRelMessage = new PubRelMessage();
|
||||
pubRelMessage.setMessageID(messageID);
|
||||
|
||||
m_clientIDs.get(clientID).getSession().write(pubRelMessage);
|
||||
}
|
||||
|
||||
void processPubComp(String clientID, int messageID) {
|
||||
//once received the PUBCOMP then remove the message from the temp memory
|
||||
String publishKey = String.format("%s%d", clientID, messageID);
|
||||
m_storageService.cleanInFlight(publishKey);
|
||||
}
|
||||
|
||||
void processDisconnect(IoSession session, String clientID, boolean cleanSession) throws InterruptedException {
|
||||
if (cleanSession) {
|
||||
//cleanup topic subscriptions
|
||||
processRemoveAllSubscriptions(clientID);
|
||||
}
|
||||
// m_notifier.disconnect(evt.getSession());
|
||||
m_clientIDs.remove(clientID);
|
||||
session.close(true);
|
||||
|
||||
//de-activate the subscriptions for this ClientID
|
||||
// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
|
||||
subscriptions.deactivate(clientID);
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove the clientID from topic subscription, if not previously subscribed,
|
||||
* doesn't reply any error
|
||||
*/
|
||||
void processUnsubscribe(IoSession session, String clientID, List<String> topics, int messageID) {
|
||||
LOG.debug("processSubscribe invoked");
|
||||
|
||||
for (String topic : topics) {
|
||||
subscriptions.removeSubscription(topic, clientID);
|
||||
}
|
||||
//ack the client
|
||||
UnsubAckMessage ackMessage = new UnsubAckMessage();
|
||||
ackMessage.setMessageID(messageID);
|
||||
|
||||
LOG.info("replying with UnsubAck to MSG ID {0}", messageID);
|
||||
session.write(ackMessage);
|
||||
}
|
||||
|
||||
|
||||
void processSubscribe(IoSession session, SubscribeMessage msg, String clientID, boolean cleanSession) {
|
||||
LOG.debug("processSubscribe invoked");
|
||||
|
||||
for (SubscribeMessage.Couple req : msg.subscriptions()) {
|
||||
AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
|
||||
Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession);
|
||||
subscribeSingleTopic(newSubscription, req.getTopic());
|
||||
}
|
||||
|
||||
//ack the client
|
||||
SubAckMessage ackMessage = new SubAckMessage();
|
||||
ackMessage.setMessageID(msg.getMessageID());
|
||||
|
||||
//TODO by now it handles only QoS 0 messages
|
||||
for (int i = 0; i < msg.subscriptions().size(); i++) {
|
||||
ackMessage.addType(AbstractMessage.QOSType.MOST_ONE);
|
||||
}
|
||||
LOG.info("replying with SubAct to MSG ID " + msg.getMessageID());
|
||||
session.write(ackMessage);
|
||||
}
|
||||
|
||||
private void subscribeSingleTopic(Subscription newSubscription, final String topic) {
|
||||
subscriptions.add(newSubscription);
|
||||
|
||||
//scans retained messages to be published to the new subscription
|
||||
Collection<HawtDBStorageService.StoredMessage> messages = m_storageService.searchMatching(new IMatchingCondition() {
|
||||
public boolean match(String key) {
|
||||
return SubscriptionsStore.matchTopics(key, topic);
|
||||
}
|
||||
});
|
||||
|
||||
for (HawtDBStorageService.StoredMessage storedMsg : messages) {
|
||||
//fire the as retained the message
|
||||
LOG.debug("send publish message for topic " + topic);
|
||||
sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
+13
-387
@@ -4,27 +4,21 @@ import com.lmax.disruptor.BatchEventProcessor;
|
||||
import com.lmax.disruptor.EventHandler;
|
||||
import com.lmax.disruptor.RingBuffer;
|
||||
import com.lmax.disruptor.SequenceBarrier;
|
||||
import java.util.Collection;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import org.apache.mina.core.session.IdleStatus;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService.StoredMessage;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
|
||||
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;
|
||||
import org.dna.mqtt.moquette.server.Constants;
|
||||
import org.dna.mqtt.moquette.server.IAuthenticator;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
@@ -44,19 +38,17 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
|
||||
|
||||
private RingBuffer<ValueEvent> m_ringBuffer;
|
||||
|
||||
// private INotifier m_notifier;
|
||||
|
||||
private IStorageService m_storageService;
|
||||
|
||||
Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<String, ConnectionDescriptor>();
|
||||
|
||||
private IAuthenticator m_authenticator;
|
||||
|
||||
private ExecutorService m_executor;
|
||||
BatchEventProcessor<ValueEvent> m_eventProcessor;
|
||||
|
||||
private static SimpleMessaging INSTANCE;
|
||||
|
||||
private ProtocolProcessor m_processor = new ProtocolProcessor();
|
||||
|
||||
private SimpleMessaging() {
|
||||
}
|
||||
|
||||
@@ -81,11 +73,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
|
||||
}
|
||||
|
||||
|
||||
/* public void setNotifier(INotifier notifier) {
|
||||
m_notifier= notifier;
|
||||
} */
|
||||
|
||||
|
||||
private void disruptorPublish(MessagingEvent msgEvent) {
|
||||
long sequence = m_ringBuffer.next();
|
||||
ValueEvent event = m_ringBuffer.get(sequence);
|
||||
@@ -100,30 +87,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
|
||||
disruptorPublish(new DisconnectEvent(session));
|
||||
}
|
||||
|
||||
//method used by hte Notifier to re-put an event on the inbound queue
|
||||
private void refill(MessagingEvent evt) {
|
||||
disruptorPublish(evt);
|
||||
}
|
||||
|
||||
public void republishStored(String clientID) {
|
||||
//create the event to push
|
||||
LOG.debug("republishStored invoked to publish soterd messages for clientID " + clientID);
|
||||
disruptorPublish(new RepublishEvent(clientID));
|
||||
}
|
||||
|
||||
public void handleProtocolMessage(IoSession session, AbstractMessage msg) {
|
||||
disruptorPublish(new ProtocolEvent(session, msg));
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* NOT SAFE Method, to be removed because used only in tests
|
||||
*/
|
||||
protected SubscriptionsStore getSubscriptions() {
|
||||
return subscriptions;
|
||||
}
|
||||
|
||||
|
||||
public void stop() {
|
||||
disruptorPublish(new StopEvent());
|
||||
}
|
||||
@@ -132,20 +99,18 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
|
||||
MessagingEvent evt = t.getEvent();
|
||||
LOG.debug("onEvent processing messaging event " + evt);
|
||||
if (evt instanceof PublishEvent) {
|
||||
processPublish((PublishEvent) evt);
|
||||
m_processor.processPublish((PublishEvent) evt);
|
||||
} else if (evt instanceof StopEvent) {
|
||||
processStop();
|
||||
} else if (evt instanceof DisconnectEvent) {
|
||||
DisconnectEvent disEvt = (DisconnectEvent) evt;
|
||||
String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID);
|
||||
processDisconnect(disEvt.getSession(), clientID);
|
||||
} else if (evt instanceof RepublishEvent) {
|
||||
processRepublish((RepublishEvent) evt);
|
||||
m_processor.processDisconnect(disEvt.getSession(), clientID, false);
|
||||
} else if (evt instanceof ProtocolEvent) {
|
||||
IoSession session = ((ProtocolEvent) evt).getSession();
|
||||
AbstractMessage message = ((ProtocolEvent) evt).getMessage();
|
||||
if (message instanceof ConnectMessage) {
|
||||
processConnect(session, (ConnectMessage) message);
|
||||
m_processor.processConnect(session, (ConnectMessage) message);
|
||||
} else if (message instanceof PublishMessage) {
|
||||
PublishMessage pubMsg = (PublishMessage) message;
|
||||
PublishEvent pubEvt;
|
||||
@@ -158,38 +123,34 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
|
||||
} else {
|
||||
pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session);
|
||||
}
|
||||
processPublish(pubEvt);
|
||||
m_processor.processPublish(pubEvt);
|
||||
} else if (message instanceof DisconnectMessage) {
|
||||
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
|
||||
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
|
||||
if (cleanSession) {
|
||||
//cleanup topic subscriptions
|
||||
processRemoveAllSubscriptions(clientID);
|
||||
}
|
||||
|
||||
//close the TCP connection
|
||||
//session.close(true);
|
||||
processDisconnect(session, clientID);
|
||||
m_processor.processDisconnect(session, clientID, cleanSession);
|
||||
} else if (message instanceof UnsubscribeMessage) {
|
||||
UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
|
||||
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
|
||||
processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID());
|
||||
m_processor.processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID());
|
||||
} else if (message instanceof SubscribeMessage) {
|
||||
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
|
||||
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
|
||||
processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession);
|
||||
m_processor.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);
|
||||
m_processor.processPubRel(clientID, messageID);
|
||||
} else if (message instanceof PubRecMessage) {
|
||||
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
|
||||
int messageID = ((PubRecMessage) message).getMessageID();
|
||||
processPubRec(clientID, messageID);
|
||||
m_processor.processPubRec(clientID, messageID);
|
||||
} else if (message instanceof PubCompMessage) {
|
||||
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
|
||||
int messageID = ((PubCompMessage) message).getMessageID();
|
||||
processPubComp(clientID, messageID);
|
||||
m_processor.processPubComp(clientID, messageID);
|
||||
} else {
|
||||
throw new RuntimeException("Illegal message received " + message);
|
||||
}
|
||||
@@ -204,265 +165,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
|
||||
m_storageService.initStore();
|
||||
|
||||
subscriptions.init(m_storageService);
|
||||
m_processor.init(m_clientIDs, subscriptions, m_storageService);
|
||||
}
|
||||
|
||||
|
||||
protected void processConnect(IoSession session, ConnectMessage msg) {
|
||||
if (msg.getProcotolVersion() != 0x03) {
|
||||
ConnAckMessage badProto = new ConnAckMessage();
|
||||
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
|
||||
session.write(badProto);
|
||||
session.close(false);
|
||||
return;
|
||||
}
|
||||
|
||||
if (msg.getClientID() == null || msg.getClientID().length() > 23) {
|
||||
ConnAckMessage okResp = new ConnAckMessage();
|
||||
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
|
||||
session.write(okResp);
|
||||
return;
|
||||
}
|
||||
|
||||
//if an old client with the same ID already exists close its session.
|
||||
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(Constants.CLEAN_SESSION);
|
||||
if (cleanSession) {
|
||||
//cleanup topic subscriptions
|
||||
processRemoveAllSubscriptions(msg.getClientID());
|
||||
}
|
||||
|
||||
m_clientIDs.get(msg.getClientID()).getSession().close(false);
|
||||
}
|
||||
|
||||
ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession());
|
||||
m_clientIDs.put(msg.getClientID(), connDescr);
|
||||
|
||||
int keepAlive = msg.getKeepAlive();
|
||||
LOG.debug(String.format("Connect with keepAlive %d s", keepAlive));
|
||||
session.setAttribute("keepAlive", keepAlive);
|
||||
session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
|
||||
//used to track the client in the subscription and publishing phases.
|
||||
session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID());
|
||||
|
||||
session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f));
|
||||
|
||||
//Handle will flag
|
||||
if (msg.isWillFlag()) {
|
||||
QOSType willQos = QOSType.values()[msg.getWillQos()];
|
||||
PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, msg.getWillMessage().getBytes(),
|
||||
msg.isWillRetain(), msg.getClientID(), session);
|
||||
processPublish(pubEvt);
|
||||
}
|
||||
|
||||
//handle user authentication
|
||||
if (msg.isUserFlag()) {
|
||||
String pwd = null;
|
||||
if (msg.isPasswordFlag()) {
|
||||
pwd = msg.getPassword();
|
||||
}
|
||||
if (!m_authenticator.checkValid(msg.getUsername(), pwd)) {
|
||||
ConnAckMessage okResp = new ConnAckMessage();
|
||||
okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
|
||||
session.write(okResp);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
subscriptions.activate(msg.getClientID());
|
||||
|
||||
//handle clean session flag
|
||||
if (msg.isCleanSession()) {
|
||||
//remove all prev subscriptions
|
||||
//cleanup topic subscriptions
|
||||
processRemoveAllSubscriptions(msg.getClientID());
|
||||
} else {
|
||||
//force the republish of stored QoS1 and QoS2
|
||||
republishStored(msg.getClientID());
|
||||
}
|
||||
|
||||
ConnAckMessage okResp = new ConnAckMessage();
|
||||
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
|
||||
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();
|
||||
final QOSType qos = evt.getQos();
|
||||
final byte[] message = evt.getMessage();
|
||||
boolean retain = evt.isRetain();
|
||||
|
||||
String publishKey = null;
|
||||
if (qos == QOSType.LEAST_ONE) {
|
||||
//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());
|
||||
}
|
||||
|
||||
publish2Subscribers(topic, qos, message, retain, evt.getMessageID());
|
||||
|
||||
if (qos == QOSType.LEAST_ONE) {
|
||||
if (publishKey == null) {
|
||||
throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + evt);
|
||||
}
|
||||
m_storageService.cleanInFlight(publishKey);
|
||||
sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
|
||||
}
|
||||
|
||||
if (retain) {
|
||||
m_storageService.storeRetained(topic, message, qos);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 {
|
||||
//if QoS 2 then store it in temp memory
|
||||
if (qos ==QOSType.EXACTLY_ONCE) {
|
||||
String publishKey = String.format("%s%d", sub.getClientId(), messageID);
|
||||
PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null);
|
||||
m_storageService.addInFlight(newPublishEvt, publishKey);
|
||||
}
|
||||
//publish
|
||||
notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void processPubRec(String clientID, int messageID) {
|
||||
//once received a PUBREC reply with a PUBREL(messageID)
|
||||
LOG.debug(String.format("processPubRec invoked for clientID %s ad messageID %d", clientID, messageID));
|
||||
PubRelMessage pubRelMessage = new PubRelMessage();
|
||||
pubRelMessage.setMessageID(messageID);
|
||||
|
||||
m_clientIDs.get(clientID).getSession().write(pubRelMessage);
|
||||
}
|
||||
|
||||
private void processPubComp(String clientID, int messageID) {
|
||||
//once received the PUBCOMP then remove the message from the temp memory
|
||||
String publishKey = String.format("%s%d", clientID, messageID);
|
||||
m_storageService.cleanInFlight(publishKey);
|
||||
}
|
||||
|
||||
private void subscribeSingleTopic(Subscription newSubscription, final String topic) {
|
||||
subscriptions.add(newSubscription);
|
||||
|
||||
//scans retained messages to be published to the new subscription
|
||||
Collection<StoredMessage> messages = m_storageService.searchMatching(new IMatchingCondition() {
|
||||
public boolean match(String key) {
|
||||
return SubscriptionsStore.matchTopics(key, topic);
|
||||
}
|
||||
});
|
||||
|
||||
for (StoredMessage storedMsg : messages) {
|
||||
//fire the as retained the message
|
||||
LOG.debug("Inserting NotifyEvent into outbound for topic " + topic);
|
||||
notify(new NotifyEvent(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true));
|
||||
}
|
||||
}
|
||||
|
||||
protected void processSubscribe(IoSession session, SubscribeMessage msg, String clientID, boolean cleanSession) {
|
||||
LOG.debug("processSubscribe invoked");
|
||||
|
||||
for (SubscribeMessage.Couple req : msg.subscriptions()) {
|
||||
QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
|
||||
Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession);
|
||||
subscribeSingleTopic(newSubscription, req.getTopic());
|
||||
}
|
||||
|
||||
//ack the client
|
||||
SubAckMessage ackMessage = new SubAckMessage();
|
||||
ackMessage.setMessageID(msg.getMessageID());
|
||||
|
||||
//TODO by now it handles only QoS 0 messages
|
||||
for (int i = 0; i < msg.subscriptions().size(); i++) {
|
||||
ackMessage.addType(QOSType.MOST_ONE);
|
||||
}
|
||||
LOG.info("replying with SubAct to MSG ID " + msg.getMessageID());
|
||||
session.write(ackMessage);
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove the clientID from topic subscription, if not previously subscribed,
|
||||
* doesn't reply any error
|
||||
*/
|
||||
protected void processUnsubscribe(IoSession session, String clientID, List<String> topics, int messageID) {
|
||||
LOG.debug("processSubscribe invoked");
|
||||
|
||||
for (String topic : topics) {
|
||||
subscriptions.removeSubscription(topic, clientID);
|
||||
}
|
||||
//ack the client
|
||||
UnsubAckMessage ackMessage = new UnsubAckMessage();
|
||||
ackMessage.setMessageID(messageID);
|
||||
|
||||
LOG.info("replying with UnsubAck to MSG ID {0}", messageID);
|
||||
session.write(ackMessage);
|
||||
}
|
||||
|
||||
protected void processRemoveAllSubscriptions(String clientID) {
|
||||
LOG.debug("processRemoveAllSubscriptions invoked");
|
||||
subscriptions.removeForClient(clientID);
|
||||
|
||||
//remove also the messages stored of type QoS1/2
|
||||
m_storageService.cleanPersistedPublishes(clientID);
|
||||
}
|
||||
|
||||
private void processDisconnect(IoSession session, String clientID) throws InterruptedException {
|
||||
// m_notifier.disconnect(evt.getSession());
|
||||
m_clientIDs.remove(clientID);
|
||||
session.close(true);
|
||||
|
||||
//de-activate the subscriptions for this ClientID
|
||||
// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
|
||||
subscriptions.deactivate(clientID);
|
||||
}
|
||||
|
||||
private void processStop() {
|
||||
LOG.debug("processStop invoked");
|
||||
m_storageService.close();
|
||||
@@ -470,84 +176,4 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
|
||||
// m_eventProcessor.halt();
|
||||
m_executor.shutdown();
|
||||
}
|
||||
|
||||
private void processRepublish(RepublishEvent evt) throws InterruptedException {
|
||||
LOG.debug("processRepublish invoked");
|
||||
List<PublishEvent> publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID());
|
||||
if (publishedEvents == null) {
|
||||
LOG.debug("processRepublish, no stored publish events");
|
||||
return;
|
||||
}
|
||||
|
||||
for (PublishEvent pubEvt : publishedEvents) {
|
||||
notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
|
||||
pubEvt.getMessage(), false, pubEvt.getMessageID()));
|
||||
}
|
||||
}
|
||||
|
||||
private void notify(NotifyEvent evt) {
|
||||
LOG.debug("notify invoked with event " + evt);
|
||||
String clientId = evt.getClientId();
|
||||
PublishMessage pubMessage = new PublishMessage();
|
||||
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());
|
||||
}
|
||||
|
||||
try {
|
||||
if (m_clientIDs == null) {
|
||||
throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!");
|
||||
}
|
||||
LOG.debug("clientIDs are " + m_clientIDs);
|
||||
if (m_clientIDs.get(clientId) == null) {
|
||||
throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs));
|
||||
}
|
||||
LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession());
|
||||
m_clientIDs.get(clientId).getSession().write(pubMessage);
|
||||
}catch(Throwable t) {
|
||||
LOG.error(null, t);
|
||||
}
|
||||
}
|
||||
|
||||
private void sendPubAck(PubAckEvent evt) {
|
||||
LOG.debug("sendPubAck invoked");
|
||||
|
||||
String clientId = evt.getClientID();
|
||||
|
||||
PubAckMessage pubAckMessage = new PubAckMessage();
|
||||
pubAckMessage.setMessageID(evt.getMessageId());
|
||||
|
||||
try {
|
||||
if (m_clientIDs == null) {
|
||||
throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!");
|
||||
}
|
||||
LOG.debug("clientIDs are " + m_clientIDs);
|
||||
if (m_clientIDs.get(clientId) == null) {
|
||||
throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs));
|
||||
}
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,69 +0,0 @@
|
||||
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
|
||||
*/
|
||||
public class NotifyEvent extends MessagingEvent {
|
||||
|
||||
private String m_clientId;
|
||||
private String m_topic;
|
||||
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) {
|
||||
m_clientId = clientId;
|
||||
m_topic = topic;
|
||||
m_qos = qos;
|
||||
m_message = message;
|
||||
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;
|
||||
}
|
||||
|
||||
public String getTopic() {
|
||||
return m_topic;
|
||||
}
|
||||
|
||||
public QOSType getQos() {
|
||||
return m_qos;
|
||||
}
|
||||
|
||||
public byte[] getMessage() {
|
||||
return m_message;
|
||||
}
|
||||
|
||||
public boolean isRetained() {
|
||||
return m_retaned;
|
||||
}
|
||||
|
||||
public int getMessageID() {
|
||||
return m_msgID;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "NotifyEvent{" +
|
||||
"m_retaned=" + m_retaned +
|
||||
", m_msgID=" + m_msgID +
|
||||
", m_qos=" + m_qos +
|
||||
", m_topic='" + m_topic + '\'' +
|
||||
", m_clientId='" + m_clientId + '\'' +
|
||||
'}';
|
||||
}
|
||||
}
|
||||
-75
@@ -1,75 +0,0 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl;
|
||||
|
||||
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.subscriptions.Subscription;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
*/
|
||||
public class DummyStorageService implements IStorageService {
|
||||
public void initStore() {
|
||||
//To change body of implemented methods use File | Settings | File Templates.
|
||||
}
|
||||
|
||||
public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) {
|
||||
//To change body of implemented methods use File | Settings | File Templates.
|
||||
}
|
||||
|
||||
public Collection<HawtDBStorageService.StoredMessage> searchMatching(IMatchingCondition condition) {
|
||||
return null; //To change body of implemented methods use File | Settings | File Templates.
|
||||
}
|
||||
|
||||
public void storePublishForFuture(PublishEvent evt) {
|
||||
//To change body of implemented methods use File | Settings | File Templates.
|
||||
}
|
||||
|
||||
public List<PublishEvent> retrivePersistedPublishes(String clientID) {
|
||||
return null; //To change body of implemented methods use File | Settings | File Templates.
|
||||
}
|
||||
|
||||
public void cleanPersistedPublishes(String clientID) {
|
||||
//To change body of implemented methods use File | Settings | File Templates.
|
||||
}
|
||||
|
||||
public void cleanInFlight(String msgID) {
|
||||
//To change body of implemented methods use File | Settings | File Templates.
|
||||
}
|
||||
|
||||
public void addInFlight(PublishEvent evt, String publishKey) {
|
||||
//To change body of implemented methods use File | Settings | File Templates.
|
||||
}
|
||||
|
||||
public void addNewSubscription(Subscription newSubscription, String clientID) {
|
||||
//To change body of implemented methods use File | Settings | File Templates.
|
||||
}
|
||||
|
||||
public void removeAllSubscriptions(String clientID) {
|
||||
//To change body of implemented methods use File | Settings | File Templates.
|
||||
}
|
||||
|
||||
public List<Subscription> retrieveAllSubscriptions() {
|
||||
return Collections.EMPTY_LIST;
|
||||
}
|
||||
|
||||
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.
|
||||
}
|
||||
}
|
||||
+142
-146
@@ -1,62 +1,52 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl;
|
||||
|
||||
import java.io.File;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import org.apache.mina.core.filterchain.IoFilter;
|
||||
import org.apache.mina.core.filterchain.IoFilterAdapter;
|
||||
import org.apache.mina.core.session.DummySession;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.apache.mina.core.write.WriteRequest;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
|
||||
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.subscriptions.Subscription;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
|
||||
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.server.ConnectionDescriptor;
|
||||
import org.dna.mqtt.moquette.server.Server;
|
||||
import org.junit.After;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Ignore;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class SimpleMessagingTest {
|
||||
|
||||
public class ProtocolProcessorTest {
|
||||
final static String FAKE_CLIENT_ID = "FAKE_123";
|
||||
final static String FAKE_TOPIC = "/news";
|
||||
SimpleMessaging messaging;
|
||||
|
||||
byte m_returnCode;
|
||||
|
||||
IoSession m_session;
|
||||
// MQTTHandler m_handler;
|
||||
byte m_returnCode;
|
||||
ConnectMessage connMsg;
|
||||
|
||||
ProtocolProcessor m_processor;
|
||||
|
||||
IStorageService m_storageService;
|
||||
SubscriptionsStore subscriptions;
|
||||
AbstractMessage m_receivedMessage;
|
||||
|
||||
@BeforeClass
|
||||
public static void beforeClass() {
|
||||
File dbFile = new File(Server.STORAGE_FILE_PATH);
|
||||
if (dbFile.exists()) {
|
||||
dbFile.delete();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@Before
|
||||
public void setUp() throws InterruptedException {
|
||||
messaging = SimpleMessaging.getInstance();
|
||||
messaging.init();
|
||||
connMsg = new ConnectMessage();
|
||||
connMsg.setProcotolVersion((byte) 0x03);
|
||||
|
||||
@@ -65,7 +55,7 @@ public class SimpleMessagingTest {
|
||||
m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
|
||||
|
||||
@Override
|
||||
public void filterWrite(NextFilter nextFilter, IoSession session,
|
||||
public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session,
|
||||
WriteRequest writeRequest) throws Exception {
|
||||
try {
|
||||
m_receivedMessage = (AbstractMessage) writeRequest.getMessage();
|
||||
@@ -81,60 +71,84 @@ public class SimpleMessagingTest {
|
||||
|
||||
//sleep to let the messaging batch processor to process the initEvent
|
||||
Thread.sleep(300);
|
||||
}
|
||||
|
||||
m_storageService = new MemoryStorageService();
|
||||
//m_storageService.initStore();
|
||||
|
||||
@After
|
||||
public void tearDown() {
|
||||
m_receivedMessage = null;
|
||||
messaging.stop();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSubscribe() {
|
||||
//Exercise
|
||||
SubscribeMessage msg = new SubscribeMessage();
|
||||
msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
|
||||
messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
|
||||
|
||||
//Verify
|
||||
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false);
|
||||
assertTrue(messaging.getSubscriptions().contains(expectedSubscription));
|
||||
subscriptions = new SubscriptionsStore();
|
||||
subscriptions.init(m_storageService);
|
||||
m_processor = new ProtocolProcessor();
|
||||
m_processor.init(new HashMap<String, ConnectionDescriptor>(), subscriptions, m_storageService);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDoubleSubscribe() {
|
||||
// SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
|
||||
SubscribeMessage msg = new SubscribeMessage();
|
||||
msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
|
||||
|
||||
messaging.getSubscriptions().clearAllSubscriptions();
|
||||
assertEquals(0, messaging.getSubscriptions().size());
|
||||
|
||||
messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
|
||||
|
||||
public void testHandleConnect_BadProtocol() {
|
||||
connMsg.setProcotolVersion((byte) 0x02);
|
||||
|
||||
//Exercise
|
||||
messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
|
||||
m_processor.processConnect(m_session, connMsg);
|
||||
|
||||
//Verify
|
||||
Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false);
|
||||
assertEquals(1, messaging.getSubscriptions().size());
|
||||
assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testConnect_badClientID() {
|
||||
connMsg.setClientID("extremely_long_clientID_greater_than_23");
|
||||
|
||||
//Exercise
|
||||
m_processor.processConnect(m_session, connMsg);
|
||||
|
||||
//Verify
|
||||
assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPublish() throws InterruptedException {
|
||||
//simulate a connect that register a clientID to an IoSession
|
||||
ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, true);
|
||||
messaging.m_clientIDs.put(FAKE_CLIENT_ID, connDescr);
|
||||
|
||||
// SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
|
||||
SubscribeMessage msg = new SubscribeMessage();
|
||||
msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
|
||||
|
||||
messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
|
||||
public void testWill() {
|
||||
connMsg.setClientID("123");
|
||||
connMsg.setWillFlag(true);
|
||||
connMsg.setWillTopic("topic");
|
||||
connMsg.setWillMessage("Topic message");
|
||||
|
||||
|
||||
//Exercise
|
||||
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null);
|
||||
messaging.processPublish(pubEvt);
|
||||
//m_handler.setMessaging(mockedMessaging);
|
||||
m_processor.processConnect(m_session, connMsg);
|
||||
|
||||
//Verify
|
||||
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode);
|
||||
//TODO verify the call
|
||||
/*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
|
||||
any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPublish() throws InterruptedException {
|
||||
final Subscription subscription = new Subscription(FAKE_CLIENT_ID,
|
||||
FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true);
|
||||
|
||||
//subscriptions.matches(topic) redefine the method to return true
|
||||
SubscriptionsStore subs = new SubscriptionsStore() {
|
||||
@Override
|
||||
public List<Subscription> matches(String topic) {
|
||||
if (topic.equals(FAKE_TOPIC)) {
|
||||
return Arrays.asList(subscription);
|
||||
} else {
|
||||
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
//simulate a connect that register a clientID to an IoSession
|
||||
Map<String, ConnectionDescriptor> connectionsMap = new HashMap<String, ConnectionDescriptor>();
|
||||
ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, subscription.isCleanSession());
|
||||
connectionsMap.put(FAKE_CLIENT_ID, connDescr);
|
||||
m_processor.init(connectionsMap, subs, m_storageService);
|
||||
|
||||
|
||||
//Exercise
|
||||
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null);
|
||||
m_processor.processPublish(pubEvt);
|
||||
|
||||
//Verify
|
||||
assertNotNull(m_receivedMessage);
|
||||
@@ -142,7 +156,40 @@ public class SimpleMessagingTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPublishOfRetainedMessage() throws Exception {
|
||||
public void testSubscribe() {
|
||||
//Exercise
|
||||
SubscribeMessage msg = new SubscribeMessage();
|
||||
msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
|
||||
m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
|
||||
|
||||
//Verify
|
||||
assertTrue(m_receivedMessage instanceof SubAckMessage);
|
||||
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false);
|
||||
assertTrue(subscriptions.contains(expectedSubscription));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDoubleSubscribe() {
|
||||
SubscribeMessage msg = new SubscribeMessage();
|
||||
msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
|
||||
|
||||
subscriptions.clearAllSubscriptions();
|
||||
assertEquals(0, subscriptions.size());
|
||||
|
||||
m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
|
||||
|
||||
//Exercise
|
||||
m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
|
||||
|
||||
//Verify
|
||||
assertEquals(1, subscriptions.size());
|
||||
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false);
|
||||
assertTrue(subscriptions.contains(expectedSubscription));
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception {
|
||||
m_session.getFilterChain().remove("MessageCatcher");
|
||||
m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
|
||||
|
||||
@@ -165,17 +212,35 @@ public class SimpleMessagingTest {
|
||||
}
|
||||
});
|
||||
|
||||
//simulate a connect that register a clientID to an IoSession
|
||||
ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, true);
|
||||
messaging.m_clientIDs.put(FAKE_CLIENT_ID, connDescr);
|
||||
// //simulate a connect that register a clientID to an IoSession
|
||||
final Subscription subscription = new Subscription(FAKE_CLIENT_ID,
|
||||
FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true);
|
||||
|
||||
//subscriptions.matches(topic) redefine the method to return true
|
||||
SubscriptionsStore subs = new SubscriptionsStore() {
|
||||
@Override
|
||||
public List<Subscription> matches(String topic) {
|
||||
if (topic.equals(FAKE_TOPIC)) {
|
||||
return Arrays.asList(subscription);
|
||||
} else {
|
||||
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
|
||||
}
|
||||
}
|
||||
};
|
||||
subs.init(new MemoryStorageService());
|
||||
|
||||
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null);
|
||||
messaging.processPublish(pubEvt);
|
||||
//simulate a connect that register a clientID to an IoSession
|
||||
Map<String, ConnectionDescriptor> connectionsMap = new HashMap<String, ConnectionDescriptor>();
|
||||
ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, subscription.isCleanSession());
|
||||
connectionsMap.put(FAKE_CLIENT_ID, connDescr);
|
||||
m_processor.init(connectionsMap, subs, m_storageService);
|
||||
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null);
|
||||
m_processor.processPublish(pubEvt);
|
||||
|
||||
//Exercise
|
||||
SubscribeMessage msg = new SubscribeMessage();
|
||||
msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#"));
|
||||
messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
|
||||
m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
|
||||
|
||||
//Verify
|
||||
assertNotNull(m_receivedMessage);
|
||||
@@ -183,73 +248,4 @@ public class SimpleMessagingTest {
|
||||
PublishMessage pubMessage = (PublishMessage) m_receivedMessage;
|
||||
assertEquals(FAKE_TOPIC, pubMessage.getTopicName());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testHandleConnect_BadProtocol() {
|
||||
connMsg.setProcotolVersion((byte) 0x02);
|
||||
|
||||
//Exercise
|
||||
messaging.processConnect(m_session, connMsg);
|
||||
|
||||
//Verify
|
||||
assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testConnect_badClientID() {
|
||||
connMsg.setClientID("extremely_long_clientID_greater_than_23");
|
||||
|
||||
//Exercise
|
||||
messaging.processConnect(m_session, connMsg);
|
||||
|
||||
//Verify
|
||||
assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWill() {
|
||||
connMsg.setClientID("123");
|
||||
connMsg.setWillFlag(true);
|
||||
connMsg.setWillTopic("topic");
|
||||
connMsg.setWillMessage("Topic message");
|
||||
//IMessaging mockedMessaging = mock(IMessaging.class);
|
||||
|
||||
//Exercise
|
||||
//m_handler.setMessaging(mockedMessaging);
|
||||
messaging.processConnect(m_session, connMsg);
|
||||
|
||||
//Verify
|
||||
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode);
|
||||
//TODO verify the call
|
||||
/*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
|
||||
any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/
|
||||
}
|
||||
|
||||
|
||||
@Ignore
|
||||
public void testHandleSubscribe() {
|
||||
/*String topicName = "/news";
|
||||
SubscribeMessage subscribeMsg = new SubscribeMessage();
|
||||
subscribeMsg.setMessageID(0x555);
|
||||
subscribeMsg.addSubscription(new SubscribeMessage.Couple(
|
||||
(byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName));
|
||||
//IMessaging mockedMessaging = mock(IMessaging.class);
|
||||
|
||||
m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID");
|
||||
m_session.setAttribute(Constants.CLEAN_SESSION, false);
|
||||
Subscription subscription = new Subscription("fakeID", "/news", QOSType.EXACTLY_ONCE, true);*/
|
||||
|
||||
//Exercise
|
||||
//m_handler.setMessaging(mockedMessaging);
|
||||
// SubscribeEvent pubEvt = new SubscribeEvent(subscription, 0);
|
||||
SubscribeMessage msg = new SubscribeMessage();
|
||||
msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.EXACTLY_ONCE.ordinal(), "/news"));
|
||||
messaging.processSubscribe(m_session, msg, "fakeID", false);
|
||||
|
||||
//Verify
|
||||
/*ArgumentCaptor<AbstractMessage.QOSType> argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class);
|
||||
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!!
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -6,7 +6,7 @@ import java.text.ParseException;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.DummyStorageService;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.MemoryStorageService;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.fusesource.hawtdb.api.MultiIndexFactory;
|
||||
import org.fusesource.hawtdb.api.PageFile;
|
||||
@@ -40,7 +40,7 @@ public class SubscriptionsStoreTest {
|
||||
pageFactory.open();
|
||||
PageFile pageFile = pageFactory.getPageFile();
|
||||
MultiIndexFactory multiIndexFactory = new MultiIndexFactory(pageFile);
|
||||
store.init(new DummyStorageService());
|
||||
store.init(new MemoryStorageService());
|
||||
}
|
||||
|
||||
@After
|
||||
|
||||
Reference in New Issue
Block a user