提交支付查询

liquidnet-bus-api/liquidnet-service-dragon-api/src/main/java/com/liquidnet/service/dragon/constant/DragonConstant.java

@@ -222,11 +222,14 @@ public class DragonConstant {
         DRAGON_REFUND_KEY ("dragon:stream:dragon-refund","同步数据-退款"),
         DRAGON_PAY_GROUP("dragon-pay-group","同步数据-支付-组"),
         DRAGON_REFUND_GROUP ("dragon-refund-group","同步数据-退款-组"),
-        //商户通知相关队列
-        DRAGON_PAY_NOTIFY_KEY("dragon:stream:dragon-pay-notify-key","支付异步通知"),
+        //三方异步通知相关队列
+        DRAGON_PAY_NOTIFY_KEY("dragon-pay-notify-key","支付异步通知"),
         DRAGON_PAY_NOTIFY_GROUP ("dragon-pay-notify-group","支付异步通知-组"),
-        PAY_MCH_NOTIFY_ERROR_KEY("dragon:stream:pay-mch-notify-error-key","商户异步通知失败"),
-        PAY_MCH_NOTIFY_ERROR_GROUP ("pay-mch-notify-error-group","商户异步通知失败");
+        //商户通知相关队列
+        PAY_MCH_NOTIFY_KEY("pay-mch-notify-key","商户异步通知"),
+        PAY_MCH_NOTIFY_GROUP ("pay-mch-notify-group","商户异步通知-组"),
+        PAY_MCH_NOTIFY_ERROR_KEY("pay-mch-notify-error-key","商户异步通知失败"),
+        PAY_MCH_NOTIFY_ERROR_GROUP ("pay-mch-notify-error-group","商户异步通知失败-组");
         private String code;
         private String message;
         MysqlRedisQueueEnum(String code, String message) {

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-dragon/src/main/java/com/liquidnet/service/consumer/dragon/service/config/RedisStreamConfig.java

 package com.liquidnet.service.consumer.dragon.service.config;
 
-import com.liquidnet.service.consumer.dragon.service.receiver.RedisPayReceiver;
-import com.liquidnet.service.consumer.dragon.service.receiver.RedisRefundReceiver;
+import com.liquidnet.service.consumer.dragon.service.receiver.*;
 import com.liquidnet.service.dragon.constant.DragonConstant;
 import lombok.var;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -23,6 +22,12 @@ public class RedisStreamConfig {
     private RedisPayReceiver redisPayReceiver;
     @Autowired
     private RedisRefundReceiver redisRefundReceiver;
+    @Autowired
+    private RedisPayNotifyReceiver redisPayNotifyReceiver;
+    @Autowired
+    private RedisMchNotifyReceiver redisMchNotifyReceiver;
+    @Autowired
+    private RedisMchNotifyFailReceiver redisMchNotifyFailReceiver;
 
     @Bean
     public Subscription subscriptionPay0(RedisConnectionFactory factory) {
@@ -79,4 +84,86 @@ public class RedisStreamConfig {
         listenerContainer.start();
         return subscription;
     }
+
+    @Bean
+    public Subscription subscriptionPayNotify0(RedisConnectionFactory factory) {
+        var options = StreamMessageListenerContainer
+                .StreamMessageListenerContainerOptions
+                .builder()
+                .pollTimeout(Duration.ofMillis(1))
+                .build();
+        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
+        var subscription = listenerContainer.receiveAutoAck(Consumer.from(DragonConstant.MysqlRedisQueueEnum.DRAGON_PAY_NOTIFY_GROUP.getCode(), "dragon-pay-notify-0"),
+                StreamOffset.create(DragonConstant.MysqlRedisQueueEnum.DRAGON_PAY_NOTIFY_KEY.getCode(), ReadOffset.lastConsumed()), redisPayNotifyReceiver);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionPayNotify1(RedisConnectionFactory factory) {
+        var options = StreamMessageListenerContainer
+                .StreamMessageListenerContainerOptions
+                .builder()
+                .pollTimeout(Duration.ofMillis(1))
+                .build();
+        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
+        var subscription = listenerContainer.receiveAutoAck(Consumer.from(DragonConstant.MysqlRedisQueueEnum.DRAGON_PAY_NOTIFY_GROUP.getCode(), "dragon-pay-notify-1"),
+                StreamOffset.create(DragonConstant.MysqlRedisQueueEnum.DRAGON_PAY_NOTIFY_KEY.getCode(), ReadOffset.lastConsumed()), redisPayNotifyReceiver);
+        listenerContainer.start();
+        return subscription;
+    }
+    @Bean
+    public Subscription subscriptionMchNotify0(RedisConnectionFactory factory) {
+        var options = StreamMessageListenerContainer
+                .StreamMessageListenerContainerOptions
+                .builder()
+                .pollTimeout(Duration.ofMillis(1))
+                .build();
+        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
+        var subscription = listenerContainer.receiveAutoAck(Consumer.from(DragonConstant.MysqlRedisQueueEnum.PAY_MCH_NOTIFY_GROUP.getCode(), "pay-mch-notify-0"),
+                StreamOffset.create(DragonConstant.MysqlRedisQueueEnum.PAY_MCH_NOTIFY_KEY.getCode(), ReadOffset.lastConsumed()), redisMchNotifyReceiver);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionMchNotify1(RedisConnectionFactory factory) {
+        var options = StreamMessageListenerContainer
+                .StreamMessageListenerContainerOptions
+                .builder()
+                .pollTimeout(Duration.ofMillis(1))
+                .build();
+        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
+        var subscription = listenerContainer.receiveAutoAck(Consumer.from(DragonConstant.MysqlRedisQueueEnum.PAY_MCH_NOTIFY_GROUP.getCode(), "pay-mch-notify-1"),
+                StreamOffset.create(DragonConstant.MysqlRedisQueueEnum.PAY_MCH_NOTIFY_KEY.getCode(), ReadOffset.lastConsumed()), redisMchNotifyReceiver);
+        listenerContainer.start();
+        return subscription;
+    }
+    @Bean
+    public Subscription subscriptionPayNotifyFail0(RedisConnectionFactory factory) {
+        var options = StreamMessageListenerContainer
+                .StreamMessageListenerContainerOptions
+                .builder()
+                .pollTimeout(Duration.ofMillis(1))
+                .build();
+        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
+        var subscription = listenerContainer.receiveAutoAck(Consumer.from(DragonConstant.MysqlRedisQueueEnum.DRAGON_PAY_NOTIFY_GROUP.getCode(), "pay-mch-notify-fail-0"),
+                StreamOffset.create(DragonConstant.MysqlRedisQueueEnum.DRAGON_PAY_NOTIFY_KEY.getCode(), ReadOffset.lastConsumed()), redisPayNotifyReceiver);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionPayNotifyFail1(RedisConnectionFactory factory) {
+        var options = StreamMessageListenerContainer
+                .StreamMessageListenerContainerOptions
+                .builder()
+                .pollTimeout(Duration.ofMillis(1))
+                .build();
+        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
+        var subscription = listenerContainer.receiveAutoAck(Consumer.from(DragonConstant.MysqlRedisQueueEnum.DRAGON_PAY_NOTIFY_GROUP.getCode(), "pay-mch-notify-fail-1"),
+                StreamOffset.create(DragonConstant.MysqlRedisQueueEnum.DRAGON_PAY_NOTIFY_KEY.getCode(), ReadOffset.lastConsumed()), redisPayNotifyReceiver);
+        listenerContainer.start();
+        return subscription;
+    }
 }

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-dragon/src/main/java/com/liquidnet/service/consumer/dragon/service/receiver/RedisMchNotifyFailReceiver.java

+package com.liquidnet.service.consumer.dragon.service.receiver;
+
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.SqlMapping;
+import com.liquidnet.service.consumer.dragon.service.IBaseDao;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * 支付商户通知消费
+ */
+@Slf4j
+@Component
+public class RedisMchNotifyFailReceiver implements StreamListener<String, MapRecord<String, String, String>> {
+
+    @Resource
+    IBaseDao baseDao;
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Override
+    public void onMessage(MapRecord<String, String, String> message) {
+        log.info("接受到来自redis pay notify 的消息");
+        System.out.println("message id " + message.getId());
+        System.out.println("stream " + message.getStream());
+        System.out.println("body " + message.getValue());
+        consumerSqlDaoHandler(message.getValue().get("message"));
+    }
+
+    private void consumerSqlDaoHandler(String msg) {
+        try {
+            SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
+            log.debug("CONSUMER SQL ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
+            log.debug("CONSUMER SQL ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
+            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls);
+            if (rstBatchSqls) {
+                //应答
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-dragon/src/main/java/com/liquidnet/service/consumer/dragon/service/receiver/RedisMchNotifyReceiver.java

+package com.liquidnet.service.consumer.dragon.service.receiver;
+
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.SqlMapping;
+import com.liquidnet.service.consumer.dragon.service.IBaseDao;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * 支付商户通知消费
+ */
+@Slf4j
+@Component
+public class RedisMchNotifyReceiver implements StreamListener<String, MapRecord<String, String, String>> {
+
+    @Resource
+    IBaseDao baseDao;
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Override
+    public void onMessage(MapRecord<String, String, String> message) {
+        log.info("接受到来自redis pay notify 的消息");
+        System.out.println("message id " + message.getId());
+        System.out.println("stream " + message.getStream());
+        System.out.println("body " + message.getValue());
+        consumerSqlDaoHandler(message.getValue().get("message"));
+    }
+
+    private void consumerSqlDaoHandler(String msg) {
+        try {
+            SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
+            log.debug("CONSUMER SQL ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
+            log.debug("CONSUMER SQL ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
+            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls);
+            if (rstBatchSqls) {
+                //应答
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-dragon/src/main/java/com/liquidnet/service/consumer/dragon/service/receiver/RedisPayNotifyReceiver.java

+package com.liquidnet.service.consumer.dragon.service.receiver;
+
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.SqlMapping;
+import com.liquidnet.service.consumer.dragon.service.IBaseDao;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * 支付商户通知消费
+ */
+@Slf4j
+@Component
+public class RedisPayNotifyReceiver implements StreamListener<String, MapRecord<String, String, String>> {
+
+    @Resource
+    IBaseDao baseDao;
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Override
+    public void onMessage(MapRecord<String, String, String> message) {
+        log.info("接受到来自redis pay notify 的消息");
+        System.out.println("message id " + message.getId());
+        System.out.println("stream " + message.getStream());
+        System.out.println("body " + message.getValue());
+        consumerSqlDaoHandler(message.getValue().get("message"));
+    }
+
+    private void consumerSqlDaoHandler(String msg) {
+        try {
+            SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
+            log.debug("CONSUMER SQL ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
+            log.debug("CONSUMER SQL ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
+            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls);
+            if (rstBatchSqls) {
+                //应答
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}

liquidnet-bus-service/liquidnet-service-dragon/liquidnet-service-dragon-impl/src/main/java/com/liquidnet/service/dragon/channel/strategy/biz/DragonPayBiz.java

 package com.liquidnet.service.dragon.channel.strategy.biz;
 
+import com.alibaba.fastjson.JSON;
 import com.liquidnet.commons.lang.util.BeanUtil;
 import com.liquidnet.commons.lang.util.HttpUtil;
 import com.liquidnet.commons.lang.util.IDGenerator;
@@ -92,6 +93,11 @@ public abstract class DragonPayBiz {
         return payNotifyReqBo;
     }
 
+    /**
+     * 三方异步通知入库
+     * @param paymentType
+     * @param content
+     */
     public void createDragonOrderLogs(String paymentType,String content){
         try {
             String orderId = IDGenerator.nextTimeId();
@@ -109,6 +115,60 @@ public abstract class DragonPayBiz {
         }
     }
 
+    /**
+     * 商户异步通知入库
+     * @param payNotifyReqBo
+     * @param notifyParam
+     */
+    public void createDragonPayNotify(PayNotifyReqBo payNotifyReqBo,String notifyParam){
+        PayNotifyDto payNotifyDto = payNotifyReqBo.getPayNotifyDto();
+        try {
+            String code = payNotifyDto.getCode();
+            String orderCode = payNotifyDto.getOrderCode();
+            String notifyUrl = payNotifyReqBo.getNotifyUrl();
+            String notifyData = notifyParam;
+            LocalDateTime createAt = LocalDateTime.now();
+            LocalDateTime updateAt = LocalDateTime.now();
+            boolean insertResult = mqHandleUtil.sendMySqlRedis(
+                    SqlMapping.get("dragon_pay_notify.insert"),
+                    new Object[]{code,orderCode,notifyUrl,notifyData, createAt, updateAt}
+                    ,DragonConstant.MysqlRedisQueueEnum.PAY_MCH_NOTIFY_KEY.getCode()
+            );
+            log.info("dragon:createDragonPayNotify:success code:{}",code);
+        } catch (Exception e) {
+            log.error("dragon:createDragonPayNotify:error msg:{}",e);
+            e.printStackTrace();
+        }
+    }
+
+    /**
+     * 商户异步通知失败入库
+     * @param payNotifyReqBo
+     * @param notifyParam
+     */
+    public void createDragonPayNotifyFail(PayNotifyReqBo payNotifyReqBo,String notifyParam){
+        PayNotifyDto payNotifyDto = payNotifyReqBo.getPayNotifyDto();
+        try {
+            String code = payNotifyDto.getCode();
+            String orderCode = payNotifyDto.getOrderCode();
+            String notifyUrl = payNotifyReqBo.getNotifyUrl();
+            String notifyData = notifyParam;
+            String failDesc = "通知失败";
+            LocalDateTime createAt = LocalDateTime.now();
+            LocalDateTime updateAt = LocalDateTime.now();
+            boolean insertResult = mqHandleUtil.sendMySqlRedis(
+                    SqlMapping.get("dragon_pay_notify_fail.insert"),
+                    new Object[]{code,orderCode,notifyUrl,notifyData,failDesc,createAt, updateAt}
+                    ,DragonConstant.MysqlRedisQueueEnum.PAY_MCH_NOTIFY_ERROR_KEY.getCode()
+            );
+            log.info("dragon:createDragonPayNotifyFail:success code:{}",code);
+        } catch (Exception e) {
+            log.error("dragon:createDragonPayNotifyFail:error msg:{}",e);
+            e.printStackTrace();
+        }
+    }
+
+
     public void sendNotify(PayNotifyReqBo payNotifyReqBo){
         PayNotifyDto payNotifyDto = payNotifyReqBo.getPayNotifyDto();
         LocalDateTime nowTime = LocalDateTime.now();
@@ -120,15 +180,15 @@ public abstract class DragonPayBiz {
         params.add("orderCode", payNotifyDto.getOrderCode());
         params.add("price", payNotifyDto.getPrice().toString());
         params.add("paymentType", payNotifyDto.getPaymentType());
+        String jsonData = JSON.toJSONString(params);
+        log.info("dragon:notify:post url:{}",payNotifyReqBo.getNotifyUrl());
+        log.info("dragon:notify:post data:{}",jsonData);
         String response = HttpUtil.post(payNotifyReqBo.getNotifyUrl(), params);
         log.debug("PAY RESPONSE=" + response);
         if (response.equals("success")) {
-
+            this.createDragonPayNotify(payNotifyReqBo,jsonData);
         } else {
-//            sendMySqlRedis(
-//                    SqlMapping.get("dragon_order_pay.update"),
-//                    new Object[]{nowTime, DateUtil.Formatter.yyyyMMddHHmmss.format(nowTime), DragonConstant.RefundStatusEnum.STATUS_FAIL.getCode(), notifyUrlDto.getRefundCode()}
-//            );
+            this.createDragonPayNotifyFail(payNotifyReqBo,jsonData);
         }
     }
 

liquidnet-bus-service/liquidnet-service-dragon/liquidnet-service-dragon-impl/src/main/java/com/liquidnet/service/dragon/channel/strategy/impl/PayChannelStrategyWepayImpl.java

@@ -71,6 +71,8 @@ public class PayChannelStrategyWepayImpl implements IPayChannelStrategy {
                 notifyMap = PayWepayUtils.parseXml(inputStream);
                 log.info("dragonNotify-->wepay json : {}", JSON.toJSONString(notifyMap));
 
+                //持久化通知记录
+                wepayBiz.createDragonOrderLogs(wepayBiz.getPaymentType(payType,deviceFrom),JSON.toJSONString(notifyMap));
                 log.info("接收到{}支付结果{}", payType, notifyMap);
 
                 String code = notifyMap.get("out_trade_no");

liquidnet-bus-service/liquidnet-service-dragon/liquidnet-service-dragon-impl/src/main/resources/sqlmap.properties

@@ -9,6 +9,7 @@ dragon_order_refund_log.insert=INSERT INTO `dragon_order_refund_logs`(`order_ref
 
 
 dragon_orders.insert=insert into dragon_orders(order_id, status, code, type, price, name, detail, order_code, client_ip, notify_url, payment_type, payment_id, payment_at, finished_at, created_at, updated_at, deleted_at) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
-
 dragon_order_logs.insert=insert into dragon_order_logs(order_id, payment_type, content, created_at, updated_at, deleted_at) values(?,?,?,?,?,?)
+dragon_pay_notify.insert=insert into dragon_pay_notify (code, order_code, notify_url, notify_data, created_at, updated_at) values(?,?,?,?,?,?)
+dragon_pay_notify_fail.insert=insert into dragon_pay_notify_fail (code, order_code, notify_url, notify_data, fail_desc, created_at, updated_at) values(?,?,?,?,?,?,?)