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

* Moved the initialization of SimpleMessaging into the event loop, with an InitEvent

Browse Source 
* Modified SimpleMessaging to be a reusable singleton, with init and close methods
This commit is contained in:
selva.andre
2012-09-03 21:52:52 +02:00
parent a2c5c83a48
commit 870eb12390
9 changed files with 111 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java
+1 -7
View File
@@ -4,13 +4,7 @@ import java.io.*;
import java.net.URISyntaxException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import org.fusesource.mqtt.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+1 -1
View File
@@ -29,7 +29,7 @@ public interface IMessaging {
/**
* Subscribe a client to a specified topic with a defined level
*/
void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession);
void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID);
/**
* Remove all subscription to any topic the client (identified by clientID)
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+5 -2
View File
@@ -38,6 +38,7 @@ public class HawtDBStorageService implements IStorageService {
private SortedIndex<String, PublishEvent> m_inflightStore;
//persistent Map of clientID, list of Subscriptions
//TODO move to Set to avoid double subscriptions
private SortedIndex<String, List<Subscription>> m_persistentSubscriptions;
public HawtDBStorageService() {
@@ -161,8 +162,10 @@ public class HawtDBStorageService implements IStorageService {
}
List<Subscription> subs = m_persistentSubscriptions.get(clientID);
subs.add(newSubscription);
m_persistentSubscriptions.put(clientID, subs);
if (!subs.contains(newSubscription)) {
subs.add(newSubscription);
m_persistentSubscriptions.put(clientID, subs);
}
}
public void removeAllSubscriptions(String clientID) {
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+62 -20
View File
@@ -20,11 +20,8 @@ import org.dna.mqtt.moquette.messaging.spi.INotifier;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
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.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.IAuthenticator;
@@ -63,7 +60,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
//private BlockingQueue<MessagingEvent> m_inboundQueue = new LinkedBlockingQueue<MessagingEvent>();
private RingBuffer<ValueEvent> m_ringBuffer;
private INotifier m_notifier;
// private INotifier m_notifier;
private IStorageService m_storageService;
@@ -71,27 +68,51 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
private IAuthenticator m_authenticator;
private ExecutorService m_executor = Executors.newFixedThreadPool(1);
private ExecutorService m_executor;
BatchEventProcessor<ValueEvent> m_eventProcessor;
private static SimpleMessaging INSTANCE;
public SimpleMessaging() {
m_ringBuffer = new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32);
private SimpleMessaging() {
/*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()); */
m_storageService = new HawtDBStorageService();
/* m_storageService = new HawtDBStorageService();
m_storageService.initStore();
subscriptions.init(m_storageService);
subscriptions.init(m_storageService); */
}
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) {
/* public void setNotifier(INotifier notifier) {
m_notifier= notifier;
}
} */
// public void run() {
// eventLoop();
@@ -115,10 +136,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
disruptorPublish(new PublishEvent(topic, qos, message, retain, clientID, messageID, session));
}
public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession) {
public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID) {
Subscription newSubscription = new Subscription(clientId, topic, qos, cleanSession);
LOG.debug("subscribe invoked for topic: " + topic);
disruptorPublish(new SubscribeEvent(newSubscription));
disruptorPublish(new SubscribeEvent(newSubscription, messageID));
}
@@ -213,11 +234,16 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
processRepublish((RepublishEvent) evt);
} else if (evt instanceof ConnectEvent) {
processConnect((ConnectEvent) evt);
} else if (evt instanceof InitEvent) {
processInit();
}
}
public void stop() {
m_eventProcessor.halt();
private void processInit() {
m_storageService = new HawtDBStorageService();
m_storageService.initStore();
subscriptions.init(m_storageService);
}
// protected void eventLoop() {
@@ -383,16 +409,29 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
//scans retained messages to be published to the new subscription
Collection<StoredMessage> messages = m_storageService.searchMatching(new IMatchingCondition() {
public boolean match(String key) {
return matchTopics(key, topic); //To change body of implemented methods use File | Settings | File Templates.
return matchTopics(key, topic);
}
});
for (StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("Inserting NotifyEvent into outbound for topic " + topic);
m_notifier.notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(),
notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(),
storedMsg.getPayload(), true));
}
// //ack the client
// SubAckMessage ackMessage = new SubAckMessage();
// ackMessage.setMessageID(evt.getMessageID());
//
// //TODO by now it handles only QoS 0 messages
// /*for (int i = 0; i < evt.subscriptions().size(); i++) {
// ackMessage.addType(QOSType.MOST_ONE);
// } */
// LOG.info("replying with SubAct to MSG ID {0}", evt.getMessageID());
// //session.write(ackMessage);
// IoSession session = m_clientIDs.get(evt.getSubscription().clientId).getSession();
// session.write(ackMessage).awaitUninterruptibly();
}
protected void processUnsubscribe(UnsubscribeEvent evt) {
@@ -423,6 +462,9 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
private void processClose() {
LOG.debug("processClose invoked");
m_storageService.close();
// m_eventProcessor.halt();
m_executor.shutdown();
}
private void processRepublish(RepublishEvent evt) throws InterruptedException {
@@ -432,7 +474,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
}
for (PublishEvent pubEvt : publishedEvents) {
m_notifier.notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID()));
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java
+6
View File
@@ -0,0 +1,6 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
/**
*/
public class InitEvent extends MessagingEvent {
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java
+8 -2
View File
@@ -9,13 +9,19 @@ import org.dna.mqtt.moquette.messaging.spi.impl.Subscription;
public class SubscribeEvent extends MessagingEvent {
Subscription m_subscription;
int m_messageID;
public SubscribeEvent(Subscription subscription) {
public SubscribeEvent(Subscription subscription, int messageID) {
m_subscription = subscription;
m_messageID = messageID;
}
public Subscription getSubscription() {
return m_subscription;
}
public int getMessageID() {
return m_messageID;
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+1 -1
View File
@@ -75,7 +75,7 @@ public class MQTTHandler extends IoHandlerAdapter/* implements INotifier*/ {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
m_messaging.subscribe(clientID, req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()],
cleanSession);
cleanSession, msg.getMessageID());
}
//ack the client
parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
+12 -4
View File
@@ -7,6 +7,7 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
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;
@@ -14,7 +15,6 @@ 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.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.proto.*;
import org.dna.mqtt.moquette.proto.messages.*;
import org.slf4j.Logger;
@@ -68,7 +68,8 @@ public class Server {
m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
MQTTHandler handler = new MQTTHandler();
messaging = new SimpleMessaging();
messaging = SimpleMessaging.getInstance();
messaging.init();
//TODO fix this hugly wiring
handler.setMessaging(messaging);
// messaging.setNotifier(handler);
@@ -97,7 +98,7 @@ public class Server {
LOG.info("Server stopping...");
messaging.close();
messaging.stop();
//messaging.stop();
// messagingEventLoop.interrupt();
// LOG.info("shutting down evet loop");
// try {
@@ -113,7 +114,14 @@ public class Server {
session.close(false);
}
}
//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()));*/
m_acceptor.unbind();
m_acceptor.dispose();
LOG.info("Server stopped");
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
+15 -10
View File
@@ -45,8 +45,9 @@ public class SimpleMessagingTest {
AbstractMessage m_receivedMessage;
@Before
public void setUp() {
messaging = new SimpleMessaging();
public void setUp() throws InterruptedException {
messaging = SimpleMessaging.getInstance();
messaging.init();
connMsg = new ConnectMessage();
connMsg.setProcotolVersion((byte) 0x03);
@@ -68,17 +69,21 @@ public class SimpleMessagingTest {
}
}
});
//sleep to let the messaging batch processor to process the initEvent
Thread.sleep(300);
}
@After
public void tearDown() {
m_receivedMessage = null;
messaging.close();
}
@Test
public void testSubscribe() {
//Exercise
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
messaging.processSubscribe(evt);
//Verify
@@ -88,7 +93,7 @@ public class SimpleMessagingTest {
@Test
public void testDoubleSubscribe() {
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
messaging.processSubscribe(evt);
//Exercise
@@ -105,7 +110,7 @@ public class SimpleMessagingTest {
ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, true);
messaging.m_clientIDs.put(FAKE_CLIENT_ID, connDescr);
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
messaging.processSubscribe(evt);
//Exercise
@@ -116,13 +121,13 @@ public class SimpleMessagingTest {
assertNotNull(m_receivedMessage);
//TODO check received message attributes
}
@Test
public void testMatchTopics_simple() {
assertTrue(messaging.matchTopics("/", "/"));
assertTrue(messaging.matchTopics("/finance", "/finance"));
}
@Test
public void testMatchTopics_multi() {
assertTrue(messaging.matchTopics("finance", "#"));
@@ -130,8 +135,8 @@ public class SimpleMessagingTest {
assertTrue(messaging.matchTopics("finance/stock", "finance/#"));
assertTrue(messaging.matchTopics("finance/stock/ibm", "finance/#"));
}
@Test
public void testMatchTopics_single() {
assertTrue(messaging.matchTopics("finance", "+"));
@@ -203,7 +208,7 @@ public class SimpleMessagingTest {
//Exercise
//m_handler.setMessaging(mockedMessaging);
SubscribeEvent pubEvt = new SubscribeEvent(subscription);
SubscribeEvent pubEvt = new SubscribeEvent(subscription, 0);
messaging.processSubscribe(pubEvt);
//Verify