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

Changed SimpleMessaging to be event driven (Incoming requests)

Browse Source
This commit is contained in:
andrea
2012-08-27 11:48:21 +02:00
parent 56431a161e
commit 739f63bc40
8 changed files with 250 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+118 -48
View File
@@ -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<MessagingEvent> m_inboundQueue = new LinkedBlockingQueue<MessagingEvent>();
private INotifier m_notifier;
private MultiIndexFactory m_multiIndexFactory;
private PageFileFactory pageFactory;
@@ -89,6 +94,10 @@ public class SimpleMessaging implements IMessaging {
m_retainedStore = (SortedIndex<String, StoredMessage>) 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<String, StoredMessage> 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<String, StoredMessage> 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();
}
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java
+9
View File
@@ -0,0 +1,9 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
/**
*
* @author andrea
*/
public class CloseEvent extends MessagingEvent {
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/MessagingEvent.java
+10
View File
@@ -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 {
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
+37
View File
@@ -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;
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RemoveAllSubscriptionsEvent.java
+19
View File
@@ -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;
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java
+21
View File
@@ -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;
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/UnsubscribeEvent.java
+25
View File
@@ -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;
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
+11 -1
View File
@@ -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()){