diff --git a/broker/src/main/java/io/moquette/interception/Interceptor.java b/broker/src/main/java/io/moquette/interception/Interceptor.java index 6687bea8..ff73e6a6 100644 --- a/broker/src/main/java/io/moquette/interception/Interceptor.java +++ b/broker/src/main/java/io/moquette/interception/Interceptor.java @@ -37,13 +37,13 @@ public interface Interceptor { void notifyClientConnected(ConnectMessage msg); - void notifyClientDisconnected(String clientID); + void notifyClientDisconnected(String clientID, String username); - void notifyTopicPublished(PublishMessage msg, String clientID); + void notifyTopicPublished(PublishMessage msg, String clientID, final String username); - void notifyTopicSubscribed(Subscription sub); + void notifyTopicSubscribed(Subscription sub, final String username); - void notifyTopicUnsubscribed(String topic, String clientID); + void notifyTopicUnsubscribed(String topic, String clientID, final String username); void notifyMessageConsumed(InterceptConsumedMessage msg); } diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java index 991ecdeb..9e73c832 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java @@ -4,12 +4,18 @@ import static io.moquette.spi.IMessagesStore.StoredMessage; public class InterceptConsumedMessage { final private StoredMessage msg; + private final String username; - public InterceptConsumedMessage( final StoredMessage msg) { + public InterceptConsumedMessage( final StoredMessage msg, String username) { this.msg = msg; + this.username = username; } public StoredMessage getMsg() { return msg; } + + public String getUsername() { + return username; + } } diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptDisconnectMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptDisconnectMessage.java index e1a7e75c..29c7cadc 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptDisconnectMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptDisconnectMessage.java @@ -1,16 +1,22 @@ -package io.moquette.interception.messages; - -/** - * @author Wagner Macedo - */ -public class InterceptDisconnectMessage { - private final String clientID; - - public InterceptDisconnectMessage(String clientID) { - this.clientID = clientID; - } - - public String getClientID() { - return clientID; - } -} +package io.moquette.interception.messages; + +/** + * @author Wagner Macedo + */ +public class InterceptDisconnectMessage { + private final String clientID; + private final String username; + + public InterceptDisconnectMessage(String clientID, String username) { + this.clientID = clientID; + this.username = username; + } + + public String getClientID() { + return clientID; + } + + public String getUsername() { + return username; + } +} diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptPublishMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptPublishMessage.java index da9c462e..5e88b2b1 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptPublishMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptPublishMessage.java @@ -1,31 +1,37 @@ -package io.moquette.interception.messages; - -import io.moquette.parser.proto.messages.PublishMessage; - -import java.nio.ByteBuffer; - -/** - * @author Wagner Macedo - */ -public class InterceptPublishMessage extends InterceptAbstractMessage { - private final PublishMessage msg; - private final String clientID; - - public InterceptPublishMessage(PublishMessage msg, String clientID) { - super(msg); - this.msg = msg; - this.clientID = clientID; - } - - public String getTopicName() { - return msg.getTopicName(); - } - - public ByteBuffer getPayload() { - return msg.getPayload(); - } - - public String getClientID() { - return clientID; - } -} +package io.moquette.interception.messages; + +import io.moquette.parser.proto.messages.PublishMessage; + +import java.nio.ByteBuffer; + +/** + * @author Wagner Macedo + */ +public class InterceptPublishMessage extends InterceptAbstractMessage { + private final PublishMessage msg; + private final String clientID; + private final String username; + + public InterceptPublishMessage(PublishMessage msg, String clientID, String username) { + super(msg); + this.msg = msg; + this.clientID = clientID; + this.username = username; + } + + public String getTopicName() { + return msg.getTopicName(); + } + + public ByteBuffer getPayload() { + return msg.getPayload(); + } + + public String getClientID() { + return clientID; + } + + public String getUsername() { + return username; + } +} diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java index 8f90256b..3e3d43e0 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java @@ -1,27 +1,33 @@ -package io.moquette.interception.messages; - -import io.moquette.parser.proto.messages.AbstractMessage; -import io.moquette.spi.impl.subscriptions.Subscription; - -/** - * @author Wagner Macedo - */ -public class InterceptSubscribeMessage { - private final Subscription subscription; - - public InterceptSubscribeMessage(Subscription subscription) { - this.subscription = subscription; - } - - public String getClientID() { - return subscription.getClientId(); - } - - public AbstractMessage.QOSType getRequestedQos() { - return subscription.getRequestedQos(); - } - - public String getTopicFilter() { - return subscription.getTopicFilter(); - } -} +package io.moquette.interception.messages; + +import io.moquette.parser.proto.messages.AbstractMessage; +import io.moquette.spi.impl.subscriptions.Subscription; + +/** + * @author Wagner Macedo + */ +public class InterceptSubscribeMessage { + private final Subscription subscription; + private final String username; + + public InterceptSubscribeMessage(Subscription subscription, String username) { + this.subscription = subscription; + this.username = username; + } + + public String getClientID() { + return subscription.getClientId(); + } + + public AbstractMessage.QOSType getRequestedQos() { + return subscription.getRequestedQos(); + } + + public String getTopicFilter() { + return subscription.getTopicFilter(); + } + + public String getUsername() { + return username; + } +} diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptUnsubscribeMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptUnsubscribeMessage.java index c8eebc01..a8be200f 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptUnsubscribeMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptUnsubscribeMessage.java @@ -1,22 +1,28 @@ -package io.moquette.interception.messages; - -/** - * @author Wagner Macedo - */ -public class InterceptUnsubscribeMessage { - private final String topicFilter; - private final String clientID; - - public InterceptUnsubscribeMessage(String topicFilter, String clientID) { - this.topicFilter = topicFilter; - this.clientID = clientID; - } - - public String getTopicFilter() { - return topicFilter; - } - - public String getClientID() { - return clientID; - } -} +package io.moquette.interception.messages; + +/** + * @author Wagner Macedo + */ +public class InterceptUnsubscribeMessage { + private final String topicFilter; + private final String clientID; + private final String username; + + public InterceptUnsubscribeMessage(String topicFilter, String clientID, String username) { + this.topicFilter = topicFilter; + this.clientID = clientID; + this.username = username; + } + + public String getTopicFilter() { + return topicFilter; + } + + public String getClientID() { + return clientID; + } + + public String getUsername() { + return username; + } +} diff --git a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java index f4ea53b4..27732fe8 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -60,48 +60,48 @@ final class BrokerInterceptor implements Interceptor { } @Override - public void notifyClientDisconnected(final String clientID) { + public void notifyClientDisconnected(final String clientID, final String username ) { for (final InterceptHandler handler : this.handlers) { executor.execute(new Runnable() { @Override public void run() { - handler.onDisconnect(new InterceptDisconnectMessage(clientID)); + handler.onDisconnect(new InterceptDisconnectMessage(clientID, username)); } }); } } @Override - public void notifyTopicPublished(final PublishMessage msg, final String clientID) { + public void notifyTopicPublished(final PublishMessage msg, final String clientID, final String username) { for (final InterceptHandler handler : this.handlers) { executor.execute(new Runnable() { @Override public void run() { - handler.onPublish(new InterceptPublishMessage(msg, clientID)); + handler.onPublish(new InterceptPublishMessage(msg, clientID, username)); } }); } } @Override - public void notifyTopicSubscribed(final Subscription sub) { + public void notifyTopicSubscribed(final Subscription sub, final String username) { for (final InterceptHandler handler : this.handlers) { executor.execute(new Runnable() { @Override public void run() { - handler.onSubscribe(new InterceptSubscribeMessage(sub)); + handler.onSubscribe(new InterceptSubscribeMessage(sub, username)); } }); } } @Override - public void notifyTopicUnsubscribed(final String topic, final String clientID) { + public void notifyTopicUnsubscribed(final String topic, final String clientID, final String username) { for (final InterceptHandler handler : this.handlers) { executor.execute(new Runnable() { @Override public void run() { - handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID)); + handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID, username)); } }); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index d0413deb..a614aa3c 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -19,14 +19,13 @@ import java.nio.ByteBuffer; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.TimeUnit; import io.moquette.server.ConnectionDescriptor; -import io.moquette.server.netty.AutoFlushHandler; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.ClientSession; import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMessagesStore; +import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.ISessionsStore; import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthorizator; @@ -35,6 +34,7 @@ import io.moquette.spi.impl.subscriptions.Subscription; import static io.moquette.parser.netty.Utils.VERSION_3_1; import static io.moquette.parser.netty.Utils.VERSION_3_1_1; +import io.moquette.interception.messages.InterceptConsumedMessage; import io.moquette.parser.proto.messages.AbstractMessage; import io.moquette.parser.proto.messages.AbstractMessage.QOSType; import io.moquette.parser.proto.messages.ConnAckMessage; @@ -209,9 +209,8 @@ public class ProtocolProcessor { byte[] willPayload = msg.getWillMessage(); ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); //save the will testament in the clientID store - WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos); + WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos ); m_willStore.put(msg.getClientID(), will); - LOG.info("Session for clientID <{}> with will to topic {}", msg.getClientID(), msg.getWillTopic()); } ConnAckMessage okResp = new ConnAckMessage(); @@ -241,19 +240,8 @@ public class ProtocolProcessor { //force the republish of stored QoS1 and QoS2 republishStoredInSession(clientSession); } - int flushIntervalMs = 500/*(keepAlive * 1000) / 2*/; - setupAutoFlusher(channel.pipeline(), flushIntervalMs); LOG.info("CONNECT processed"); - } - - private void setupAutoFlusher(ChannelPipeline pipeline, int flushIntervalMs) { - AutoFlushHandler autoFlushHandler = new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS); - try { - pipeline.addAfter("idleEventHandler", "autoFlusher", autoFlushHandler); - } catch (NoSuchElementException nseex) { - //the idleEventHandler is not present on the pipeline - pipeline.addFirst("autoFlusher", autoFlushHandler); - } +// LOG.info("CONNECT clients descriptors {}", m_clientIDs); } private void setIdleTime(ChannelPipeline pipeline, int idleTime) { @@ -289,13 +277,18 @@ public class ProtocolProcessor { } } - public void processPubAck(Channel channel, PubAckMessage msg) { - String clientID = NettyUtils.clientID(channel); + public void processPubAck(Channel session, PubAckMessage msg) { + String clientID = NettyUtils.clientID(session); int messageID = msg.getMessageID(); + String username = NettyUtils.userName(session); + StoredMessage inflightMsg = m_sessionsStore.getInflightMessage(clientID, messageID); + //Remove the message from message store ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); verifyToActivate(clientID, targetSession); targetSession.inFlightAcknowledged(messageID); + + m_interceptor.notifyMessageConsumed(new InterceptConsumedMessage(inflightMsg, username)); } private void verifyToActivate(String clientID, ClientSession targetSession) { @@ -317,13 +310,13 @@ public class ProtocolProcessor { return pub; } - public void processPublish(Channel channel, PublishMessage msg) { + public void processPublish(Channel session, PublishMessage msg) { LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg); - String clientID = NettyUtils.clientID(channel); + String clientID = NettyUtils.clientID(session); final String topic = msg.getTopicName(); //check if the topic can be wrote - String user = NettyUtils.userName(channel); - if (!m_authorizator.canWrite(topic, user, clientID)) { + String username = NettyUtils.userName(session); + if (!m_authorizator.canWrite(topic, username, clientID)) { LOG.debug("topic {} doesn't have write credentials", topic); return; } @@ -363,7 +356,7 @@ public class ProtocolProcessor { } } } - m_interceptor.notifyTopicPublished(msg, clientID); + m_interceptor.notifyTopicPublished(msg, clientID, username); } /** @@ -471,11 +464,9 @@ public class ProtocolProcessor { } } - protected void directSend(ClientSession clientsession, String topic, AbstractMessage.QOSType qos, - ByteBuffer message, boolean retained, Integer messageID) { + protected void directSend(ClientSession clientsession, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) { String clientId = clientsession.clientID; - LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", - clientId, topic, qos, retained, messageID); + LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", clientId, topic, qos, retained, messageID); PublishMessage pubMessage = new PublishMessage(); pubMessage.setRetainFlag(retained); pubMessage.setTopicName(topic); @@ -491,32 +482,22 @@ public class ProtocolProcessor { pubMessage.setMessageID(messageID); } else { if (messageID != null) { - throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message " + - "with PacketIdentifier: " + messageID); + throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message with PacketIdentifier: " + messageID); } } if (m_clientIDs == null) { - throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be " + - "initialized, somewhere it's overwritten!!"); + throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!"); } - //LOG.trace("clientIDs are {}", m_clientIDs); + LOG.debug("clientIDs are {}", m_clientIDs); if (m_clientIDs.get(clientId) == null) { //TODO while we were publishing to the target client, that client disconnected, // 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)); } Channel channel = m_clientIDs.get(clientId).channel; - LOG.trace("Session for clientId {}", clientId); -// channel.writeAndFlush(pubMessage); - if (channel.isWritable()) { - //if channel is writable don't enqueue - channel.write(pubMessage); - } else { - //enqueue to the client session - clientsession.enqueue(pubMessage); - } + LOG.debug("Session for clientId {} is {}", clientId, channel); + channel.writeAndFlush(pubMessage); } private void sendPubRec(String clientID, int messageID) { @@ -598,15 +579,18 @@ public class ProtocolProcessor { public void processPubComp(Channel channel, PubCompMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); + StoredMessage inflightMsg = m_sessionsStore.getInflightMessage( clientID, messageID ); + LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID); //once received the PUBCOMP then remove the message from the temp memory ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); verifyToActivate(clientID, targetSession); targetSession.secondPhaseAcknowledged(messageID); + String username = NettyUtils.userName(channel); + m_interceptor.notifyMessageConsumed( new InterceptConsumedMessage(inflightMsg, username) ); } - public void processDisconnect(Channel channel) throws InterruptedException { - channel.flush(); + public void processDisconnect(Channel channel) { String clientID = NettyUtils.clientID(channel); boolean cleanSession = NettyUtils.cleanSession(channel); LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession); @@ -619,7 +603,8 @@ public class ProtocolProcessor { //cleanup the will store m_willStore.remove(clientID); - m_interceptor.notifyClientDisconnected(clientID); + String username = NettyUtils.userName(channel); + m_interceptor.notifyClientDisconnected(clientID, username); LOG.info("DISCONNECT client <{}> finished", clientID, cleanSession); } @@ -665,7 +650,8 @@ public class ProtocolProcessor { subscriptions.removeSubscription(topic, clientID); clientSession.unsubscribeFrom(topic); - m_interceptor.notifyTopicUnsubscribed(topic, clientID); + String username = NettyUtils.userName(channel); + m_interceptor.notifyTopicUnsubscribed(topic, clientID, username); } //ack the client @@ -686,10 +672,10 @@ public class ProtocolProcessor { SubAckMessage ackMessage = new SubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); - String user = NettyUtils.userName(channel); + String username = NettyUtils.userName(channel); List newSubscriptions = new ArrayList<>(); for (SubscribeMessage.Couple req : msg.subscriptions()) { - if (!m_authorizator.canRead(req.topicFilter, user, clientSession.clientID)) { + if (!m_authorizator.canRead(req.topicFilter, username, clientSession.clientID)) { //send SUBACK with 0x80, the user hasn't credentials to read the topic LOG.debug("topic {} doesn't have read credentials", req.topicFilter); ackMessage.addType( AbstractMessage.QOSType.FAILURE); @@ -714,17 +700,18 @@ public class ProtocolProcessor { //fire the publish for(Subscription subscription : newSubscriptions) { - subscribeSingleTopic(subscription); + subscribeSingleTopic(subscription, username); } } - private boolean subscribeSingleTopic(final Subscription newSubscription) { + private boolean subscribeSingleTopic(final Subscription newSubscription, String username) { subscriptions.add(newSubscription.asClientTopicCouple()); //scans retained messages to be published to the new subscription //TODO this is ugly, it does a linear scan on potential big dataset Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() { - public boolean match(String key) { + @Override + public boolean match(String key) { return SubscriptionsStore.matchTopics(key, newSubscription.getTopicFilter()); } }); @@ -741,22 +728,7 @@ public class ProtocolProcessor { } //notify the Observables - m_interceptor.notifyTopicSubscribed(newSubscription); + m_interceptor.notifyTopicSubscribed(newSubscription, username); return true; } - - public void notifyChannelWritable(Channel channel) { - String clientID = NettyUtils.clientID(channel); - ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); - boolean emptyQueue = false; - while (channel.isWritable() && !emptyQueue) { - AbstractMessage msg = clientSession.dequeue(); - if (msg == null) { - emptyQueue = true; - } else { - channel.write(msg); - } - } - channel.flush(); - } } diff --git a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java index aded6043..f91e68dc 100644 --- a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java @@ -97,28 +97,28 @@ public class BrokerInterceptorTest { @Test public void testNotifyClientDisconnected() throws Exception { - interceptor.notifyClientDisconnected("cli1234"); + interceptor.notifyClientDisconnected("cli1234","cli1234"); interval(); assertEquals(50, n.get()); } @Test public void testNotifyTopicPublished() throws Exception { - interceptor.notifyTopicPublished(new PublishMessage(), "cli1234"); + interceptor.notifyTopicPublished(new PublishMessage(), "cli1234", "cli1234"); interval(); assertEquals(60, n.get()); } @Test public void testNotifyTopicSubscribed() throws Exception { - interceptor.notifyTopicSubscribed(new Subscription("cli1", "o2", AbstractMessage.QOSType.MOST_ONE)); + interceptor.notifyTopicSubscribed(new Subscription("cli1", "o2", AbstractMessage.QOSType.MOST_ONE), "cli1234"); interval(); assertEquals(70, n.get()); } @Test public void testNotifyTopicUnsubscribed() throws Exception { - interceptor.notifyTopicUnsubscribed("o2", "cli1234"); + interceptor.notifyTopicUnsubscribed("o2", "cli1234", "cli1234"); interval(); assertEquals(80, n.get()); }