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

First clean of Moquette's channel abstraction to user Netty's one

Browse Source
This commit is contained in:
andsel
2016-01-24 19:21:24 +01:00
parent c7e82497dd
commit beb556c73c
6 changed files with 60 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/ConnectionDescriptor.java
+4 -2
View File
@@ -15,6 +15,8 @@
*/ */
package io.moquette.server; package io.moquette.server;
import io.netty.channel.Channel;
/** /**
* Value object to maintain the information of single connection, like ClientID, IoSession, * Value object to maintain the information of single connection, like ClientID, IoSession,
* and other clean session fla. * and other clean session fla.
@@ -25,10 +27,10 @@ package io.moquette.server;
public class ConnectionDescriptor { public class ConnectionDescriptor {
public final String clientID; public final String clientID;
public final ServerChannel session; public final Channel session;
public final boolean cleanSession; public final boolean cleanSession;
public ConnectionDescriptor(String clientID, ServerChannel session, boolean cleanSession) { public ConnectionDescriptor(String clientID, Channel session, boolean cleanSession) {
this.clientID = clientID; this.clientID = clientID;
this.session = session; this.session = session;
this.cleanSession = cleanSession; this.cleanSession = cleanSession;
broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java
+2 -2
View File
@@ -49,7 +49,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
try { try {
switch (msg.getMessageType()) { switch (msg.getMessageType()) {
case CONNECT: case CONNECT:
m_processor.processConnect(new NettyChannel(ctx), (ConnectMessage) msg); m_processor.processConnect(ctx.channel(), (ConnectMessage) msg);
break; break;
case SUBSCRIBE: case SUBSCRIBE:
m_processor.processSubscribe(new NettyChannel(ctx), (SubscribeMessage) msg); m_processor.processSubscribe(new NettyChannel(ctx), (SubscribeMessage) msg);
@@ -96,7 +96,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
if (stolenAttr != null && stolenAttr == Boolean.TRUE) { if (stolenAttr != null && stolenAttr == Boolean.TRUE) {
stolen = stolenAttr; stolen = stolenAttr;
} }
m_processor.processConnectionLost(clientID, stolen, new NettyChannel(ctx)); m_processor.processConnectionLost(clientID, stolen, ctx.channel());
} }
ctx.close(); ctx.close();
} }
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+33 -23
View File
@@ -47,6 +47,9 @@ import io.moquette.proto.messages.SubAckMessage;
import io.moquette.proto.messages.SubscribeMessage; import io.moquette.proto.messages.SubscribeMessage;
import io.moquette.proto.messages.UnsubAckMessage; import io.moquette.proto.messages.UnsubAckMessage;
import io.moquette.proto.messages.UnsubscribeMessage; import io.moquette.proto.messages.UnsubscribeMessage;
import io.netty.channel.Channel;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@@ -133,14 +136,14 @@ public class ProtocolProcessor {
m_sessionsStore = sessionsStore; m_sessionsStore = sessionsStore;
} }
public void processConnect(ServerChannel session, ConnectMessage msg) { public void processConnect(Channel session, ConnectMessage msg) {
LOG.debug("CONNECT for client <{}>", msg.getClientID()); LOG.debug("CONNECT for client <{}>", msg.getClientID());
if (msg.getProtocolVersion() != VERSION_3_1 && msg.getProtocolVersion() != VERSION_3_1_1) { if (msg.getProtocolVersion() != VERSION_3_1 && msg.getProtocolVersion() != VERSION_3_1_1) {
ConnAckMessage badProto = new ConnAckMessage(); ConnAckMessage badProto = new ConnAckMessage();
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
LOG.warn("processConnect sent bad proto ConnAck"); LOG.warn("processConnect sent bad proto ConnAck");
session.write(badProto); session.write(badProto);
session.close(false); session.close();
return; return;
} }
@@ -165,7 +168,7 @@ public class ProtocolProcessor {
failedCredentials(session); failedCredentials(session);
return; return;
} }
session.setAttribute(NettyChannel.ATTR_KEY_USERNAME, msg.getUsername()); session.attr(NettyChannel.ATTR_KEY_USERNAME).set(msg.getUsername());
} else if (!this.allowAnonymous) { } else if (!this.allowAnonymous) {
failedCredentials(session); failedCredentials(session);
return; return;
@@ -175,11 +178,11 @@ public class ProtocolProcessor {
if (m_clientIDs.containsKey(msg.getClientID())) { if (m_clientIDs.containsKey(msg.getClientID())) {
LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID()); LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID());
//clean the subscriptions if the old used a cleanSession = true //clean the subscriptions if the old used a cleanSession = true
ServerChannel oldSession = m_clientIDs.get(msg.getClientID()).session; Channel oldSession = m_clientIDs.get(msg.getClientID()).session;
ClientSession oldClientSession = m_sessionsStore.sessionForClient(msg.getClientID()); ClientSession oldClientSession = m_sessionsStore.sessionForClient(msg.getClientID());
oldClientSession.disconnect(); oldClientSession.disconnect();
oldSession.setAttribute(NettyChannel.ATTR_KEY_SESSION_STOLEN, true); oldSession.attr(NettyChannel.ATTR_KEY_SESSION_STOLEN).set(true);
oldSession.close(false); oldSession.close();
LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientID()); LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientID());
} }
@@ -188,13 +191,13 @@ public class ProtocolProcessor {
int keepAlive = msg.getKeepAlive(); int keepAlive = msg.getKeepAlive();
LOG.debug("Connect with keepAlive {} s", keepAlive); LOG.debug("Connect with keepAlive {} s", keepAlive);
session.setAttribute(NettyChannel.ATTR_KEY_KEEPALIVE, keepAlive); session.attr(NettyChannel.ATTR_KEY_KEEPALIVE).set(keepAlive);
session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, msg.isCleanSession()); session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession());
//used to track the client in the subscription and publishing phases. //used to track the client in the subscription and publishing phases.
session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, msg.getClientID()); session.attr(NettyChannel.ATTR_KEY_CLIENTID).set(msg.getClientID());
LOG.debug("Connect create session <{}>", session); LOG.debug("Connect create session <{}>", session);
session.setIdleTime(Math.round(keepAlive * 1.5f)); setIdleTime(session.pipeline(), Math.round(keepAlive * 1.5f));
//Handle will flag //Handle will flag
if (msg.isWillFlag()) { if (msg.isWillFlag()) {
@@ -234,11 +237,17 @@ public class ProtocolProcessor {
// LOG.info("CONNECT clients descriptors {}", m_clientIDs); // LOG.info("CONNECT clients descriptors {}", m_clientIDs);
} }
private void failedCredentials(ServerChannel session) { private void setIdleTime(ChannelPipeline pipeline, int idleTime) {
if (pipeline.names().contains("idleStateHandler")) {
pipeline.remove("idleStateHandler");
}
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
}
private void failedCredentials(Channel session) {
ConnAckMessage okResp = new ConnAckMessage(); ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
session.write(okResp); session.write(okResp);
session.close(false);
} }
/** /**
@@ -289,12 +298,12 @@ public class ProtocolProcessor {
return pub; return pub;
} }
public void processPublish(ServerChannel session, PublishMessage msg) { public void processPublish(Channel session, PublishMessage msg) {
LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg); LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg);
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get();
final String topic = msg.getTopicName(); final String topic = msg.getTopicName();
//check if the topic can be wrote //check if the topic can be wrote
String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME); String user = (String) session.attr(NettyChannel.ATTR_KEY_USERNAME).get();
if (!m_authorizator.canWrite(topic, user, clientID)) { if (!m_authorizator.canWrite(topic, user, clientID)) {
LOG.debug("topic {} doesn't have write credentials", topic); LOG.debug("topic {} doesn't have write credentials", topic);
return; return;
@@ -518,10 +527,10 @@ public class ProtocolProcessor {
// could happen is not an error HANDLE IT // could happen is not an error HANDLE IT
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs)); throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs));
} }
ServerChannel session = m_clientIDs.get(clientId).session; Channel session = m_clientIDs.get(clientId).session;
LOG.debug("Session for clientId {} is {}", clientId, session); LOG.debug("Session for clientId {} is {}", clientId, session);
String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME); String user = (String) session.attr(NettyChannel.ATTR_KEY_USERNAME).get();
if (!m_authorizator.canRead(topic, user, clientId)) { if (!m_authorizator.canRead(topic, user, clientId)) {
LOG.debug("topic {} doesn't have read credentials", topic); LOG.debug("topic {} doesn't have read credentials", topic);
return; return;
@@ -632,7 +641,7 @@ public class ProtocolProcessor {
LOG.info("DISCONNECT client <{}> finished", clientID, cleanSession); LOG.info("DISCONNECT client <{}> finished", clientID, cleanSession);
} }
public void processConnectionLost(String clientID, boolean sessionStolen, NettyChannel channel) { public void processConnectionLost(String clientID, boolean sessionStolen, Channel channel) {
ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true); ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true);
m_clientIDs.remove(clientID, oldConnDescr); m_clientIDs.remove(clientID, oldConnDescr);
//If already removed a disconnect message was already processed for this clientID //If already removed a disconnect message was already processed for this clientID
@@ -654,10 +663,11 @@ public class ProtocolProcessor {
* Remove the clientID from topic subscription, if not previously subscribed, * Remove the clientID from topic subscription, if not previously subscribed,
* doesn't reply any error * doesn't reply any error
*/ */
public void processUnsubscribe(ServerChannel session, UnsubscribeMessage msg) { public void processUnsubscribe(Channel session, UnsubscribeMessage msg) {
List<String> topics = msg.topicFilters(); List<String> topics = msg.topicFilters();
int messageID = msg.getMessageID(); int messageID = msg.getMessageID();
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get();
LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
@@ -666,7 +676,7 @@ public class ProtocolProcessor {
boolean validTopic = SubscriptionsStore.validate(topic); boolean validTopic = SubscriptionsStore.validate(topic);
if (!validTopic) { if (!validTopic) {
//close the connection, not valid topicFilter is a protocol violation //close the connection, not valid topicFilter is a protocol violation
session.close(true); session.close();
LOG.warn("UNSUBSCRIBE found an invalid topic filter <{}> for clientID <{}>", topic, clientID); LOG.warn("UNSUBSCRIBE found an invalid topic filter <{}> for clientID <{}>", topic, clientID);
return; return;
} }
@@ -684,8 +694,8 @@ public class ProtocolProcessor {
session.write(ackMessage); session.write(ackMessage);
} }
public void processSubscribe(ServerChannel session, SubscribeMessage msg) { public void processSubscribe(Channel session, SubscribeMessage msg) {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get();
LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID()); LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID());
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
broker/src/test/java/io/moquette/spi/impl/DummyChannel.java
+13 -13
View File
@@ -15,7 +15,7 @@
*/ */
package io.moquette.spi.impl; package io.moquette.spi.impl;
import io.moquette.server.ServerChannel; import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.util.AttributeKey; import io.netty.util.AttributeKey;
import io.moquette.proto.messages.AbstractMessage; import io.moquette.proto.messages.AbstractMessage;
import io.moquette.proto.messages.ConnAckMessage; import io.moquette.proto.messages.ConnAckMessage;
@@ -27,7 +27,7 @@ import java.util.Map;
* *
* @author andrea * @author andrea
*/ */
public class DummyChannel implements ServerChannel { public class DummyChannel extends EmbeddedChannel {
AbstractMessage m_receivedMessage; AbstractMessage m_receivedMessage;
@@ -56,17 +56,17 @@ public class DummyChannel implements ServerChannel {
return this.m_channelClosed; return this.m_channelClosed;
} }
public void write(Object value) { // public void write(Object value) {
try { // try {
m_receivedMessage = (AbstractMessage) value; // m_receivedMessage = (AbstractMessage) value;
if (m_receivedMessage instanceof ConnAckMessage) { // if (m_receivedMessage instanceof ConnAckMessage) {
ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; // ConnAckMessage buf = (ConnAckMessage) m_receivedMessage;
m_returnCode = buf.getReturnCode(); // m_returnCode = buf.getReturnCode();
} // }
} catch (Exception ex) { // } catch (Exception ex) {
throw new AssertionError("Wrong return code"); // throw new AssertionError("Wrong return code");
} // }
} // }
public byte getReturnCode() { public byte getReturnCode() {
return m_returnCode; return m_returnCode;
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+4 -3
View File
@@ -28,6 +28,7 @@ import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.security.PermitAllAuthorizator;
import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore; import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.netty.channel.embedded.EmbeddedChannel;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
@@ -163,8 +164,8 @@ public class ProtocolProcessorTest {
storageService.initStore(); storageService.initStore();
subs.init(storageService.sessionsStore()); subs.init(storageService.sessionsStore());
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); EmbeddedChannel firstReceiverSession = new EmbeddedChannel();
ConnectMessage connectMessage = new ConnectMessage(); ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setProtocolVersion((byte) 3); connectMessage.setProtocolVersion((byte) 3);
connectMessage.setClientID(FAKE_CLIENT_ID); connectMessage.setClientID(FAKE_CLIENT_ID);
@@ -172,7 +173,7 @@ public class ProtocolProcessorTest {
m_processor.processConnect(firstReceiverSession, connectMessage); m_processor.processConnect(firstReceiverSession, connectMessage);
//connect the second fake subscriber //connect the second fake subscriber
MockReceiverChannel secondReceiverSession = new MockReceiverChannel(); EmbeddedChannel secondReceiverSession = new EmbeddedChannel();
ConnectMessage connectMessage2 = new ConnectMessage(); ConnectMessage connectMessage2 = new ConnectMessage();
connectMessage2.setProtocolVersion((byte) 3); connectMessage2.setProtocolVersion((byte) 3);
connectMessage2.setClientID(FAKE_CLIENT_ID2); connectMessage2.setClientID(FAKE_CLIENT_ID2);
broker/src/test/resources/log4j.properties
+4 -7
View File
@@ -4,8 +4,8 @@ log4j.rootLogger=ERROR, stdout, file
log4j.logger.io.moquette=WARN log4j.logger.io.moquette=WARN
#log4j.logger.io.moquette=DEBUG #log4j.logger.io.moquette=DEBUG
log4j.logger.io.moquette.spi.impl.ProtocolProcessor=TRACE #log4j.logger.io.moquette.spi.impl.ProtocolProcessor=TRACE
#log4j.logger.io.moquette.spi.impl.ProtocolProcessor=WARN log4j.logger.io.moquette.spi.impl.ProtocolProcessor=WARN
#log4j.logger.io.moquette.spi.impl.SimpleMessaging=DEBUG #log4j.logger.io.moquette.spi.impl.SimpleMessaging=DEBUG
#Protocol parsing #Protocol parsing
@@ -17,15 +17,12 @@ log4j.logger.io.moquette.server.netty.NettyMQTTHandler=WARN
#log4j.logger.io.moquette.spi.impl.subscriptions.SubscriptionsStore=DEBUG #log4j.logger.io.moquette.spi.impl.subscriptions.SubscriptionsStore=DEBUG
#log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=DEBUG #log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=DEBUG
log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=WARN log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=WARN
log4j.logger.io.moquette.spi.persistence.MapDBSessionsStore=DEBUG log4j.logger.io.moquette.spi.persistence.MapDBSessionsStore=WARN
# stdout appender is set to be a ConsoleAppender. # stdout appender is set to be a ConsoleAppender.
log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout=org.apache.log4j.ConsoleAppender
#log4j.appender.stdout.Threshold=WARN log4j.appender.stdout.Threshold=INFO
#log4j.appender.stdout.Threshold=INFO
log4j.appender.stdout.Threshold=TRACE
# for debug trace # for debug trace
#log4j.appender.stdout.Threshold=TRACE
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
#log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n #log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n
log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1} %M %L %x - %m%n log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1} %M %L %x - %m%n