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

Fixed merge

Browse Source
This commit is contained in:
andsel
2016-03-23 23:13:01 +01:00
parent 1e7f6d98e7
commit ed6ad315d1
1 changed files with 54 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+54 -41
View File
@@ -57,9 +57,9 @@ import org.slf4j.LoggerFactory;
/**
* Class responsible to handle the logic of MQTT protocol it's the director of
* the protocol execution.
*
*
* Used by the front facing class SimpleMessaging.
*
*
* @author andrea
*/
public class ProtocolProcessor {
@@ -92,11 +92,11 @@ public class ProtocolProcessor {
public QOSType getQos() {
return qos;
}
}
private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
protected ConcurrentMap<String, ConnectionDescriptor> m_clientIDs;
private SubscriptionsStore subscriptions;
private boolean allowAnonymous;
@@ -108,7 +108,7 @@ public class ProtocolProcessor {
//maps clientID to Will testament, if specified on CONNECT
private ConcurrentMap<String, WillMessage> m_willStore = new ConcurrentHashMap<>();
ProtocolProcessor() {}
/**
@@ -209,8 +209,9 @@ 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();
@@ -276,18 +277,18 @@ public class ProtocolProcessor {
clientSession.removeEnqueued(pubEvt.getGuid());
}
}
public void processPubAck(Channel session, PubAckMessage msg) {
String clientID = NettyUtils.clientID(session);
public void processPubAck(Channel channel, PubAckMessage msg) {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.getMessageID();
String username = NettyUtils.userName(session);
StoredMessage inflightMsg = m_sessionsStore.getInflightMessage(clientID, messageID);
String username = NettyUtils.userName(channel);
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);
String topic = inflightMsg.getTopic();
m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username));
@@ -311,13 +312,13 @@ public class ProtocolProcessor {
pub.setRetained(will.isRetained());
return pub;
}
public void processPublish(Channel session, PublishMessage msg) {
public void processPublish(Channel channel, PublishMessage msg) {
LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg);
String clientID = NettyUtils.clientID(session);
String clientID = NettyUtils.clientID(channel);
final String topic = msg.getTopicName();
//check if the topic can be wrote
String username = NettyUtils.userName(session);
String username = NettyUtils.userName(channel);
if (!m_authorizator.canWrite(topic, username, clientID)) {
LOG.debug("topic {} doesn't have write credentials", topic);
return;
@@ -397,7 +398,7 @@ public class ProtocolProcessor {
}
m_messagesStore.storeRetained(topic, guid);
}
/**
* Specialized version to publish will testament message.
*/
@@ -466,15 +467,17 @@ 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);
pubMessage.setQos(qos);
pubMessage.setPayload(message);
LOG.info("send publish message to <{}> on topic <{}>", clientId, topic);
if (LOG.isDebugEnabled()) {
LOG.debug("content <{}>", DebugUtils.payload2Str(message));
@@ -484,31 +487,40 @@ 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.debug("clientIDs are {}", m_clientIDs);
//LOG.trace("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.debug("Session for clientId {} is {}", clientId, channel);
channel.writeAndFlush(pubMessage);
LOG.trace("Session for clientId {}", clientId);
if (channel.isWritable()) {
//if channel is writable don't enqueue
channel.write(pubMessage);
} else {
//enqueue to the client session
clientsession.enqueue(pubMessage);
}
}
private void sendPubRec(String clientID, int messageID) {
LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID);
PubRecMessage pubRecMessage = new PubRecMessage();
pubRecMessage.setMessageID(messageID);
m_clientIDs.get(clientID).channel.writeAndFlush(pubRecMessage);
}
private void sendPubAck(String clientId, int messageID) {
LOG.trace("sendPubAck invoked");
PubAckMessage pubAckMessage = new PubAckMessage();
@@ -527,7 +539,7 @@ public class ProtocolProcessor {
LOG.error(null, t);
}
}
/**
* Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish
* to all interested subscribers.
@@ -552,7 +564,7 @@ public class ProtocolProcessor {
sendPubComp(clientID, messageID);
}
private void sendPubComp(String clientID, int messageID) {
LOG.debug("PUB <--PUBCOMP-- SRV sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
PubCompMessage pubCompMessage = new PubCompMessage();
@@ -560,7 +572,7 @@ public class ProtocolProcessor {
m_clientIDs.get(clientID).channel.writeAndFlush(pubCompMessage);
}
public void processPubRec(Channel channel, PubRecMessage msg) {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.getMessageID();
@@ -581,8 +593,8 @@ 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 );
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);
@@ -592,8 +604,9 @@ public class ProtocolProcessor {
String topic = inflightMsg.getTopic();
m_interceptor.notifyMessageAcknowledged( new InterceptAcknowledgedMessage(inflightMsg, topic, username) );
}
public void processDisconnect(Channel channel) {
public void processDisconnect(Channel channel) throws InterruptedException {
channel.flush();
String clientID = NettyUtils.clientID(channel);
boolean cleanSession = NettyUtils.cleanSession(channel);
LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession);
@@ -605,7 +618,7 @@ public class ProtocolProcessor {
//cleanup the will store
m_willStore.remove(clientID);
String username = NettyUtils.userName(channel);
m_interceptor.notifyClientDisconnected(clientID, username);
LOG.info("DISCONNECT client <{}> finished", clientID, cleanSession);
@@ -628,7 +641,7 @@ public class ProtocolProcessor {
m_willStore.remove(clientID);
}
}
/**
* Remove the clientID from topic subscription, if not previously subscribed,
* doesn't reply any error
@@ -706,7 +719,7 @@ public class ProtocolProcessor {
subscribeSingleTopic(subscription, username);
}
}
private boolean subscribeSingleTopic(final Subscription newSubscription, String username) {
subscriptions.add(newSubscription.asClientTopicCouple());
@@ -714,7 +727,7 @@ public class ProtocolProcessor {
//TODO this is ugly, it does a linear scan on potential big dataset
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
@Override
public boolean match(String key) {
public boolean match(String key) {
return SubscriptionsStore.matchTopics(key, newSubscription.getTopicFilter());
}
});