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

Removed back notifier loop

Browse Source
This commit is contained in:
selva.andre
2012-09-01 12:58:40 +02:00
parent 42c04b074c
commit 9276ea8c98
3 changed files with 54 additions and 31 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
+31 -15
View File
@@ -12,6 +12,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import org.apache.mina.core.session.IoSession;
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.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
@@ -55,9 +56,9 @@ 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 BlockingQueue<MessagingEvent> m_outboundQueue = new LinkedBlockingQueue<MessagingEvent>();
// private INotifier m_notifier;
private INotifier m_notifier;
private MultiIndexFactory m_multiIndexFactory;
private PageFileFactory pageFactory;
private SortedIndex<String, StoredMessage> m_retainedStore;
@@ -115,6 +116,10 @@ public class SimpleMessaging implements IMessaging, Runnable {
m_persistentMessageStore = (SortedIndex<String, List<PublishEvent>>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory);
}
public void setNotifier(INotifier notifier) {
m_notifier= notifier;
}
public void run() {
eventLoop();
@@ -233,9 +238,9 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
}
public BlockingQueue<MessagingEvent> getNotifyEventQueue() {
/*public BlockingQueue<MessagingEvent> getNotifyEventQueue() {
return m_outboundQueue;
}
}*/
protected void eventLoop() {
@@ -288,7 +293,8 @@ public class SimpleMessaging implements IMessaging, Runnable {
for (final Subscription sub : subscriptions.matches(topic)) {
if (qos == QOSType.MOST_ONE) {
//QoS 0
m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false));
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
@@ -303,13 +309,17 @@ public class SimpleMessaging implements IMessaging, Runnable {
storedEvents.add(evt);
m_persistentMessageStore.put(clientID, storedEvents);
}
m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false, evt.getMessageID()));
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) {
m_outboundQueue.put(cleanEvt);
m_outboundQueue.put(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
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) {
@@ -336,14 +346,16 @@ public class SimpleMessaging implements IMessaging, Runnable {
for (Map.Entry<String, StoredMessage> entry : m_retainedStore) {
StoredMessage storedMsg = entry.getValue();
if (matchTopics(entry.getKey(), topic)) {
try {
// try {
//fire the as retained the message
LOG.debug("Inserting NotifyEvent into outbound for topic " + topic + " entry " + entry.getKey());
m_outboundQueue.put(new NotifyEvent(newSubscription.clientId,
m_notifier.notify(new NotifyEvent(newSubscription.clientId,
topic, storedMsg.getQos(), storedMsg.getPayload(), true));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
// m_outboundQueue.put(new NotifyEvent(newSubscription.clientId,
// topic, storedMsg.getQos(), storedMsg.getPayload(), true));
// } catch (InterruptedException ex) {
// LOG.error(null, ex);
// }
}
}
}
@@ -362,7 +374,8 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
private void processDisconnect(DisconnectEvent evt) throws InterruptedException {
m_outboundQueue.put(evt);
// m_outboundQueue.put(evt);
m_notifier.disconnect(evt.getSession());
//de-activate the subscriptions for this ClientID
String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
@@ -385,8 +398,11 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
for (PublishEvent pubEvt : publishedEvents) {
m_outboundQueue.put(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
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/Server.java
+6 -6
View File
@@ -33,11 +33,11 @@ public class Server {
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 static final int NOTIFIER_POOL_SIZE = 1;
private IoAcceptor m_acceptor;
SimpleMessaging messaging;
Thread messagingEventLoop;
private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/;
//private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/;
public static void main(String[] args) throws IOException {
new Server().startServer();
@@ -71,18 +71,18 @@ public class Server {
messaging = new SimpleMessaging();
//TODO fix this hugly wiring
handler.setMessaging(messaging);
// messaging.setNotifier(handler);
messaging.setNotifier(handler);
messagingEventLoop = new Thread(messaging);
messagingEventLoop.setName("Event Loop" + System.currentTimeMillis());
messagingEventLoop.start();
m_notifierPool = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE);
/*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);
@@ -113,7 +113,7 @@ public class Server {
LOG.error(null, ex);
}
m_notifierPool.shutdown();
/*m_notifierPool.shutdown();*/
for(IoSession session: m_acceptor.getManagedSessions().values()) {
if(session.isConnected() && !session.isClosing()){
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
+17 -10
View File
@@ -1,6 +1,8 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.util.concurrent.BlockingQueue;
import org.dna.mqtt.moquette.messaging.spi.INotifier;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
@@ -9,6 +11,8 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import static org.mockito.Mockito.*;
/**
*
@@ -53,25 +57,28 @@ public class SimpleMessagingTest {
public void testPublish() throws InterruptedException {
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
messaging.processSubscribe(evt);
// INotifier notifier = mock(INotifier.class);
// messaging.setNotifier(notifier);
INotifier notifier = mock(INotifier.class);
messaging.setNotifier(notifier);
//Exercise
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null);
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());
// 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()));
// assertEquals("Hello", new String(notifyEvt.getMessage()));
ArgumentCaptor<NotifyEvent> argument = ArgumentCaptor.forClass(NotifyEvent.class);
verify(notifier).notify(argument.capture());
}
@Test