From 43677cca892e978e31569081f3ad9889ea445634 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sun, 17 Aug 2014 16:14:35 +0200 Subject: [PATCH] Introduced annotation based dispacth to invoke ProtocolProcessor methods --- .../messaging/spi/impl/AnnotationHelper.java | 75 +++++++++++++++++++ .../messaging/spi/impl/MQTTMessage.java | 33 ++++++++ .../messaging/spi/impl/ProtocolProcessor.java | 1 + .../messaging/spi/impl/SimpleMessaging.java | 9 ++- .../spi/impl/AnnotationHelperTest.java | 56 ++++++++++++++ 5 files changed, 172 insertions(+), 2 deletions(-) create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelper.java create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MQTTMessage.java create mode 100644 broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelperTest.java 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/AnnotationHelper.java new file mode 100644 index 00000000..5c4a3f86 --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelper.java @@ -0,0 +1,75 @@ +/* + * Copyright (c) 2012-2014 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +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; + +/** + * Annotation @MQTTMessage helper class. + * + * It has the responsibility to identify the @MQTTMessage method of a given instance + * or class and define an easy access to invoke them. + * + * @author andrea + */ +class AnnotationHelper { + + private Map messageClassToMethod = new HashMap(); + private Object targetInstance; + + void processAnnotations(Object instance) { + targetInstance = instance; + Class instanceCls = instance.getClass(); + Method[] allMethods = instanceCls.getDeclaredMethods(); + for (Method m : allMethods) { + MQTTMessage annotation = m.getAnnotation(MQTTMessage.class); + if (annotation == null) { + continue; + } + Class mqttMessageClass = annotation.message(); + messageClassToMethod.put(mqttMessageClass, m); + } + } + + /** + * Dispatch a call to the wrapped target invoking the method that match the + * msg class. + * + * @return true iff the dispatched method was found. + */ + boolean dispatch(ServerChannel session, AbstractMessage msg) { + Method targetMethod = this.messageClassToMethod.get(msg.getClass()); + if (targetMethod == null) { + return false; + } + try { + targetMethod.invoke(targetInstance, session, msg); + } catch (Exception ex) { + throw new RuntimeException(ex); + } + + return true; + } + + +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MQTTMessage.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MQTTMessage.java new file mode 100644 index 00000000..78cc0d79 --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MQTTMessage.java @@ -0,0 +1,33 @@ +/* + * Copyright (c) 2012-2014 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import java.lang.annotation.Documented; +import java.lang.annotation.Retention; +import static java.lang.annotation.ElementType.METHOD; +import static java.lang.annotation.RetentionPolicy.RUNTIME; +import java.lang.annotation.Target; + +/** + * + * @author andrea + */ +@Documented +@Target(METHOD) +@Retention(RUNTIME) +@interface MQTTMessage { + Class message(); +} 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 8cdc9a52..574fc96f 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 @@ -142,6 +142,7 @@ class ProtocolProcessor implements EventHandler { m_executor.submit(m_eventProcessor); } + @MQTTMessage(message = ConnectMessage.class) void processConnect(ServerChannel session, ConnectMessage msg) { LOG.debug("processConnect for client {}", msg.getClientID()); if (msg.getProcotolVersion() != VERSION_3_1 && msg.getProcotolVersion() != VERSION_3_1_1) { 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 0dea1bf9..1669dc0d 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 @@ -69,6 +69,8 @@ public class SimpleMessaging implements IMessaging, EventHandler { CountDownLatch m_stopLatch; + private AnnotationHelper annotationHelper = new AnnotationHelper(); + private SimpleMessaging() { } @@ -87,9 +89,11 @@ public class SimpleMessaging implements IMessaging, EventHandler { SequenceBarrier barrier = m_ringBuffer.newBarrier(); m_eventProcessor = new BatchEventProcessor(m_ringBuffer, barrier, this); - //TODO in a presentation is said to don't do the followinf line!! + //TODO in a presentation is said to don't do the following line!! m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence()); m_executor.submit(m_eventProcessor); + + annotationHelper.processAnnotations(m_processor); disruptorPublish(new InitEvent(configProps)); } @@ -147,7 +151,8 @@ public class SimpleMessaging implements IMessaging, EventHandler { ServerChannel session = ((ProtocolEvent) evt).getSession(); AbstractMessage message = ((ProtocolEvent) evt).getMessage(); if (message instanceof ConnectMessage) { - m_processor.processConnect(session, (ConnectMessage) message); +// m_processor.processConnect(session, (ConnectMessage) message); + annotationHelper.dispatch(session, message); } else if (message instanceof PublishMessage) { PublishEvent pubEvt; String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); 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/AnnotationHelperTest.java new file mode 100644 index 00000000..52ba01a0 --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/AnnotationHelperTest.java @@ -0,0 +1,56 @@ +/* + * Copyright (c) 2012-2014 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.dna.mqtt.moquette.server.ServerChannel; +import static org.junit.Assert.*; +import org.junit.Before; +import org.junit.Test; + +/** + * + * @author andrea + */ +public class AnnotationHelperTest { + + class SimplyAnnotatedFixture { + + @MQTTMessage(message = ConnectMessage.class) + void methodToBeDispatched(ServerChannel session, ConnectMessage msg) { + System.out.println("Just a sample method name"); + methodInvoked = true; + } + } + + SimplyAnnotatedFixture annotationTarget = new SimplyAnnotatedFixture(); + boolean methodInvoked; + + @Before + public void setUp() { + methodInvoked = false; + } + + @Test + public void testAnnotationDiscovery() { + AnnotationHelper annHelper = new AnnotationHelper(); + annHelper.processAnnotations(annotationTarget); + annHelper.dispatch(null, new ConnectMessage()); + + //Verify + assertTrue(methodInvoked); + } +}