Added check to avoid creation of LostConnectionEvents for not yet completed connects (fix issue #88)

This commit is contained in:
andsel
2015-09-22 15:29:23 +02:00
parent 56f122b8b9
commit f7466fd968
5 changed files with 65 additions and 3 deletions
+1
View File
@@ -9,6 +9,7 @@ Version 0.8:
Switched from commit persistence on every change to time based to fix delay in handling many msgs/sec (fix issue #79) Switched from commit persistence on every change to time based to fix delay in handling many msgs/sec (fix issue #79)
Changed willMessage and password in CONNECT message from String to byte[] (fix issue #94 and #95) Changed willMessage and password in CONNECT message from String to byte[] (fix issue #94 and #95)
Fixed ugly empty session storage, avoid fake NullObject empty subscription subscription Fixed ugly empty session storage, avoid fake NullObject empty subscription subscription
Added check to avoid creation of LostConnectionEvents for not yet completed connects (fix issue #88)
Version 0.7: Version 0.7:
Fix issue #38, added aligned to MQTT 3.1.1 Fix issue #38, added aligned to MQTT 3.1.1
+1 -1
View File
@@ -1,2 +1,2 @@
#the password is sha256(passwd) #the password is sha256(passwd)
testuser:36767030e79073783c8e9b299d9f8f1d4cd66baa82cd8869caad8dbb55204099 testuser:0d6be69b264717f2dd33652e212b173104b4a647b7c11ae72e9885f11cd312fb
@@ -22,6 +22,7 @@ import java.util.HashMap;
import java.util.Map; import java.util.Map;
import io.netty.handler.codec.CorruptedFrameException; import io.netty.handler.codec.CorruptedFrameException;
import org.apache.commons.codec.binary.StringUtils;
import org.eclipse.moquette.spi.IMessaging; import org.eclipse.moquette.spi.IMessaging;
import org.eclipse.moquette.proto.Utils; import org.eclipse.moquette.proto.Utils;
import org.eclipse.moquette.proto.messages.AbstractMessage; import org.eclipse.moquette.proto.messages.AbstractMessage;
@@ -70,7 +71,11 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
@Override @Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception { public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String clientID = (String) NettyUtils.getAttribute(ctx, NettyChannel.ATTR_KEY_CLIENTID); String clientID = (String) NettyUtils.getAttribute(ctx, NettyChannel.ATTR_KEY_CLIENTID);
m_messaging.lostConnection(clientID); if (clientID != null && !clientID.isEmpty()) {
//if the channel was of a correctly connected client, inform messagin
//else it was of a not completed CONNECT message
m_messaging.lostConnection(clientID);
}
ctx.close(/*false*/); ctx.close(/*false*/);
} }
@@ -1,5 +1,4 @@
# Set root logger level to DEBUG and its only appender to A1. # Set root logger level to DEBUG and its only appender to A1.
#log4j.rootLogger=ERROR, A1
log4j.rootLogger=ERROR, stdout, file log4j.rootLogger=ERROR, stdout, file
#log4j.logger.org.eclipse.moquette=WARN #log4j.logger.org.eclipse.moquette=WARN
@@ -0,0 +1,57 @@
//@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10')
@GrabResolver(name='Paho', root="https://repo.eclipse.org/content/repositories/paho-releases/")
@Grab(group='org.eclipse.paho', module='org.eclipse.paho.client.mqttv3', version='1.0.2')
import org.eclipse.paho.client.mqttv3.MqttClient
import org.eclipse.paho.client.mqttv3.MqttConnectOptions
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence
import org.eclipse.paho.client.mqttv3.MqttException
MqttDefaultFilePersistence subDataStore = new MqttDefaultFilePersistence("/tmp/sub")
MqttClient subClient = new MqttClient("tcp://localhost:1883", "ConnectTestCLI", subDataStore)
MqttConnectOptions connectOptions = new MqttConnectOptions()
connectOptions.setUserName("gino")
connectOptions.setPassword("fake".toCharArray())
connectOptions.setKeepAliveInterval(5)
println "Before connect with bad"
try {
subClient.connect(connectOptions)
} catch (MqttException mex) {
println "OK, catched connect with bad credentials, wait 4x keepalive"
sleep(20 * 1000)
}
println "Before connect with ok credentials"
connectOptions = new MqttConnectOptions()
connectOptions.setUserName("testuser")
connectOptions.setPassword("passwd".toCharArray())
subClient.connect(connectOptions)
println "Successfull connected"
subClient.disconnect()
println "Finished!"
//import org.fusesource.mqtt.client.BlockingConnection
//import org.fusesource.mqtt.client.MQTT
//"Usage: groovy issue88_reconnect_after_bad_credentials.groovy"
//MQTT mqtt = new MQTT()
//def host = "localhost"
//mqtt.setCleanSession(true)
//mqtt.setHost(host, 1883)
//mqtt.setUserName("gino")
//mqtt.setPassword("fake")
//
//mqtt.setClientId("BadConnecter_client")
//BlockingConnection connection = mqtt.blockingConnection()
//connection.connect()
//
////try a valid conn
//mqtt.setUserName("testuser")
//mqtt.setPassword("passwd")
//connection = mqtt.blockingConnection()
//connection.connect()
//
//connection.disconnect()
//println "exited"