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

Renamed AnnotationHelper to AnnotationSupport

Browse Source
This commit is contained in:
selva.andre
2014-08-17 22:03:25 +02:00
parent 2c90b1386d
commit 43d19e02a6
5 changed files with 19 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelper.java → broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSupport.java
+2 -5
View File
@@ -15,12 +15,9 @@
*/
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.server.ServerChannel;
@@ -32,9 +29,9 @@ import org.dna.mqtt.moquette.server.ServerChannel;
*
* @author andrea
*/
class AnnotationHelper {
class AnnotationSupport {
private Map<Class, Method> messageClassToMethod = new HashMap<Class, Method>();
private final Map<Class, Method> messageClassToMethod = new HashMap<Class, Method>();
private Object targetInstance;
void processAnnotations(Object instance) {
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+5 -1
View File
@@ -41,6 +41,7 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.dna.mqtt.moquette.proto.messages.DisconnectMessage;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
import org.dna.mqtt.moquette.proto.messages.PubRecMessage;
@@ -487,7 +488,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
m_messagesStore.cleanInFlight(publishKey);
}
void processDisconnect(ServerChannel session, String clientID, boolean cleanSession) throws InterruptedException {
@MQTTMessage(message = DisconnectMessage.class)
void processDisconnect(ServerChannel session, DisconnectMessage msg) throws InterruptedException {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
cleanSession(clientID);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+6 -11
View File
@@ -66,7 +66,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
private static SimpleMessaging INSTANCE;
private final ProtocolProcessor m_processor = new ProtocolProcessor();
private final AnnotationHelper annotationHelper = new AnnotationHelper();
private final AnnotationSupport annotationSupport = new AnnotationSupport();
CountDownLatch m_stopLatch;
@@ -92,7 +92,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
m_executor.submit(m_eventProcessor);
annotationHelper.processAnnotations(m_processor);
annotationSupport.processAnnotations(m_processor);
processInit(configProps);
// disruptorPublish(new InitEvent(configProps));
}
@@ -133,6 +133,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
}
}
@Override
public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
MessagingEvent evt = t.getEvent();
LOG.info("onEvent processing messaging event from input ringbuffer {}", evt);
@@ -149,20 +150,14 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
message instanceof PubRelMessage ||
message instanceof PubRecMessage ||
message instanceof PubCompMessage ||
message instanceof PubAckMessage) {
annotationHelper.dispatch(session, message);
message instanceof PubAckMessage ||
message instanceof DisconnectMessage) {
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 if (message instanceof DisconnectMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
//close the TCP connection
//session.close(true);
m_processor.processDisconnect(session, clientID, cleanSession);
} else {
throw new RuntimeException("Illegal message received " + message);
}
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelperTest.java → broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSuportTest.java
+2 -2
View File
@@ -25,7 +25,7 @@ import org.junit.Test;
*
* @author andrea
*/
public class AnnotationHelperTest {
public class AnnotationSuportTest {
class SimplyAnnotatedFixture {
@@ -46,7 +46,7 @@ public class AnnotationHelperTest {
@Test
public void testAnnotationDiscovery() {
AnnotationHelper annHelper = new AnnotationHelper();
AnnotationSupport annHelper = new AnnotationSupport();
annHelper.processAnnotations(annotationTarget);
annHelper.dispatch(null, new ConnectMessage());
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java
+4 -1
View File
@@ -35,6 +35,7 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.dna.mqtt.moquette.proto.messages.DisconnectMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
@@ -259,11 +260,13 @@ public class ProtocolProcessorTest {
connMsg.setProcotolVersion(VERSION_3_1_1);
connMsg.setClientID("CliID");
connMsg.setCleanSession(false);
m_session.setAttribute(Constants.ATTR_CLIENTID, "CliID");
m_session.setAttribute(Constants.CLEAN_SESSION, false);
//Connect a first time
m_processor.processConnect(m_session, connMsg);
//disconnect
m_processor.processDisconnect(m_session, "CliID", false);
m_processor.processDisconnect(m_session, new DisconnectMessage());
//Exercise, reconnect
MockReceiverChannel firstReceiverSession = new MockReceiverChannel();