First raw code move from parser to newly created broker module
This commit is contained in:
selva.andre
@@ -0,0 +1,35 @@
|
||||
package org.dna.mqtt.moquette.server;
|
||||
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
|
||||
/**
|
||||
* Maintains the information of single connection, like ClientID, IoSession,
|
||||
* and other connection related flags.
|
||||
*
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class ConnectionDescriptor {
|
||||
|
||||
private String m_clientID;
|
||||
private IoSession m_session;
|
||||
private boolean m_cleanSession;
|
||||
|
||||
public ConnectionDescriptor(String clientID, IoSession session, boolean cleanSession) {
|
||||
this.m_clientID = clientID;
|
||||
this.m_session = session;
|
||||
this.m_cleanSession = cleanSession;
|
||||
}
|
||||
|
||||
public boolean isCleanSession() {
|
||||
return m_cleanSession;
|
||||
}
|
||||
|
||||
public String getClientID() {
|
||||
return m_clientID;
|
||||
}
|
||||
|
||||
public IoSession getSession() {
|
||||
return m_session;
|
||||
}
|
||||
}
|
||||
@@ -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";
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package org.dna.mqtt.moquette.server;
|
||||
|
||||
/**
|
||||
* username and password checker
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public interface IAuthenticator {
|
||||
|
||||
boolean checkValid(String username, String password);
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package org.dna.mqtt.moquette.server;
|
||||
|
||||
import org.apache.mina.core.service.IoHandlerAdapter;
|
||||
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.proto.Utils;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
|
||||
|
||||
/**
|
||||
* MINA MQTT Handler used to route messages to protocol logic
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class MQTTHandler extends IoHandlerAdapter {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class);
|
||||
private IMessaging m_messaging;
|
||||
|
||||
@Override
|
||||
public void messageReceived(IoSession session, Object message) throws Exception {
|
||||
AbstractMessage msg = (AbstractMessage) message;
|
||||
LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType())));
|
||||
try {
|
||||
switch (msg.getMessageType()) {
|
||||
case CONNECT:
|
||||
case SUBSCRIBE:
|
||||
case UNSUBSCRIBE:
|
||||
case PUBLISH:
|
||||
case PUBREC:
|
||||
case PUBCOMP:
|
||||
case PUBREL:
|
||||
case PINGREQ:
|
||||
case DISCONNECT:
|
||||
m_messaging.handleProtocolMessage(session, msg);
|
||||
break;
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
LOG.error("Bad error in processing the message", ex);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sessionIdle(IoSession session, IdleStatus status) {
|
||||
session.close(false);
|
||||
}
|
||||
|
||||
public void setMessaging(IMessaging messaging) {
|
||||
m_messaging = messaging;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,136 @@
|
||||
package org.dna.mqtt.moquette.server;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
import org.apache.mina.core.service.IoAcceptor;
|
||||
import org.apache.mina.core.service.IoServiceStatistics;
|
||||
import org.apache.mina.core.session.IdleStatus;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.apache.mina.filter.codec.ProtocolCodecFilter;
|
||||
import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder;
|
||||
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.proto.*;
|
||||
import org.dna.mqtt.moquette.proto.messages.*;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
/**
|
||||
* Launch a configured version of the server.
|
||||
* @author andrea
|
||||
*/
|
||||
public class Server {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Server.class);
|
||||
|
||||
// public static final int PORT = 1883;
|
||||
// public static final int DEFAULT_CONNECT_TIMEOUT = 10;
|
||||
public static final String STORAGE_FILE_PATH = System.getProperty("user.home") +
|
||||
File.separator + "moquette_store.hawtdb";
|
||||
|
||||
//TODO one notifier per outbound queue, else no serialization of message flow is granted!!
|
||||
//private static final int NOTIFIER_POOL_SIZE = 1;
|
||||
private IoAcceptor m_acceptor;
|
||||
SimpleMessaging messaging;
|
||||
// Thread messagingEventLoop;
|
||||
//private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/;
|
||||
|
||||
public static void main(String[] args) throws IOException {
|
||||
new Server().startServer();
|
||||
|
||||
}
|
||||
|
||||
protected void startServer() throws IOException {
|
||||
DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
|
||||
decoder.addMessageDecoder(new ConnectDecoder());
|
||||
decoder.addMessageDecoder(new PublishDecoder());
|
||||
decoder.addMessageDecoder(new PubAckDecoder());
|
||||
decoder.addMessageDecoder(new PubRelDecoder());
|
||||
decoder.addMessageDecoder(new PubRecDecoder());
|
||||
decoder.addMessageDecoder(new PubCompDecoder());
|
||||
decoder.addMessageDecoder(new SubscribeDecoder());
|
||||
decoder.addMessageDecoder(new UnsubscribeDecoder());
|
||||
decoder.addMessageDecoder(new DisconnectDecoder());
|
||||
decoder.addMessageDecoder(new PingReqDecoder());
|
||||
|
||||
DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
|
||||
// encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
|
||||
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(PubRecMessage.class, new PubRecEncoder());
|
||||
encoder.addMessageEncoder(PubCompMessage.class, new PubCompEncoder());
|
||||
encoder.addMessageEncoder(PubRelMessage.class, new PubRelEncoder());
|
||||
encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder());
|
||||
encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder());
|
||||
|
||||
m_acceptor = new NioSocketAcceptor();
|
||||
|
||||
m_acceptor.getFilterChain().addLast( "logger", new MQTTLoggingFilter("SERVER LOG") );
|
||||
m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
|
||||
|
||||
MQTTHandler handler = new MQTTHandler();
|
||||
messaging = SimpleMessaging.getInstance();
|
||||
messaging.init();
|
||||
//TODO fix this hugly wiring
|
||||
handler.setMessaging(messaging);
|
||||
// messaging.setNotifier(handler);
|
||||
// messagingEventLoop = new Thread(messaging);
|
||||
// messagingEventLoop.setName("Event Loop" + System.currentTimeMillis());
|
||||
// messagingEventLoop.start();
|
||||
|
||||
m_acceptor.setHandler(handler);
|
||||
((NioSocketAcceptor)m_acceptor).setReuseAddress(true);
|
||||
((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true);
|
||||
m_acceptor.getSessionConfig().setReadBufferSize( 2048 );
|
||||
m_acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, DEFAULT_CONNECT_TIMEOUT );
|
||||
m_acceptor.getStatistics().setThroughputCalculationInterval(10);
|
||||
m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis());
|
||||
m_acceptor.bind( new InetSocketAddress(PORT) );
|
||||
LOG.info("Server binded");
|
||||
|
||||
//Bind a shutdown hook
|
||||
Runtime.getRuntime().addShutdownHook(new Thread() {
|
||||
@Override
|
||||
public void run() {
|
||||
stopServer();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
protected void stopServer() {
|
||||
LOG.info("Server stopping...");
|
||||
|
||||
messaging.stop();
|
||||
//messaging.stop();
|
||||
// messagingEventLoop.interrupt();
|
||||
// LOG.info("shutting down evet loop");
|
||||
// try {
|
||||
// messagingEventLoop.join();
|
||||
// } catch (InterruptedException ex) {
|
||||
// LOG.error(null, ex);
|
||||
// }
|
||||
|
||||
/*m_notifierPool.shutdown();*/
|
||||
|
||||
//log statistics
|
||||
IoServiceStatistics statistics = m_acceptor.getStatistics();
|
||||
statistics.updateThroughput(System.currentTimeMillis());
|
||||
System.out.println(String.format("Total read bytes: %d, read throughtput: %f (b/s)", statistics.getReadBytes(), statistics.getReadBytesThroughput()));
|
||||
System.out.println(String.format("Total read msgs: %d, read msg throughtput: %f (msg/s)", statistics.getReadMessages(), statistics.getReadMessagesThroughput()));
|
||||
|
||||
for(IoSession session: m_acceptor.getManagedSessions().values()) {
|
||||
if(session.isConnected() && !session.isClosing()){
|
||||
session.close(false);
|
||||
}
|
||||
}
|
||||
|
||||
m_acceptor.unbind();
|
||||
m_acceptor.dispose();
|
||||
LOG.info("Server stopped");
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi;
|
||||
|
||||
/**
|
||||
*/
|
||||
public interface IMatchingCondition {
|
||||
boolean match(String key);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
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.
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public interface IMessaging {
|
||||
|
||||
void stop();
|
||||
|
||||
void disconnect(IoSession session);
|
||||
|
||||
void republishStored(String clientID);
|
||||
|
||||
void handleProtocolMessage(IoSession session, AbstractMessage msg);
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
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);
|
||||
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi;
|
||||
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import static org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService.StoredMessage;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Defines the SPI to be implemented by a StorageService that handle persistence of messages and subscriptions.
|
||||
*/
|
||||
public interface IStorageService {
|
||||
|
||||
/**
|
||||
* Used to initialize all persistent store structures
|
||||
* */
|
||||
void initStore();
|
||||
|
||||
void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos);
|
||||
|
||||
Collection<StoredMessage> searchMatching(IMatchingCondition condition);
|
||||
|
||||
void storePublishForFuture(PublishEvent evt);
|
||||
|
||||
List<PublishEvent> retrivePersistedPublishes(String clientID);
|
||||
|
||||
void cleanPersistedPublishes(String clientID);
|
||||
|
||||
void cleanInFlight(String msgID);
|
||||
|
||||
void addInFlight(PublishEvent evt, String publishKey);
|
||||
|
||||
void addNewSubscription(Subscription newSubscription, String clientID);
|
||||
|
||||
void removeAllSubscriptions(String clientID);
|
||||
|
||||
List<Subscription> retrieveAllSubscriptions();
|
||||
|
||||
void close();
|
||||
|
||||
void persistQoS2Message(String publishKey, PublishEvent evt);
|
||||
|
||||
void removeQoS2Message(String publishKey);
|
||||
|
||||
PublishEvent retrieveQoS2Message(String publishKey);
|
||||
}
|
||||
@@ -0,0 +1,227 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl;
|
||||
|
||||
import org.dna.mqtt.moquette.MQTTException;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.dna.mqtt.moquette.server.Server;
|
||||
import org.fusesource.hawtbuf.codec.StringCodec;
|
||||
import org.fusesource.hawtdb.api.*;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.io.Serializable;
|
||||
|
||||
|
||||
/**
|
||||
* Implementation of IStorageService backed by HawtDB
|
||||
*/
|
||||
public class HawtDBStorageService implements IStorageService {
|
||||
|
||||
public static class StoredMessage implements Serializable {
|
||||
AbstractMessage.QOSType m_qos;
|
||||
byte[] m_payload;
|
||||
|
||||
StoredMessage(byte[] message, AbstractMessage.QOSType qos) {
|
||||
m_qos = qos;
|
||||
m_payload = message;
|
||||
}
|
||||
|
||||
AbstractMessage.QOSType getQos() {
|
||||
return m_qos;
|
||||
}
|
||||
|
||||
byte[] getPayload() {
|
||||
return m_payload;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(HawtDBStorageService.class);
|
||||
|
||||
private MultiIndexFactory m_multiIndexFactory;
|
||||
private PageFileFactory pageFactory;
|
||||
|
||||
//maps clientID to the list of pending messages stored
|
||||
private SortedIndex<String, List<PublishEvent>> m_persistentMessageStore;
|
||||
private SortedIndex<String, StoredMessage> m_retainedStore;
|
||||
//bind clientID+MsgID -> evt message published
|
||||
private SortedIndex<String, PublishEvent> m_inflightStore;
|
||||
//bind clientID+MsgID -> evt message published
|
||||
private SortedIndex<String, PublishEvent> m_qos2Store;
|
||||
|
||||
//persistent Map of clientID, set of Subscriptions
|
||||
private SortedIndex<String, Set<Subscription>> m_persistentSubscriptions;
|
||||
|
||||
public HawtDBStorageService() {
|
||||
String storeFile = Server.STORAGE_FILE_PATH;
|
||||
|
||||
pageFactory = new PageFileFactory();
|
||||
File tmpFile;
|
||||
try {
|
||||
tmpFile = new File(storeFile);
|
||||
tmpFile.createNewFile();
|
||||
} catch (IOException ex) {
|
||||
LOG.error(null, ex);
|
||||
throw new MQTTException("Can't create temp file for subscriptions storage [" + storeFile + "]", ex);
|
||||
}
|
||||
pageFactory.setFile(tmpFile);
|
||||
pageFactory.open();
|
||||
PageFile pageFile = pageFactory.getPageFile();
|
||||
m_multiIndexFactory = new MultiIndexFactory(pageFile);
|
||||
}
|
||||
|
||||
|
||||
public void initStore() {
|
||||
initRetainedStore();
|
||||
//init the message store for QoS 1/2 messages in clean sessions
|
||||
initPersistentMessageStore();
|
||||
initInflightMessageStore();
|
||||
initPersistentSubscriptions();
|
||||
initPersistentQoS2MessageStore();
|
||||
}
|
||||
|
||||
private void initRetainedStore() {
|
||||
BTreeIndexFactory<String, StoredMessage> indexFactory = new BTreeIndexFactory<String, StoredMessage>();
|
||||
indexFactory.setKeyCodec(StringCodec.INSTANCE);
|
||||
|
||||
m_retainedStore = (SortedIndex<String, StoredMessage>) m_multiIndexFactory.openOrCreate("retained", 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);
|
||||
}
|
||||
|
||||
private void initPersistentSubscriptions() {
|
||||
BTreeIndexFactory<String, Set<Subscription>> indexFactory = new BTreeIndexFactory<String, Set<Subscription>>();
|
||||
indexFactory.setKeyCodec(StringCodec.INSTANCE);
|
||||
|
||||
m_persistentSubscriptions = (SortedIndex<String, Set<Subscription>>) m_multiIndexFactory.openOrCreate("subscriptions", 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 initPersistentQoS2MessageStore() {
|
||||
BTreeIndexFactory<String, PublishEvent> indexFactory = new BTreeIndexFactory<String, PublishEvent>();
|
||||
indexFactory.setKeyCodec(StringCodec.INSTANCE);
|
||||
|
||||
m_qos2Store = (SortedIndex<String, PublishEvent>) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory);
|
||||
}
|
||||
|
||||
public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) {
|
||||
if (message.length == 0) {
|
||||
//clean the message from topic
|
||||
m_retainedStore.remove(topic);
|
||||
} else {
|
||||
//store the message to the topic
|
||||
m_retainedStore.put(topic, new StoredMessage(message, qos));
|
||||
}
|
||||
}
|
||||
|
||||
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
|
||||
LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size());
|
||||
|
||||
List<StoredMessage> results = new ArrayList<StoredMessage>();
|
||||
|
||||
for (Map.Entry<String, StoredMessage> entry : m_retainedStore) {
|
||||
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) {
|
||||
allSubscriptions.addAll(entry.getValue());
|
||||
}
|
||||
return allSubscriptions;
|
||||
}
|
||||
|
||||
public void close() {
|
||||
try {
|
||||
pageFactory.close();
|
||||
} catch (IOException ex) {
|
||||
LOG.error(null, ex);
|
||||
}
|
||||
}
|
||||
|
||||
/*-------- QoS 2 storage management --------------*/
|
||||
public void persistQoS2Message(String publishKey, PublishEvent evt) {
|
||||
LOG.debug(String.format("persistQoS2Message store pubKey %s, evt %s", publishKey, evt));
|
||||
m_qos2Store.put(publishKey, evt);
|
||||
}
|
||||
|
||||
public void removeQoS2Message(String publishKey) {
|
||||
m_qos2Store.remove(publishKey);
|
||||
}
|
||||
|
||||
public PublishEvent retrieveQoS2Message(String publishKey) {
|
||||
return m_qos2Store.get(publishKey);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,544 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl;
|
||||
|
||||
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.subscriptions.SubscriptionsStore;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
|
||||
import org.dna.mqtt.moquette.proto.PubCompMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.*;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
|
||||
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
|
||||
import org.dna.mqtt.moquette.server.Constants;
|
||||
import org.dna.mqtt.moquette.server.IAuthenticator;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService.StoredMessage;
|
||||
|
||||
/**
|
||||
*
|
||||
* Singleton class that orchestrate the execution of the protocol.
|
||||
*
|
||||
* Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion.
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class);
|
||||
|
||||
private SubscriptionsStore subscriptions = new SubscriptionsStore();
|
||||
|
||||
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 SimpleMessaging() {
|
||||
}
|
||||
|
||||
public static SimpleMessaging getInstance() {
|
||||
if (INSTANCE == null) {
|
||||
INSTANCE = new SimpleMessaging();
|
||||
}
|
||||
return INSTANCE;
|
||||
}
|
||||
|
||||
public void init() {
|
||||
m_executor = Executors.newFixedThreadPool(1);
|
||||
|
||||
m_ringBuffer = new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32);
|
||||
|
||||
SequenceBarrier barrier = m_ringBuffer.newBarrier();
|
||||
m_eventProcessor = new BatchEventProcessor<ValueEvent>(m_ringBuffer, barrier, this);
|
||||
m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
|
||||
m_executor.submit(m_eventProcessor);
|
||||
|
||||
disruptorPublish(new InitEvent());
|
||||
}
|
||||
|
||||
|
||||
/* public void setNotifier(INotifier notifier) {
|
||||
m_notifier= notifier;
|
||||
} */
|
||||
|
||||
|
||||
private void disruptorPublish(MessagingEvent msgEvent) {
|
||||
long sequence = m_ringBuffer.next();
|
||||
ValueEvent event = m_ringBuffer.get(sequence);
|
||||
|
||||
event.setEvent(msgEvent);
|
||||
|
||||
m_ringBuffer.publish(sequence);
|
||||
}
|
||||
|
||||
|
||||
public void disconnect(IoSession session) {
|
||||
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());
|
||||
}
|
||||
|
||||
public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
|
||||
MessagingEvent evt = t.getEvent();
|
||||
LOG.debug("onEvent processing messaging event " + evt);
|
||||
if (evt instanceof PublishEvent) {
|
||||
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);
|
||||
} else if (evt instanceof ProtocolEvent) {
|
||||
IoSession session = ((ProtocolEvent) evt).getSession();
|
||||
AbstractMessage message = ((ProtocolEvent) evt).getMessage();
|
||||
if (message instanceof ConnectMessage) {
|
||||
processConnect(session, (ConnectMessage) message);
|
||||
} else if (message instanceof PublishMessage) {
|
||||
PublishMessage pubMsg = (PublishMessage) message;
|
||||
PublishEvent pubEvt;
|
||||
|
||||
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
|
||||
|
||||
if (message.getQos() == QOSType.MOST_ONE) {
|
||||
pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, session);
|
||||
|
||||
} else {
|
||||
pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session);
|
||||
}
|
||||
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);
|
||||
} else if (message instanceof UnsubscribeMessage) {
|
||||
UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
|
||||
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
|
||||
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);
|
||||
} else if (message instanceof PubRelMessage) {
|
||||
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
|
||||
int messageID = ((PubRelMessage) message).getMessageID();
|
||||
processPubRel(clientID, messageID);
|
||||
} else if (message instanceof PubRecMessage) {
|
||||
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
|
||||
int messageID = ((PubRecMessage) message).getMessageID();
|
||||
processPubRec(clientID, messageID);
|
||||
} else if (message instanceof PubCompMessage) {
|
||||
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
|
||||
int messageID = ((PubCompMessage) message).getMessageID();
|
||||
processPubComp(clientID, messageID);
|
||||
} else {
|
||||
throw new RuntimeException("Illegal message received " + message);
|
||||
}
|
||||
|
||||
} else if (evt instanceof InitEvent) {
|
||||
processInit();
|
||||
}
|
||||
}
|
||||
|
||||
private void processInit() {
|
||||
m_storageService = new HawtDBStorageService();
|
||||
m_storageService.initStore();
|
||||
|
||||
subscriptions.init(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();
|
||||
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) {
|
||||
assert publishKey != null;
|
||||
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(), topic, 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();
|
||||
|
||||
// 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 {
|
||||
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(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 {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl;
|
||||
|
||||
import com.lmax.disruptor.EventFactory;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
|
||||
|
||||
/**
|
||||
* Carrier value object for the RingBuffer.
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public final class ValueEvent {
|
||||
|
||||
private MessagingEvent m_event;
|
||||
|
||||
public MessagingEvent getEvent() {
|
||||
return m_event;
|
||||
}
|
||||
|
||||
public void setEvent(MessagingEvent event) {
|
||||
m_event = event;
|
||||
}
|
||||
|
||||
public final static EventFactory<ValueEvent> EVENT_FACTORY = new EventFactory<ValueEvent>() {
|
||||
|
||||
public ValueEvent newInstance() {
|
||||
return new ValueEvent();
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl.events;
|
||||
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class DisconnectEvent extends MessagingEvent {
|
||||
|
||||
IoSession m_session;
|
||||
|
||||
public DisconnectEvent(IoSession session) {
|
||||
m_session = session;
|
||||
}
|
||||
|
||||
public IoSession getSession() {
|
||||
return m_session;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl.events;
|
||||
|
||||
/**
|
||||
*/
|
||||
public class InitEvent extends MessagingEvent {
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl.events;
|
||||
|
||||
/**
|
||||
*Event used to push work request on the SimpleMessaging.
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public abstract class MessagingEvent {
|
||||
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
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 + '\'' +
|
||||
'}';
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl.events;
|
||||
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
|
||||
/**
|
||||
* Event used to carry ProtocolMessages from front handler to event processor
|
||||
*/
|
||||
public class ProtocolEvent extends MessagingEvent {
|
||||
IoSession m_session;
|
||||
AbstractMessage message;
|
||||
|
||||
public ProtocolEvent(IoSession session, AbstractMessage message) {
|
||||
this.m_session = session;
|
||||
this.message = message;
|
||||
}
|
||||
|
||||
public IoSession getSession() {
|
||||
return m_session;
|
||||
}
|
||||
|
||||
public AbstractMessage getMessage() {
|
||||
return message;
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
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 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,
|
||||
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() {
|
||||
return m_topic;
|
||||
}
|
||||
|
||||
public QOSType getQos() {
|
||||
return m_qos;
|
||||
}
|
||||
|
||||
public byte[] getMessage() {
|
||||
return m_message;
|
||||
}
|
||||
|
||||
public boolean isRetain() {
|
||||
return m_retain;
|
||||
}
|
||||
|
||||
public String getClientID() {
|
||||
return m_clientID;
|
||||
}
|
||||
|
||||
public int getMessageID() {
|
||||
return m_msgID;
|
||||
}
|
||||
|
||||
public IoSession getSession() {
|
||||
return m_session;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "PublishEvent{" +
|
||||
"m_msgID=" + m_msgID +
|
||||
", m_clientID='" + m_clientID + '\'' +
|
||||
", m_retain=" + m_retain +
|
||||
", m_qos=" + m_qos +
|
||||
", m_topic='" + m_topic + '\'' +
|
||||
'}';
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl.events;
|
||||
|
||||
/**
|
||||
* Poison pill to shutdown the event loop
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class StopEvent extends MessagingEvent {
|
||||
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl.events;
|
||||
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class SubscribeEvent extends MessagingEvent {
|
||||
|
||||
Subscription m_subscription;
|
||||
|
||||
int m_messageID;
|
||||
|
||||
public SubscribeEvent(Subscription subscription, int messageID) {
|
||||
m_subscription = subscription;
|
||||
m_messageID = messageID;
|
||||
}
|
||||
|
||||
public Subscription getSubscription() {
|
||||
return m_subscription;
|
||||
}
|
||||
|
||||
public int getMessageID() {
|
||||
return m_messageID;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions;
|
||||
|
||||
import java.io.Serializable;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
|
||||
|
||||
/**
|
||||
* Maintain the information about which Topic a certain ClientID is subscribed
|
||||
* and at which QoS
|
||||
*
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class Subscription implements Serializable {
|
||||
|
||||
QOSType requestedQos;
|
||||
String clientId;
|
||||
String topic;
|
||||
boolean cleanSession;
|
||||
boolean active = true;
|
||||
|
||||
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() {
|
||||
return clientId;
|
||||
}
|
||||
|
||||
public QOSType getRequestedQos() {
|
||||
return requestedQos;
|
||||
}
|
||||
|
||||
public String getTopic() {
|
||||
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) {
|
||||
return false;
|
||||
}
|
||||
if (getClass() != obj.getClass()) {
|
||||
return false;
|
||||
}
|
||||
final Subscription other = (Subscription) obj;
|
||||
if (this.requestedQos != other.requestedQos) {
|
||||
return false;
|
||||
}
|
||||
if ((this.clientId == null) ? (other.clientId != null) : !this.clientId.equals(other.clientId)) {
|
||||
return false;
|
||||
}
|
||||
if ((this.topic == null) ? (other.topic != null) : !this.topic.equals(other.topic)) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
int hash = 3;
|
||||
hash = 37 * hash + (this.requestedQos != null ? this.requestedQos.hashCode() : 0);
|
||||
hash = 37 * hash + (this.clientId != null ? this.clientId.hashCode() : 0);
|
||||
hash = 37 * hash + (this.topic != null ? this.topic.hashCode() : 0);
|
||||
return hash;
|
||||
}
|
||||
|
||||
/**
|
||||
* Trivial match method
|
||||
*/
|
||||
boolean match(String topic) {
|
||||
return this.topic.equals(topic);
|
||||
}
|
||||
}
|
||||
+210
@@ -0,0 +1,210 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions;
|
||||
|
||||
import java.text.ParseException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Queue;
|
||||
import java.util.concurrent.LinkedBlockingDeque;
|
||||
|
||||
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* Represents a tree of topics subscriptions.
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class SubscriptionsStore {
|
||||
|
||||
private TreeNode subscriptions = new TreeNode(null);
|
||||
private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
|
||||
|
||||
private IStorageService m_storageService;
|
||||
|
||||
/**
|
||||
* Initialize basic store structures, like the FS storage to maintain
|
||||
* client's topics subscriptions
|
||||
*/
|
||||
public void init(IStorageService storageService) {
|
||||
LOG.debug("init invoked");
|
||||
|
||||
m_storageService = storageService;
|
||||
|
||||
//reload any subscriptions persisted
|
||||
LOG.debug("Reloading all stored subscriptions...");
|
||||
for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) {
|
||||
LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic());
|
||||
addDirect(subscription);
|
||||
}
|
||||
LOG.debug("Finished loading");
|
||||
}
|
||||
|
||||
protected void addDirect(Subscription newSubscription) {
|
||||
TreeNode current = findMatchingNode(newSubscription.topic);
|
||||
current.addSubcription(newSubscription);
|
||||
}
|
||||
|
||||
private TreeNode findMatchingNode(String topic) {
|
||||
List<Token> tokens = new ArrayList<Token>();
|
||||
try {
|
||||
tokens = splitTopic(topic);
|
||||
} catch (ParseException ex) {
|
||||
//TODO handle the parse exception
|
||||
LOG.error(null, ex);
|
||||
// return;
|
||||
}
|
||||
|
||||
TreeNode current = subscriptions;
|
||||
for (Token token : tokens) {
|
||||
TreeNode matchingChildren;
|
||||
|
||||
//check if a children with the same token already exists
|
||||
if ((matchingChildren = current.childWithToken(token)) != null) {
|
||||
current = matchingChildren;
|
||||
} else {
|
||||
//create a new node for the newly inserted token
|
||||
matchingChildren = new TreeNode(current);
|
||||
matchingChildren.setToken(token);
|
||||
current.addChild(matchingChildren);
|
||||
current = matchingChildren;
|
||||
}
|
||||
}
|
||||
return current;
|
||||
}
|
||||
|
||||
public void add(Subscription newSubscription) {
|
||||
addDirect(newSubscription);
|
||||
|
||||
//log the subscription
|
||||
String clientID = newSubscription.getClientId();
|
||||
m_storageService.addNewSubscription(newSubscription, clientID);
|
||||
// pageFile.flush();
|
||||
}
|
||||
|
||||
|
||||
public void removeSubscription(String topic, String clientID) {
|
||||
TreeNode matchNode = findMatchingNode(topic);
|
||||
|
||||
//search fr the subscription to remove
|
||||
Subscription toBeRemoved = null;
|
||||
for (Subscription sub : matchNode.subscriptions()) {
|
||||
if (sub.topic.equals(topic)) {
|
||||
toBeRemoved = sub;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (toBeRemoved != null) {
|
||||
matchNode.subscriptions().remove(toBeRemoved);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Visit the topics tree to remove matching subscriptions with clientID
|
||||
*/
|
||||
public void removeForClient(String clientID) {
|
||||
subscriptions.removeClientSubscriptions(clientID);
|
||||
|
||||
//remove from log all subscriptions
|
||||
m_storageService.removeAllSubscriptions(clientID);
|
||||
}
|
||||
|
||||
public void deactivate(String clientID) {
|
||||
subscriptions.deactivate(clientID);
|
||||
}
|
||||
|
||||
public void activate(String clientID) {
|
||||
LOG.debug("activate re-activating subscriptions for clientID " + clientID);
|
||||
subscriptions.activate(clientID);
|
||||
}
|
||||
|
||||
public List<Subscription> matches(String topic) {
|
||||
List<Token> tokens = new ArrayList<Token>();
|
||||
try {
|
||||
tokens = splitTopic(topic);
|
||||
} catch (ParseException ex) {
|
||||
//TODO handle the parse exception
|
||||
LOG.error(null, ex);
|
||||
}
|
||||
|
||||
Queue<Token> tokenQueue = new LinkedBlockingDeque<Token>(tokens);
|
||||
List<Subscription> matchingSubs = new ArrayList<Subscription>();
|
||||
subscriptions.matches(tokenQueue, matchingSubs);
|
||||
return matchingSubs;
|
||||
}
|
||||
|
||||
public boolean contains(Subscription sub) {
|
||||
return !matches(sub.topic).isEmpty();
|
||||
}
|
||||
|
||||
public int size() {
|
||||
return subscriptions.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* Verify if the 2 topics matching respecting the rules of MQTT Appendix A
|
||||
*/
|
||||
public static boolean matchTopics(String msgTopic, String subscriptionTopic) {
|
||||
try {
|
||||
List<Token> msgTokens = SubscriptionsStore.splitTopic(msgTopic);
|
||||
List<Token> subscriptionTokens = SubscriptionsStore.splitTopic(subscriptionTopic);
|
||||
int i = 0;
|
||||
for (; i< subscriptionTokens.size(); i++) {
|
||||
Token subToken = subscriptionTokens.get(i);
|
||||
if (subToken != Token.MULTI && subToken != Token.SINGLE) {
|
||||
Token msgToken = msgTokens.get(i);
|
||||
if (!msgToken.equals(subToken)) {
|
||||
return false;
|
||||
}
|
||||
} else {
|
||||
if (subToken == Token.MULTI) {
|
||||
return true;
|
||||
}
|
||||
if (subToken == Token.SINGLE) {
|
||||
//skip a step forward
|
||||
}
|
||||
}
|
||||
}
|
||||
return i == msgTokens.size();
|
||||
} catch (ParseException ex) {
|
||||
LOG.error(null, ex);
|
||||
throw new RuntimeException(ex);
|
||||
}
|
||||
}
|
||||
|
||||
protected static List<Token> splitTopic(String topic) throws ParseException {
|
||||
List res = new ArrayList<Token>();
|
||||
String[] splitted = topic.split("/");
|
||||
|
||||
if (splitted.length == 0) {
|
||||
res.add(Token.EMPTY);
|
||||
}
|
||||
|
||||
for (int i = 0; i < splitted.length; i++) {
|
||||
String s = splitted[i];
|
||||
if (s.isEmpty()) {
|
||||
if (i != 0) {
|
||||
throw new ParseException("Bad format of topic, expetec topic name between separators", i);
|
||||
}
|
||||
res.add(Token.EMPTY);
|
||||
} else if (s.equals("#")) {
|
||||
//check that multi is the last symbol
|
||||
if (i != splitted.length - 1) {
|
||||
throw new ParseException("Bad format of topic, the multi symbol (#) has to be the last one after a separator", i);
|
||||
}
|
||||
res.add(Token.MULTI);
|
||||
} else if (s.contains("#")) {
|
||||
throw new ParseException("Bad format of topic, invalid subtopic name: " + s, i);
|
||||
} else if (s.equals("+")) {
|
||||
res.add(Token.SINGLE);
|
||||
} else if (s.contains("+")) {
|
||||
throw new ParseException("Bad format of topic, invalid subtopic name: " + s, i);
|
||||
} else {
|
||||
res.add(new Token(s));
|
||||
}
|
||||
}
|
||||
|
||||
return res;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions;
|
||||
|
||||
class Token {
|
||||
|
||||
static final Token EMPTY = new Token("");
|
||||
static final Token MULTI = new Token("#");
|
||||
static final Token SINGLE = new Token("+");
|
||||
String name;
|
||||
|
||||
protected Token(String s) {
|
||||
name = s;
|
||||
}
|
||||
|
||||
protected String name() {
|
||||
return name;
|
||||
}
|
||||
|
||||
protected boolean match(Token t) {
|
||||
if (t == MULTI || t == SINGLE) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (this == MULTI || this == SINGLE) {
|
||||
return true;
|
||||
}
|
||||
|
||||
return equals(t);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
int hash = 7;
|
||||
hash = 29 * hash + (this.name != null ? this.name.hashCode() : 0);
|
||||
return hash;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (obj == null) {
|
||||
return false;
|
||||
}
|
||||
if (getClass() != obj.getClass()) {
|
||||
return false;
|
||||
}
|
||||
final Token other = (Token) obj;
|
||||
if ((this.name == null) ? (other.name != null) : !this.name.equals(other.name)) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return name;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,154 @@
|
||||
package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Queue;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
|
||||
class TreeNode {
|
||||
|
||||
TreeNode m_parent;
|
||||
Token m_token;
|
||||
List<TreeNode> m_children = new ArrayList<TreeNode>();
|
||||
List<Subscription> m_subscriptions = new ArrayList<Subscription>();
|
||||
|
||||
TreeNode(TreeNode parent) {
|
||||
this.m_parent = parent;
|
||||
}
|
||||
|
||||
Token getToken() {
|
||||
return m_token;
|
||||
}
|
||||
|
||||
void setToken(Token topic) {
|
||||
this.m_token = topic;
|
||||
}
|
||||
|
||||
void addSubcription(Subscription s) {
|
||||
//avoid double registering
|
||||
if (m_subscriptions.contains(s)) {
|
||||
return;
|
||||
}
|
||||
m_subscriptions.add(s);
|
||||
}
|
||||
|
||||
void addChild(TreeNode child) {
|
||||
m_children.add(child);
|
||||
}
|
||||
|
||||
boolean isLeaf() {
|
||||
return m_children.isEmpty();
|
||||
}
|
||||
|
||||
/**
|
||||
* Search for children that has the specified token, if not found return
|
||||
* null;
|
||||
*/
|
||||
TreeNode childWithToken(Token token) {
|
||||
for (TreeNode child : m_children) {
|
||||
if (child.getToken().equals(token)) {
|
||||
return child;
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
List<Subscription> subscriptions() {
|
||||
return m_subscriptions;
|
||||
}
|
||||
|
||||
void matches(Queue<Token> tokens, List<Subscription> matchingSubs) {
|
||||
Token t = tokens.poll();
|
||||
|
||||
//check if t is null <=> tokens finished
|
||||
if (t == null) {
|
||||
matchingSubs.addAll(m_subscriptions);
|
||||
//check if it has got a MULTI child and add its subscriptions
|
||||
for (TreeNode n : m_children) {
|
||||
if (n.getToken() == Token.MULTI) {
|
||||
matchingSubs.addAll(n.subscriptions());
|
||||
}
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
//we are on MULTI, than add subscriptions and return
|
||||
if (m_token == Token.MULTI) {
|
||||
matchingSubs.addAll(m_subscriptions);
|
||||
return;
|
||||
}
|
||||
|
||||
for (TreeNode n : m_children) {
|
||||
if (n.getToken().match(t)) {
|
||||
//Create a copy of token, alse if navigate 2 sibling it
|
||||
//consumes 2 elements on the queue instead of one
|
||||
n.matches(new LinkedBlockingQueue<Token>(tokens), matchingSubs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the number of registered subscriptions
|
||||
*/
|
||||
int size() {
|
||||
int res = m_subscriptions.size();
|
||||
for (TreeNode child : m_children) {
|
||||
res += child.size();
|
||||
}
|
||||
return res;
|
||||
}
|
||||
|
||||
void removeClientSubscriptions(String clientID) {
|
||||
//collect what to delete and then delete to avoid ConcurrentModification
|
||||
List<Subscription> subsToRemove = new ArrayList<Subscription>();
|
||||
for (Subscription s : m_subscriptions) {
|
||||
if (s.clientId.equals(clientID)) {
|
||||
subsToRemove.add(s);
|
||||
}
|
||||
}
|
||||
|
||||
for (Subscription s : subsToRemove) {
|
||||
m_subscriptions.remove(s);
|
||||
}
|
||||
|
||||
//go deep
|
||||
for (TreeNode child : m_children) {
|
||||
child.removeClientSubscriptions(clientID);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Deactivate all topic subscriptions for the given clientID.
|
||||
* */
|
||||
void deactivate(String clientID) {
|
||||
for (Subscription s : m_subscriptions) {
|
||||
if (s.clientId.equals(clientID)) {
|
||||
s.setActive(false);
|
||||
}
|
||||
}
|
||||
|
||||
//go deep
|
||||
for (TreeNode child : m_children) {
|
||||
child.deactivate(clientID);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Activate all topic subscriptions for the given clientID.
|
||||
* */
|
||||
public void activate(String clientID) {
|
||||
for (Subscription s : m_subscriptions) {
|
||||
if (s.clientId.equals(clientID)) {
|
||||
s.setActive(true);
|
||||
}
|
||||
}
|
||||
|
||||
//go deep
|
||||
for (TreeNode child : m_children) {
|
||||
child.deactivate(clientID);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user