修改nft消费为接口调用

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/pom.xml

@@ -32,7 +32,11 @@
         </dependency>
 
         <!-- liquidnet-bus-api -->
-
+        <dependency>
+            <groupId>com.liquidnet</groupId>
+            <artifactId>liquidnet-service-galaxy-api</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
     </dependencies>
 
     <build>

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/ConsumerCommonBizRedisStreamConfig.java

@@ -2,6 +2,8 @@ package com.liquidnet.service.consumer.base.config;
 
 import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
 import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.base.receiver.ConsumerGalaxyJsonNftPublishAndBuyReceiver;
+import com.liquidnet.service.consumer.base.receiver.ConsumerGalaxyJsonNftUserRegisterReceiver;
 import com.liquidnet.service.consumer.base.receiver.ConsumerGoblinBizArtworkUplReceiver;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -30,6 +32,10 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
     StringRedisTemplate stringRedisTemplate;
     @Autowired
     ConsumerGoblinBizArtworkUplReceiver consumerGoblinBizArtworkUplReceiver;
+    @Autowired
+    private ConsumerGalaxyJsonNftPublishAndBuyReceiver jsonNftPublishAndBuyReceiver;
+    @Autowired
+    private ConsumerGalaxyJsonNftUserRegisterReceiver jsonNftUserRegisterReceiver;
 
     @Bean// 藏品上传声明
     public List<Subscription> subscriptionGoblinBizArtworkUpl(RedisConnectionFactory factory) {
@@ -46,4 +52,48 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
         }
         return subscriptionList;
     }
+
+    /**
+     * galaxy发行和购买
+     * @param factory
+     * @return
+     */
+    @Bean
+    public List<Subscription> subscriptionGalaxyJsonNftPublishAndBuy(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY;
+
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 20; 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()), jsonNftPublishAndBuyReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+
+    /**
+     * Galaxy数字账户开通
+     * @param factory
+     * @return
+     */
+    @Bean
+    public List<Subscription> subscriptionGalaxyJsonNftUserRegister(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.JSON_NFT_USER_REGISTER;
+
+        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()), jsonNftUserRegisterReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
 }

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/ConsumerGalaxyJsonNftPublishAndBuyReceiver.java

+package com.liquidnet.service.consumer.base.receiver;
+
+import com.liquidnet.commons.lang.util.CollectionUtil;
+import com.liquidnet.commons.lang.util.HttpUtil;
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.ResponseDto;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.galaxy.dto.param.GalaxyNftPublishAndBuyReqDto;
+import com.liquidnet.service.galaxy.dto.param.GalaxyNftPublishAndBuyRespDto;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.connection.stream.StreamRecords;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+
+@Slf4j
+@Component
+public class ConsumerGalaxyJsonNftPublishAndBuyReceiver extends AbstractBizRedisReceiver {
+    @Value("${liquidnet.service.goblin.url}")
+    private String serviceGoblinUrl;
+
+    @Override
+    protected boolean consumerMessageHandler(String msg) {
+        boolean aBoolean = false;
+        try {
+            GalaxyNftPublishAndBuyReqDto textMessage = JsonUtils.fromJson(msg, GalaxyNftPublishAndBuyReqDto.class);
+            if (textMessage == null) {
+                aBoolean = true;
+            } else {
+                //执行计数
+                ResponseDto<GalaxyNftPublishAndBuyRespDto> responseDto = this.nftPublishAndBuy(textMessage);
+                if(responseDto.isSuccess()){
+                    aBoolean = true;
+                }
+            }
+        } catch (Exception e) {
+            log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
+        } finally {
+            if (!aBoolean) {
+                HashMap<String, String> map = CollectionUtil.mapStringString();
+                map.put("message", msg);
+                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
+            }
+        }
+        return aBoolean;
+    }
+
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY.getGroup();
+    }
+
+    /**
+     * 执行购买处理
+     * @param reqDto
+     * @return
+     */
+    private ResponseDto<GalaxyNftPublishAndBuyRespDto> nftPublishAndBuy(GalaxyNftPublishAndBuyReqDto reqDto) {
+        String postUrl = serviceGoblinUrl + "/goblin/nftTrade/nftPublishAndBuy";
+        try {
+            String postRespStr = HttpUtil.postJson(postUrl, JsonUtils.toJson(reqDto));
+            ResponseDto responseDto = JsonUtils.fromJson(postRespStr, ResponseDto.class);
+            return  responseDto;
+        } catch (Exception e) {
+            log.error("Ex.NFT发行和购买:请求异常[url={},paramsStr={}],ex:{}", postUrl, JsonUtils.toJson(reqDto), e.getMessage());
+            return ResponseDto.failure();
+        }
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/ConsumerGalaxyJsonNftUserRegisterReceiver.java

+package com.liquidnet.service.consumer.base.receiver;
+
+import com.liquidnet.commons.lang.util.CollectionUtil;
+import com.liquidnet.commons.lang.util.HttpUtil;
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.ResponseDto;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.galaxy.dto.param.GalaxyUserRegisterReqDto;
+import com.liquidnet.service.galaxy.dto.param.GalaxyUserRegisterRespDto;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.connection.stream.StreamRecords;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+
+@Slf4j
+@Component
+public class ConsumerGalaxyJsonNftUserRegisterReceiver extends AbstractBizRedisReceiver {
+    @Value("${liquidnet.service.goblin.url}")
+    private String serviceGoblinUrl;
+
+    @Override
+    protected boolean consumerMessageHandler(String msg) {
+        boolean aBoolean = false;
+        try {
+            GalaxyUserRegisterReqDto textMessage = JsonUtils.fromJson(msg, GalaxyUserRegisterReqDto.class);
+            if (textMessage == null) {
+                aBoolean = true;
+            } else {
+                //执行计数
+                ResponseDto<GalaxyUserRegisterRespDto> responseDto = this.userRegister(textMessage);
+//                if(responseDto.isSuccess()){
+                    aBoolean = true;
+//                }
+            }
+        } catch (Exception e) {
+            log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
+        } finally {
+            if (!aBoolean) {
+                HashMap<String, String> map = CollectionUtil.mapStringString();
+                map.put("message", msg);
+                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
+            }
+        }
+        return aBoolean;
+    }
+
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.GalaxyQueue.JSON_NFT_USER_REGISTER.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.GalaxyQueue.JSON_NFT_USER_REGISTER.getGroup();
+    }
+
+    /**
+     * 执行用户注册
+     * @param reqDto
+     * @return
+     */
+    private ResponseDto<GalaxyUserRegisterRespDto> userRegister(GalaxyUserRegisterReqDto reqDto) {
+        String postUrl = serviceGoblinUrl + "/goblin/nftUser/register";
+        try {
+            String postRespStr = HttpUtil.postJson(postUrl, JsonUtils.toJson(reqDto));
+            ResponseDto responseDto = JsonUtils.fromJson(postRespStr, ResponseDto.class);
+            return  responseDto;
+        } catch (Exception e) {
+            log.error("Ex.NFT开通数字账户:请求异常[url={},paramsStr={}],ex:{}", postUrl, JsonUtils.toJson(reqDto), e.getMessage());
+            return ResponseDto.failure();
+        }
+    }
+}