add 集群

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/JetLinksConfiguration.java

@@ -16,8 +16,6 @@ import org.jetlinks.community.device.service.AutoDiscoverDeviceRegistry;
 import org.jetlinks.community.device.service.data.DeviceDataService;
 import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric;
 import org.jetlinks.community.micrometer.MeterRegistryManager;
-import org.jetlinks.community.standalone.configuration.cluster.ClusterDeviceMessageBrokeMessageBroker;
-import org.jetlinks.community.standalone.configuration.cluster.ClusterDeviceMessageConnector;
 import org.jetlinks.core.ProtocolSupports;
 import org.jetlinks.core.cluster.ClusterManager;
 import org.jetlinks.core.config.ConfigStorageManager;
@@ -33,7 +31,6 @@ import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.core.spi.ServiceContext;
 import org.jetlinks.supports.cluster.ClusterDeviceRegistry;
 import org.jetlinks.supports.cluster.redis.RedisClusterManager;
-import org.jetlinks.supports.config.ClusterConfigStorageManager;
 import org.jetlinks.supports.config.EventBusStorageManager;
 import org.jetlinks.supports.event.BrokerEventBus;
 import org.jetlinks.supports.protocol.ServiceLoaderProtocolSupports;
@@ -45,8 +42,6 @@ import org.jetlinks.supports.server.DefaultSendToDeviceMessageHandler;
 import org.jetlinks.supports.server.monitor.MicrometerGatewayServerMetrics;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.config.BeanPostProcessor;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/cluster/ClusterDeviceMessageBrokeMessageBroker.java

@@ -13,6 +13,7 @@ import reactor.core.publisher.*;
 import reactor.core.scheduler.Schedulers;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 /**
@@ -29,8 +30,6 @@ public class ClusterDeviceMessageBrokeMessageBroker extends StandaloneDeviceMess
 
     private ClusterManager clusterManager;
 
-
-
     private List<ServerNode> getAllNode(){
         return clusterManager.getHaManager().getAllNode();
     }

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/cluster/ClusterDeviceMessageConnector.java

@@ -21,7 +21,10 @@ import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 import javax.annotation.Nonnull;
 import java.time.Duration;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.util.*;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author lifang
@@ -35,6 +38,9 @@ public class ClusterDeviceMessageConnector extends DeviceMessageConnector {
 
     private LinkedHashMap<String,ClusterMessage> messageServerMap=new LinkedHashMap<>(1024,0.75f,true);
 
+
+    private static final long DEFAULT_TIMEOUT = TimeUnit.SECONDS.toMillis(Integer.getInteger("jetlinks.device.message.default-timeout", 10));
+
     public ClusterDeviceMessageConnector(EventBus eventBus, DeviceRegistry registry, MessageHandler messageHandler, DeviceSessionManager sessionManager, String serverId, ClusterManager clusterManager) {
         super(eventBus, registry, messageHandler, sessionManager);
         this.serverId = serverId;
@@ -91,23 +97,25 @@ public class ClusterDeviceMessageConnector extends DeviceMessageConnector {
         /**
          * 清理过期消息
          */
-//        Flux.interval(Duration.ofSeconds(5))
-//            .doOnNext(ignore->{
-//                LocalDateTime now = LocalDateTime.now();
-//                Set<String> keySet = messageServerMap.keySet();
-//                for (String key : keySet) {
-//                    Message message = messageServerMap.get(key);
-//                    if(message!=null){
-//                        long messageTime = message.getTimestamp();
-//                        if(Duration.between(LocalDateTime.ofEpochSecond(messageTime,0, ZoneOffset.ofHours(8)),now).toMillis()>3000){
-//                            messageServerMap.remove(key);
-//                        }else {
-//                            break;
-//                        }
-//                    }
-//                }
-//            })
-//            .subscribe();
+        Flux.interval(Duration.ofSeconds(5))
+            .doOnNext(ignore->{
+                LocalDateTime now = LocalDateTime.now();
+                Set<String> keySet = messageServerMap.keySet();
+                for (String key : keySet) {
+                    ClusterMessage message = messageServerMap.get(key);
+                    if(message!=null){
+                        long messageTime = message.getPayload().getTimestamp();
+                        //超过10s将过期消息删除
+                        if(Duration.between(LocalDateTime.ofEpochSecond(messageTime/1000,0, ZoneOffset.ofHours(8)),now)
+                            .toMillis()>DEFAULT_TIMEOUT){
+                            messageServerMap.remove(key);
+                        }else {
+                            break;
+                        }
+                    }
+                }
+            })
+            .subscribe();
 
 
     }
@@ -118,7 +126,9 @@ public class ClusterDeviceMessageConnector extends DeviceMessageConnector {
             return  Flux.fromStream(getAllNode().stream())
                 .flatMap(node-> clusterManager
                     .getTopic(ClusterMessageType.topicOf(node.getId(), ClusterMessageType.up))
-                    .publish(Mono.just(message))
+                    .publish(Mono.just(message)
+                        .doOnNext(msg->msg.addHeader(Headers.serverId,serverId))
+                        .map(msg->new ClusterMessage(msg,serverId,ClusterMessageType.topicOf(serverId,ClusterMessageType.up))))
                 )
                 .flatMap(result->result==0?Mono.just(false):Mono.just(true))
                 .reduce((a1,a2)->a1&&a2);
@@ -145,7 +155,9 @@ public class ClusterDeviceMessageConnector extends DeviceMessageConnector {
                         }
                         return clusterManager
                             .getTopic(ClusterMessageType.topicOf(serverId,ClusterMessageType.up))
-                            .publish(Mono.just(message))
+                            .publish(Mono.just(message)
+                                .doOnNext(msg->msg.addHeader(Headers.serverId,serverId))
+                                .map(msg->new ClusterMessage(msg,serverId,ClusterMessageType.topicOf(serverId,ClusterMessageType.up))))
                             .flatMap(result->result==0?Mono.just(false):Mono.just(true));
                     }
                 });