anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added username to all messages. From a consumer perspective the username is more useful than a clientID which is often autogenerated.

Browse Source
This commit is contained in:
Dee Eldredge
2016-03-23 23:10:24 +01:00
committed by andsel
parent f44283d5f0
commit 7dab8a367f
9 changed files with 183 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/interception/Interceptor.java
+4 -4
View File
@@ -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);
}
broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java
+7 -1
View File
@@ -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;
}
}
broker/src/main/java/io/moquette/interception/messages/InterceptDisconnectMessage.java
+22 -16
View File
@@ -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;
}
}
broker/src/main/java/io/moquette/interception/messages/InterceptPublishMessage.java
+37 -31
View File
@@ -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;
}
}
broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java
+33 -27
View File
@@ -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;
}
}
broker/src/main/java/io/moquette/interception/messages/InterceptUnsubscribeMessage.java
+28 -22
View File
@@ -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;
}
}
broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java
+8 -8
View File
@@ -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));
}
});
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+40 -68
View File
@@ -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<Subscription> 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<IMessagesStore.StoredMessage> 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();
}
}
broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java
+4 -4
View File
@@ -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());
}