package com.alibaba.rocketmq.client.consumer.cacheable;

import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.client.producer.concurrent.DefaultLocalMessageStore;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;

import java.util.concurrent.LinkedBlockingQueue;

public class ProcessMessageTask implements Runnable {

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

    private static final String NEXT_TIME_KEY = "next_time";

    private MessageExt message;

    private MessageHandler messageHandler;

    private DefaultLocalMessageStore localMessageStore;

    private LinkedBlockingQueue<MessageExt> messageQueue;

    public ProcessMessageTask(MessageExt message, //Message to process.
                              MessageHandler messageHandler, //Message handler.
                              DefaultLocalMessageStore localMessageStore, //Local message store.
                              LinkedBlockingQueue<MessageExt> messageQueue //Buffered message queue.
    ) {
        this.message = message;
        this.messageHandler = messageHandler;
        this.localMessageStore = localMessageStore;
        this.messageQueue = messageQueue;
    }

    @Override
    public void run() {
        try {
            int result = messageHandler.handle(message);
            //Remove the message from in-progress queue.
            if (null != messageQueue && messageQueue.contains(message)) {
                messageQueue.remove(message);
            }

            if (result > 0) {
                Message me = TranslateMsg.getMessageFromMessageExt(message);
                me.putUserProperty(NEXT_TIME_KEY, String.valueOf(System.currentTimeMillis() + result));
                localMessageStore.stash(me);
            }
        } catch (Exception e) {
            LOGGER.error("ProcessMessageTask error", e);
        }
    }

}

@@ -1347,52 +1347,460 @@
     
-int result = messageHandler.handle(message);
+long start = System.currentTimeMillis();%0A            LOGGER.info(%22Request business client to process message%5BmsgId=%22 + message.getMsgId() + %22%5D. CurrentTimeInMilliseconds:%22 + start);%0A            int result = messageHandler.handle(message);%0A            long end = System.currentTimeMillis();%0A            LOGGER.info(%22Business processing completes. CurrentTimeInMilliseconds:%22 + System.currentTimeMillis() + %22. Cost: %22 + (end - start) + %22 milliseconds%22);%0A
 %0A   
@@ -1965,32 +1965,115 @@
 emove(message);%0A
+                LOGGER.info(%22Now %22 + messageQueue.size() + %22 messages in queue.%22);%0A
             %7D%0A%0A 
@@ -2302,34 +2302,238 @@
     
-localMessageStore.stash(me
+LOGGER.info(%22Stashing message%5BmsgId=%22 + message.getMsgId() + %22%5D for later retry in %22 + result%0A                        + %22 milliseconds.%22);%0A                localMessageStore.stash(me);%0A                LOGGER.info(%22Message stashed.%22
 );%0A 

add more log to cacheable client