// Path: mq-aliyun-consumer-spring-boot-starter/src/main/java/com/github/mq/consumer/parms/ArgumentExtractorWrapper.java
// public class ArgumentExtractorWrapper {
//     private ArgumentExtractor<?> argumentExtractor;
//     private Annotation annotation;
//     private Class<?> parameterClass;
// 
// 
//     public Result excuteExtract(Message message){
//         return argumentExtractor.extract(message,parameterClass,annotation);
//     }
// 
//     public static ArgumentExtractorWrapper build(ArgumentExtractor argumentExtractor,Annotation annotation,Class<?> parameterClass){
//         ArgumentExtractorWrapper extractorWrapper = new ArgumentExtractorWrapper();
//         extractorWrapper.setArgumentExtractor(argumentExtractor);
//         extractorWrapper.setAnnotation(annotation);
//         extractorWrapper.setParameterClass(parameterClass);
//         return extractorWrapper;
//     }
// 
//     public ArgumentExtractor<?> getArgumentExtractor() {
//         return argumentExtractor;
//     }
// 
//     public void setArgumentExtractor(ArgumentExtractor<?> argumentExtractor) {
//         this.argumentExtractor = argumentExtractor;
//     }
// 
//     public Annotation getAnnotation() {
//         return annotation;
//     }
// 
//     public void setAnnotation(Annotation annotation) {
//         this.annotation = annotation;
//     }
// 
//     public Class<?> getParameterClass() {
//         return parameterClass;
//     }
// 
//     public void setParameterClass(Class<?> parameterClass) {
//         this.parameterClass = parameterClass;
//     }
// }


import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;
import com.github.mq.consumer.models.*;
import com.github.mq.consumer.parms.ArgumentExtractorWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Properties;
                    dispatch(consumerId, message, consumerOptional.getMaxReconsume()).equals(Action.commit)?
                            OrderAction.Success : OrderAction.Suspend
            ));
            orderConsumer.start();
        } else {
            Consumer consumer = ONSFactory.createConsumer(properties);
            consumer.subscribe(topic, tags, (message, context) ->
                    dispatch(consumerId, message, consumerOptional.getMaxReconsume()).equals(Action.commit)?
                            com.aliyun.openservices.ons.api.Action.CommitMessage : com.aliyun.openservices.ons.api.Action.ReconsumeLater
            );
            consumer.start();
       }
        logger.info("消费者启动成功: {}({}) , 消费模式: {} , 消费线程数: {} , 最大重试次数: {} {} , 订阅 TOPIC: {} TAG: {}",
                cid, consumerId.isOrdered()?"有序":"无序",
                consumerOptional.getConsumerModel(), consumerOptional.getConsumeThread(), consumerOptional.getMaxReconsume(),
                consumerId.isOrdered()?", 重试前的等待时间:"+consumerOptional.getSuspendTime()+"毫秒":"",topic,tags);
    }


    private static Action dispatch(ConsumerId consumerId, Message message, int defaultReconsume) {
        Tag tag = consumerId.getTagMap().get(message.getTag());
        if (null == tag) {
            throw new RuntimeException(String.format("%s 没有注册 请检查MqConsumer的init配置", message.getTag()));
        }
        int reconsume = tag.getReconsume() == null ? defaultReconsume : tag.getReconsume();
        if (message.getReconsumeTimes() > reconsume) {
            logger.warn("超过允许的最大重试次数。 允许的重试次数: {}, 当前重试次数: {} MessageId: {} topic: {} , tag: {}",
                    reconsume, message.getReconsumeTimes(),message.getMsgID(),message.getTopic(),message.getTag()) ;
            return Action.commit;
        }
// Path: mq-aliyun-consumer-spring-boot-starter/src/main/java/com/github/mq/consumer/parms/ArgumentExtractorWrapper.java
// public class ArgumentExtractorWrapper {
//     private ArgumentExtractor<?> argumentExtractor;
//     private Annotation annotation;
//     private Class<?> parameterClass;
// 
// 
//     public Result excuteExtract(Message message){
//         return argumentExtractor.extract(message,parameterClass,annotation);
//     }
// 
//     public static ArgumentExtractorWrapper build(ArgumentExtractor argumentExtractor,Annotation annotation,Class<?> parameterClass){
//         ArgumentExtractorWrapper extractorWrapper = new ArgumentExtractorWrapper();
//         extractorWrapper.setArgumentExtractor(argumentExtractor);
//         extractorWrapper.setAnnotation(annotation);
//         extractorWrapper.setParameterClass(parameterClass);
//         return extractorWrapper;
//     }
// 
//     public ArgumentExtractor<?> getArgumentExtractor() {
//         return argumentExtractor;
//     }
// 
//     public void setArgumentExtractor(ArgumentExtractor<?> argumentExtractor) {
//         this.argumentExtractor = argumentExtractor;
//     }
// 
//     public Annotation getAnnotation() {
//         return annotation;
//     }
// 
//     public void setAnnotation(Annotation annotation) {
//         this.annotation = annotation;
//     }
// 
//     public Class<?> getParameterClass() {
//         return parameterClass;
//     }
// 
//     public void setParameterClass(Class<?> parameterClass) {
//         this.parameterClass = parameterClass;
//     }
// }


// Path: mq-aliyun-consumer-spring-boot-starter/src/main/java/com/github/mq/consumer/impls/ConsumerRun.java
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;
import com.github.mq.consumer.models.*;
import com.github.mq.consumer.parms.ArgumentExtractorWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Properties;

                    dispatch(consumerId, message, consumerOptional.getMaxReconsume()).equals(Action.commit)?
                            OrderAction.Success : OrderAction.Suspend
            ));
            orderConsumer.start();
        } else {
            Consumer consumer = ONSFactory.createConsumer(properties);
            consumer.subscribe(topic, tags, (message, context) ->
                    dispatch(consumerId, message, consumerOptional.getMaxReconsume()).equals(Action.commit)?
                            com.aliyun.openservices.ons.api.Action.CommitMessage : com.aliyun.openservices.ons.api.Action.ReconsumeLater
            );
            consumer.start();
       }
        logger.info("消费者启动成功: {}({}) , 消费模式: {} , 消费线程数: {} , 最大重试次数: {} {} , 订阅 TOPIC: {} TAG: {}",
                cid, consumerId.isOrdered()?"有序":"无序",
                consumerOptional.getConsumerModel(), consumerOptional.getConsumeThread(), consumerOptional.getMaxReconsume(),
                consumerId.isOrdered()?", 重试前的等待时间:"+consumerOptional.getSuspendTime()+"毫秒":"",topic,tags);
    }


    private static Action dispatch(ConsumerId consumerId, Message message, int defaultReconsume) {
        Tag tag = consumerId.getTagMap().get(message.getTag());
        if (null == tag) {
            throw new RuntimeException(String.format("%s 没有注册 请检查MqConsumer的init配置", message.getTag()));
        }
        int reconsume = tag.getReconsume() == null ? defaultReconsume : tag.getReconsume();
        if (message.getReconsumeTimes() > reconsume) {
            logger.warn("超过允许的最大重试次数。 允许的重试次数: {}, 当前重试次数: {} MessageId: {} topic: {} , tag: {}",
                    reconsume, message.getReconsumeTimes(),message.getMsgID(),message.getTopic(),message.getTag()) ;
            return Action.commit;
        }
        ArgumentExtractorWrapper[] extractors = tag.getArgumentExtractors();