Introduced annotation based dispacth to invoke ProtocolProcessor methods

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<Class, Method> messageClassToMethod = new HashMap<Class, Method>();
+    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;
+    }
+
+    
+}

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();
+}

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java

@@ -142,6 +142,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         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) {

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java

@@ -69,6 +69,8 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     
     CountDownLatch m_stopLatch;
     
+    private AnnotationHelper annotationHelper = new AnnotationHelper();
+    
     private SimpleMessaging() {
     }
 
@@ -87,9 +89,11 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
 
         SequenceBarrier barrier = m_ringBuffer.newBarrier();
         m_eventProcessor = new BatchEventProcessor<ValueEvent>(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<ValueEvent> {
             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);