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

Moved IoSessions management into the event loop

Browse Source
This commit is contained in:
selva.andre
2012-09-02 11:31:18 +02:00
parent 21d6d5880c
commit 217e754de2
5 changed files with 249 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+3
View File
@@ -4,6 +4,7 @@ import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
/**
* Interface to the underling messaging system used to publish, subscribe.
@@ -47,4 +48,6 @@ public interface IMessaging {
void disconnect(IoSession session);
void republishStored(String clientID);
void connect(IoSession session, ConnectMessage msg);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+154 -5
View File
@@ -1,5 +1,6 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
@@ -8,14 +9,22 @@ 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.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;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Serializable;
import java.text.ParseException;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
@@ -53,6 +62,10 @@ public class SimpleMessaging implements IMessaging, Runnable {
private INotifier m_notifier;
private IStorageService m_storageService;
Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<String, ConnectionDescriptor>();
private IAuthenticator m_authenticator;
public SimpleMessaging() {
m_storageService = new HawtDBStorageService();
@@ -129,6 +142,14 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
}
public void connect(IoSession session, ConnectMessage msg) {
try {
m_inboundQueue.put(new ConnectEvent(session, msg));
} catch (InterruptedException iex) {
LOG.error(null, iex);
}
}
/**
* Verify if the 2 topics matching respecting the rules of MQTT Appendix A
*/
@@ -207,6 +228,8 @@ public class SimpleMessaging implements IMessaging, Runnable {
m_storageService.cleanInFlight(((CleanInFlightEvent) evt).getMsgId());
} else if (evt instanceof RepublishEvent) {
processRepublish((RepublishEvent) evt);
} else if (evt instanceof ConnectEvent) {
processConnect((ConnectEvent) evt);
}
} catch (InterruptedException ex) {
processClose();
@@ -215,6 +238,85 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
}
private void processConnect(ConnectEvent evt) {
ConnectMessage msg = evt.getMessage();
IoSession session = evt.getSession();
if (msg.getProcotolVersion() != 0x03) {
ConnAckMessage badProto = new ConnAckMessage();
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
session.write(badProto);
session.close(false);
return;
}
if (msg.getClientID() == null || msg.getClientID().length() > 23) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
session.write(okResp);
return;
}
//if an old client with the same ID already exists close its session.
if (m_clientIDs.containsKey(msg.getClientID())) {
//clean the subscriptions if the old used a cleanSession = true
IoSession oldSession = m_clientIDs.get(msg.getClientID()).getSession();
boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
removeSubscriptions(msg.getClientID());
}
m_clientIDs.get(msg.getClientID()).getSession().close(false);
}
ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession());
m_clientIDs.put(msg.getClientID(), connDescr);
int keepAlive = msg.getKeepAlive();
session.setAttribute("keepAlive", keepAlive);
session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
//used to track the client in the subscription and publishing phases.
session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID());
session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f));
//Handle will flag
if (msg.isWillFlag()) {
QOSType willQos = QOSType.values()[msg.getWillQos()];
publish(msg.getWillTopic(), msg.getWillMessage().getBytes(),
willQos, msg.isWillRetain(), msg.getClientID(), session);
}
//handle user authentication
if (msg.isUserFlag()) {
String pwd = null;
if (msg.isPasswordFlag()) {
pwd = msg.getPassword();
}
if (!m_authenticator.checkValid(msg.getUsername(), pwd)) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
session.write(okResp);
return;
}
}
//handle clean session flag
if (msg.isCleanSession()) {
//remove all prev subscriptions
//cleanup topic subscriptions
removeSubscriptions(msg.getClientID());
} else {
//force the republish of stored QoS1 and QoS2
republishStored(msg.getClientID());
}
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
session.write(okResp);
}
protected void processPublish(PublishEvent evt) throws InterruptedException {
LOG.debug("processPublish invoked");
final String topic = evt.getTopic();
@@ -234,20 +336,20 @@ public class SimpleMessaging implements IMessaging, Runnable {
for (final Subscription sub : subscriptions.matches(topic)) {
if (qos == QOSType.MOST_ONE) {
//QoS 0
m_notifier.notify(new NotifyEvent(sub.clientId, topic, qos, message, false));
notify(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
if (!sub.isCleanSession() && !sub.isActive()) {
m_storageService.storePublishForFuture(evt);
}
m_notifier.notify(new NotifyEvent(sub.clientId, topic, qos, message, false));
notify(new NotifyEvent(sub.clientId, topic, qos, message, false));
}
}
if (cleanEvt != null) {
refill(cleanEvt);
m_notifier.sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
}
if (retain) {
@@ -292,10 +394,14 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
private void processDisconnect(DisconnectEvent evt) throws InterruptedException {
m_notifier.disconnect(evt.getSession());
// m_notifier.disconnect(evt.getSession());
IoSession session = evt.getSession();
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
m_clientIDs.remove(clientID);
session.close(true);
//de-activate the subscriptions for this ClientID
String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
subscriptions.disconnect(clientID);
}
@@ -315,4 +421,47 @@ public class SimpleMessaging implements IMessaging, Runnable {
pubEvt.getMessage(), false, pubEvt.getMessageID()));
}
}
private void notify(NotifyEvent evt) {
LOG.debug("notify invoked with event " + evt);
String clientId = evt.getClientId();
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(evt.isRetained());
pubMessage.setTopicName(evt.getTopic());
pubMessage.setQos(evt.getQos());
pubMessage.setPayload(evt.getMessage());
if (pubMessage.getQos() != QOSType.MOST_ONE) {
pubMessage.setMessageID(evt.getMessageID());
}
LOG.debug("notify invoked");
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);
}
}
private void sendPubAck(PubAckEvent evt) {
LOG.debug("sendPubAck invoked");
String clientId = evt.getClientID();
PubAckMessage pubAckMessage = new PubAckMessage();
pubAckMessage.setMessageID(evt.getMessageId());
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(pubAckMessage);
}catch(Throwable t) {
LOG.error(null, t);
}
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ConnectEvent.java
+25
View File
@@ -0,0 +1,25 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
/**
*/
public class ConnectEvent extends MessagingEvent {
IoSession m_session;
ConnectMessage message;
public ConnectEvent(IoSession session, ConnectMessage message) {
this.m_session = session;
this.message = message;
}
public IoSession getSession() {
return m_session;
}
public ConnectMessage getMessage() {
return message;
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+66 -142
View File
@@ -22,16 +22,16 @@ import org.slf4j.LoggerFactory;
*
* @author andrea
*/
public class MQTTHandler extends IoHandlerAdapter implements INotifier {
public class MQTTHandler extends IoHandlerAdapter/* implements INotifier*/ {
private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class);
/**
* Maps CLIENT_ID to the IoSession that represents the connection
*/
Lock m_clientIDsLock = new ReentrantLock();
Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<String, ConnectionDescriptor>();
// Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<String, ConnectionDescriptor>();
private IMessaging m_messaging;
private IAuthenticator m_authenticator;
// private IAuthenticator m_authenticator;
@Override
public void messageReceived(IoSession session, Object message) throws Exception {
@@ -65,84 +65,8 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
protected void handleConnect(IoSession session, ConnectMessage msg) {
LOG.info("handleConnect invoked");
if (msg.getProcotolVersion() != 0x03) {
ConnAckMessage badProto = new ConnAckMessage();
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
session.write(badProto);
session.close(false);
return;
}
if (msg.getClientID() == null || msg.getClientID().length() > 23) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
session.write(okResp);
return;
}
m_clientIDsLock.lock();
try {
//if an old client with the same ID already exists close its session.
if (m_clientIDs.containsKey(msg.getClientID())) {
//clean the subscriptions if the old used a cleanSession = true
IoSession oldSession = m_clientIDs.get(msg.getClientID()).getSession();
boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
m_messaging.removeSubscriptions(msg.getClientID());
}
m_clientIDs.get(msg.getClientID()).getSession().close(false);
}
ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession());
m_clientIDs.put(msg.getClientID(), connDescr);
} finally {
m_clientIDsLock.unlock();
}
int keepAlive = msg.getKeepAlive();
session.setAttribute("keepAlive", keepAlive);
session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
//used to track the client in the subscription and publishing phases.
session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID());
session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f));
//Handle will flag
if (msg.isWillFlag()) {
QOSType willQos = QOSType.values()[msg.getWillQos()];
m_messaging.publish(msg.getWillTopic(), msg.getWillMessage().getBytes(),
willQos, msg.isWillRetain(), msg.getClientID(), session);
}
//handle user authentication
if (msg.isUserFlag()) {
String pwd = null;
if (msg.isPasswordFlag()) {
pwd = msg.getPassword();
}
if (!m_authenticator.checkValid(msg.getUsername(), pwd)) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
session.write(okResp);
return;
}
}
//handle clean session flag
if (msg.isCleanSession()) {
//remove all prev subscriptions
//cleanup topic subscriptions
m_messaging.removeSubscriptions(msg.getClientID());
} else {
//force the republish of stored QoS1 and QoS2
m_messaging.republishStored(msg.getClientID());
}
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
session.write(okResp);
m_messaging.connect(session, msg);
}
protected void handleSubscribe(IoSession session, SubscribeMessage msg) {
@@ -220,69 +144,69 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
m_messaging = messaging;
}
public void setAuthenticator(IAuthenticator authenticator) {
m_authenticator = authenticator;
}
// public void setAuthenticator(IAuthenticator authenticator) {
// m_authenticator = authenticator;
// }
public void notify(NotifyEvent evt) {
LOG.debug("notify invoked with event " + evt);
String clientId = evt.getClientId();
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(evt.isRetained());
pubMessage.setTopicName(evt.getTopic());
pubMessage.setQos(evt.getQos());
pubMessage.setPayload(evt.getMessage());
if (pubMessage.getQos() != QOSType.MOST_ONE) {
pubMessage.setMessageID(evt.getMessageID());
}
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(Constants.ATTR_CLIENTID);
m_clientIDsLock.lock();
try {
m_clientIDs.remove(clientID);
} finally {
m_clientIDsLock.unlock();
}
session.close(true);
}
public void sendPubAck(PubAckEvent evt) {
LOG.debug("sendPubAck invoked");
String clientId = evt.getClientID();
PubAckMessage pubAckMessage = new PubAckMessage();
pubAckMessage.setMessageID(evt.getMessageId());
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(pubAckMessage);
}catch(Throwable t) {
LOG.error(null, t);
} finally {
m_clientIDsLock.unlock();
}
}
// public void notify(NotifyEvent evt) {
// LOG.debug("notify invoked with event " + evt);
// String clientId = evt.getClientId();
// PublishMessage pubMessage = new PublishMessage();
// pubMessage.setRetainFlag(evt.isRetained());
// pubMessage.setTopicName(evt.getTopic());
// pubMessage.setQos(evt.getQos());
// pubMessage.setPayload(evt.getMessage());
// if (pubMessage.getQos() != QOSType.MOST_ONE) {
// pubMessage.setMessageID(evt.getMessageID());
// }
//
// 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(Constants.ATTR_CLIENTID);
// m_clientIDsLock.lock();
// try {
// m_clientIDs.remove(clientID);
// } finally {
// m_clientIDsLock.unlock();
// }
// session.close(true);
// }
//
// public void sendPubAck(PubAckEvent evt) {
// LOG.debug("sendPubAck invoked");
//
// String clientId = evt.getClientID();
//
// PubAckMessage pubAckMessage = new PubAckMessage();
// pubAckMessage.setMessageID(evt.getMessageId());
//
// 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(pubAckMessage);
// }catch(Throwable t) {
// LOG.error(null, t);
// } finally {
// m_clientIDsLock.unlock();
// }
// }
}
parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
+1 -1
View File
@@ -71,7 +71,7 @@ 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();