Support turms-gateway in sending notifications to each other for futu…

…re use
turms-im · Jun 10, 2024 · abd1149 · abd1149
1 parent afe4536
commit abd1149
Show file tree

Hide file tree

Showing 6 changed files with 64 additions and 32 deletions.
diff --git a/...ce/infra/logging/NotificationLogging.java → ...gging/BaseNotificationLoggingManager.java b/...ce/infra/logging/NotificationLogging.java → ...gging/BaseNotificationLoggingManager.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package im.turms.service.infra.logging;
+package im.turms.server.common.infra.logging;
 
 import io.netty.buffer.ByteBuf;
 
@@ -29,21 +29,14 @@
 
 /**
  * @author James Chen
- * @implNote Don't use StringBuilder because String#join is more efficient
  */
-public final class NotificationLogging {
-
-    private NotificationLogging() {
-    }
+public abstract class BaseNotificationLoggingManager {
 
     /**
      * Note that although TurmsNotification can represent a "response" or "notification", the method
      * is only designed to log "notification" instead of "response"
      */
-    public static void log(
-            int recipientCount,
-            int onlineRecipientCount,
-            TurmsNotification notification) {
+    public void log(TurmsNotification notification, int recipientCount, int onlineRecipientCount) {
         TurmsRequest relayedRequest = notification.getRelayedRequest();
         ByteBuf buffer = ByteBufUtil.join(64,
                 LOG_FIELD_DELIMITER,

diff --git a/...ssage/service/OutboundMessageService.java → ...infra/message/OutboundMessageManager.java b/...ssage/service/OutboundMessageService.java → ...infra/message/OutboundMessageManager.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package im.turms.service.domain.message.service;
+package im.turms.server.common.infra.message;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -41,15 +41,17 @@
 import im.turms.server.common.infra.cluster.node.Node;
 import im.turms.server.common.infra.collection.CollectionUtil;
 import im.turms.server.common.infra.collection.CollectorUtil;
+import im.turms.server.common.infra.logging.BaseApiLoggingContext;
+import im.turms.server.common.infra.logging.BaseNotificationLoggingManager;
 import im.turms.server.common.infra.proto.ProtoEncoder;
 import im.turms.server.common.infra.reactor.PublisherPool;
 import im.turms.server.common.infra.tracing.TracingCloseableContext;
 import im.turms.server.common.infra.tracing.TracingContext;
 import im.turms.server.common.storage.mongo.IMongoCollectionInitializer;
-import im.turms.service.infra.logging.ApiLoggingContext;
-import im.turms.service.infra.logging.NotificationLogging;
 
 /**
+ * The class is responsible for sending the outbound message buffer to any users on any servers.
+ *
  * @author James Chen
  * @implNote 1. All operations that send the outbound message buffer to other servers need to ensure
  *           that the buffer will be released by 1.
@@ -60,18 +62,21 @@
  */
 @Service
 @DependsOn(IMongoCollectionInitializer.BEAN_NAME)
-public class OutboundMessageService {
+public class OutboundMessageManager {
 
     private final Node node;
-    private final ApiLoggingContext apiLoggingContext;
+    private final BaseApiLoggingContext apiLoggingContext;
+    private final BaseNotificationLoggingManager notificationLoggingManager;
     private final UserStatusService userStatusService;
 
-    public OutboundMessageService(
+    public OutboundMessageManager(
             Node node,
-            ApiLoggingContext apiLoggingContext,
+            BaseApiLoggingContext apiLoggingContext,
+            BaseNotificationLoggingManager notificationLoggingManager,
             UserStatusService userStatusService) {
         this.node = node;
         this.apiLoggingContext = apiLoggingContext;
+        this.notificationLoggingManager = notificationLoggingManager;
         this.userStatusService = userStatusService;
     }
 
@@ -402,8 +407,8 @@ private Mono<Set<Long>> tryLogNotification(
             int offlineRecipientCount = CollectionUtil.getSize(signal.get());
             try (TracingCloseableContext ignored =
                     TracingContext.getCloseableContext(signal.getContextView())) {
-                NotificationLogging
-                        .log(recipientCount, recipientCount - offlineRecipientCount, notification);
+                notificationLoggingManager
+                        .log(notification, recipientCount, recipientCount - offlineRecipientCount);
             }
         });
     }

diff --git a/...main/java/im/turms/service/access/servicerequest/dispatcher/ServiceRequestDispatcher.java b/...main/java/im/turms/service/access/servicerequest/dispatcher/ServiceRequestDispatcher.java
@@ -56,6 +56,7 @@
 import im.turms.server.common.infra.lang.ByteArrayWrapper;
 import im.turms.server.common.infra.logging.core.logger.Logger;
 import im.turms.server.common.infra.logging.core.logger.LoggerFactory;
+import im.turms.server.common.infra.message.OutboundMessageManager;
 import im.turms.server.common.infra.plugin.PluginManager;
 import im.turms.server.common.infra.property.TurmsPropertiesManager;
 import im.turms.server.common.infra.proto.ProtoDecoder;
@@ -64,7 +65,6 @@
 import im.turms.server.common.infra.tracing.TracingContext;
 import im.turms.service.access.servicerequest.dto.ClientRequest;
 import im.turms.service.access.servicerequest.dto.RequestHandlerResult;
-import im.turms.service.domain.message.service.OutboundMessageService;
 import im.turms.service.infra.logging.ApiLoggingContext;
 import im.turms.service.infra.logging.ClientApiLogging;
 import im.turms.service.infra.plugin.extension.ClientRequestTransformer;
@@ -91,8 +91,8 @@ public class ServiceRequestDispatcher implements IServiceRequestDispatcher {
 
     private final ApiLoggingContext apiLoggingContext;
     private final BlocklistService blocklistService;
+    private final OutboundMessageManager outboundMessageManager;
     private final ServerStatusManager serverStatusManager;
-    private final OutboundMessageService outboundMessageService;
     private final PluginManager pluginManager;
 
     private final FastEnumMap<TurmsRequest.KindCase, ClientRequestHandler> requestTypeToHandler;
@@ -123,14 +123,14 @@ public ServiceRequestDispatcher(
             ApiLoggingContext apiLoggingContext,
             ApplicationContext context,
             BlocklistService blocklistService,
+            OutboundMessageManager outboundMessageManager,
             ServerStatusManager serverStatusManager,
-            OutboundMessageService outboundMessageService,
             PluginManager pluginManager,
             TurmsPropertiesManager propertiesManager) {
         this.apiLoggingContext = apiLoggingContext;
         this.blocklistService = blocklistService;
+        this.outboundMessageManager = outboundMessageManager;
         this.serverStatusManager = serverStatusManager;
-        this.outboundMessageService = outboundMessageService;
         this.pluginManager = pluginManager;
         Set<TurmsRequest.KindCase> disabledEndpoints = propertiesManager.getLocalProperties()
                 .getService()
@@ -435,19 +435,19 @@ private Mono<Void> notifyRelatedUsersOfAction0(
         Mono<Set<Long>> mono;
         if (forwardNotificationToRequesterOtherOnlineSessions) {
             if (noRecipient) {
-                mono = outboundMessageService.forwardNotification(notificationForRecipients,
+                mono = outboundMessageManager.forwardNotification(notificationForRecipients,
                         notificationByteBuf,
                         requesterId,
                         requesterDevice);
             } else {
                 notificationByteBuf.retain(2);
                 Mono<Set<Long>> notifyRequesterMono =
-                        outboundMessageService.forwardNotification(notificationForRecipients,
+                        outboundMessageManager.forwardNotification(notificationForRecipients,
                                 notificationByteBuf,
                                 requesterId,
                                 requesterDevice);
                 Mono<Set<Long>> notifyRecipientsMono =
-                        outboundMessageService.forwardNotification(notificationForRecipients,
+                        outboundMessageManager.forwardNotification(notificationForRecipients,
                                 notificationByteBuf,
                                 recipients);
                 mono = Flux.mergeDelayError(2, notifyRequesterMono, notifyRecipientsMono)
@@ -456,7 +456,7 @@ private Mono<Void> notifyRelatedUsersOfAction0(
                         .doFinally(signal -> notificationByteBuf.release());
             }
         } else {
-            mono = outboundMessageService.forwardNotification(notificationForRecipients,
+            mono = outboundMessageManager.forwardNotification(notificationForRecipients,
                     notificationByteBuf,
                     recipients);
         }

diff --git a/turms-service/src/main/java/im/turms/service/domain/message/service/MessageService.java b/turms-service/src/main/java/im/turms/service/domain/message/service/MessageService.java
@@ -63,6 +63,7 @@
 import im.turms.server.common.infra.lang.LongUtil;
 import im.turms.server.common.infra.logging.core.logger.Logger;
 import im.turms.server.common.infra.logging.core.logger.LoggerFactory;
+import im.turms.server.common.infra.message.OutboundMessageManager;
 import im.turms.server.common.infra.net.InetAddressUtil;
 import im.turms.server.common.infra.netty.ReferenceCountUtil;
 import im.turms.server.common.infra.plugin.PluginManager;
@@ -126,11 +127,13 @@ public class MessageService {
     private static final Method GET_MESSAGES_TO_DELETE_METHOD;
 
     private final MessageRepository messageRepository;
+
+    private final OutboundMessageManager outboundMessageManager;
     @Nullable
     private final TurmsRedisClientManager redisClientManager;
     private final Node node;
+
     private final ConversationService conversationService;
-    private final OutboundMessageService outboundMessageService;
     private final GroupService groupService;
     private final GroupMemberService groupMemberService;
     private final UserService userService;
@@ -181,6 +184,7 @@ public class MessageService {
     @Autowired
     public MessageService(
             MessageRepository messageRepository,
+            OutboundMessageManager outboundMessageManager,
             @Nullable @Autowired(
                     required = false) @Qualifier("sequenceIdRedisClientManager") TurmsRedisClientManager sequenceIdRedisClientManager,
 
@@ -191,25 +195,25 @@ public MessageService(
             GroupService groupService,
             GroupMemberService groupMemberService,
             UserService userService,
-            OutboundMessageService outboundMessageService,
             MetricsService metricsService,
 
             PluginManager pluginManager,
             TaskManager taskManager) {
         this.messageRepository = messageRepository;
+        this.outboundMessageManager = outboundMessageManager;
         this.redisClientManager = sequenceIdRedisClientManager;
         this.node = node;
         this.conversationService = conversationService;
         this.groupService = groupService;
         this.groupMemberService = groupMemberService;
         this.userService = userService;
-        this.outboundMessageService = outboundMessageService;
         this.pluginManager = pluginManager;
 
         TurmsProperties globalProperties = propertiesManager.getGlobalProperties();
         MessageProperties messageProperties = globalProperties.getService()
                 .getMessage();
         useConversationId = messageProperties.isUseConversationId();
+
         SequenceIdProperties sequenceIdProperties = messageProperties.getSequenceId();
         useSequenceIdForGroupConversation =
                 sequenceIdProperties.isUseSequenceIdForGroupConversation();
@@ -1446,7 +1450,7 @@ private Mono<Void> sendMessage(
         } else {
             excludedUserSessionIds = Collections.emptySet();
         }
-        return outboundMessageService
+        return outboundMessageManager
                 .forwardNotification(notification, recipientIds, excludedUserSessionIds)
                 // TODO: Should trigger extension points
                 // https://github.com/turms-im/turms/issues/1189

diff --git a/turms-service/src/main/java/im/turms/service/infra/logging/ApiLoggingContext.java b/turms-service/src/main/java/im/turms/service/infra/logging/ApiLoggingContext.java
@@ -57,8 +57,9 @@ public boolean shouldLogRequest(TurmsRequest.KindCase requestType) {
         return shouldLog(requestType, typeToSupportedLoggingRequestProperties);
     }
 
+    @Override
     public boolean shouldLogNotification(TurmsRequest.KindCase requestType) {
         return shouldLog(requestType, typeToSupportedLoggingNotificationProperties);
     }
 
-}
+}
diff --git a/turms-service/src/main/java/im/turms/service/infra/logging/NotificationLoggingManager.java b/turms-service/src/main/java/im/turms/service/infra/logging/NotificationLoggingManager.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright (C) 2019 The Turms Project
+ * https://github.com/turms-im/turms
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package im.turms.service.infra.logging;
+
+import org.springframework.stereotype.Component;
+
+import im.turms.server.common.infra.logging.BaseNotificationLoggingManager;
+
+/**
+ * @author James Chen
+ */
+@Component
+public class NotificationLoggingManager extends BaseNotificationLoggingManager {
+}