Merge remote-tracking branch 'origin/dev_nft_411' into dev_nft_411

liquidnet-bus-common/liquidnet-common-cache/liquidnet-common-cache-redis/src/main/java/com.liquidnet.common.cache/redis/config/RedisStreamConfig.java

@@ -33,6 +33,7 @@ public class RedisStreamConfig {
         try {
             if (stringRedisTemplate.hasKey(key)) {
                 log.info("redis stream exist[{},{}]", key, group);
+                stringRedisTemplate.opsForStream().createGroup(key, group);
             } else {
                 log.info("redis stream init[{},{}]", key, group);
                 HashMap<String, String> map = new HashMap<>();
@@ -43,7 +44,7 @@ public class RedisStreamConfig {
                 stringObjectObjectStreamOperations.delete(key, recordId.getValue());
             }
         } catch (Exception e) {
-            log.error("Ex:redis stream init [{},{}]", key, group, e);
+            log.error("Ex:redis stream init [{},{}],ex:{}", key, group, e.getMessage());
         }
     }
 

liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/impl/AdamUserBusiAcctServiceImpl.java

@@ -45,6 +45,8 @@ public class AdamUserBusiAcctServiceImpl implements IAdamUserBusiAcctService {
 //    @Value("${liquidnet.service.galaxy.url}")// TODO: 2022/3/29 ==zhanggb
 //    private String sevGalaxyUrl;
 
+    private static final String BIZ_ACC_OPEN_NFT_ZXIN_BODY = "{\"routerType\":\"zxinchain\",\"idCardType\":\"1\",\"idCard\":\"P3\",\"userName\":\"P4\",\"mobile\":\"P5\",\"userId\":\"P6\"}";
+
     @Override
     public ResponseDto<AdamUserBizAcctVo> openAccount(String uid, String bizCode, String name, String idCard, String mobile) {
         List<AdamUserBizAcctVo> userBizAcctVoList = adamRdmService.getBizAcctVosByUid(uid);
@@ -101,29 +103,20 @@ public class AdamUserBusiAcctServiceImpl implements IAdamUserBusiAcctService {
             }
         }
 
-        LinkedMultiValueMap<String, String> paramsMap = CollectionUtil.linkedMultiValueMapStringString();
-        paramsMap.add("routerType", "zxinchain");// TODO: 2022/3/29 ==zhanggb
-        paramsMap.add("userId", uid);
-        paramsMap.add("idCardType", "1");
-        paramsMap.add("idCard", idCard);
-        paramsMap.add("userName", name);
-        paramsMap.add("mobile", mobile);
-
+        String postBody = BIZ_ACC_OPEN_NFT_ZXIN_BODY.replace("P3", idCard).replace("P4", name).replace("P5", mobile).replace("P6", uid);
 //        String postUrl = sevGalaxyUrl + "/user/register", blockChainAddress;// TODO: 2022/3/29 ==zhanggb
         String postUrl = "https://ENVgalaxy.zhengzai.tv/galaxy/user/register".replace("ENV", env.getProperty(CurrentUtil.CK_ENV_ACTIVE)), blockChainAddress;
         try {
-            LinkedMultiValueMap<String, String> headerMap = CollectionUtil.linkedMultiValueMapStringString();
-            headerMap.add("Accept", MediaType.APPLICATION_JSON_VALUE);
-            String postRespStr = HttpUtil.post(postUrl, paramsMap, headerMap);
+            String postRespStr = HttpUtil.postRaw(postUrl, postBody, null);
             JsonNode postRespJNode = JsonUtils.fromJson(postRespStr, JsonNode.class), postRespCode;
             if (null == postRespJNode || null == (postRespCode = postRespJNode.get("code")) || !postRespCode.asText().equals("0")) {
-                log.warn("#开通业务账号:至信数字藏品账号开通失败[paramsMap={},postRespStr={}]", paramsMap, postRespStr);
+                log.warn("#开通业务账号:至信数字藏品账号开通失败[postBody={},postRespStr={}]", postBody, postRespStr);
                 return ResponseDto.failure(ErrorMapping.get("10113"));
             }
             JsonNode postRespDataJNode = postRespJNode.get("data");
             blockChainAddress = postRespDataJNode.get("blockChainAddress").asText();
         } catch (Exception e) {
-            log.error("Ex.开通业务账号:至信数字藏品账号开通请求异常[UID={},url={},paramsMap={}],ex:{}", uid, postUrl, paramsMap, e.getMessage());
+            log.error("Ex.开通业务账号:至信数字藏品账号开通请求异常[url={},postBody={}],ex:{}", postUrl, postBody, e.getMessage());
             return ResponseDto.failure(ErrorMapping.get("10113"));
         }
 

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerCommonSqlRedisStreamConfig.java

@@ -2,8 +2,7 @@ package com.liquidnet.service.consumer.kylin.config;
 
 import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
 import com.liquidnet.service.base.constant.MQConst;
-import com.liquidnet.service.consumer.kylin.receiver.ConsumerGoblinNftOrderRdsReceiver;
-import com.liquidnet.service.consumer.kylin.receiver.ConsumerGoblinSqlArtworkGenRdsReceiver;
+import com.liquidnet.service.consumer.kylin.receiver.*;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -24,10 +23,64 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
     @Autowired
     StringRedisTemplate stringRedisTemplate;
     @Autowired
+    ConsumerGoblinSqlStoreRdsReceiver consumerGoblinSqlStoreRdsReceiver;
+    @Autowired
+    ConsumerGoblinSqlGoodsRdsReceiver consumerGoblinSqlGoodsRdsReceiver;
+    @Autowired
+    ConsumerGoblinSqlCouponRdsReceiver consumerGoblinSqlCouponRdsReceiver;
+    @Autowired
     ConsumerGoblinSqlArtworkGenRdsReceiver consumerGoblinSqlArtworkGenRdsReceiver;
     @Autowired
     ConsumerGoblinNftOrderRdsReceiver consumerGoblinNftOrderRdsReceiver;
 
+    @Bean// 店铺相关
+    public List<Subscription> subscriptionSqlStore(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GoblinQueue stream = MQConst.GoblinQueue.SQL_STORE;
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 2; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGoblinSqlStoreRdsReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+
+    @Bean// 商品相关
+    public List<Subscription> subscriptionSqlGoods(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GoblinQueue stream = MQConst.GoblinQueue.SQL_GOODS;
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 3; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGoblinSqlGoodsRdsReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+
+    @Bean// 店铺优惠券相关
+    public List<Subscription> subscriptionSqlStoreCoupon(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GoblinQueue stream = MQConst.GoblinQueue.SQL_COUPON;
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 10; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGoblinSqlCouponRdsReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+
     @Bean// 藏品生成
     public List<Subscription> subscriptionSqlArtworkGen(RedisConnectionFactory factory) {
         List<Subscription> subscriptionList = new ArrayList<>();
@@ -45,7 +98,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
     }
 
     @Bean// NFT订单
-    public List<Subscription> subscriptionBizNftOrder(RedisConnectionFactory factory) {
+    public List<Subscription> subscriptionSqlNftOrder(RedisConnectionFactory factory) {
         List<Subscription> subscriptionList = new ArrayList<>();
         MQConst.GoblinQueue stream = MQConst.GoblinQueue.GOBLIN_NFT_ORDER;
         this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());

liquidnet-bus-service/liquidnet-service-goblin/liquidnet-service-goblin-impl/src/main/java/com/liquidnet/service/goblin/service/impl/inner/GoblinGalaxyService.java

 package com.liquidnet.service.goblin.service.impl.inner;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.commons.lang.util.CurrentUtil;
 import com.liquidnet.commons.lang.util.HttpUtil;
@@ -89,20 +90,18 @@ public class GoblinGalaxyService {
      * @return
      */
     public List<String> checkNftTradeResult(String uid, String orderId, String routerType) {
-        LinkedMultiValueMap<String, String> paramsMap = CollectionUtil.linkedMultiValueMapStringString();
-        paramsMap.add("userId", uid);
-        paramsMap.add("nftOrderPayId", orderId);// TODO: 2022/3/31 ==zhanggb
-        paramsMap.add("routerType", routerType);
+        ObjectNode postBodyNode = JsonUtils.OM().createObjectNode();
+        postBodyNode.put("userId", uid);
+        postBodyNode.put("nftOrderPayId", orderId);
+        postBodyNode.put("routerType", routerType);
 
 //        String postUrl = sevGalaxyUrl + "/user/register", blockChainAddress;// TODO: 2022/3/31 ==zhanggb
         String postUrl = "https://ENVgalaxy.zhengzai.tv/galaxy/nftTrade/nftPublishAndBuyResultQuery".replace("ENV", env.getProperty(CurrentUtil.CK_ENV_ACTIVE)), blockChainAddress;
         try {
-            LinkedMultiValueMap<String, String> headerMap = CollectionUtil.linkedMultiValueMapStringString();
-            headerMap.add("Accept", MediaType.APPLICATION_JSON_VALUE);
-            String postRespStr = HttpUtil.post(postUrl, paramsMap, headerMap);
+            String postRespStr = HttpUtil.postRaw(postUrl, postBodyNode.toString(), null);
             JsonNode postRespJNode = JsonUtils.fromJson(postRespStr, JsonNode.class), postRespCode;
             if (null == postRespJNode || null == (postRespCode = postRespJNode.get("code")) || !postRespCode.asText().equals("0")) {
-                log.warn("#NFT发行购买结果查询:查询失败[paramsMap={},postRespStr={}]", paramsMap, postRespStr);
+                log.warn("#NFT发行购买结果查询:查询失败[postBody={},postRespStr={}]", postBodyNode, postRespStr);
                 return null;
             }
             JsonNode postRespDataJNode = postRespJNode.get("data");
@@ -115,7 +114,7 @@ public class GoblinGalaxyService {
             respDataList.add(postRespDataJNode.get("nftBuyTxHash").asText());
             return respDataList;
         } catch (Exception e) {
-            log.error("Ex.NFT发行购买结果查询:请求异常[UID={},url={},paramsMap={}],ex:{}", uid, postUrl, paramsMap, e.getMessage());
+            log.error("Ex.NFT发行购买结果查询:请求异常[url={},postBody={}],ex:{}", postUrl, postBodyNode, e.getMessage());
             return null;
         }
     }