[集群] 优化国标级联目录更新

src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java

@@ -35,10 +35,10 @@ public class EventPublisher {
     private ApplicationEventPublisher applicationEventPublisher;
 
 	@Autowired
-    private UserSetting userSetting;
+	private UserSetting userSetting;
 
 	@Autowired
-    private IRedisRpcService redisRpcService;
+	private IRedisRpcService redisRpcService;
 	
 	/**
 	 * 设备报警事件
@@ -76,15 +76,15 @@ public class EventPublisher {
 	}
 
 	public void catalogEventPublish(Platform platform, List<CommonGBChannel> deviceChannels, String type) {
-		if (!userSetting.getServerId().equals(platform.getServerId())) {
+		if (platform != null && !userSetting.getServerId().equals(platform.getServerId())) {
-			List<Integer> ids = new ArrayList<>();
+			// 指定了上级平台的推送，则发送到指定的设备，未指定的则全部发送， 接收后各自处理自己的
-			for (int i = 0; i < deviceChannels.size(); i++) {
+			CatalogEvent outEvent = new CatalogEvent(this);
-				ids.add(deviceChannels.get(i).getGbId());
+			outEvent.setChannels(deviceChannels);
-			}
+			outEvent.setType(type);
-			redisRpcService.catalogEventPublish(platform.getServerId(), platform.getId(), ids, type);
+			outEvent.setPlatform(platform);
+			redisRpcService.catalogEventPublish(platform.getServerId(), outEvent);
 			return;
 		}
-
 		CatalogEvent outEvent = new CatalogEvent(this);
 		List<CommonGBChannel> channels = new ArrayList<>();
 		if (deviceChannels.size() > 1) {
@@ -103,6 +103,14 @@ public class EventPublisher {
 		outEvent.setType(type);
 		outEvent.setPlatform(platform);
 		applicationEventPublisher.publishEvent(outEvent);
+		if (platform == null) {
+			// 如果没指定上级平台，则推送消息到所有在线的wvp处理自己含有的平台的目录更新
+			redisRpcService.catalogEventPublish(null, outEvent);
+		}
+	}
+
+	public void catalogEventPublish(CatalogEvent event) {
+		applicationEventPublisher.publishEvent(event);
 	}
 
 	public void mobilePositionEventPublish(MobilePosition mobilePosition) {

src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java

@@ -1,5 +1,6 @@
 package com.genersoft.iot.vmp.gb28181.event.subscribe.catalog;
 
+import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
 import com.genersoft.iot.vmp.gb28181.bean.Platform;
 import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
@@ -7,6 +8,7 @@ import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo;
 import com.genersoft.iot.vmp.gb28181.service.IPlatformChannelService;
 import com.genersoft.iot.vmp.gb28181.service.IPlatformService;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
+import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationListener;
@@ -15,10 +17,7 @@ import org.springframework.stereotype.Component;
 import javax.sip.InvalidArgumentException;
 import javax.sip.SipException;
 import java.text.ParseException;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
 
 /**
  * catalog事件
@@ -30,9 +29,6 @@ public class CatalogEventLister implements ApplicationListener<CatalogEvent> {
     @Autowired
     private IPlatformChannelService platformChannelService;
 
-    @Autowired
-    private IPlatformService platformService;
-
     @Autowired
     private ISIPCommanderForPlatform sipCommanderFroPlatform;
 

src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java

@@ -4,6 +4,7 @@ import com.genersoft.iot.vmp.common.CommonCallback;
 import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.gb28181.bean.Platform;
 import com.genersoft.iot.vmp.gb28181.bean.SendRtpInfo;
+import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
 import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 
 import java.util.List;
@@ -33,5 +34,5 @@ public interface IRedisRpcService {
 
     boolean updatePlatform(String serverId, Platform platform);
 
-    void catalogEventPublish(String serverId, Integer platformId, List<Integer> channelIds, String type);
+    void catalogEventPublish(String serverId, CatalogEvent catalogEvent);
 }

src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcPlatformController.java

@@ -9,6 +9,7 @@ import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcResponse;
 import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
 import com.genersoft.iot.vmp.gb28181.bean.Platform;
 import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
+import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
 import com.genersoft.iot.vmp.gb28181.service.IPlatformChannelService;
 import com.genersoft.iot.vmp.gb28181.service.IPlatformService;
 import com.genersoft.iot.vmp.service.redisMsg.dto.RedisRpcController;
@@ -70,21 +71,9 @@ public class RedisRpcPlatformController extends RpcController {
      */
     @RedisRpcMapping("catalogEventPublish")
     public RedisRpcResponse catalogEventPublish(RedisRpcRequest request) {
-        JSONObject jsonObject = JSONObject.parseObject(request.getParam().toString());
+        CatalogEvent event = JSONObject.parseObject(request.getParam().toString(), CatalogEvent.class);
-        int platformId = jsonObject.getIntValue("platformId");
+        eventPublisher.catalogEventPublish(event);
-        Integer[] channelIds = jsonObject.getJSONArray("channelIds").toArray(Integer.class);
-        String type = jsonObject.getString("type");
         RedisRpcResponse response = request.getResponse();
-
-        Platform platform = platformService.queryOne(platformId);
-        if (platform == null ) {
-            log.warn("[]");
-            response.setStatusCode(ErrorCode.ERROR400.getCode());
-            response.setBody(ErrorCode.ERROR400.getMsg());
-            return response;
-        }
-        List<CommonGBChannel> commonGBChannels = platformChannelService.queryChannelByPlatformIdAndChannelIds(platformId, Arrays.asList(channelIds));
-        eventPublisher.catalogEventPublish(platform, commonGBChannels, type);
         response.setStatusCode(ErrorCode.SUCCESS.getCode());
         return response;
     }

src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java

@@ -10,6 +10,7 @@ import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcRequest;
 import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcResponse;
 import com.genersoft.iot.vmp.gb28181.bean.Platform;
 import com.genersoft.iot.vmp.gb28181.bean.SendRtpInfo;
+import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
 import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
 import com.genersoft.iot.vmp.media.event.hook.Hook;
 import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
@@ -234,13 +235,11 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
     }
 
     @Override
-    public void catalogEventPublish(String serverId, Integer platformId, List<Integer> channelIds, String type) {
+    public void catalogEventPublish(String serverId, CatalogEvent event) {
-        JSONObject jsonObject = new JSONObject();
+        RedisRpcRequest request = buildRequest("platform/catalogEventPublish", event);
-        jsonObject.put("platformId", platformId);
+        if (serverId != null) {
-        jsonObject.put("channelIds", channelIds);
+            request.setToId(serverId);
-        jsonObject.put("type", type);
+        }
-        RedisRpcRequest request = buildRequest("platform/catalogEventPublish", jsonObject);
-        request.setToId(serverId);
         redisRpcConfig.request(request, 100);
     }
 }