From 739f63bc40ca3427f2222f1728aaeec87438d3cb Mon Sep 17 00:00:00 2001 From: andrea Date: Mon, 27 Aug 2012 11:48:21 +0200 Subject: [PATCH] Changed SimpleMessaging to be event driven (Incoming requests) --- .../messaging/spi/impl/SimpleMessaging.java | 166 +++++++++++++----- .../messaging/spi/impl/events/CloseEvent.java | 9 + .../spi/impl/events/MessagingEvent.java | 10 ++ .../spi/impl/events/PublishEvent.java | 37 ++++ .../events/RemoveAllSubscriptionsEvent.java | 19 ++ .../spi/impl/events/SubscribeEvent.java | 21 +++ .../spi/impl/events/UnsubscribeEvent.java | 25 +++ .../org/dna/mqtt/moquette/server/Server.java | 12 +- 8 files changed, 250 insertions(+), 49 deletions(-) create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/MessagingEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RemoveAllSubscriptionsEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/UnsubscribeEvent.java diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index d59a5074..f58b759a 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -1,17 +1,23 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +import org.dna.mqtt.moquette.messaging.spi.impl.events.RemoveAllSubscriptionsEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; import java.io.File; import java.io.IOException; import java.io.Serializable; import java.text.ParseException; import java.util.List; import java.util.Map; -import java.util.concurrent.locks.ReadWriteLock; -import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.INotifier; import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token; +import org.dna.mqtt.moquette.messaging.spi.impl.events.CloseEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.SubscribeEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.UnsubscribeEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.server.Server; import org.fusesource.hawtbuf.codec.StringCodec; @@ -27,8 +33,8 @@ import org.slf4j.LoggerFactory; * * @author andrea */ -public class SimpleMessaging implements IMessaging { - +public class SimpleMessaging implements IMessaging, Runnable { + private static class StoredMessage implements Serializable { QOSType m_qos; byte[] m_payload; @@ -49,11 +55,10 @@ public class SimpleMessaging implements IMessaging { private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class); - //TODO is poorly performace on concurrent load, use at least ReadWriteLock - - private ReadWriteLock rwLock = new ReentrantReadWriteLock(); private SubscriptionsStore subscriptions = new SubscriptionsStore(); + private BlockingQueue m_inboundQueue = new LinkedBlockingQueue(); + private INotifier m_notifier; private MultiIndexFactory m_multiIndexFactory; private PageFileFactory pageFactory; @@ -89,6 +94,10 @@ public class SimpleMessaging implements IMessaging { m_retainedStore = (SortedIndex) m_multiIndexFactory.openOrCreate("retained", indexFactory); } + public void run() { + eventLoop(); + } + public void setNotifier(INotifier notifier) { m_notifier = notifier; } @@ -97,54 +106,28 @@ public class SimpleMessaging implements IMessaging { //TODO for this moment no qos > 0 trivial implementation QOSType defQos = QOSType.MOST_ONE; - rwLock.readLock().lock(); try { - for (Subscription sub : subscriptions.matches(topic)) { - m_notifier.notify(sub.clientId, topic, defQos, message, false); - } - if (retain) { - 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)); - } - } - } finally { - rwLock.readLock().unlock(); + m_inboundQueue.put(new PublishEvent(topic, defQos, message, retain)); + } catch (InterruptedException ex) { + LOG.error(null, ex); } } public void subscribe(String clientId, String topic, QOSType qos) { Subscription newSubscription = new Subscription(clientId, topic, qos); - rwLock.writeLock().lock(); try { - subscriptions.add(newSubscription); - - //scans retained messages to be published to the new subscription - LOG.debug("Scanning all retained messages..."); - for (Map.Entry entry : m_retainedStore) { - StoredMessage storedMsg = entry.getValue(); - if (matchTopics(entry.getKey(), topic)) { - //fire the as retained the message - m_notifier.notify(newSubscription.clientId, topic, storedMsg.getQos(), - storedMsg.getPayload(), true); - } - } - } finally { - LOG.debug("Finished firing retained messages"); - rwLock.writeLock().unlock(); + m_inboundQueue.put(new SubscribeEvent(newSubscription)); + } catch (InterruptedException ex) { + LOG.error(null, ex); } } public void unsubscribe(String topic, String clientID) { - rwLock.writeLock().lock(); try { - subscriptions.removeSubscription(topic, clientID); - } finally { - rwLock.writeLock().unlock(); + m_inboundQueue.put(new UnsubscribeEvent(topic, clientID)); + } catch (InterruptedException ex) { + LOG.error(null, ex); } } @@ -180,24 +163,111 @@ public class SimpleMessaging implements IMessaging { } } + /** + * NOT SAFE Method, to be removed because used only in tests + */ protected SubscriptionsStore getSubscriptions() { return subscriptions; } public void removeSubscriptions(String clientID) { - rwLock.writeLock().lock(); - subscriptions.removeForClient(clientID); - rwLock.writeLock().unlock(); + try { + m_inboundQueue.put(new RemoveAllSubscriptionsEvent(clientID)); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } } public void close() { - rwLock.writeLock().lock(); + try { + m_inboundQueue.put(new CloseEvent()); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } + } + + + protected void eventLoop() { + LOG.debug("Started event loop"); + boolean interrupted = false; + while (!interrupted) { + try { + MessagingEvent evt = m_inboundQueue.take(); + if (evt instanceof PublishEvent) { + processPublish((PublishEvent) evt); + } else if (evt instanceof SubscribeEvent) { + processSubscribe((SubscribeEvent) evt); + } else if (evt instanceof UnsubscribeEvent) { + processUnsubscribe((UnsubscribeEvent) evt); + } else if (evt instanceof RemoveAllSubscriptionsEvent) { + processRemoveAllSubscriptions((RemoveAllSubscriptionsEvent) evt); + } else if (evt instanceof CloseEvent) { + processClose(); + } + } catch (InterruptedException ex) { + processClose(); + interrupted = true; + } + } + } + + protected void processPublish(PublishEvent evt) { + LOG.debug("processPublish invoked"); + String topic = evt.getTopic(); + QOSType qos = evt.getQos(); + byte[] message = evt.getMessage(); + boolean retain = evt.isRetain(); + + for (Subscription sub : subscriptions.matches(topic)) { + m_notifier.notify(sub.clientId, topic, qos, message, false); + } + if (retain) { + 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)); + } + } + } + + + protected void processSubscribe(SubscribeEvent evt) { + LOG.debug("processSubscribe invoked"); + Subscription newSubscription = evt.getSubscription(); + String topic = newSubscription.getTopic(); + + subscriptions.add(newSubscription); + + //scans retained messages to be published to the new subscription + LOG.debug("Scanning all retained messages..."); + for (Map.Entry entry : m_retainedStore) { + StoredMessage storedMsg = entry.getValue(); + if (matchTopics(entry.getKey(), topic)) { + //fire the as retained the message + m_notifier.notify(newSubscription.clientId, topic, storedMsg.getQos(), + storedMsg.getPayload(), true); + } + } + } + + protected void processUnsubscribe(UnsubscribeEvent evt) { + LOG.debug("processSubscribe invoked"); + subscriptions.removeSubscription(evt.getTopic(), evt.getClientID()); + } + + protected void processRemoveAllSubscriptions(RemoveAllSubscriptionsEvent evt) { + LOG.debug("processRemoveAllSubscriptions invoked"); + subscriptions.removeForClient(evt.getClientID()); + } + + private void processClose() { + LOG.debug("processClose invoked"); try { pageFactory.close(); } catch (IOException ex) { LOG.error(null, ex); - } finally { - rwLock.writeLock().unlock(); } } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java new file mode 100644 index 00000000..8ca2119e --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java @@ -0,0 +1,9 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + * + * @author andrea + */ +public class CloseEvent extends MessagingEvent { + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/MessagingEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/MessagingEvent.java new file mode 100644 index 00000000..6fcdc55b --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/MessagingEvent.java @@ -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 { + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java new file mode 100644 index 00000000..355a56a3 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java @@ -0,0 +1,37 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; + +/** + * + * @author andrea + */ +public class PublishEvent extends MessagingEvent { + String m_topic; + QOSType m_qos; + byte[] m_message; + boolean m_retain; + + public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain) { + m_topic = topic; + m_qos = qos; + m_message = message; + m_retain = retain; + } + + public String getTopic() { + return m_topic; + } + + public QOSType getQos() { + return m_qos; + } + + public byte[] getMessage() { + return m_message; + } + + public boolean isRetain() { + return m_retain; + } +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RemoveAllSubscriptionsEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RemoveAllSubscriptionsEvent.java new file mode 100644 index 00000000..e2832185 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RemoveAllSubscriptionsEvent.java @@ -0,0 +1,19 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + * + * @author andrea + */ +public class RemoveAllSubscriptionsEvent extends MessagingEvent { + + String m_clientID; + + public RemoveAllSubscriptionsEvent(String clientID) { + m_clientID = clientID; + } + + public String getClientID() { + return m_clientID; + } + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java new file mode 100644 index 00000000..334e7e48 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java @@ -0,0 +1,21 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +import org.dna.mqtt.moquette.messaging.spi.impl.Subscription; + +/** + * + * @author andrea + */ +public class SubscribeEvent extends MessagingEvent { + + Subscription m_subscription; + + public SubscribeEvent(Subscription subscription) { + m_subscription = subscription; + } + + public Subscription getSubscription() { + return m_subscription; + } + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/UnsubscribeEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/UnsubscribeEvent.java new file mode 100644 index 00000000..ab752088 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/UnsubscribeEvent.java @@ -0,0 +1,25 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + * + * @author andrea + */ +public class UnsubscribeEvent extends MessagingEvent { + + String m_topic; + String m_clientID; + + public UnsubscribeEvent(String topic, String clientID) { + m_topic = topic; + m_clientID = clientID; + } + + public String getTopic() { + return m_topic; + } + + public String getClientID() { + return m_clientID; + } + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java index c28236fd..644e64d4 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.server; import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; +import java.util.logging.Level; import org.apache.mina.core.service.IoAcceptor; import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; @@ -44,6 +45,7 @@ public class Server { File.separator + "moquette_store.hawtdb"; private IoAcceptor m_acceptor; SimpleMessaging messaging; + Thread messagingEventLoop; public static void main(String[] args) throws IOException { new Server().startServer(); @@ -77,7 +79,9 @@ public class Server { //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); @@ -100,6 +104,12 @@ public class Server { LOG.info("Server stopping..."); messaging.close(); + messagingEventLoop.interrupt(); + try { + messagingEventLoop.join(); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } for(IoSession session: m_acceptor.getManagedSessions().values()) { if(session.isConnected() && !session.isClosing()){