实现百度链转让-消费

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

@@ -36,11 +36,14 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
     ConsumerGoblinBizArtworkUplReceiver consumerGoblinBizArtworkUplReceiver;
     @Autowired
     ConsumerGoblinBizArtworkClqReceiver consumerGoblinBizArtworkClqReceiver;
+    /*------galaxy------*/
     @Autowired
     ConsumerGalaxyJsonNftPublishAndBuyReceiver jsonNftPublishAndBuyReceiver;
     @Autowired
     ConsumerGalaxyJsonNftUserRegisterReceiver jsonNftUserRegisterReceiver;
     @Autowired
+    ConsumerGalaxyJsonNftTransferReceiver jsonNftTransferReceiver;
+    @Autowired
     ConsumerGoblinBizIntegralReceiver consumerGoblinBizIntegralReceiver;
     /*------sweet------*/
     @Autowired
@@ -177,6 +180,28 @@ public class ConsumerCommonBizRedisStreamConfig extends RedisStreamConfig {
         return subscriptionList;
     }
 
+    /**
+     * Galaxy nft转让
+     * @param factory
+     * @return
+     */
+    @Bean
+    public List<Subscription> subscriptionGalaxyJsonNftTransfer(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.JSON_NFT_USER_NFT_TRANSFER;
+
+        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()), jsonNftTransferReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+
     @Bean // 增减积分
     public List<Subscription> subscriptionGoblinBizIntegral(RedisConnectionFactory factory) {
         List<Subscription> subscriptionList = new ArrayList<>();

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

@@ -71,6 +71,7 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
     ConsumerSlimeSqlLineReceiver consumerSlimeSqlLineReceiver;
     @Autowired
     ConsumerSlimeSqlPerformanceInsertReceiver consumerSlimeSqlPerformanceInsertReceiver;
+    /*------galaxy------*/
     @Autowired
     ConsumerGalaxySqlUserInfoReceiver consumerGalaxySqlUserInfoReceiver;
     @Autowired

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/ConsumerGalaxyJsonNftTransferReceiver.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.GalaxyNftTransferReqDto;
+import com.liquidnet.service.galaxy.dto.param.GalaxyNftTransferRespDto;
+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 ConsumerGalaxyJsonNftTransferReceiver extends AbstractBizRedisReceiver {
+    @Value("${liquidnet.service.goblin.url}")
+    private String serviceGoblinUrl;
+
+    @Override
+    protected boolean consumerMessageHandler(String msg) {
+        boolean aBoolean = false;
+        try {
+            GalaxyNftTransferReqDto textMessage = JsonUtils.fromJson(msg, GalaxyNftTransferReqDto.class);
+            if (textMessage == null) {
+                aBoolean = true;
+            } else {
+                //执行计数
+                ResponseDto<GalaxyNftTransferRespDto> responseDto = this.nftTransfer(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_NFT_TRANSFER.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.GalaxyQueue.JSON_NFT_USER_NFT_TRANSFER.getGroup();
+    }
+
+    /**
+     * 执行nft转让
+     * @param reqDto
+     * @return
+     */
+    private ResponseDto<GalaxyNftTransferRespDto> nftTransfer(GalaxyNftTransferReqDto reqDto) {
+        String postUrl = serviceGoblinUrl + "/goblin/nftTrade/que/nftTransfer";
+        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();
+        }
+    }
+}