package org.nlab.article.fairconsuming;

import java.util.List;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;


public class DwrrMainLoop implements Runnable {

    public static final Logger LOG = LoggerFactory.getLogger(DwrrMainLoop.class);

    public static final int MESSAGE_WEIGHT = 4;

    private final List<DwrrSlot> slots;
    private final Semaphore consumerToken;
    private final Channel channel;
    private final Consumer<QueueingConsumer.Delivery> consumer;

    private volatile boolean stopProcessing;

    public DwrrMainLoop(List<DwrrSlot> slots, Semaphore consumerToken, Channel channel, Consumer<QueueingConsumer.Delivery> consumer) {
        this.slots = slots;
        this.consumerToken = consumerToken;
        this.channel = channel;
        this.consumer = consumer;
    }

    public void run() {

        try {

            while (!stopProcessing) {
                int processedMessageCounter = 0;

                //Try to acquire a consumer token, wait for 5 seconds
                if (consumerToken.tryAcquire(1, 5, TimeUnit.SECONDS)) {
                    //We got a token, a delivery is available
                    for (DwrrSlot slot : slots) {
                        //The deficit is reduced per iteration
                        slot.reduceDeficit();

                        //Loop until the slot does not contain any deliveries or until the slot deficit is bellow the message weight
                        while (!slot.getConsumer().getDeliveries().isEmpty() && slot.getDeficit() >= MESSAGE_WEIGHT) {
                            QueueingConsumer.Delivery delivery = slot.getConsumer().getDeliveries().poll();
                            slot.increaseDeficit(MESSAGE_WEIGHT);
                            //Consume
                            consumer.accept(delivery);
                            //Finally ack the message, so RabbitMQ will push a new one to this slot
                            slot.getConsumer().getChannel().basicAck(delivery.getEnvelope().getDeliveryTag(), false);

                        }
                        //If the slot does not contain any deliveries, reset the deficit
                        if (slot.getConsumer().getDeliveries().isEmpty()) {
                            slot.reset();
                        }
                    }

                    if (processedMessageCounter > 0) {
                        //If we have processed message, we must acquire the number of processed message minus the first acquire
                        consumerToken.acquire(processedMessageCounter - 1);
                    } else {
                        //If we do not have processed message (because all deficit where < weight) we must release the token
                        consumerToken.release();
                    }

                } else {
                    LOG.info("No message");
                }
            }
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }
    }

    public void stop() {
        stopProcessing = true;
    }
}

@@ -1085,56 +1085,8 @@
 ) %7B%0A
-                int processedMessageCounter = 0;
 %0A%0A  
@@ -1221,24 +1221,108 @@
 SECONDS)) %7B%0A
+                    int processedMessageCounter = 0;%0A                    try %7B%0A%0A    
             
@@ -1391,16 +1391,20 @@
         
+    
 for (Dwr
@@ -1421,24 +1421,28 @@
  : slots) %7B%0A
+    
             
@@ -1512,24 +1512,28 @@
             
+    
 slot.reduceD
@@ -1539,24 +1539,28 @@
 Deficit();%0A%0A
+    
             
@@ -1680,16 +1680,20 @@
  weight%0A
+    
         
@@ -1831,16 +1831,20 @@
         
+    
 Queueing
@@ -1939,24 +1939,28 @@
             
+    
 slot.increas
@@ -2013,16 +2013,20 @@
         
+   
+ 
 //Consum
@@ -2027,16 +2027,20 @@
 Consume%0A
+    
         
@@ -2086,16 +2086,20 @@
 ivery);%0A
+    
         
@@ -2214,24 +2214,28 @@
             
+   
+ 
 slot.getCons
@@ -2309,24 +2309,163 @@
 ), false);%0A%0A
+                                //Increment the number of processed message%0A                                processedMessageCounter++;%0A    
             
@@ -2470,32 +2470,36 @@
               %7D%0A
+    
                 
@@ -2591,24 +2591,28 @@
             
+    
 if (slot.get
@@ -2675,24 +2675,28 @@
             
+ 
+   
 slot.reset()
@@ -2693,24 +2693,28 @@
 ot.reset();%0A
+    
             
@@ -2739,35 +2739,74 @@
                 
-%7D%0A%0A
+    %7D%0A                    %7D finally %7B%0A    
                 
@@ -2860,32 +2860,36 @@
                 
+    
 //If we have pro
@@ -2992,32 +2992,36 @@
                 
+    
 consumerToken.ac
@@ -3056,16 +3056,20 @@
 r - 1);%0A
+    
         
@@ -3105,32 +3105,36 @@
                 
+    
 //If we do not h
@@ -3234,32 +3234,36 @@
                 
+    
 consumerToken.re
@@ -3267,24 +3267,50 @@
 .release();%0A
+                        %7D%0A
             

Fix the processedMessageCounter increment issue