From 43d19e02a6c50b3d4e7a50793de2b34a3991e7b3 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sun, 17 Aug 2014 22:03:25 +0200 Subject: [PATCH] Renamed AnnotationHelper to AnnotationSupport --- ...tationHelper.java => AnnotationSupport.java} | 7 ++----- .../messaging/spi/impl/ProtocolProcessor.java | 6 +++++- .../messaging/spi/impl/SimpleMessaging.java | 17 ++++++----------- ...elperTest.java => AnnotationSuportTest.java} | 4 ++-- .../spi/impl/ProtocolProcessorTest.java | 5 ++++- 5 files changed, 19 insertions(+), 20 deletions(-) rename broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/{AnnotationHelper.java => AnnotationSupport.java} (90%) rename broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/{AnnotationHelperTest.java => AnnotationSuportTest.java} (94%) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelper.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSupport.java similarity index 90% rename from broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelper.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSupport.java index 5c4a3f86..e676d82e 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelper.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSupport.java @@ -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 messageClassToMethod = new HashMap(); + private final Map messageClassToMethod = new HashMap(); private Object targetInstance; void processAnnotations(Object instance) { 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 3a74fd21..b80ede8f 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 @@ -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 { 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); 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 8ff8e28d..943f7f4a 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 @@ -66,7 +66,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { 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 { 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 { } } + @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 { 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); } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelperTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSuportTest.java similarity index 94% rename from broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelperTest.java rename to broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSuportTest.java index 52ba01a0..aeb20138 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelperTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationSuportTest.java @@ -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()); 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 76917689..baff74b8 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 @@ -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();