Fixed issue 5, was due to bad read timeout management in IoSession.idle.
This commit is contained in:
andrea
+544
-544
File diff suppressed because it is too large
Load Diff
@@ -1,56 +1,62 @@
|
||||
package org.dna.mqtt.moquette.server;
|
||||
|
||||
import org.apache.mina.core.service.IoHandlerAdapter;
|
||||
import org.apache.mina.core.session.IdleStatus;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
|
||||
import org.dna.mqtt.moquette.proto.Utils;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
|
||||
|
||||
/**
|
||||
* MINA MQTT Handler used to route messages to protocol logic
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class MQTTHandler extends IoHandlerAdapter {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class);
|
||||
private IMessaging m_messaging;
|
||||
|
||||
@Override
|
||||
public void messageReceived(IoSession session, Object message) throws Exception {
|
||||
AbstractMessage msg = (AbstractMessage) message;
|
||||
LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType())));
|
||||
try {
|
||||
switch (msg.getMessageType()) {
|
||||
case CONNECT:
|
||||
case SUBSCRIBE:
|
||||
case UNSUBSCRIBE:
|
||||
case PUBLISH:
|
||||
case PUBREC:
|
||||
case PUBCOMP:
|
||||
case PUBREL:
|
||||
case PINGREQ:
|
||||
case DISCONNECT:
|
||||
m_messaging.handleProtocolMessage(session, msg);
|
||||
break;
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
LOG.error("Bad error in processing the message", ex);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sessionIdle(IoSession session, IdleStatus status) {
|
||||
session.close(false);
|
||||
}
|
||||
|
||||
public void setMessaging(IMessaging messaging) {
|
||||
m_messaging = messaging;
|
||||
}
|
||||
|
||||
}
|
||||
package org.dna.mqtt.moquette.server;
|
||||
|
||||
import org.apache.mina.core.service.IoHandlerAdapter;
|
||||
import org.apache.mina.core.session.IdleStatus;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
|
||||
import org.dna.mqtt.moquette.proto.Utils;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
|
||||
import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* MINA MQTT Handler used to route messages to protocol logic
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class MQTTHandler extends IoHandlerAdapter {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class);
|
||||
private IMessaging m_messaging;
|
||||
|
||||
@Override
|
||||
public void messageReceived(IoSession session, Object message) throws Exception {
|
||||
AbstractMessage msg = (AbstractMessage) message;
|
||||
LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType())));
|
||||
try {
|
||||
switch (msg.getMessageType()) {
|
||||
case CONNECT:
|
||||
case SUBSCRIBE:
|
||||
case UNSUBSCRIBE:
|
||||
case PUBLISH:
|
||||
case PUBREC:
|
||||
case PUBCOMP:
|
||||
case PUBREL:
|
||||
case DISCONNECT:
|
||||
m_messaging.handleProtocolMessage(session, msg);
|
||||
break;
|
||||
case PINGREQ:
|
||||
PingRespMessage pingResp = new PingRespMessage();
|
||||
session.write(pingResp);
|
||||
break;
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
LOG.error("Bad error in processing the message", ex);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sessionIdle(IoSession session, IdleStatus status) {
|
||||
if (status == IdleStatus.READER_IDLE) {
|
||||
session.close(false);
|
||||
//TODO send a notification to messaging part to remove the bining clientID-ConnConfig
|
||||
}
|
||||
}
|
||||
|
||||
public void setMessaging(IMessaging messaging) {
|
||||
m_messaging = messaging;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,17 +1,19 @@
|
||||
# Set root logger level to DEBUG and its only appender to A1.
|
||||
log4j.rootLogger=ERROR, A1
|
||||
#log4j.rootLogger=DEBUG, A1
|
||||
#log4j.rootLogger=WARN, A1
|
||||
#log4j.logger.org.dna.mqtt.moquette.client=WARN
|
||||
#log4j.logger.org.mina=WARN
|
||||
#log4j.logger.org.dna.mqtt.moquette.server=DEBUG
|
||||
#log4j.logger.org.dna.mqtt.moquette.messaging=DEBUG
|
||||
|
||||
log4j.logger.org.dna.mqtt.bechnmark=INFO
|
||||
|
||||
# A1 is set to be a ConsoleAppender.
|
||||
log4j.appender.A1=org.apache.log4j.ConsoleAppender
|
||||
|
||||
# A1 uses PatternLayout.
|
||||
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
|
||||
# Set root logger level to DEBUG and its only appender to A1.
|
||||
#log4j.rootLogger=ERROR, A1
|
||||
log4j.rootLogger=INFO, A1
|
||||
#log4j.rootLogger=WARN, A1
|
||||
#log4j.logger.org.dna.mqtt.moquette.client=WARN
|
||||
#log4j.logger.org.mina=WARN
|
||||
#log4j.logger.org.dna.mqtt.moquette.server=DEBUG
|
||||
#log4j.logger.org.dna.mqtt.moquette.messaging=DEBUG
|
||||
#log4j.logger.org.dna.mqtt.moquette.proto.MQTTLoggingFilter=DEBUG
|
||||
#log4j.logger.org.dna.mqtt.moquette.proto.PublishDecoder=DEBUG
|
||||
|
||||
log4j.logger.org.dna.mqtt.bechnmark=INFO
|
||||
|
||||
# A1 is set to be a ConsoleAppender.
|
||||
log4j.appender.A1=org.apache.log4j.ConsoleAppender
|
||||
|
||||
# A1 uses PatternLayout.
|
||||
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
|
||||
log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n
|
||||
@@ -1,26 +1,25 @@
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class UtilsTests {
|
||||
|
||||
@Test
|
||||
public void testEncodeFlags() {
|
||||
UnsubscribeMessage msg = new UnsubscribeMessage();
|
||||
msg.setRetainFlag(true);
|
||||
msg.setQos(AbstractMessage.QOSType.MOST_ONE);
|
||||
|
||||
//Exercise
|
||||
assertEquals(1, Utils.encodeFlags(msg));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
|
||||
import static org.junit.Assert.*;
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class UtilsTests {
|
||||
|
||||
@Test
|
||||
public void testEncodeFlags() {
|
||||
UnsubscribeMessage msg = new UnsubscribeMessage();
|
||||
msg.setRetainFlag(true);
|
||||
msg.setQos(AbstractMessage.QOSType.MOST_ONE);
|
||||
|
||||
//Exercise
|
||||
assertEquals(1, Utils.encodeFlags(msg));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user