diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java index 8cfaee63..d8909b28 100644 --- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java +++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java @@ -171,6 +171,8 @@ final class MQTTConnection { NettyUtils.clientID(channel, clientId); LOG.trace("CONNACK sent, channel: {}", channel); + postOffice.dispatchConnection(msg); + LOG.trace("dispatch connection: {}", msg.toString()); } catch (SessionCorruptedException scex) { LOG.warn("MQTT session for client ID {} cannot be created, channel: {}", clientId, channel); abortConnection(CONNECTION_REFUSED_SERVER_UNAVAILABLE); @@ -257,6 +259,10 @@ final class MQTTConnection { sessionRegistry.disconnect(clientID); } 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) { @@ -284,6 +290,9 @@ final class MQTTConnection { connected = false; channel.close().addListener(FIRE_EXCEPTION_ON_FAILURE); 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) { diff --git a/broker/src/main/java/io/moquette/broker/PostOffice.java b/broker/src/main/java/io/moquette/broker/PostOffice.java index 08a61096..6d31885e 100644 --- a/broker/src/main/java/io/moquette/broker/PostOffice.java +++ b/broker/src/main/java/io/moquette/broker/PostOffice.java @@ -301,4 +301,20 @@ class PostOffice { } 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); + } } diff --git a/broker/src/main/java/io/moquette/broker/security/DenyAllAuthorizatorPolicy.java b/broker/src/main/java/io/moquette/broker/security/DenyAllAuthorizatorPolicy.java index b15a0bee..f59cb7ee 100644 --- a/broker/src/main/java/io/moquette/broker/security/DenyAllAuthorizatorPolicy.java +++ b/broker/src/main/java/io/moquette/broker/security/DenyAllAuthorizatorPolicy.java @@ -21,12 +21,12 @@ import io.moquette.broker.subscriptions.Topic; public class DenyAllAuthorizatorPolicy implements IAuthorizatorPolicy { @Override - public boolean canWrite(Topic topic, String user, String client) { - return true; + public boolean canRead(Topic topic, String user, String client) { + return false; } @Override - public boolean canRead(Topic topic, String user, String client) { - return true; + public boolean canWrite(Topic topic, String user, String client) { + return false; } }