Fixes moquette-io#583: inflightSlots counting error

When processing an ACK, the inflightSlots should only be increased if it
is the first time the ACK is received for a given message. The ACK may
be received multiple times, the subsequent ACKs should not cause the
count to be increased.

broker/src/main/java/io/moquette/broker/Session.java

@@ -189,9 +189,9 @@ public void processPubRec(int packetId) {
         SessionRegistry.EnqueuedMessage removed = inflightWindow.remove(packetId);
         if (removed != null) {
             removed.release();
+            inflightSlots.incrementAndGet();
         }
 
-        inflightSlots.incrementAndGet();
         if (canSkipQueue()) {
             inflightSlots.decrementAndGet();
             int pubRelPacketId = packetId/*mqttConnection.nextPacketId()*/;
@@ -211,10 +211,9 @@ public void processPubComp(int messageID) {
         SessionRegistry.EnqueuedMessage removed = inflightWindow.remove(messageID);
         if (removed != null) {
             removed.release();
+            inflightSlots.incrementAndGet();
         }
 
-        inflightSlots.incrementAndGet();
-
         drainQueueToConnection();
 
         // TODO notify the interceptor
@@ -315,9 +314,9 @@ void pubAckReceived(int ackPacketId) {
         SessionRegistry.EnqueuedMessage removed = inflightWindow.remove(ackPacketId);
         if (removed != null) {
             removed.release();
+            inflightSlots.incrementAndGet();
         }
 
-        inflightSlots.incrementAndGet();
         drainQueueToConnection();
     }