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; /** * @author wzy * @date 2021-09-25 **/ @Slf4j @Component public class AgentConsumer { @Autowired 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) { log.info("消费者:{}", new String(message.getBody())); } @RabbitListener(queues = "queue_order_delay") public void orderCancelDelay(String id) { try { orderInfoService.autoCancelOrder(Long.parseLong(id)); } catch (Exception e) { log.error("订单超时支付异常", e); } } @RabbitListener(queues = QueueConstants.AGENT_AUTO_LEVEL_UP) public void agentAutoLevelUp(String id) { log.info("收到代理自动升级消息:{}", id); try { agentService.autoUpAgentLevel(Long.parseLong(id)); } catch (Exception e) { log.error("代理自动升级异常", e); } } @RabbitListener(queues = QueueConstants.AGENT_RETURN_MONEY) public void agentReturnMoney(String orderId) { log.info("收到返利消息:{}", orderId); 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(); } } }