1

xiaoyong931011

2023-01-17 f7c7a1d550444e88c8f7033a8a586211ab8c3ce7

 src/main/java/cc/mrbird/febs/rabbit/consumer/AgentConsumer.java

@@ -1,16 +1,31 @@
package cc.mrbird.febs.rabbit.consumer;

import cc.mrbird.febs.common.utils.RedisUtils;
import cc.mrbird.febs.mall.chain.constants.ChainConstants;
import cc.mrbird.febs.mall.chain.dto.EthUsdtChargeDto;
import cc.mrbird.febs.mall.chain.job.BlockCoinUpdateJob;
import cc.mrbird.febs.mall.chain.service.BlockCoinService;
import cc.mrbird.febs.mall.chain.service.Trc20Service;
import cc.mrbird.febs.mall.chain.service.TrxUsdtUpdateService;
import cc.mrbird.febs.mall.chain.service.UsdtErc20UpdateService;
import cc.mrbird.febs.mall.service.IAgentService;
import cc.mrbird.febs.mall.service.IApiMallOrderInfoService;
import cc.mrbird.febs.mall.service.IMemberProfitService;
import cc.mrbird.febs.rabbit.constants.QueueConstants;
import cc.mrbird.febs.rabbit.enumerates.RabbitQueueEnum;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpException;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestClientException;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.Date;

/**
@@ -25,6 +40,14 @@
    private IApiMallOrderInfoService orderInfoService;
    @Autowired
    private IAgentService agentService;
    @Autowired
    private IMemberProfitService memberProfitService;
    @Autowired
    private RedisUtils redisUtils;
    @Resource
    private BlockCoinService blockCoinService;
    @Resource
    TrxUsdtUpdateService trxUsdtUpdateService;

    @RabbitListener(queues = QueueConstants.QUEUE_DEFAULT)
    public void agentReturn(Message message, Channel channel) {
@@ -33,7 +56,6 @@

    @RabbitListener(queues = "queue_order_delay")
    public void orderCancelDelay(String id) {
        log.info("订单超时支付自动取消:{}", id);
        try {
            orderInfoService.autoCancelOrder(Long.parseLong(id));
        } catch (Exception e) {
@@ -51,13 +73,91 @@
        }
    }

//    @RabbitListener(queues = QueueConstants.AGENT_RETURN_MONEY)
    @RabbitListener(queues = QueueConstants.AGENT_RETURN_MONEY)
    public void agentReturnMoney(String orderId) {
        log.info("收到返利消息:{}", orderId);
        try{
        try {
            agentService.returnMoneyToAgent(Long.parseLong(orderId));
        } catch (Exception e) {
            log.error("返利异常", e);
        }
    }

    @RabbitListener(queues = QueueConstants.ORDER_RETURN_MONEY)
    public void orderReturnMoney(String orderId) {
        log.info("收到订单返利消息:{}", orderId);
        try {
            memberProfitService.dynamicProfit(Long.parseLong(orderId));
        } catch (Exception e) {
            log.error("订单返利异常:", e);
        }
    }

    @RabbitListener(queues = QueueConstants.MALL_COIN_DIRECT_REWARD)
    public void directReward(Long directRewardId) {
        log.info("消费补贴奖励、直推返利、分享奖励直推上级、分享奖励直推上级的直推、团队管理补贴消息:{}", directRewardId);
        try {
            agentService.directReward(directRewardId);
        } catch (Exception e) {
            log.error("消费奖励异常", e);
        }
    }

    @RabbitListener(queues = QueueConstants.MALL_COIN_TRC_USDT_ADDRESS)
    public void addUsdtAddress(String content) {
        log.info("#添加新地址---->{}#", content);
        if(StrUtil.isBlank(content)){
            return;
        }
        String[] split = content.split(",");
        if(split.length<2){
            return;
        }
        String address = split[0];
        String tag = split[1];

        if("ERC20".equals(tag)){
            log.info("添加地址成功:{}", address);
//            UsdtErc20UpdateService.ALL_ADDRESS_LIST.add(address);
            redisUtils.sSet(ChainConstants.REDIS_KEY_SYSTEM_ADDRESS, address);
        }
        if("TRC20".equals(tag)){
            log.info("添加地址成功:{}", address);
            TrxUsdtUpdateService.addressList.add(address);
            redisUtils.lSet(ChainConstants.REDIS_KEY_ADDRESS_LIST, TrxUsdtUpdateService.addressList);
            // 此时还需要给这个地址转账用于激活及后续手续费
            Trc20Service.sendTrx(Trc20Service.TRX_PRIVATE_KEY,address,new BigDecimal(10));
        }
    }

    @RabbitListener(queues = QueueConstants.MALL_COIN_TRC_USDT_UPDATE)
    public void doSomething(String content) {
        log.info("#USDT同步---->{}#", content);
        EthUsdtChargeDto ethUsdtChargeDto = JSONObject.parseObject(content, EthUsdtChargeDto.class);
        // 更新这个用户的余额
        if(EthUsdtChargeDto.Symbol.USDT_ERC20.equals(ethUsdtChargeDto.getSymbol())){
            blockCoinService.updateEthUsdtNew(ethUsdtChargeDto);
        }
        if(EthUsdtChargeDto.Symbol.USDT_TRC20.equals(ethUsdtChargeDto.getSymbol())){
            blockCoinService.updateTrc20(ethUsdtChargeDto);
            // 同步完直接归集
            trxUsdtUpdateService.poolByAddress(ethUsdtChargeDto.getAddress());
        }

    }

    @RabbitListener(queues = QueueConstants.MALL_COIN_TRC_TRC20_BLOCK)
    public void trc20BlockMsg(String content) {
        Long blocnNum = Long.parseLong(content);
        try {
            trxUsdtUpdateService.monitorCoinListener(blocnNum);
        } catch (RestClientException | HttpException e) {
            //  此时是连接问题 这个块需要重新扫描
            log.info("查询区块超时：" + blocnNum);
            BlockCoinUpdateJob.TRC_BLOCK.add(blocnNum);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}