package com.alibaba.rocketmq.client.producer.concurrent;

import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.common.message.Message;
import org.slf4j.Logger;

public class ResendMessageTask implements Runnable {

    /**
     * Indicate number of messages to retrieve from local message store each time.
     */
    private static final int BATCH_FETCH_MESSAGE_FROM_STORE_SIZE = 100;

    /**
     * Logger instance.
     */
    private static final Logger LOGGER = ClientLogger.getLog();

    private LocalMessageStore localMessageStore;

    private MultiThreadMQProducer multiThreadMQProducer;

    public ResendMessageTask(LocalMessageStore localMessageStore, MultiThreadMQProducer multiThreadMQProducer) {
        this.localMessageStore = localMessageStore;
        this.multiThreadMQProducer = multiThreadMQProducer;
    }

    @Override
    public void run() {
        LOGGER.info("Start to re-send");
        if (localMessageStore.getNumberOfMessageStashed() == 0) {
            LOGGER.info("No stashed messages to re-send");
            return;
        }

        int popSize = Math.min(multiThreadMQProducer.getSemaphore().availablePermits(),
                BATCH_FETCH_MESSAGE_FROM_STORE_SIZE);

        if (popSize == 0) {
            LOGGER.info("No permits available in semaphore. Yield and wait for next round.");
            return;
        }

        Message[] messages = localMessageStore.pop(popSize);
        if (null == messages || messages.length == 0) {
            LOGGER.info("No stashed messages to re-send");
            return;
        }

        while (null != messages && messages.length > 0) {
            //Acquire tokens from semaphore.
            multiThreadMQProducer.getSemaphore().acquireUninterruptibly(messages.length);
            //Send messages with tokens.
            multiThreadMQProducer.send(messages, true);

            LOGGER.info(messages.length + " stashed messages re-sending completes: scheduled job submitted.");

            //Prepare for next loop step.
            popSize = Math.min(multiThreadMQProducer.getSemaphore().availablePermits(),
                    BATCH_FETCH_MESSAGE_FROM_STORE_SIZE);
            if (popSize == 0) {
                LOGGER.info("No permits available in semaphore. Break looping.");
                break;
            }
            messages = localMessageStore.pop(popSize);
        }
    }
}

@@ -887,24 +887,42 @@
 oid run() %7B%0A
+        try %7B%0A    
         LOGG
@@ -942,32 +942,36 @@
 t to re-send%22);%0A
+    
         if (loca
@@ -1024,32 +1024,36 @@
 ) %7B%0A            
+  
+  
 LOGGER.info(%22No 
@@ -1087,32 +1087,36 @@
 %22);%0A            
+    
 return;%0A        
@@ -1107,35 +1107,43 @@
 return;%0A        
+  
+  
 %7D%0A%0A
+    
         int popS
@@ -1222,32 +1222,36 @@
                 
+    
 BATCH_FETCH_MESS
@@ -1273,32 +1273,36 @@
 SIZE);%0A%0A        
+    
 if (popSize == 0
@@ -1309,32 +1309,36 @@
 ) %7B%0A            
+  
+  
 LOGGER.info(%22No 
@@ -1407,32 +1407,36 @@
 %22);%0A            
+    
 return;%0A        
@@ -1427,35 +1427,43 @@
 return;%0A        
+  
+  
 %7D%0A%0A
+    
         Message%5B
@@ -1511,24 +1511,28 @@
 e);%0A        
+ 
+   
 if (null == 
@@ -1571,32 +1571,36 @@
 ) %7B%0A            
+    
 LOGGER.info(%22No 
@@ -1634,32 +1634,36 @@
 %22);%0A            
+ 
+   
 return;%0A        
@@ -1654,32 +1654,36 @@
 return;%0A        
+    
 %7D%0A%0A        while
@@ -1677,16 +1677,20 @@
         
+ 
+   
 while (n
@@ -1743,16 +1743,20 @@
         
+    
 //Acquir
@@ -1784,32 +1784,36 @@
 re.%0A            
+ 
+   
 multiThreadMQPro
@@ -1874,16 +1874,20 @@
 ength);%0A
+    
         
@@ -1923,32 +1923,36 @@
 ns.%0A            
+    
 multiThreadMQPro
@@ -1976,24 +1976,28 @@
 es, true);%0A%0A
+    
             
@@ -2108,16 +2108,20 @@
         
+    
 //Prepar
@@ -2142,16 +2142,20 @@
 p step.%0A
+    
         
@@ -2246,32 +2246,36 @@
                 
+    
 BATCH_FETCH_MESS
@@ -2296,16 +2296,20 @@
 _SIZE);%0A
+    
         
@@ -2340,32 +2340,36 @@
                 
+    
 LOGGER.info(%22No 
@@ -2434,16 +2434,20 @@
         
+   
+ 
 break;%0A 
@@ -2453,26 +2453,34 @@
             
-%7D%0A
+    %7D%0A    
             
@@ -2526,22 +2526,137 @@
 e);%0A        
-%7D
+    %7D%0A        %7D catch (Throwable e) %7B%0A            LOGGER.error(%22ResendMessageTask got an exception!%22, e);%0A        %7D%0A
 %0A    %7D%0A%7D%0A

Debug resend service stalling.