package cc.mrbird.febs.rabbit.consumer; import cc.mrbird.febs.common.utils.RedisUtils; import cc.mrbird.febs.dapp.chain.*; import cc.mrbird.febs.dapp.chain.entity.EthUsdtChargeDto; import cc.mrbird.febs.dapp.service.ApiIgtOnHookPlanService; import cc.mrbird.febs.job.BlockCoinUpdateJob; import cc.mrbird.febs.rabbit.RabbitMqConfig; import cn.hutool.core.util.StrUtil; import cn.hutool.http.HttpException; import com.alibaba.fastjson.JSONObject; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Component; import org.springframework.web.client.RestClientException; import javax.annotation.Resource; import java.math.BigDecimal; /** * @author wzy * @date 2020-05-25 **/ @Slf4j @Component @ConditionalOnProperty(prefix = "system", name = "block-job", havingValue = "true") public class UsdtUpdateConsumer { @Resource private BlockCoinService blockCoinService; @Resource private ApiIgtOnHookPlanService apiIgtOnHookPlanService; @Resource TrxUsdtUpdateService trxUsdtUpdateService; @Resource private RedisUtils redisUtils; @RabbitListener(queues = RabbitMqConfig.MEMBER_COIN_INSIDE_QUEUE) public void memberCoinInside(Long withdrawId) { log.info("处理用户内部转账的消息:{}#", withdrawId); apiIgtOnHookPlanService.memberCoinInsideMessage(withdrawId); } @RabbitListener(queues = RabbitMqConfig.END_ON_HOOK_QUEUE) public void endOnHook(Long orderId) { log.info("处理用户结束挂机的消息:{}#", orderId); apiIgtOnHookPlanService.endOnHookMessage(orderId); } @RabbitListener(queues = RabbitMqConfig.QUEUE_USDT_UPDATE) public void doSomething(String content) { log.info("#USDT同步---->{}#", content); EthUsdtChargeDto ethUsdtChargeDto = JSONObject.parseObject(content, EthUsdtChargeDto.class); if(EthUsdtChargeDto.Symbol.USDT_TRC20.equals(ethUsdtChargeDto.getSymbol())){ blockCoinService.updateTrc20(ethUsdtChargeDto); // 同步完直接归集 trxUsdtUpdateService.poolByAddress(ethUsdtChargeDto.getAddress()); } } @RabbitListener(queues = RabbitMqConfig.QUEUE_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("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 = RabbitMqConfig.QUEUE_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(); } } }