xy-device/device-api-service/src/main/java/com/xy/consumer/data/DataConsumer.java

package com.xy.consumer.data;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.xy.annotate.MqttConsumerAsyn;
import com.xy.config.DeviceThreadPoolConfig;
import com.xy.consumer.MqttConsumer;
import com.xy.dto.DataMqttVo;
import com.xy.dto.DeviceDataDto;
import com.xy.service.DeviceDataServiceImpl;
import com.xy.service.DeviceStatusServiceImpl;
import com.xy.utils.Emptys;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * <p>
 * 数据统计消费者
 * </p>
 *
 * @author lijin
 * @since 2022-03-11
 */
@Slf4j
@Service
@AllArgsConstructor
@MqttConsumerAsyn(DeviceThreadPoolConfig.DEVICE_DATA_POLL)
public class DataConsumer implements MqttConsumer {

    private DeviceDataServiceImpl deviceDataService;

    private DeviceStatusServiceImpl deviceStatusService;

    @Override
    public boolean message(String topic, String payload) {
        JSONObject jsonObject = JSONUtil.parseObj(payload);
        DataMqttVo dataMqttVo = jsonObject.toBean(DataMqttVo.class);
        Boolean type = dataMqttVo.getType();
        List<DataMqttVo.Goods> goods = dataMqttVo.getGoods();
        Integer money = dataMqttVo.getMoney();
        int number = 0;
        if (Emptys.check(goods)) {
            for (DataMqttVo.Goods good : goods) {
                number = number + good.getNumber();
            }
        }
        //设备统计
        DeviceDataDto.SaveOrAccum saveOrAccum = new DeviceDataDto.SaveOrAccum()
                .setDeviceId(dataMqttVo.getDeviceId());
        saveOrAccum.setMercId(dataMqttVo.getMercId());
        if (type) {
            //购买
            if (dataMqttVo.getIsError()) {
                //异常单
                saveOrAccum.setRiskCount(1);
            } else {
                if (money == null || money == 0) {
                    //0元单
                    saveOrAccum.setZeroCount(1);
                } else {
                    //正常单，累计商品数量
                    saveOrAccum.setSalesMoney(money);
                    saveOrAccum.setSalesCount(1);
                    saveOrAccum.setGoodsCount(number);
                }
            }
        } else {
            //退款
            saveOrAccum.setRefundMoney(money);
            saveOrAccum.setRefundCount(1);
        }
        deviceDataService.saveOrAccum(saveOrAccum);
        //修改库存
        if (number > 0) {
            deviceStatusService.upStock(dataMqttVo.getDeviceId(), type ? (~(number - 1)) : number);
        }
        return true;
    }
}