update 提现和充值测试

src/main/java/com/m2pool/lease/mq/MessageReceiverListener.java

@@ -11,7 +11,7 @@ import com.m2pool.lease.exception.PayRechargeException;
 import com.m2pool.lease.mapper.*;
 import com.m2pool.lease.mq.message.*;
 import com.m2pool.lease.service.LeaseOrderItemService;
-import com.m2pool.lease.utils.UuidGeneratorUtil;
+import com.m2pool.lease.service.WalletCallbackService;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.messaging.handler.annotation.Payload;
@@ -237,33 +237,37 @@ public class MessageReceiverListener {
     //}
 
 
-
         /**
      * 余额充值 消费者 （正常流程一笔充值会有两个消息 2 充值中 1充值成功）
+     *
      * @param payRechargeReturnMessage
      */
     @LedgerLog(value = "余额充值")
-    @RabbitListener(queues = RabbitmqConstant.PAY_RECHARGE_RETURN_QUEUE,containerFactory ="rabbitListenerContainerFactory")
+    @RabbitListener(queues = RabbitmqConstant.PAY_RECHARGE_RETURN_QUEUE, containerFactory = "rabbitListenerContainerFactory")
     @Transactional(rollbackFor = Exception.class)
     public void listenerPayRechargeStatusQueueMessage(@Payload RabbitmqPayRechargeReturnMessage payRechargeReturnMessage) {
-        System.out.println("充值消费者---"+JSONUtil.toJsonPrettyStr(payRechargeReturnMessage));
+        System.out.println("充值消费者---" + JSONUtil.toJsonPrettyStr(payRechargeReturnMessage));
-        //获取到需要操作的钱包
+        if(payRechargeReturnMessage.getStatus() == 4){
+            // 触发钱包监听回调，表示监听成功
+            WalletCallbackService.triggerCallback(payRechargeReturnMessage.getQueue_id(), true);
+        }
+        // 获取到需要操作的钱包
         List<LeaseUserWalletData> leaseUserWalletDataList = leaseUserWalletDataMapper.selectList(new LambdaQueryWrapper<LeaseUserWalletData>()
                 .eq(LeaseUserWalletData::getFromAddress, payRechargeReturnMessage.getAddress())
                 .eq(LeaseUserWalletData::getDel, false)
         );
-        if (leaseUserWalletDataList.isEmpty()){
+        if (leaseUserWalletDataList.isEmpty()) {
             return;
         }
         LeaseUserWalletData leaseUserWalletData = null;
         for (LeaseUserWalletData item : leaseUserWalletDataList) {
-            if (item.getFromSymbol().equals(payRechargeReturnMessage.getSymbol())){
+            if (item.getFromSymbol().equals(payRechargeReturnMessage.getSymbol())) {
                 leaseUserWalletData = item;
                 break;
             }
         }
-        //需要新增该币种的钱包
+        // 需要新增该币种的钱包
-        if (leaseUserWalletData == null){
+        if (leaseUserWalletData == null) {
             LeaseUserWalletData leaseUserWalletData1 = leaseUserWalletDataList.get(0);
             leaseUserWalletData = LeaseUserWalletData.builder()
                     .queueId(payRechargeReturnMessage.getQueue_id())
@@ -280,12 +284,11 @@ public class MessageReceiverListener {
         LeasePayRechargeMessage leasePayRechargeMessage = leasePayRechargeMessageMapper.selectOne(new LambdaQueryWrapper<LeasePayRechargeMessage>()
                 .eq(LeasePayRechargeMessage::getTxHash, payRechargeReturnMessage.getTx_hash()));
         BigDecimal balance = leaseUserWalletData.getBalance();
-        //处理支付中消息
+        // 处理支付中消息
-        if(payRechargeReturnMessage.getStatus() == 2 || leasePayRechargeMessage == null){
+        if (payRechargeReturnMessage.getStatus() == 2 || leasePayRechargeMessage == null || leasePayRechargeMessage.getStatus() == 3) {
-            if (leasePayRechargeMessage != null){
+            if (leasePayRechargeMessage != null) {
                 return;
             }
-
             LeasePayRechargeMessage build = LeasePayRechargeMessage.builder()
                     .queueId(payRechargeReturnMessage.getQueue_id())
                     .chain(payRechargeReturnMessage.getChain())
@@ -298,24 +301,24 @@ public class MessageReceiverListener {
                     .status(payRechargeReturnMessage.getStatus())
                     .build();
             leasePayRechargeMessageMapper.saveOrUpdateByIndex(build);
-        }else {
+        } else {
-            //已存在已完成充值的消息
+            // 已存在已完成充值的消息
-            if (leasePayRechargeMessage.getStatus() == 1){
+            if (leasePayRechargeMessage.getStatus() == 1) {
                 return;
             }
             leasePayRechargeMessage.setStatus(payRechargeReturnMessage.getStatus());
             leasePayRechargeMessage.setUpdateTime(LocalDateTime.now());
             leasePayRechargeMessageMapper.updateById(leasePayRechargeMessage);
         }
-        //处理支付成功消息
+        // 处理支付成功消息
-        if (payRechargeReturnMessage.getStatus() == 1){
+        if (payRechargeReturnMessage.getStatus() == 1) {
             leaseUserWalletData.setBalance(balance.add(payRechargeReturnMessage.getAmount()));
-            //使用乐观锁防止余额少加
+            // 使用乐观锁防止余额少加
             int update = leaseUserWalletDataMapper.update(leaseUserWalletData, new LambdaQueryWrapper<LeaseUserWalletData>()
                     .eq(LeaseUserWalletData::getId, leaseUserWalletData.getId())
                     .eq(LeaseUserWalletData::getBalance, balance)
             );
-            if (update < 1){
+            if (update < 1) {
                 throw new PayRechargeException("余额充值失败,余额已被修改");
             }
         }
@@ -484,7 +487,7 @@ public class MessageReceiverListener {
         if (status == 1){
             leaseUserWalletData.setBalance(balance);
             leaseUserWalletData.setBlockedBalance(blockBalance);
-        }else if (status == 0 ){
+        }else if (status == 0 || status == 3){
             //余额提现失败 冻结金额 需要减去提现手续费和提现金额
             leaseUserWalletData.setBlockedBalance(blockBalance);
         }
@@ -554,59 +557,64 @@ public class MessageReceiverListener {
     //}
     //
     //测试 开发环境 充值测试
-    @RabbitListener(queues = RabbitmqConstant.PAY_RECHARGE_QUEUE,containerFactory ="rabbitListenerContainerFactory")
+    //@RabbitListener(queues = RabbitmqConstant.PAY_RECHARGE_QUEUE,containerFactory ="rabbitListenerContainerFactory")
-    public void listenerPayRechargeQueueMessage(@Payload RabbitmqPayRechargeMessage payAutoReturnMessage) {
+    //public void listenerPayRechargeQueueMessage(@Payload RabbitmqPayRechargeMessage payAutoReturnMessage,
-        String s = UuidGeneratorUtil.generateUuidWithoutHyphen();
+    //                                             com.rabbitmq.client.Channel channel,
-        ////发送充值消息
+    //                                             org.springframework.amqp.core.Message message) {
-        //RabbitmqPayRechargeReturnMessage rabbitmqPayRechargeReturnMessage = RabbitmqPayRechargeReturnMessage.builder()
+    //    try {
+    //        System.out.println("收到钱包监听请求: " + JSONUtil.toJsonPrettyStr(payAutoReturnMessage));
+    //
+    //        // 这里可以添加实际的监听逻辑
+    //        // 例如：调用外部 API 启动钱包监听
+    //
+    //        // 模拟监听成功
+    //        String response = "钱包监听启动成功: " + payAutoReturnMessage.getAddress();
+    //
+    //        // 发送响应给调用者
+    //        channel.basicPublish(
+    //                "",
+    //                message.getMessageProperties().getReplyTo(),
+    //                null,
+    //                response.getBytes()
+    //        );
+    //
+    //        // 手动确认消息
+    //        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+    //
+    //        System.out.println("钱包监听响应已发送: " + response);
+    //    } catch (Exception e) {
+    //        System.out.println("处理钱包监听请求失败: " + e.getMessage());
+    //        try {
+    //            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
+    //        } catch (Exception ex) {
+    //            ex.printStackTrace();
+    //        }
+    //    }
+    //}
+    //}
+    //
+    //
+    ////提现
+    //@RabbitListener(queues = RabbitmqConstant.PAY_WITHDRAW_QUEUE,containerFactory ="rabbitListenerContainerFactory")
+    //public void listenerWithdrawQueueMessage(@Payload RabbitmqPayWithdrawMessage payAutoReturnMessage) throws InterruptedException {
+    //    Thread.sleep(10000);
+    //    RabbitmqPayWithdrawReturnMessage rabbitmqPayRechargeReturnMessage = RabbitmqPayWithdrawReturnMessage.builder()
     //            .queue_id(payAutoReturnMessage.getQueue_id())
-        //        .status(2)
+    //            .status(1)
-        //        .amount(BigDecimal.valueOf(20))
+    //            .amount(payAutoReturnMessage.getAmount())
     //            .chain(payAutoReturnMessage.getChain())
     //            .symbol(payAutoReturnMessage.getSymbol())
-        //        .address(payAutoReturnMessage.getAddress())
+    //            .from_address(payAutoReturnMessage.getFrom_address())
-        //        .tx_hash(s)
+    //            .tx_hash(UuidGeneratorUtil.generateUuidWithoutHyphen())
+    //            .fee(payAutoReturnMessage.getFee())
+    //            .user_email(payAutoReturnMessage.getUser_email())
     //            .build();
-        //rabbitTemplate.convertAndSend(RabbitmqConstant.PAY_RECHARGE_RETURN_QUEUE,rabbitmqPayRechargeReturnMessage);
+    //    //提现100 提现失败
-
+    //    if (payAutoReturnMessage.getAmount().equals(BigDecimal.valueOf(100))){
-
+    //        rabbitmqPayRechargeReturnMessage.setStatus(0);
-        //发送充值消息
+    //    }
-        RabbitmqPayRechargeReturnMessage rabbitmqPayRechargeReturnMessage1 = RabbitmqPayRechargeReturnMessage.builder()
+    //    rabbitTemplate.convertAndSend(RabbitmqConstant.PAY_WITHDRAW_RETURN_QUEUE,rabbitmqPayRechargeReturnMessage);
-                .queue_id(payAutoReturnMessage.getQueue_id())
+    //}
-                .status(1)
-                .amount(BigDecimal.valueOf(100))
-                .chain(payAutoReturnMessage.getChain())
-                .symbol(payAutoReturnMessage.getSymbol())
-                .address(payAutoReturnMessage.getAddress())
-                .user_email(payAutoReturnMessage.getUser_email())
-                .fromAddress(payAutoReturnMessage.getAddress())
-                .tx_hash(s)
-                .build();
-        rabbitTemplate.convertAndSend(RabbitmqConstant.PAY_RECHARGE_RETURN_QUEUE,rabbitmqPayRechargeReturnMessage1);
-    }
-
-
-    //提现
-    @RabbitListener(queues = RabbitmqConstant.PAY_WITHDRAW_QUEUE,containerFactory ="rabbitListenerContainerFactory")
-    public void listenerWithdrawQueueMessage(@Payload RabbitmqPayWithdrawMessage payAutoReturnMessage) throws InterruptedException {
-        Thread.sleep(10000);
-        RabbitmqPayWithdrawReturnMessage rabbitmqPayRechargeReturnMessage = RabbitmqPayWithdrawReturnMessage.builder()
-                .queue_id(payAutoReturnMessage.getQueue_id())
-                .status(1)
-                .amount(payAutoReturnMessage.getAmount())
-                .chain(payAutoReturnMessage.getChain())
-                .symbol(payAutoReturnMessage.getSymbol())
-                .from_address(payAutoReturnMessage.getFrom_address())
-                .tx_hash(UuidGeneratorUtil.generateUuidWithoutHyphen())
-                .fee(payAutoReturnMessage.getFee())
-                .user_email(payAutoReturnMessage.getUser_email())
-                .build();
-        //提现100 提现失败
-        if (payAutoReturnMessage.getAmount().equals(BigDecimal.valueOf(100))){
-            rabbitmqPayRechargeReturnMessage.setStatus(0);
-        }
-        rabbitTemplate.convertAndSend(RabbitmqConstant.PAY_WITHDRAW_RETURN_QUEUE,rabbitmqPayRechargeReturnMessage);
-    }
 
 
     ////测试 开发环境 删除钱包测试

src/main/java/com/m2pool/lease/mq/message/RabbitmqPayWithdrawReturnMessage.java

@@ -46,9 +46,8 @@ public class RabbitmqPayWithdrawReturnMessage {
 
 
     /**
-     * 提现金额
+     * 提现金额 这个金额是减去手续费的
      */
-    @LedgerLogParam(value = "提现金额")
     private BigDecimal amount;
 
     /**

src/main/java/com/m2pool/lease/service/WalletCallbackService.java

@@ -0,0 +1,48 @@
+package com.m2pool.lease.service;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 钱包监听回调服务
+ * 用于存储和管理钱包监听的回调结果
+ */
+public class WalletCallbackService {
+
+    // 用于存储钱包监听回调结果，key: queueId, value: 回调结果
+    private static final Map<String, Boolean> walletListenCallbackMap = new ConcurrentHashMap<>();
+
+    /**
+     * 注册回调（清除之前的回调结果）
+     * @param queueId 队列ID
+     */
+    public static void registerCallback(String queueId) {
+        walletListenCallbackMap.remove(queueId);
+    }
+
+    /**
+     * 触发回调
+     * @param queueId 队列ID
+     * @param success 是否成功
+     */
+    public static void triggerCallback(String queueId, boolean success) {
+        walletListenCallbackMap.put(queueId, success);
+    }
+
+    /**
+     * 获取回调结果
+     * @param queueId 队列ID
+     * @return 回调结果，如果不存在则返回 null
+     */
+    public static Boolean getCallbackResult(String queueId) {
+        return walletListenCallbackMap.get(queueId);
+    }
+
+    /**
+     * 移除回调
+     * @param queueId 队列ID
+     */
+    public static void removeCallback(String queueId) {
+        walletListenCallbackMap.remove(queueId);
+    }
+}

src/main/java/com/m2pool/lease/service/impl/LeaseShopServiceImpl.java

@@ -727,6 +727,7 @@ public class LeaseShopServiceImpl extends ServiceImpl<LeaseShopMapper, LeaseShop
             try {
                 //发送提现消息
                 rabbitTemplate.convertAndSend(PAY_WITHDRAW_QUEUE,message);
+                OperationLogAspect.addExtraLogParam("提现金额", balanceVo.getAmount());
                 OperationLogAspect.addExtraLogParam("手续费", serviceCharge.toString());
                 OperationLogAspect.addExtraLogParam("消息ID", queueId);
                 return Result.success("提现申请成功");

src/main/java/com/m2pool/lease/service/impl/LeaseUserServiceImpl.java

@@ -1,6 +1,7 @@
 package com.m2pool.lease.service.impl;
 
 
+import cn.hutool.json.JSONUtil;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.nacos.common.utils.StringUtils;
 import com.baomidou.dynamic.datasource.annotation.DSTransactional;
@@ -19,12 +20,14 @@ import com.m2pool.lease.exception.AuthException;
 import com.m2pool.lease.exception.PaymentException;
 import com.m2pool.lease.mapper.*;
 import com.m2pool.lease.mq.message.RabbitmqPayRechargeMessage;
+import com.m2pool.lease.mq.message.RabbitmqPayRechargeReturnMessage;
 import com.m2pool.lease.mq.message.RabbitmqPayWithdrawMessage;
 import com.m2pool.lease.redis.EmailCodeValue;
 import com.m2pool.lease.redis.RedisAuthKey;
 import com.m2pool.lease.redis.service.RedisService;
 import com.m2pool.lease.service.LeaseShopService;
 import com.m2pool.lease.service.LeaseUserService;
+import com.m2pool.lease.service.WalletCallbackService;
 import com.m2pool.lease.utils.*;
 import com.m2pool.lease.utils.email.EmailTemplateEntity;
 import com.m2pool.lease.utils.text.Convert;
@@ -57,6 +60,7 @@ import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.time.LocalDateTime;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.zip.ZipEntry;
@@ -1040,19 +1044,18 @@ public class LeaseUserServiceImpl extends ServiceImpl<LeaseUserMapper, LeaseUser
             return Result.fail("请求服务器失败,请重试");
         }
     }
-
     @Override
     @Transactional
     public Result<UserWalletDataDto> bindWallet(ChainAndCoinVo chainAndCoinVo) {
         Long authId = SecurityUtils.getUserId();
         String username = SecurityUtils.getUsername();
         //同一用户同一个链只会存在一个钱包
-        List<UserWalletDataDto> userWalletDataDtoList = leaseUserWalletDataMapper.getWalletInfoByChain(chainAndCoinVo.getChain(),authId);
+        List<UserWalletDataDto> userWalletDataDtoList = leaseUserWalletDataMapper.getWalletInfoByChain(chainAndCoinVo.getChain(), authId);
         UserWalletDataDto result;
         //如果某个链 未查找到钱包地址 去新绑定一个
-        if (userWalletDataDtoList.isEmpty()){
+        if (userWalletDataDtoList.isEmpty()) {
             LeaseAutoAddress bindAddress = getBindAddress(chainAndCoinVo.getChain(), chainAndCoinVo.getCoin());
-            if (bindAddress != null){
+            if (bindAddress != null) {
                 String address = bindAddress.getAddress();
                 String qrcode = QrCodeUtils.creatRrCode(bindAddress.getAddress(), 200, 200);
                 String queueId = UuidGeneratorUtil.generateUuidWithoutHyphen();
@@ -1078,25 +1081,37 @@ public class LeaseUserServiceImpl extends ServiceImpl<LeaseUserMapper, LeaseUser
                         .build());
                 bindAddress.setStatus(1);
                 leaseAutoAddressMapper.updateById(bindAddress);
-                //监听钱包
+                // 清除之前的回调结果（如果有）
-                sendMessage(result,username,authId);
+                WalletCallbackService.registerCallback(queueId);
-
+                // 异步监听钱包，发送消息
-            }else{
+                sendMessage(result, username, authId);
+                // 阻塞等待回调，最多等待10秒
+                try {
+                    long startTime = System.currentTimeMillis();
+                    long timeout = 5000; // 5秒超时
+                    Boolean callbackResult = false;
+                    while (System.currentTimeMillis() - startTime < timeout) {
+                        callbackResult = WalletCallbackService.getCallbackResult(queueId);
+                        // 每隔100毫秒检查一次
+                        Thread.sleep(100);
+                    }
+                    // 超时未收到回调
+                    WalletCallbackService.removeCallback(queueId);
+                    if (callbackResult != null &&  callbackResult) {
+                        return Result.success(result, "钱包监听启动成功");
+                    } else {
+                        throw new AuthException("钱包监听启动失败，请稍后重试");
+                    }
+                } catch (InterruptedException e) {
+                    WalletCallbackService.removeCallback(queueId);
+                    return Result.success(result, "钱包监听启动中，请稍后查看");
+                }
+            } else {
                 return Result.fail("没有可用的地址,请联系管理员生成可用的钱包地址");
             }
-        }else{
+        } else {
             for (UserWalletDataDto userWalletDataDto : userWalletDataDtoList) {
-                if (userWalletDataDto.getFromSymbol().equals(chainAndCoinVo.getCoin())){
+                if (userWalletDataDto.getFromSymbol().equals(chainAndCoinVo.getCoin())) {
-                    //开发环境
-                    sendMessage(UserWalletDataDto.builder()
-                            .queueId(userWalletDataDto.getQueueId())
-                            .fromAddress(userWalletDataDto.getFromAddress())
-                            .fromChain(chainAndCoinVo.getChain())
-                            .fromSymbol(chainAndCoinVo.getCoin())
-                            .balance(BigDecimal.ZERO)
-                            .userId(userWalletDataDto.getUserId())
-                            .qrcode(userWalletDataDto.getQrcode())
-                            .build(),username,authId);
                     return Result.success(userWalletDataDto);
                 }
             }
@@ -1110,7 +1125,6 @@ public class LeaseUserServiceImpl extends ServiceImpl<LeaseUserMapper, LeaseUser
                     .userId(userWalletDataDto.getUserId())
                     .qrcode(userWalletDataDto.getQrcode())
                     .build();
-
             LeaseUserWalletData build = LeaseUserWalletData.builder()
                     .queueId(userWalletDataDto.getQueueId())
                     .fromChain(userWalletDataDto.getFromChain())
@@ -1124,7 +1138,6 @@ public class LeaseUserServiceImpl extends ServiceImpl<LeaseUserMapper, LeaseUser
             //链钱包存在 但是新增币种
             leaseUserWalletDataMapper.insert(build);
         }
-
         return Result.success(result);
     }
 
@@ -1176,7 +1189,7 @@ public class LeaseUserServiceImpl extends ServiceImpl<LeaseUserMapper, LeaseUser
     }
 
     /**
-     * 发送mq 消息，从而启动钱包监听
+     * 发送mq 消息，从而启动钱包监听（异步）
      */
     public void sendMessage(UserWalletDataDto userWalletDataDto,String  username,Long authId){
         try {
@@ -1291,8 +1304,10 @@ public class LeaseUserServiceImpl extends ServiceImpl<LeaseUserMapper, LeaseUser
                     .user_email(username+"--"+authId)
                     .build();
             try {
+
                 //发送 提现 消息
                 rabbitTemplate.convertAndSend(PAY_WITHDRAW_QUEUE,message);
+                OperationLogAspect.addExtraLogParam("提现金额", balanceVo.getAmount().toString());
                 OperationLogAspect.addExtraLogParam("手续费", serviceCharge.toString());
                 OperationLogAspect.addExtraLogParam("消息ID", queueId);
                 return Result.success("提现申请成功");