Fix issue #439 and #443

This commit is contained in:
zhangliangbo
2019-01-20 18:01:37 +01:00
committed by andsel
parent 1de5322a00
commit ea063159f0
3 changed files with 29 additions and 4 deletions
@@ -171,6 +171,8 @@ final class MQTTConnection {
NettyUtils.clientID(channel, clientId); NettyUtils.clientID(channel, clientId);
LOG.trace("CONNACK sent, channel: {}", channel); LOG.trace("CONNACK sent, channel: {}", channel);
postOffice.dispatchConnection(msg);
LOG.trace("dispatch connection: {}", msg.toString());
} catch (SessionCorruptedException scex) { } catch (SessionCorruptedException scex) {
LOG.warn("MQTT session for client ID {} cannot be created, channel: {}", clientId, channel); LOG.warn("MQTT session for client ID {} cannot be created, channel: {}", clientId, channel);
abortConnection(CONNECTION_REFUSED_SERVER_UNAVAILABLE); abortConnection(CONNECTION_REFUSED_SERVER_UNAVAILABLE);
@@ -257,6 +259,10 @@ final class MQTTConnection {
sessionRegistry.disconnect(clientID); sessionRegistry.disconnect(clientID);
} }
connected = false; connected = false;
//dispatch connection lost to intercept.
String userName= NettyUtils.userName(channel);
postOffice.dispatchConnectionLost(clientID,userName);
LOG.trace("dispatch disconnection: clientId={}, userName={}", clientID, userName);
} }
void sendConnAck(boolean isSessionAlreadyPresent) { void sendConnAck(boolean isSessionAlreadyPresent) {
@@ -284,6 +290,9 @@ final class MQTTConnection {
connected = false; connected = false;
channel.close().addListener(FIRE_EXCEPTION_ON_FAILURE); channel.close().addListener(FIRE_EXCEPTION_ON_FAILURE);
LOG.trace("Processed DISCONNECT CId={}, channel: {}", clientID, channel); LOG.trace("Processed DISCONNECT CId={}, channel: {}", clientID, channel);
String userName=NettyUtils.userName(channel);
postOffice.dispatchDisconnection(clientID,userName);
LOG.trace("dispatch disconnection: clientId={}, userName={}", clientID, userName);
} }
void processSubscribe(MqttSubscribeMessage msg) { void processSubscribe(MqttSubscribeMessage msg) {
@@ -301,4 +301,20 @@ class PostOffice {
} }
retainedRepository.retain(topic, msg); retainedRepository.retain(topic, msg);
} }
/**
* notify MqttConnectMessage after connection established (already pass login).
* @param msg
*/
void dispatchConnection(MqttConnectMessage msg){
interceptor.notifyClientConnected(msg);
}
void dispatchDisconnection(String clientId,String userName){
interceptor.notifyClientDisconnected(clientId,userName);
}
void dispatchConnectionLost(String clientId,String userName){
interceptor.notifyClientConnectionLost(clientId,userName);
}
} }
@@ -21,12 +21,12 @@ import io.moquette.broker.subscriptions.Topic;
public class DenyAllAuthorizatorPolicy implements IAuthorizatorPolicy { public class DenyAllAuthorizatorPolicy implements IAuthorizatorPolicy {
@Override @Override
public boolean canWrite(Topic topic, String user, String client) { public boolean canRead(Topic topic, String user, String client) {
return true; return false;
} }
@Override @Override
public boolean canRead(Topic topic, String user, String client) { public boolean canWrite(Topic topic, String user, String client) {
return true; return false;
} }
} }