package com.doumee.config.rocketmq5;


import com.alibaba.fastjson.JSONObject;
import com.doumee.core.utils.Constants;
import com.doumee.dao.business.model.Activity;
import com.doumee.service.business.ActivityService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collections;
import java.util.List;

@Component
@Slf4j
public class NormalMsgConsumer {
    @Autowired
    public  MqConfig mqConfig;

    @Autowired
    private ActivityService activityService;

    private  SimpleConsumer consumer ;

    @Async("asyncPoolTaskExecutor")
    public void buildConsumer()  {
            final ClientServiceProvider provider = ClientServiceProvider.loadService();

            // Credential provider is optional for client configuration.
            StaticSessionCredentialsProvider staticSessionCredentialsProvider =
                    new StaticSessionCredentialsProvider(mqConfig.getUsername(), mqConfig.getPassword());
            String endpoints = mqConfig.getNameSrvAddr();
            ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                    .setEndpoints(endpoints)
                    .setCredentialProvider(staticSessionCredentialsProvider)
                    .build();
            String consumerGroup = mqConfig.getGroupId();
            Duration awaitDuration = Duration.ofSeconds(30);
            String topic = mqConfig.getTopic();
            String tag = "";
            FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
            // In most case, you don't need to create too many consumers, singleton pattern is recommended.
            do {
                try {
                    log.info("start rocketmq connect.....");
                    consumer = provider.newSimpleConsumerBuilder()
                            .setClientConfiguration(clientConfiguration)
                            // Set the consumer group name.
                            .setConsumerGroup(consumerGroup)
                            // set await duration for long-polling.
                            .setAwaitDuration(awaitDuration)
                            // Set the subscription for the consumer.
                            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                            .build();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }while (consumer==null);
            try {
                startService();
            }catch (Exception e){
                e.printStackTrace();
            }
    }


    public  void startService() throws  Exception {
        if(consumer==null){
            return;
        }
        // Max message num for each long polling.
        int maxMessageNum = 16;
        // Set message invisible duration after it is received.
        Duration invisibleDuration = Duration.ofSeconds(15);
        // Receive message, multi-threading is more recommended.
        do {
            try {
                final List<MessageView> messages = consumer.receive(maxMessageNum, invisibleDuration);
                log.info("Received {} message(s)", messages.size());
                for (MessageView message : messages) {
                    final MessageId messageId = message.getMessageId();
                    try {
                        if(doBusiness(message)){
                            consumer.ack(message);
                            log.info("Message is acknowledged successfully, messageId={}", messageId);
                        }else{
                            log.error("Message messageId={}=======业务处理失败", messageId);
                        }
                    } catch (Throwable t) {
                        log.error("Message is failed to be acknowledged, messageId={}", messageId, t);
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        } while (true);
        // Close the simple consumer when you don't need it anymore.
        // consumer.close();
    }

    /**
     * 业务处理
     * @param message
     */
    private boolean doBusiness(MessageView message) {
        try {
            String tag =(String) message.getTag().get();
            String body = StandardCharsets.UTF_8.decode(message.getBody()).toString();
            if(StringUtils.equals(tag, Constants.MQ_TAG.activity)){
                Activity model = JSONObject.toJavaObject(JSONObject.parseObject(body),Activity.class);
                if(model != null){
                    activityService.updateLooknum(model);
                }
            }
            return true;
        }catch (Exception e){
            return false;
        }

    }

}