Added QoS 1 management to Client part

parser/src/main/java/org/dna/mqtt/moquette/client/Client.java

@@ -24,17 +24,7 @@ import org.dna.mqtt.moquette.ConnectionException;
 import org.dna.mqtt.moquette.MQTTException;
 import org.dna.mqtt.moquette.PublishException;
 import org.dna.mqtt.moquette.SubscribeException;
-import org.dna.mqtt.moquette.proto.ConnAckDecoder;
+import org.dna.mqtt.moquette.proto.*;
-import org.dna.mqtt.moquette.proto.ConnectEncoder;
-import org.dna.mqtt.moquette.proto.DisconnectEncoder;
-import org.dna.mqtt.moquette.proto.PingReqEncoder;
-import org.dna.mqtt.moquette.proto.PingRespDecoder;
-import org.dna.mqtt.moquette.proto.PublishDecoder;
-import org.dna.mqtt.moquette.proto.PublishEncoder;
-import org.dna.mqtt.moquette.proto.SubAckDecoder;
-import org.dna.mqtt.moquette.proto.SubscribeEncoder;
-import org.dna.mqtt.moquette.proto.UnsubAckDecoder;
-import org.dna.mqtt.moquette.proto.UnsubscribeEncoder;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
 import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
@@ -120,6 +110,7 @@ public final class Client {
         decoder.addMessageDecoder(new SubAckDecoder());
         decoder.addMessageDecoder(new UnsubAckDecoder());
         decoder.addMessageDecoder(new PublishDecoder());
+        decoder.addMessageDecoder(new PubAckDecoder());
         decoder.addMessageDecoder(new PingRespDecoder());
 
         DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
@@ -230,27 +221,46 @@ public final class Client {
     public void publish(String topic, byte[] payload) throws PublishException {
         publish(topic, payload, false);
     }
-    
+
+    /**
+     * Publish by default with QoS 0
+     * */
     public void publish(String topic, byte[] payload, boolean retain) throws PublishException {
+        publish(topic, payload, AbstractMessage.QOSType.MOST_ONE, retain);
+    }
+
+    public void publish(String topic, byte[] payload, AbstractMessage.QOSType qos, boolean retain) throws PublishException {
         PublishMessage msg = new PublishMessage();
         msg.setRetainFlag(retain);
         msg.setTopicName(topic);
         msg.setPayload(payload);
 
-        //Untill the server could handle all the Qos levels
+        //Untill the server could handle all the Qos 2 level
-        msg.setQos(AbstractMessage.QOSType.MOST_ONE);
+        if (qos != AbstractMessage.QOSType.MOST_ONE) {
+            msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
+            int messageID = m_messageIDGenerator.next();
+            msg.setMessageID(messageID);
 
-        WriteFuture wf = m_session.write(msg);
+            try {
-        try {
+                manageSendQoS1(msg);
-            wf.await();
+            } catch (Throwable ex) {
-        } catch (InterruptedException ex) {
+                throw new MQTTException(ex);
-            LOG.debug(null, ex);
+            }
-            throw new PublishException(ex);
+        } else {
-        }
+            //QoS 0 case
+            msg.setQos(AbstractMessage.QOSType.MOST_ONE);
+            WriteFuture wf = m_session.write(msg);
+            try {
+                wf.await();
+            } catch (InterruptedException ex) {
+                LOG.debug(null, ex);
+                throw new PublishException(ex);
+            }
 
-        Throwable ex = wf.getException();
+            Throwable ex = wf.getException();
-        if (ex != null) {
+            if (ex != null) {
-            throw new PublishException(ex);
+                throw new PublishException(ex);
+            }
         }
 
         updatePinger();
@@ -368,6 +378,12 @@ public final class Client {
         m_receivedSubAckMessageID = messageID;
     }
 
+    void publishAckCallback(Integer messageID) {
+        LOG.info("publishAckCallback invoked");
+        m_subscribeBarrier.countDown();
+        m_receivedSubAckMessageID = messageID;
+    }
+
     protected void publishCallback(String topic, byte[] payload) {
         IPublishCallback callback = m_subscribersList.get(topic);
         if (callback == null) {

parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java

@@ -2,12 +2,10 @@ package org.dna.mqtt.moquette.client;
 
 import org.apache.mina.core.service.IoHandlerAdapter;
 import org.apache.mina.core.session.IoSession;
-import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
+import org.dna.mqtt.moquette.proto.messages.*;
+
 import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
-import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
+
-import org.dna.mqtt.moquette.proto.messages.PublishMessage;
-import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
-import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,11 +43,18 @@ public class ClientMQTTHandler extends IoHandlerAdapter {
             case PUBLISH:
                 handlePublish(session, (PublishMessage) msg);
                 break;
+            case PUBACK:
+                handlePublishAck(session, (PubAckMessage) msg);
+                break;
             case PINGRESP:    
                 break;
         }
     }
 
+    private void handlePublishAck(IoSession session, PubAckMessage msg) {
+        m_callback.publishAckCallback(msg.getMessageID());
+    }
+
     private void handleConnectAck(IoSession session, ConnAckMessage connAckMessage) {
         m_callback.connectionAckCallback(connAckMessage.getReturnCode());
     }