Send task center events through MQ

nacos_config/rc-service-live/application.yml

@@ -7,4 +7,9 @@ task:
       base-url: ${TASK_CENTER_GO_URL:${LIKEI_GATEWAY_ROUTE_GO_URI:http://golang:2900}}
       internal-token: ${GAME_INTERNAL_CALLBACK_SECRET:}
       timeout-millis: ${TASK_CENTER_GO_TIMEOUT_MILLIS:3000}
+    event:
+      mq:
+        enabled: ${TASK_CENTER_EVENT_MQ_ENABLED:false}
+        topic: ${TASK_CENTER_EVENT_MQ_TOPIC:RC_DEFAULT_APP_ORDINARY}
+        tag: ${TASK_CENTER_EVENT_MQ_TAG:task_center_event}
   

nacos_config/rc-service-order/application.yml

@@ -79,3 +79,8 @@ task:
       base-url: ${TASK_CENTER_GO_URL:${LIKEI_GATEWAY_ROUTE_GO_URI:http://golang:2900}}
       internal-token: ${GAME_INTERNAL_CALLBACK_SECRET:}
       timeout-millis: ${TASK_CENTER_GO_TIMEOUT_MILLIS:3000}
+    event:
+      mq:
+        enabled: ${TASK_CENTER_EVENT_MQ_ENABLED:false}
+        topic: ${TASK_CENTER_EVENT_MQ_TOPIC:RC_DEFAULT_APP_ORDINARY}
+        tag: ${TASK_CENTER_EVENT_MQ_TAG:task_center_event}

nacos_config/rc-service-other/application.yml

@@ -122,6 +122,11 @@ task:
       base-url: ${TASK_CENTER_GO_URL:${LIKEI_GATEWAY_ROUTE_GO_URI:http://golang:2900}}
       internal-token: ${GAME_INTERNAL_CALLBACK_SECRET:}
       timeout-millis: ${TASK_CENTER_GO_TIMEOUT_MILLIS:3000}
+    event:
+      mq:
+        enabled: ${TASK_CENTER_EVENT_MQ_ENABLED:false}
+        topic: ${TASK_CENTER_EVENT_MQ_TOPIC:RC_DEFAULT_APP_ORDINARY}
+        tag: ${TASK_CENTER_EVENT_MQ_TAG:task_center_event}
 
 voice-room:
   region-broadcast:

nacos_config/rc-service-wallet/application.yml

@@ -18,3 +18,8 @@ task:
       base-url: ${TASK_CENTER_GO_URL:${LIKEI_GATEWAY_ROUTE_GO_URI:http://golang:2900}}
       internal-token: ${GAME_INTERNAL_CALLBACK_SECRET:}
       timeout-millis: ${TASK_CENTER_GO_TIMEOUT_MILLIS:3000}
+    event:
+      mq:
+        enabled: ${TASK_CENTER_EVENT_MQ_ENABLED:false}
+        topic: ${TASK_CENTER_EVENT_MQ_TOPIC:RC_DEFAULT_APP_ORDINARY}
+        tag: ${TASK_CENTER_EVENT_MQ_TAG:task_center_event}

rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/client/TaskCenterGoClient.java

@@ -1,7 +1,9 @@
 package com.red.circle.live.infra.client;
 
 import com.alibaba.fastjson.JSON;
+import com.red.circle.component.mq.MessageEvent;
 import com.red.circle.live.domain.constant.ActiveSecondConstant;
+import com.red.circle.mq.rocket.business.service.MessageSenderService;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.HttpURLConnection;
@@ -14,6 +16,7 @@ import java.util.Objects;
 import lombok.Data;
 import lombok.experimental.Accessors;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
@@ -23,6 +26,8 @@ public class TaskCenterGoClient {
 
   private static final String EVENT_TYPE_MIC_DURATION_SECONDS = "MIC_DURATION_SECONDS";
 
+  private final ObjectProvider<MessageSenderService> messageSenderServiceProvider;
+
   @Value("${task.center.go.enabled:true}")
   private boolean enabled;
 
@@ -35,6 +40,19 @@ public class TaskCenterGoClient {
   @Value("${task.center.go.timeout-millis:3000}")
   private Integer timeoutMillis;
 
+  @Value("${task.center.event.mq.enabled:false}")
+  private boolean mqEnabled;
+
+  @Value("${task.center.event.mq.topic:RC_DEFAULT_APP_ORDINARY}")
+  private String mqTopic;
+
+  @Value("${task.center.event.mq.tag:task_center_event}")
+  private String mqTag;
+
+  public TaskCenterGoClient(ObjectProvider<MessageSenderService> messageSenderServiceProvider) {
+    this.messageSenderServiceProvider = messageSenderServiceProvider;
+  }
+
   public void reportMicDuration(String sysOrigin,
       Long roomId,
       Integer micIndex,
@@ -81,6 +99,9 @@ public class TaskCenterGoClient {
   }
 
   private void reportEvent(TaskCenterEventRequest request) {
+    if (mqEnabled) {
+      publishMqEvent(request);
+    }
     if (!enabled) {
       return;
     }
@@ -134,6 +155,24 @@ public class TaskCenterGoClient {
     }
   }
 
+  private void publishMqEvent(TaskCenterEventRequest request) {
+    try {
+      MessageSenderService senderService = messageSenderServiceProvider.getIfAvailable();
+      if (Objects.isNull(senderService)) {
+        log.warn("Task center mq sender missing, skip event report. eventId={}", request.getEventId());
+        return;
+      }
+      senderService.sendEventMessage(MessageEvent.builder()
+          .consumeId(request.getEventId())
+          .topicString(defaultIfBlank(mqTopic, "RC_DEFAULT_APP_ORDINARY"))
+          .tag(defaultIfBlank(mqTag, "task_center_event"))
+          .body(request)
+          .build());
+    } catch (Exception e) {
+      log.warn("Task center mq report error. eventId={}", request.getEventId(), e);
+    }
+  }
+
   private String readResponse(HttpURLConnection connection) {
     try (InputStream inputStream = connection.getErrorStream() != null
         ? connection.getErrorStream()
@@ -166,6 +205,11 @@ public class TaskCenterGoClient {
     return value;
   }
 
+  private String defaultIfBlank(String value, String fallback) {
+    String normalized = Objects.toString(value, "").trim();
+    return normalized.isEmpty() ? fallback : normalized;
+  }
+
   @Data
   @Accessors(chain = true)
   private static class TaskCenterEventRequest {

rc-service/rc-service-order/order-application/src/main/java/com/red/circle/order/app/common/task/TaskCenterGoClient.java

@@ -1,6 +1,8 @@
 package com.red.circle.order.app.common.task;
 
 import com.alibaba.fastjson.JSON;
+import com.red.circle.component.mq.MessageEvent;
+import com.red.circle.mq.rocket.business.service.MessageSenderService;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.math.BigDecimal;
@@ -15,6 +17,7 @@ import java.util.Objects;
 import lombok.Data;
 import lombok.experimental.Accessors;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
@@ -24,6 +27,8 @@ public class TaskCenterGoClient {
 
   private static final String EVENT_TYPE_RECHARGE_GOLD = "RECHARGE_GOLD";
 
+  private final ObjectProvider<MessageSenderService> messageSenderServiceProvider;
+
   @Value("${task.center.go.enabled:true}")
   private boolean enabled;
 
@@ -36,6 +41,19 @@ public class TaskCenterGoClient {
   @Value("${task.center.go.timeout-millis:3000}")
   private Integer timeoutMillis;
 
+  @Value("${task.center.event.mq.enabled:false}")
+  private boolean mqEnabled;
+
+  @Value("${task.center.event.mq.topic:RC_DEFAULT_APP_ORDINARY}")
+  private String mqTopic;
+
+  @Value("${task.center.event.mq.tag:task_center_event}")
+  private String mqTag;
+
+  public TaskCenterGoClient(ObjectProvider<MessageSenderService> messageSenderServiceProvider) {
+    this.messageSenderServiceProvider = messageSenderServiceProvider;
+  }
+
   public void reportRechargeGold(String sysOrigin,
       Long userId,
       Long purchaseHistoryId,
@@ -68,6 +86,9 @@ public class TaskCenterGoClient {
   }
 
   private void reportEvent(TaskCenterEventRequest request) {
+    if (mqEnabled) {
+      publishMqEvent(request);
+    }
     if (!enabled) {
       return;
     }
@@ -109,6 +130,24 @@ public class TaskCenterGoClient {
     }
   }
 
+  private void publishMqEvent(TaskCenterEventRequest request) {
+    try {
+      MessageSenderService senderService = messageSenderServiceProvider.getIfAvailable();
+      if (Objects.isNull(senderService)) {
+        log.warn("Task center mq sender missing, skip event report. eventId={}", request.getEventId());
+        return;
+      }
+      senderService.sendEventMessage(MessageEvent.builder()
+          .consumeId(request.getEventId())
+          .topicString(defaultIfBlank(mqTopic, "RC_DEFAULT_APP_ORDINARY"))
+          .tag(defaultIfBlank(mqTag, "task_center_event"))
+          .body(request)
+          .build());
+    } catch (Exception e) {
+      log.warn("Task center mq report error. eventId={}", request.getEventId(), e);
+    }
+  }
+
   private String readResponse(HttpURLConnection connection) {
     try (InputStream inputStream = connection.getErrorStream() != null
         ? connection.getErrorStream()
@@ -129,6 +168,11 @@ public class TaskCenterGoClient {
     return value;
   }
 
+  private String defaultIfBlank(String value, String fallback) {
+    String normalized = Objects.toString(value, "").trim();
+    return normalized.isEmpty() ? fallback : normalized;
+  }
+
   @Data
   @Accessors(chain = true)
   private static class TaskCenterEventRequest {

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/common/task/TaskCenterGoClient.java

@@ -3,6 +3,8 @@ package com.red.circle.other.app.common.task;
 import cn.hutool.http.HttpResponse;
 import cn.hutool.http.HttpUtil;
 import com.alibaba.fastjson.JSON;
+import com.red.circle.component.mq.MessageEvent;
+import com.red.circle.mq.rocket.business.service.MessageSenderService;
 import java.math.BigDecimal;
 import java.time.Instant;
 import java.util.HashMap;
@@ -11,6 +13,7 @@ import java.util.Objects;
 import lombok.Data;
 import lombok.experimental.Accessors;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
@@ -21,6 +24,8 @@ public class TaskCenterGoClient {
   private static final String EVENT_TYPE_GAME_CONSUME_GOLD = "GAME_CONSUME_GOLD";
   private static final String EVENT_TYPE_GIFT_CONSUME_GOLD = "GIFT_CONSUME_GOLD";
 
+  private final ObjectProvider<MessageSenderService> messageSenderServiceProvider;
+
   @Value("${task.center.go.enabled:true}")
   private boolean enabled;
 
@@ -33,6 +38,19 @@ public class TaskCenterGoClient {
   @Value("${task.center.go.timeout-millis:3000}")
   private Integer timeoutMillis;
 
+  @Value("${task.center.event.mq.enabled:false}")
+  private boolean mqEnabled;
+
+  @Value("${task.center.event.mq.topic:RC_DEFAULT_APP_ORDINARY}")
+  private String mqTopic;
+
+  @Value("${task.center.event.mq.tag:task_center_event}")
+  private String mqTag;
+
+  public TaskCenterGoClient(ObjectProvider<MessageSenderService> messageSenderServiceProvider) {
+    this.messageSenderServiceProvider = messageSenderServiceProvider;
+  }
+
   public void reportGameConsumeGold(String sysOrigin,
       Long userId,
       String provider,
@@ -101,6 +119,9 @@ public class TaskCenterGoClient {
   }
 
   private void reportEvent(TaskCenterEventRequest request) {
+    if (mqEnabled) {
+      publishMqEvent(request);
+    }
     if (!enabled) {
       return;
     }
@@ -127,6 +148,24 @@ public class TaskCenterGoClient {
     }
   }
 
+  private void publishMqEvent(TaskCenterEventRequest request) {
+    try {
+      MessageSenderService senderService = messageSenderServiceProvider.getIfAvailable();
+      if (Objects.isNull(senderService)) {
+        log.warn("Task center mq sender missing, skip event report. eventId={}", request.getEventId());
+        return;
+      }
+      senderService.sendEventMessage(MessageEvent.builder()
+          .consumeId(request.getEventId())
+          .topicString(defaultIfBlank(mqTopic, "RC_DEFAULT_APP_ORDINARY"))
+          .tag(defaultIfBlank(mqTag, "task_center_event"))
+          .body(request)
+          .build());
+    } catch (Exception e) {
+      log.warn("Task center mq report error. eventId={}", request.getEventId(), e);
+    }
+  }
+
   private String trimRightSlash(String value) {
     while (value.endsWith("/")) {
       value = value.substring(0, value.length() - 1);
@@ -134,6 +173,11 @@ public class TaskCenterGoClient {
     return value;
   }
 
+  private String defaultIfBlank(String value, String fallback) {
+    String normalized = Objects.toString(value, "").trim();
+    return normalized.isEmpty() ? fallback : normalized;
+  }
+
   @Data
   @Accessors(chain = true)
   private static class TaskCenterEventRequest {

rc-service/rc-service-wallet/wallet-application/src/main/java/com/red/circle/wallet/app/common/task/TaskCenterGoClient.java

@@ -1,6 +1,8 @@
 package com.red.circle.wallet.app.common.task;
 
 import com.alibaba.fastjson.JSON;
+import com.red.circle.component.mq.MessageEvent;
+import com.red.circle.mq.rocket.business.service.MessageSenderService;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.math.BigDecimal;
@@ -15,6 +17,7 @@ import java.util.Objects;
 import lombok.Data;
 import lombok.experimental.Accessors;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
@@ -24,6 +27,8 @@ public class TaskCenterGoClient {
 
   private static final String EVENT_TYPE_RECHARGE_GOLD = "RECHARGE_GOLD";
 
+  private final ObjectProvider<MessageSenderService> messageSenderServiceProvider;
+
   @Value("${task.center.go.enabled:true}")
   private boolean enabled;
 
@@ -36,6 +41,19 @@ public class TaskCenterGoClient {
   @Value("${task.center.go.timeout-millis:3000}")
   private Integer timeoutMillis;
 
+  @Value("${task.center.event.mq.enabled:false}")
+  private boolean mqEnabled;
+
+  @Value("${task.center.event.mq.topic:RC_DEFAULT_APP_ORDINARY}")
+  private String mqTopic;
+
+  @Value("${task.center.event.mq.tag:task_center_event}")
+  private String mqTag;
+
+  public TaskCenterGoClient(ObjectProvider<MessageSenderService> messageSenderServiceProvider) {
+    this.messageSenderServiceProvider = messageSenderServiceProvider;
+  }
+
   public void reportShippingAgentRechargeGold(String sysOrigin,
       Long userId,
       Long runningWaterId,
@@ -99,6 +117,9 @@ public class TaskCenterGoClient {
   }
 
   private void reportEvent(TaskCenterEventRequest request) {
+    if (mqEnabled) {
+      publishMqEvent(request);
+    }
     if (!enabled) {
       return;
     }
@@ -140,6 +161,24 @@ public class TaskCenterGoClient {
     }
   }
 
+  private void publishMqEvent(TaskCenterEventRequest request) {
+    try {
+      MessageSenderService senderService = messageSenderServiceProvider.getIfAvailable();
+      if (Objects.isNull(senderService)) {
+        log.warn("Task center mq sender missing, skip event report. eventId={}", request.getEventId());
+        return;
+      }
+      senderService.sendEventMessage(MessageEvent.builder()
+          .consumeId(request.getEventId())
+          .topicString(defaultIfBlank(mqTopic, "RC_DEFAULT_APP_ORDINARY"))
+          .tag(defaultIfBlank(mqTag, "task_center_event"))
+          .body(request)
+          .build());
+    } catch (Exception e) {
+      log.warn("Task center mq report error. eventId={}", request.getEventId(), e);
+    }
+  }
+
   private String readResponse(HttpURLConnection connection) {
     try (InputStream inputStream = connection.getErrorStream() != null
         ? connection.getErrorStream()
@@ -160,6 +199,11 @@ public class TaskCenterGoClient {
     return value;
   }
 
+  private String defaultIfBlank(String value, String fallback) {
+    String normalized = Objects.toString(value, "").trim();
+    return normalized.isEmpty() ? fallback : normalized;
+  }
+
   @Data
   @Accessors(chain = true)
   private static class TaskCenterEventRequest {