package com.bailuntec.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.annotation.JSONField;
import com.bailuntec.domain.entity.DcBaseQueue;
import com.bailuntec.domain.entity.DcBaseStock;
import com.bailuntec.domain.example.DcBaseStockExample;
import com.bailuntec.mapper.DcBaseQueueMapper;
import com.bailuntec.mapper.DcBaseStockMapper;
import com.bailuntec.utils.SessionUtil;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.SqlSession;

import java.util.Arrays;
import java.util.List;

/**
 * <p>
 *
 * </p>
 *
 * @author robbendev
 * @since 2020/11/24 4:10 下午
 */
@Slf4j
public class QueueConsumerJob implements SimpleJob {


    @Data
    public static class BaseQueueMessage {
        @JSONField(name = "bailun_sku")
        private String bailunSku;
        @JSONField(name = "warehouse_code")
        private String warehouseCode;
    }

    @Override
    public void execute(ShardingContext shardingContext) {

        try (SqlSession sqlSession = SessionUtil.getFactory().openSession(true)) {
            AutoTurnoverJob autoTurnoverJob = new AutoTurnoverJob();
            DcBaseQueueMapper dcBaseQueueMapper = sqlSession.getMapper(DcBaseQueueMapper.class);
            DcBaseStockMapper dcBaseStockMapper = sqlSession.getMapper(DcBaseStockMapper.class);

            List<DcBaseQueue> dcBaseQueueList = dcBaseQueueMapper.poll(shardingContext);

            dcBaseQueueList.parallelStream().forEach(dcBaseQueue -> {
                BaseQueueMessage baseQueueMessage = JSON.parseObject(dcBaseQueue.getMessage(), BaseQueueMessage.class);

                DcBaseStock dcBaseStock = dcBaseStockMapper.selectOneByExample(DcBaseStockExample.newAndCreateCriteria()
                        .andBailunSkuEqualTo(baseQueueMessage.getBailunSku())
                        .andWarehouseCodeEqualTo(baseQueueMessage.getWarehouseCode())
                        .example());
                try {

                    autoTurnoverJob.autoTurnoverFromStock(dcBaseStock);
                    dcBaseQueueMapper.deleteByPrimaryKey(dcBaseQueue.getId());
                } catch (Exception e) {
                    try {
                        dcBaseQueue.setQueueType(2);
                        dcBaseQueue.setErrorMessage(e.getMessage());
                        dcBaseQueue.setErrorStackTrace(Arrays.toString(e.getStackTrace()));
                        dcBaseQueueMapper.deleteByPrimaryKey(dcBaseQueue.getId());
                    } catch (Exception ex) {
                        log.error(ex.getMessage());
                    }
                } finally {
                    log.info("消费一条数据 message:{}", dcBaseQueue.getMessage());
                }

            });
        }
    }
}
