From aee85a7cb07905ba5043c1447a5c0ff7f7671a3d Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Mon, 18 Aug 2014 23:02:56 +0200 Subject: [PATCH] Added also PublishMessage to be dispatched by annotation based dispather --- .../moquette/messaging/spi/IMessaging.java | 2 +- .../messaging/spi/impl/ProtocolProcessor.java | 55 +++++++++++++++- .../messaging/spi/impl/SimpleMessaging.java | 11 ++-- .../spi/impl/events/PublishEvent.java | 1 - .../spi/impl/AnnotationSupportTest.java | 8 +++ .../spi/impl/ProtocolProcessorTest.java | 63 ++++++++++++++----- broker/src/test/resources/log4j.properties | 4 +- pom.xml | 2 +- 8 files changed, 117 insertions(+), 29 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index ecadfe19..6c4560bf 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -19,7 +19,7 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.server.ServerChannel; /** - * Interface to the underling messaging system used to publish, subscribe. + * Callback interface used to be notified of some events from the input event queue. * * It's the abstraction of the messaging stuff attached in after the front protocol * parsing stuff. 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 b80ede8f..722fa657 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 @@ -272,6 +272,55 @@ class ProtocolProcessor implements EventHandler { m_messagesStore.cleanPersistedPublishes(clientID); } + @MQTTMessage(message = PublishMessage.class) + void processPublish(ServerChannel session, PublishMessage msg) { + LOG.trace("PUB --PUBLISH--> SRV processPublish invoked with {}", msg); + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + final String topic = msg.getTopicName(); + final AbstractMessage.QOSType qos = msg.getQos(); + final ByteBuffer message = msg.getPayload(); + boolean retain = msg.isRetainFlag(); + + LOG.info("Publish recieved from clientID <{}> on topic <{}> with QoS {}", + clientID, topic, qos); + + String publishKey = null; + if (qos == AbstractMessage.QOSType.LEAST_ONE) { + //store the temporary message + publishKey = String.format("%s%d", clientID, msg.getMessageID()); + m_messagesStore.addInFlight(new PublishEvent(msg, clientID), publishKey); + } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { + publishKey = String.format("%s%d", clientID, msg.getMessageID()); + //store the message in temp store + m_messagesStore.persistQoS2Message(publishKey, new PublishEvent(msg, clientID)); + sendPubRec(clientID, msg.getMessageID()); + } + + //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher + if (qos != AbstractMessage.QOSType.EXACTLY_ONCE) { + publish2Subscribers(topic, qos, message, retain, msg.getMessageID()); + } + + if (qos == AbstractMessage.QOSType.LEAST_ONE) { + if (publishKey == null) { + throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + msg); + } + m_messagesStore.cleanInFlight(publishKey); + sendPubAck(new PubAckEvent(msg.getMessageID(), clientID)); + LOG.debug("replying with PubAck to MSG ID {}", msg.getMessageID()); + } + + if (retain) { + if (qos == AbstractMessage.QOSType.MOST_ONE) { + //QoS == 0 && retain => clean old retained + m_messagesStore.cleanRetained(topic); + } else { + m_messagesStore.storeRetained(topic, message, qos); + } + } + } + + @Deprecated protected void processPublish(PublishEvent evt) { LOG.trace("PUB --PUBLISH--> SRV processPublish invoked with {}", evt); final String topic = evt.getTopic(); @@ -344,13 +393,13 @@ class ProtocolProcessor implements EventHandler { //if the target subscription is not clean session and is not connected => store it if (!sub.isCleanSession() && !sub.isActive()) { //clone the event with matching clientID - PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID); + PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0); m_messagesStore.storePublishForFuture(newPublishEvt); } else { //if QoS 2 then store it in temp memory - if (qos ==AbstractMessage.QOSType.EXACTLY_ONCE) { + if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { String publishKey = String.format("%s%d", sub.getClientId(), messageID); - PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID); + PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0); m_messagesStore.addInFlight(newPublishEvt, publishKey); } //publish 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 943f7f4a..a70c1343 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 @@ -138,7 +138,8 @@ public class SimpleMessaging implements IMessaging, EventHandler { MessagingEvent evt = t.getEvent(); LOG.info("onEvent processing messaging event from input ringbuffer {}", evt); if (evt instanceof PublishEvent) { - m_processor.processPublish((PublishEvent) evt); +// m_processor.processPublish((PublishEvent) evt); +// System.out.println("onEvent invoked with PublishEvent ##############################"); } else if (evt instanceof StopEvent) { processStop(); } else if (evt instanceof ProtocolEvent) { @@ -151,13 +152,9 @@ public class SimpleMessaging implements IMessaging, EventHandler { message instanceof PubRecMessage || message instanceof PubCompMessage || message instanceof PubAckMessage || - message instanceof DisconnectMessage) { + message instanceof DisconnectMessage || + message instanceof PublishMessage) { annotationSupport.dispatch(session, message); - } else if (message instanceof PublishMessage) { - PublishEvent pubEvt; - String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - pubEvt = new PublishEvent((PublishMessage) message, clientID); - m_processor.processPublish(pubEvt); } else { throw new RuntimeException("Illegal message received " + message); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java index c51ece56..1cd4df4a 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java @@ -19,7 +19,6 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import java.nio.ByteBuffer; import org.dna.mqtt.moquette.proto.messages.PublishMessage; -import org.dna.mqtt.moquette.server.ServerChannel; /** * diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSupportTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSupportTest.java index 7b13c739..0464fdd8 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSupportTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSupportTest.java @@ -15,6 +15,7 @@ */ package org.dna.mqtt.moquette.messaging.spi.impl; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.server.ServerChannel; import static org.junit.Assert.*; @@ -53,4 +54,11 @@ public class AnnotationSupportTest { //Verify assertTrue(methodInvoked); } + + @Test(expected = RuntimeException.class) + public void testAnnotationFailsDiscovery() { + AnnotationSupport annHelper = new AnnotationSupport(); + annHelper.processAnnotations(annotationTarget); + annHelper.dispatch(null, new ConnAckMessage()); + } } 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 baff74b8..a4f5324d 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 @@ -27,7 +27,6 @@ import java.util.concurrent.TimeUnit; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IMessagesStore; import org.dna.mqtt.moquette.messaging.spi.ISessionsStore; -import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; import static org.dna.mqtt.moquette.parser.netty.Utils.VERSION_3_1_1; @@ -310,8 +309,13 @@ public class ProtocolProcessorTest { //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, false, "FakeCLI"); - m_processor.processPublish(pubEvt); + PublishMessage msg = new PublishMessage(); + msg.setTopicName(FAKE_TOPIC); + msg.setQos(QOSType.MOST_ONE); + msg.setPayload(buffer); + msg.setRetainFlag(false); + m_session.setAttribute(Constants.ATTR_CLIENTID, "FakeCLI"); + m_processor.processPublish(m_session, msg); //Verify assertNotNull(m_receivedMessage); @@ -359,8 +363,13 @@ public class ProtocolProcessorTest { //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); buffer.rewind(); - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, false, "FakeCLI"); - m_processor.processPublish(pubEvt); + PublishMessage msg = new PublishMessage(); + msg.setTopicName(FAKE_TOPIC); + msg.setQos(QOSType.MOST_ONE); + msg.setPayload(buffer); + msg.setRetainFlag(false); + m_session.setAttribute(Constants.ATTR_CLIENTID, "FakeCLI"); + m_processor.processPublish(m_session, msg); //Verify Thread.sleep(100); //ugly but we dependend on the asynch that pull data from back disruptor @@ -459,9 +468,13 @@ public class ProtocolProcessorTest { connectMessage.setCleanSession(subscription.isCleanSession()); m_processor.processConnect(m_session, connectMessage); 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); + PublishMessage pubmsg = new PublishMessage(); + pubmsg.setTopicName(FAKE_TOPIC); + pubmsg.setQos(QOSType.MOST_ONE); + pubmsg.setPayload(buffer); + pubmsg.setRetainFlag(true); m_session.setAttribute(Constants.ATTR_CLIENTID, FAKE_PUBLISHER_ID); + m_processor.processPublish(m_session, pubmsg); m_session.setAttribute(Constants.CLEAN_SESSION, false); //Exercise @@ -490,8 +503,13 @@ public class ProtocolProcessorTest { //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); - PublishEvent pubEvt = new PublishEvent("/topic", AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher"); - m_processor.processPublish(pubEvt); + PublishMessage msg = new PublishMessage(); + msg.setTopicName("/topic"); + msg.setQos(QOSType.MOST_ONE); + msg.setPayload(buffer); + msg.setRetainFlag(true); + m_session.setAttribute(Constants.ATTR_CLIENTID, "Publisher"); + m_processor.processPublish(m_session, msg); //Verify no message is received assertNull(m_receivedMessage); @@ -510,8 +528,13 @@ public class ProtocolProcessorTest { //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); - PublishEvent pubEvt = new PublishEvent("/topic", AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher"); - m_processor.processPublish(pubEvt); + PublishMessage msg = new PublishMessage(); + msg.setTopicName("/topic"); + msg.setQos(QOSType.MOST_ONE); + msg.setPayload(buffer); + msg.setRetainFlag(true); + m_session.setAttribute(Constants.ATTR_CLIENTID, "Publisher"); + m_processor.processPublish(m_session, msg); //Verify no message is received assertNull(m_receivedMessage); @@ -528,13 +551,23 @@ public class ProtocolProcessorTest { connMsg.setClientID("Publisher"); m_processor.processConnect(m_session, connMsg); //prepare and existing retained store + m_session.setAttribute(Constants.ATTR_CLIENTID, "Publisher"); ByteBuffer payload = ByteBuffer.allocate(5).put("Hello".getBytes()); - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.LEAST_ONE, payload, true, "Publisher"); - m_processor.processPublish(pubEvt); + PublishMessage msg = new PublishMessage(); + msg.setTopicName(FAKE_TOPIC); + msg.setQos(QOSType.LEAST_ONE); + msg.setPayload(payload); + msg.setRetainFlag(true); + msg.setMessageID(100); + m_processor.processPublish(m_session, msg); //Exercise - PublishEvent cleanPubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, payload, true, "Publisher"); - m_processor.processPublish(cleanPubEvt); + PublishMessage cleanPubMsg = new PublishMessage(); + cleanPubMsg.setTopicName(FAKE_TOPIC); + cleanPubMsg.setPayload(payload); + cleanPubMsg.setQos(QOSType.MOST_ONE); + cleanPubMsg.setRetainFlag(true); + m_processor.processPublish(m_session, cleanPubMsg); //Verify Collection messages = m_storageService.searchMatching(new IMatchingCondition() { diff --git a/broker/src/test/resources/log4j.properties b/broker/src/test/resources/log4j.properties index 9eb69665..e0f3c5bf 100644 --- a/broker/src/test/resources/log4j.properties +++ b/broker/src/test/resources/log4j.properties @@ -4,6 +4,7 @@ log4j.rootLogger=ERROR, stdout, file log4j.logger.org.dna.mqtt=INFO log4j.logger.org.dna.mqtt.moquette.server.ServerIntegrationFuseTest=DEBUG +log4j.logger.org.dna.mqtt.moquette.server.ServerIntegrationQoSValidationTest=INFO log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=INFO #log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=DEBUG #log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=TRACE @@ -21,7 +22,8 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.Threshold=WARN #log4j.appender.stdout.Threshold=INFO #log4j.appender.stdout.Threshold=DEBUG -#log4j.appender.stdout.Threshold=TRACE +# for debug trace +#log4j.appender.stdout.Threshold=TRACE log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n diff --git a/pom.xml b/pom.xml index e99078d6..ab07e5cd 100644 --- a/pom.xml +++ b/pom.xml @@ -11,7 +11,7 @@ pom 0.7-SNAPSHOT - Moquette MQTT library + Moquette MQTT parent http://code.google.com/p/moquette-mqtt/