diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 574fc96f..3a74fd21 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -49,6 +49,7 @@ import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.proto.messages.SubAckMessage; import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; import org.dna.mqtt.moquette.server.ConnectionDescriptor; import org.dna.mqtt.moquette.server.Constants; import org.dna.mqtt.moquette.server.IAuthenticator; @@ -252,7 +253,10 @@ class ProtocolProcessor implements EventHandler { } } - void processPubAck(String clientID, int messageID) { + @MQTTMessage(message = PubAckMessage.class) + void processPubAck(ServerChannel session, PubAckMessage msg) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + int messageID = msg.getMessageID(); //Remove the message from message store m_messagesStore.cleanPersistedPublishMessage(clientID, messageID); } @@ -428,7 +432,10 @@ class ProtocolProcessor implements EventHandler { * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish * to all interested subscribers. * */ - void processPubRel(String clientID, int messageID) { + @MQTTMessage(message = PubRelMessage.class) + void processPubRel(ServerChannel session, PubRelMessage msg) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + int messageID = msg.getMessageID(); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); String publishKey = String.format("%s%d", clientID, messageID); PublishEvent evt = m_messagesStore.retrieveQoS2Message(publishKey); @@ -456,7 +463,10 @@ class ProtocolProcessor implements EventHandler { disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubCompMessage)); } - void processPubRec(String clientID, int messageID) { + @MQTTMessage(message = PubRecMessage.class) + void processPubRec(ServerChannel session, PubRecMessage msg) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + int messageID = msg.getMessageID(); //once received a PUBREC reply with a PUBREL(messageID) LOG.debug("\t\tSRV <--PUBREC-- SUB processPubRec invoked for clientID {} ad messageID {}", clientID, messageID); PubRelMessage pubRelMessage = new PubRelMessage(); @@ -467,7 +477,10 @@ class ProtocolProcessor implements EventHandler { disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRelMessage)); } - void processPubComp(String clientID, int messageID) { + @MQTTMessage(message = PubCompMessage.class) + void processPubComp(ServerChannel session, PubCompMessage msg) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + int messageID = msg.getMessageID(); 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 String publishKey = String.format("%s%d", clientID, messageID); @@ -513,7 +526,11 @@ class ProtocolProcessor implements EventHandler { * Remove the clientID from topic subscription, if not previously subscribed, * doesn't reply any error */ - void processUnsubscribe(ServerChannel session, String clientID, List topics, int messageID) { + @MQTTMessage(message = UnsubscribeMessage.class) + void processUnsubscribe(ServerChannel session, UnsubscribeMessage msg) { + List topics = msg.topicFilters(); + int messageID = msg.getMessageID(); + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); LOG.debug("processUnsubscribe invoked, removing subscription on topics {}, for clientID <{}>", topics, clientID); for (String topic : topics) { @@ -527,8 +544,10 @@ class ProtocolProcessor implements EventHandler { session.write(ackMessage); } - - void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) { + @MQTTMessage(message = SubscribeMessage.class) + void processSubscribe(ServerChannel session, SubscribeMessage msg) { + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); LOG.debug("processSubscribe invoked from client {} with msgID {}", clientID, msg.getMessageID()); for (SubscribeMessage.Couple req : msg.subscriptions()) { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 1669dc0d..1ec4f9c9 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -150,19 +150,18 @@ public class SimpleMessaging implements IMessaging, EventHandler { } else if (evt instanceof ProtocolEvent) { ServerChannel session = ((ProtocolEvent) evt).getSession(); AbstractMessage message = ((ProtocolEvent) evt).getMessage(); - if (message instanceof ConnectMessage) { -// m_processor.processConnect(session, (ConnectMessage) message); + if (message instanceof ConnectMessage || + message instanceof UnsubscribeMessage || + message instanceof SubscribeMessage || + message instanceof PubRelMessage || + message instanceof PubRecMessage || + message instanceof PubCompMessage || + message instanceof PubAckMessage) { annotationHelper.dispatch(session, message); } else if (message instanceof PublishMessage) { PublishEvent pubEvt; String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); pubEvt = new PublishEvent((PublishMessage) message, clientID); -// if (message.getQos() == QOSType.MOST_ONE) { -// pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, session); -// -// } else { -// pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session); -// } m_processor.processPublish(pubEvt); } else if (message instanceof DisconnectMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); @@ -171,30 +170,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { //close the TCP connection //session.close(true); m_processor.processDisconnect(session, clientID, cleanSession); - } else if (message instanceof UnsubscribeMessage) { - UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message; - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - m_processor.processUnsubscribe(session, clientID, unsubMsg.topicFilters(), unsubMsg.getMessageID()); - } else if (message instanceof SubscribeMessage) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); - m_processor.processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession); - } else if (message instanceof PubRelMessage) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - int messageID = ((PubRelMessage) message).getMessageID(); - m_processor.processPubRel(clientID, messageID); - } else if (message instanceof PubRecMessage) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - int messageID = ((PubRecMessage) message).getMessageID(); - m_processor.processPubRec(clientID, messageID); - } else if (message instanceof PubCompMessage) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - int messageID = ((PubCompMessage) message).getMessageID(); - m_processor.processPubComp(clientID, messageID); - } else if (message instanceof PubAckMessage) { - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - int messageID = ((PubAckMessage) message).getMessageID(); - m_processor.processPubAck(clientID, messageID); } else { throw new RuntimeException("Illegal message received " + message); } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index d475c01f..76917689 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -38,6 +38,7 @@ import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.proto.messages.SubAckMessage; import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; +import org.dna.mqtt.moquette.server.Constants; import org.dna.mqtt.moquette.server.ServerChannel; import static org.junit.Assert.*; import org.junit.Before; @@ -376,7 +377,9 @@ public class ProtocolProcessorTest { //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); - m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + m_session.setAttribute(Constants.ATTR_CLIENTID, FAKE_CLIENT_ID); + m_session.setAttribute(Constants.CLEAN_SESSION, false); + m_processor.processSubscribe(m_session, msg/*, FAKE_CLIENT_ID, false*/); //Verify assertTrue(m_receivedMessage instanceof SubAckMessage); @@ -388,14 +391,15 @@ public class ProtocolProcessorTest { public void testDoubleSubscribe() { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); - + m_session.setAttribute(Constants.ATTR_CLIENTID, FAKE_CLIENT_ID); + m_session.setAttribute(Constants.CLEAN_SESSION, false); subscriptions.clearAllSubscriptions(); assertEquals(0, subscriptions.size()); - m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + m_processor.processSubscribe(m_session, msg/*, FAKE_CLIENT_ID, false*/); //Exercise - m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + m_processor.processSubscribe(m_session, msg/*, FAKE_CLIENT_ID, false*/); //Verify assertEquals(1, subscriptions.size()); @@ -454,11 +458,13 @@ public class ProtocolProcessorTest { ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher"); m_processor.processPublish(pubEvt); + m_session.setAttribute(Constants.ATTR_CLIENTID, FAKE_PUBLISHER_ID); + m_session.setAttribute(Constants.CLEAN_SESSION, false); //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#")); - m_processor.processSubscribe(m_session, msg, FAKE_PUBLISHER_ID, false); + m_processor.processSubscribe(m_session, msg/*, FAKE_PUBLISHER_ID, false*/); //Verify //wait the latch