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

Merge branch 'seda_migration'

Browse Source
This commit is contained in:
selva.andre
2012-08-29 22:35:58 +02:00
parent 719b8f1894 87791963b6
commit 802d07c894
21 changed files with 503 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarking/src/main/java/org/dna/mqtt/bechnmark/Consumer.java
+1 -3
View File
@@ -15,8 +15,6 @@ import org.slf4j.LoggerFactory;
*/
public class Consumer implements Runnable {
static final int PUB_LOOP = 10000;
private static final Logger LOG = LoggerFactory.getLogger(Consumer.class);
private String m_clientID;
@@ -54,7 +52,7 @@ public class Consumer implements Runnable {
}
Message message = null;
for (int i = 0; i < PUB_LOOP; i++) {
for (int i = 0; i < Producer.PUB_LOOP; i++) {
try {
message = connection.receive();
} catch (Exception ex) {
benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java
+1 -1
View File
@@ -19,7 +19,7 @@ public class Producer implements Runnable {
private String m_clientID;
static final int PUB_LOOP = 10000;
public static final int PUB_LOOP = 1000000;
public Producer(String clientID) {
m_clientID = clientID;
parser/src/main/java/org/dna/mqtt/moquette/client/Client.java
+1 -44
View File
@@ -266,51 +266,8 @@ public final class Client {
// m_inflightIDs.add(messageID);
register(topic, publishCallback);
// boolean unlocked = false; //true when a SUBACK is received
try {
manageSendQoS1(msg);
// for (int retries = 0; retries < RETRIES_QOS_GT0 || !unlocked; retries++) {
// if (retries > 0) {
// msg.setDupFlag(true);
// }
//
// WriteFuture wf = m_session.write(msg);
// try {
// wf.await();
// } catch (InterruptedException ex) {
// LOG.error(null, ex);
// throw new SubscribeException(ex);
// }
// LOG.info("subscribe message sent");
//
// Throwable ex = wf.getException();
// if (ex != null) {
// throw new SubscribeException(ex);
// }
//
// //wait for the SubAck
// m_subscribeBarrier = new CountDownLatch(1);
//
// //suspend until the server respond with CONN_ACK
// try {
// LOG.info("subscribe waiting for suback");
// unlocked = m_subscribeBarrier.await(SUBACK_TIMEOUT, TimeUnit.MILLISECONDS); //TODO parametrize
// } catch (InterruptedException iex) {
// throw new SubscribeException(iex);
// }
// }
//
// //if not arrive into certain limit, raise an error
// if (!unlocked) {
// throw new SubscribeException(String.format("Server doesn't replyed with a SUB_ACK after %d replies", RETRIES_QOS_GT0));
// } else {
// //check if message ID match
// if (m_receivedSubAckMessageID != messageID) {
// throw new SubscribeException(String.format("Server replyed with "
// + "a broken MessageID in SUB_ACK, expected %d but received %d",
// messageID, m_receivedSubAckMessageID));
// }
// }
} catch(Throwable ex) {
//in case errors arise, remove the registration because the subscription
// hasn't get well
@@ -318,7 +275,6 @@ public final class Client {
throw new MQTTException(ex);
}
// register(topic, publishCallback);
updatePinger();
}
@@ -355,6 +311,7 @@ public final class Client {
int messageID = msg.getMessageID();
boolean unlocked = false;
for (int retries = 0; retries < RETRIES_QOS_GT0 || !unlocked; retries++) {
LOG.debug("manageSendQoS1 retry " + retries);
if (retries > 0) {
msg.setDupFlag(true);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+3
View File
@@ -1,5 +1,6 @@
package org.dna.mqtt.moquette.messaging.spi;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
/**
@@ -35,4 +36,6 @@ public interface IMessaging {
* doesn't reply any error
*/
void unsubscribe(String topic, String clientID);
void disconnect(IoSession session);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java
+3
View File
@@ -1,5 +1,6 @@
package org.dna.mqtt.moquette.messaging.spi;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
/**
@@ -9,5 +10,7 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
public interface INotifier {
public void notify(String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained);
public void disconnect(IoSession session);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+147 -53
View File
@@ -6,12 +6,20 @@ 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.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.CloseEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.DisconnectEvent;
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;
import org.dna.mqtt.moquette.messaging.spi.impl.events.RemoveAllSubscriptionsEvent;
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 +35,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,17 +57,16 @@ 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 INotifier m_notifier;
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;
private SortedIndex<String, StoredMessage> m_retainedStore;
public SimpleMessaging() {
String storeFile = Server.STORAGE_FILE_PATH;
@@ -89,62 +96,47 @@ public class SimpleMessaging implements IMessaging {
m_retainedStore = (SortedIndex<String, StoredMessage>) m_multiIndexFactory.openOrCreate("retained", indexFactory);
}
public void setNotifier(INotifier notifier) {
m_notifier = notifier;
public void run() {
eventLoop();
}
public void publish(String topic, byte[] message, QOSType qos, boolean retain) {
//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();
LOG.debug("subscribe invoked for topic: " + topic);
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);
}
}
public void disconnect(IoSession session) {
try {
m_inboundQueue.put(new DisconnectEvent(session));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
@@ -180,24 +172,126 @@ 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);
}
}
public BlockingQueue<MessagingEvent> getNotifyEventQueue() {
return m_outboundQueue;
}
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();
} else if (evt instanceof DisconnectEvent) {
m_outboundQueue.put(evt);
}
} catch (InterruptedException ex) {
processClose();
interrupted = true;
}
}
}
protected void processPublish(PublishEvent evt) {
LOG.debug("processPublish invoked");
final String topic = evt.getTopic();
final QOSType qos = evt.getQos();
final byte[] message = evt.getMessage();
boolean retain = evt.isRetain();
for (final Subscription sub : subscriptions.matches(topic)) {
try {
m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
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)/* throws InterruptedException*/ {
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, presents are " + m_retainedStore.size());
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_outboundQueue.put(new NotifyEvent(newSubscription.clientId,
topic, storedMsg.getQos(), storedMsg.getPayload(), true));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
}
}
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/SubscriptionsStore.java
+2 -24
View File
@@ -200,6 +200,7 @@ public class SubscriptionsStore {
* client's topics subscriptions
*/
public void init(MultiIndexFactory multiFactory) {
LOG.debug("init invoked");
BTreeIndexFactory<String, List<Subscription>> indexFactory =
new BTreeIndexFactory<String, List<Subscription>>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
@@ -210,6 +211,7 @@ public class SubscriptionsStore {
LOG.debug("Reloading all stored subscriptions...");
for (Map.Entry<String, List<Subscription>> entry : m_persistent) {
for (Subscription subscription : entry.getValue()) {
LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic());
addDirect(subscription);
}
}
@@ -217,30 +219,6 @@ public class SubscriptionsStore {
}
protected void addDirect(Subscription newSubscription) {
// List<Token> tokens = new ArrayList<Token>();
// try {
// tokens = splitTopic(newSubscription.topic);
// } catch (ParseException ex) {
// //TODO handle the parse exception
// LOG.error(null, ex);
//// return;
// }
//
// TreeNode current = subscriptions;
// for (Token token : tokens) {
// TreeNode matchingChildren;
//
// //check if a children with the same token already exists
// if ((matchingChildren = current.childWithToken(token)) != null) {
// current = matchingChildren;
// } else {
// //create a new node for the newly inserted token
// matchingChildren = new TreeNode(current);
// matchingChildren.setToken(token);
// current.addChild(matchingChildren);
// current = matchingChildren;
// }
// }
TreeNode current = findMatchingNode(newSubscription.topic);
current.addSubcription(newSubscription);
}
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/DisconnectEvent.java
+22
View File
@@ -0,0 +1,22 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
import org.apache.mina.core.session.IoSession;
/**
*
* @author andrea
*/
public class DisconnectEvent extends MessagingEvent {
IoSession m_session;
public DisconnectEvent(IoSession session) {
m_session = session;
}
public IoSession getSession() {
return m_session;
}
}
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/NotifyEvent.java
+46
View File
@@ -0,0 +1,46 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
/**
*
* @author andrea
*/
public class NotifyEvent extends MessagingEvent {
private String m_clientId;
private String m_topic;
private QOSType m_qos;
private byte[] m_message;
private boolean m_retaned;
public NotifyEvent(String clientId, String topic, QOSType qos, byte[] message, boolean retained) {
m_clientId = clientId;
m_topic = topic;
m_qos = qos;
m_message = message;
m_retaned = retained;
}
public String getClientId() {
return m_clientId;
}
public String getTopic() {
return m_topic;
}
public QOSType getQos() {
return m_qos;
}
public byte[] getMessage() {
return m_message;
}
public boolean isRetained() {
return m_retaned;
}
}
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/MQTTHandler.java
+35 -13
View File
@@ -63,7 +63,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
break;
case DISCONNECT:
handleDisconnect(session, (DisconnectMessage) msg);
break;
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
@@ -152,7 +152,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
}
protected void handleSubscribe(IoSession session, SubscribeMessage msg) {
LOG.info("registering the subscriptions");
LOG.debug("handleSubscribe, registering the subscriptions");
for (SubscribeMessage.Couple req : msg.subscriptions()) {
m_messaging.subscribe((String) session.getAttribute(ATTR_CLIENTID),
req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()]);
@@ -191,12 +191,12 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
protected void handleDisconnect(IoSession session, DisconnectMessage disconnectMessage) {
String clientID = (String) session.getAttribute(ATTR_CLIENTID);
//remove from clientIDs
m_clientIDsLock.lock();
try {
m_clientIDs.remove(clientID);
} finally {
m_clientIDsLock.unlock();
}
// m_clientIDsLock.lock();
// try {
// m_clientIDs.remove(clientID);
// } finally {
// m_clientIDsLock.unlock();
// }
boolean cleanSession = (Boolean) session.getAttribute("cleanSession");
if (cleanSession) {
//cleanup topic subscriptions
@@ -204,7 +204,8 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
}
//close the TCP connection
session.close(true);
//session.close(true);
m_messaging.disconnect(session);
}
@Override
@@ -226,9 +227,30 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
pubMessage.setTopicName(topic);
pubMessage.setQos(qOSType);
pubMessage.setPayload(payload);
assert m_clientIDs != null;
assert m_clientIDs.get(clientId) != null;
LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession());
m_clientIDs.get(clientId).getSession().write(pubMessage);
LOG.debug("notify invoked");
m_clientIDsLock.lock();
try {
assert m_clientIDs != null;
LOG.debug("clientIDs are " + m_clientIDs);
assert m_clientIDs.get(clientId) != null;
LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession());
m_clientIDs.get(clientId).getSession().write(pubMessage);
}catch(Throwable t) {
LOG.error(null, t);
} finally {
m_clientIDsLock.unlock();
}
}
public void disconnect(IoSession session) {
String clientID = (String) session.getAttribute(ATTR_CLIENTID);
m_clientIDsLock.lock();
try {
m_clientIDs.remove(clientID);
} finally {
m_clientIDsLock.unlock();
}
session.close(true);
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java
+48
View File
@@ -0,0 +1,48 @@
package org.dna.mqtt.moquette.server;
import java.util.concurrent.BlockingQueue;
import org.dna.mqtt.moquette.messaging.spi.INotifier;
import org.dna.mqtt.moquette.messaging.spi.impl.events.DisconnectEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent;
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;
Notifier(BlockingQueue<MessagingEvent> queue, INotifier notifier) {
m_queue = queue;
m_notifier = notifier;
}
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.getClientId(), evtN.getTopic(), evtN.getQos(), evtN.getMessage(), evtN.isRetained());
} else if (evt instanceof DisconnectEvent) {
DisconnectEvent evtD = (DisconnectEvent) evt;
m_notifier.disconnect(evtD.getSession());
}
} catch (InterruptedException ex) {
interrupted = true;
}
}
LOG.debug("Notifier stopped");
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
+28 -1
View File
@@ -3,6 +3,9 @@ package org.dna.mqtt.moquette.server;
import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
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.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
@@ -11,6 +14,7 @@ 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.ConnAckEncoder;
import org.dna.mqtt.moquette.proto.ConnectDecoder;
import org.dna.mqtt.moquette.proto.DisconnectDecoder;
@@ -42,8 +46,13 @@ public class Server {
public static final int DEFAULT_CONNECT_TIMEOUT = 10;
public static final String STORAGE_FILE_PATH = System.getProperty("user.home") +
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 IoAcceptor m_acceptor;
SimpleMessaging messaging;
Thread messagingEventLoop;
private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/;
public static void main(String[] args) throws IOException {
new Server().startServer();
@@ -76,8 +85,17 @@ 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);
BlockingQueue<MessagingEvent> notifyQueue = messaging.getNotifyEventQueue();
for (int i = 0; i < NOTIFIER_POOL_SIZE; i++) {
m_notifierPool.submit(new Notifier(notifyQueue, handler));
}
m_acceptor.setHandler(handler);
((NioSocketAcceptor)m_acceptor).setReuseAddress(true);
@@ -100,6 +118,15 @@ public class Server {
LOG.info("Server stopping...");
messaging.close();
messagingEventLoop.interrupt();
LOG.info("shutting down evet loop");
try {
messagingEventLoop.join();
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
m_notifierPool.shutdown();
for(IoSession session: m_acceptor.getManagedSessions().values()) {
if(session.isConnected() && !session.isClosing()){
parser/src/main/resources/log4j.properties
+7 -1
View File
@@ -1,5 +1,11 @@
# Set root logger level to DEBUG and its only appender to A1.
log4j.rootLogger=DEBUG, A1
log4j.rootLogger=ERROR, A1
#log4j.rootLogger=DEBUG, A1
#log4j.rootLogger=WARN, A1
#log4j.logger.org.dna.mqtt.moquette.client=WARN, A1
#log4j.logger.org.mina=WARN, A1
#log4j.logger.org.dna.mqtt.moquette.server=DEBUG, A1
#log4j.logger.org.dna.mqtt.moquette.messaging=DEBUG, A1
# A1 is set to be a ConsoleAppender.
log4j.appender.A1=org.apache.log4j.ConsoleAppender
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
+28 -14
View File
@@ -1,12 +1,14 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import org.dna.mqtt.moquette.messaging.spi.INotifier;
import java.util.concurrent.BlockingQueue;
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;
import org.dna.mqtt.moquette.messaging.spi.impl.events.SubscribeEvent;
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.*;
/**
*
@@ -26,7 +28,8 @@ public class SimpleMessagingTest {
@Test
public void testSubscribe() {
//Exercise
messaging.subscribe(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE);
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE));
messaging.processSubscribe(evt);
//Verify
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE);
@@ -35,10 +38,11 @@ public class SimpleMessagingTest {
@Test
public void testDoubleSubscribe() {
messaging.subscribe(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE);
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE));
messaging.processSubscribe(evt);
//Exercise
messaging.subscribe(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE);
messaging.processSubscribe(evt);
//Verify
Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE);
@@ -46,18 +50,28 @@ public class SimpleMessagingTest {
}
@Test
public void testPublish() {
messaging.subscribe(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE);
INotifier notifier = mock(INotifier.class);
messaging.setNotifier(notifier);
public void testPublish() throws InterruptedException {
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE));
messaging.processSubscribe(evt);
// INotifier notifier = mock(INotifier.class);
// messaging.setNotifier(notifier);
//Exercise
messaging.publish(FAKE_TOPIC, "Hello".getBytes(), QOSType.MOST_ONE, false);
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false);
messaging.processPublish(pubEvt);
//Verify
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(argument.getValue()));
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()));
}
@Test
parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java
+10 -4
View File
@@ -9,6 +9,7 @@ import org.dna.mqtt.moquette.client.IPublishCallback;
import org.junit.After;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
/**
@@ -163,7 +164,8 @@ public class ServerIntegrationTest {
@Test
public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws IOException {
public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws IOException, InterruptedException {
final CountDownLatch barrier = new CountDownLatch(1);
Client client = new Client("localhost", Server.PORT, "CLID_123");
client.connect(false); //without session cleanup
@@ -185,6 +187,7 @@ public class ServerIntegrationTest {
public void published(String topic, byte[] message) {
received = true;
barrier.countDown();
}
});
client.connect(false);
@@ -192,6 +195,7 @@ public class ServerIntegrationTest {
client.close();
//Verify
barrier.await(1, TimeUnit.SECONDS);
assertTrue(received);
//TearDown
@@ -200,7 +204,8 @@ public class ServerIntegrationTest {
@Test
public void testRetain_maintainMessage_againstClientDestruction() {
public void testRetain_maintainMessage_againstClientDestruction() throws InterruptedException {
final CountDownLatch barrier = new CountDownLatch(1);
Client client = new Client("localhost", Server.PORT, "CLID_123");
client.connect();
@@ -215,11 +220,13 @@ public class ServerIntegrationTest {
public void published(String topic, byte[] message) {
received = true;
barrier.countDown();
}
});
client.close();
//Verify
barrier.await(1, TimeUnit.SECONDS);
assertTrue(received);
//TearDown
@@ -227,7 +234,7 @@ public class ServerIntegrationTest {
}
@Test
@Ignore
public void testUnsubscribe_do_not_notify_anymore_same_session() throws InterruptedException {
Client client = new Client("localhost", Server.PORT, "CLID_123");
// Client client = new Client("test.mosquitto.org", 1883);
@@ -305,7 +312,6 @@ public class ServerIntegrationTest {
client.close();
client.shutdown();
}
}