Added unit test and fix for issue #12, due to bad topic in publish message after a late subscription and messages retained in DB.

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

@@ -26,10 +26,12 @@ public class HawtDBStorageService implements IStorageService {
     public static class StoredMessage implements Serializable {
         AbstractMessage.QOSType m_qos;
         byte[] m_payload;
+        String m_topic;
 
-        StoredMessage(byte[] message, AbstractMessage.QOSType qos) {
+        StoredMessage(byte[] message, AbstractMessage.QOSType qos, String topic) {
             m_qos = qos;
             m_payload = message;
+            m_topic = topic;
         }
 
         AbstractMessage.QOSType getQos() {
@@ -39,6 +41,10 @@ public class HawtDBStorageService implements IStorageService {
         byte[] getPayload() {
             return m_payload;
         }
+        
+        String getTopic() {
+            return m_topic;
+        }
     }
 
 
@@ -132,7 +138,7 @@ public class HawtDBStorageService implements IStorageService {
             m_retainedStore.remove(topic);
         } else {
             //store the message to the topic
-            m_retainedStore.put(topic, new StoredMessage(message, qos));
+            m_retainedStore.put(topic, new StoredMessage(message, qos, topic));
         }
     }
 

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

@@ -400,7 +400,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
         for (StoredMessage storedMsg : messages) {
             //fire the as retained the message
             LOG.debug("Inserting NotifyEvent into outbound for topic " + topic);
-            notify(new NotifyEvent(newSubscription.getClientId(), topic, storedMsg.getQos(), storedMsg.getPayload(), true));
+            notify(new NotifyEvent(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true));
         }
     }
 
@@ -408,7 +408,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
         LOG.debug("processSubscribe invoked");
 
         for (SubscribeMessage.Couple req : msg.subscriptions()) {
-            QOSType qos = AbstractMessage.QOSType.fromByte(req.getQos());
+            QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
             Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession);
             subscribeSingleTopic(newSubscription, req.getTopic());
         }

broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java

@@ -1,5 +1,6 @@
 package org.dna.mqtt.moquette.messaging.spi.impl;
 
+import org.apache.mina.core.filterchain.IoFilter;
 import org.apache.mina.core.filterchain.IoFilterAdapter;
 import org.apache.mina.core.session.DummySession;
 import org.apache.mina.core.session.IoSession;
@@ -124,6 +125,49 @@ public class SimpleMessagingTest {
         //TODO check received message attributes
     }
     
+    @Test
+    public void testPublishOfRetainedMessage() throws Exception {
+        m_session.getFilterChain().remove("MessageCatcher");
+        m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
+
+            @Override
+            public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session,
+                                    WriteRequest writeRequest) throws Exception {
+                try {
+                    System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName());
+                    if (writeRequest.getMessage() instanceof PublishMessage) {
+                        m_receivedMessage = (AbstractMessage) writeRequest.getMessage();
+                    }
+                    
+                    if (m_receivedMessage instanceof ConnAckMessage) {
+                        ConnAckMessage buf = (ConnAckMessage) m_receivedMessage;
+                        m_returnCode = buf.getReturnCode();
+                    }
+                } catch (Exception ex) {
+                    throw new AssertionError("Wrong return code");
+                }
+            }
+        });    
+            
+        //simulate a connect that register a clientID to an IoSession
+        ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, true);
+        messaging.m_clientIDs.put(FAKE_CLIENT_ID, connDescr);
+        
+        PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null);
+        messaging.processPublish(pubEvt);
+        
+        //Exercise
+        SubscribeMessage msg = new SubscribeMessage();
+        msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#"));
+        messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
+        
+        //Verify
+        assertNotNull(m_receivedMessage); 
+        assertTrue(m_receivedMessage instanceof PublishMessage);
+        PublishMessage pubMessage = (PublishMessage) m_receivedMessage;
+        assertEquals(FAKE_TOPIC, pubMessage.getTopicName());
+    }
+    
     @Test
     public void testHandleConnect_BadProtocol() {
         connMsg.setProcotolVersion((byte) 0x02);