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

Finished to remove back notifier loop

Browse Source
This commit is contained in:
selva.andre
2012-09-01 13:16:40 +02:00
parent 9285691c62
commit 25efc898a0
4 changed files with 6 additions and 100 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
+6 -26
View File
@@ -56,8 +56,7 @@ public class SimpleMessaging implements IMessaging, Runnable {
private SubscriptionsStore subscriptions = new SubscriptionsStore();
private BlockingQueue<MessagingEvent> m_inboundQueue = new LinkedBlockingQueue<MessagingEvent>();
// private BlockingQueue<MessagingEvent> m_outboundQueue = new LinkedBlockingQueue<MessagingEvent>();
private INotifier m_notifier;
private MultiIndexFactory m_multiIndexFactory;
private PageFileFactory pageFactory;
@@ -238,10 +237,6 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
}
/*public BlockingQueue<MessagingEvent> getNotifyEventQueue() {
return m_outboundQueue;
}*/
protected void eventLoop() {
LOG.debug("Started event loop");
@@ -294,7 +289,6 @@ public class SimpleMessaging implements IMessaging, Runnable {
if (qos == QOSType.MOST_ONE) {
//QoS 0
m_notifier.notify(new NotifyEvent(sub.clientId, topic, qos, message, false));
// m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false));
} else {
//QoS 1 or 2
//if the target subscription is not clean session and is not connected => store it
@@ -310,16 +304,12 @@ public class SimpleMessaging implements IMessaging, Runnable {
m_persistentMessageStore.put(clientID, storedEvents);
}
m_notifier.notify(new NotifyEvent(sub.clientId, topic, qos, message, false));
// m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false, evt.getMessageID()));
}
}
if (cleanEvt != null) {
refill(cleanEvt);
m_notifier.sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
// m_outboundQueue.put(cleanEvt);
// m_outboundQueue.put(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
}
if (retain) {
@@ -334,7 +324,7 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
protected void processSubscribe(SubscribeEvent evt)/* throws InterruptedException*/ {
protected void processSubscribe(SubscribeEvent evt) {
LOG.debug("processSubscribe invoked");
Subscription newSubscription = evt.getSubscription();
String topic = newSubscription.getTopic();
@@ -346,16 +336,10 @@ public class SimpleMessaging implements IMessaging, Runnable {
for (Map.Entry<String, StoredMessage> entry : m_retainedStore) {
StoredMessage storedMsg = entry.getValue();
if (matchTopics(entry.getKey(), topic)) {
// try {
//fire the as retained the message
LOG.debug("Inserting NotifyEvent into outbound for topic " + topic + " entry " + entry.getKey());
m_notifier.notify(new NotifyEvent(newSubscription.clientId,
topic, storedMsg.getQos(), storedMsg.getPayload(), true));
// m_outboundQueue.put(new NotifyEvent(newSubscription.clientId,
// topic, storedMsg.getQos(), storedMsg.getPayload(), true));
// } catch (InterruptedException ex) {
// LOG.error(null, ex);
// }
//fire the as retained the message
LOG.debug("Inserting NotifyEvent into outbound for topic " + topic + " entry " + entry.getKey());
m_notifier.notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(),
storedMsg.getPayload(), true));
}
}
}
@@ -374,7 +358,6 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
private void processDisconnect(DisconnectEvent evt) throws InterruptedException {
// m_outboundQueue.put(evt);
m_notifier.disconnect(evt.getSession());
//de-activate the subscriptions for this ClientID
@@ -400,9 +383,6 @@ public class SimpleMessaging implements IMessaging, Runnable {
for (PublishEvent pubEvt : publishedEvents) {
m_notifier.notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID()));
// m_outboundQueue.put(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
// pubEvt.getMessage(), false, pubEvt.getMessageID()));
}
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java
-54
View File
@@ -1,54 +0,0 @@
package org.dna.mqtt.moquette.server;
import java.util.concurrent.BlockingQueue;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.INotifier;
import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
class Notifier implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(Notifier.class);
BlockingQueue<MessagingEvent> m_queue;
INotifier m_notifier;
IMessaging m_messaging;
Notifier(BlockingQueue<MessagingEvent> queue, INotifier notifier, IMessaging messaging) {
m_queue = queue;
m_notifier = notifier;
m_messaging = messaging;
}
public void run() {
LOG.debug("Notifier started");
boolean interrupted = false;
while (!interrupted) {
try {
MessagingEvent evt = m_queue.take();
LOG.debug("Notifing event " + evt);
if (evt instanceof NotifyEvent) {
NotifyEvent evtN = (NotifyEvent) evt;
m_notifier.notify(evtN);
} else if (evt instanceof DisconnectEvent) {
DisconnectEvent evtD = (DisconnectEvent) evt;
m_notifier.disconnect(evtD.getSession());
} else if (evt instanceof CleanInFlightEvent) {
m_messaging.refill(evt);
} else if (evt instanceof PubAckEvent) {
m_notifier.sendPubAck((PubAckEvent) evt);
}
} catch (InterruptedException ex) {
interrupted = true;
}
}
LOG.debug("Notifier stopped");
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
-8
View File
@@ -76,14 +76,6 @@ public class Server {
messagingEventLoop.setName("Event Loop" + System.currentTimeMillis());
messagingEventLoop.start();
/*m_notifierPool = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE);
BlockingQueue<MessagingEvent> notifyQueue = messaging.getNotifyEventQueue();
for (int i = 0; i < NOTIFIER_POOL_SIZE; i++) {
Notifier notifier = new Notifier(notifyQueue, handler, messaging);
m_notifierPool.submit(notifier);
}*/
m_acceptor.setHandler(handler);
((NioSocketAcceptor)m_acceptor).setReuseAddress(true);
((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true);
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
-12
View File
@@ -65,18 +65,6 @@ public class SimpleMessagingTest {
messaging.processPublish(pubEvt);
//Verify
// BlockingQueue<MessagingEvent> queue = messaging.getNotifyEventQueue();
// MessagingEvent msgEvt = queue.take();
// assertTrue(msgEvt instanceof NotifyEvent);
// NotifyEvent notifyEvt = (NotifyEvent) msgEvt;
// assertEquals(FAKE_CLIENT_ID, notifyEvt.getClientId());
// assertEquals(FAKE_TOPIC, notifyEvt.getTopic());
// assertFalse(notifyEvt.isRetained());
// ArgumentCaptor<byte[]> argument = ArgumentCaptor.forClass(byte[].class);
// verify(notifier).notify(eq(FAKE_CLIENT_ID), eq(FAKE_TOPIC), any(QOSType.class), argument.capture(), eq(false));
// assertEquals("Hello", new String(notifyEvt.getMessage()));
ArgumentCaptor<NotifyEvent> argument = ArgumentCaptor.forClass(NotifyEvent.class);
verify(notifier).notify(argument.capture());
}