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

Added also PublishMessage to be dispatched by annotation based dispather

Browse Source
This commit is contained in:
selva.andre
2014-08-18 23:02:56 +02:00
parent 98f04859ab
commit aee85a7cb0
8 changed files with 117 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+1 -1
View File
@@ -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.
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+52 -3
View File
@@ -272,6 +272,55 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
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<ValueEvent> {
//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
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+4 -7
View File
@@ -138,7 +138,8 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
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<ValueEvent> {
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);
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
-1
View File
@@ -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;
/**
*
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSupportTest.java
+8
View File
@@ -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());
}
}
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java
+48 -15
View File
@@ -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<HawtDBPersistentStore.StoredMessage> messages = m_storageService.searchMatching(new IMatchingCondition() {
broker/src/test/resources/log4j.properties
+3 -1
View File
@@ -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
pom.xml
+1 -1
View File
@@ -11,7 +11,7 @@
<packaging>pom</packaging>
<version>0.7-SNAPSHOT</version>
<name>Moquette MQTT library</name>
<name>Moquette MQTT parent</name>
<url>http://code.google.com/p/moquette-mqtt/</url>