[集群] 支持通道刷新

src/main/java/com/genersoft/iot/vmp/gb28181/controller/DeviceQuery.java

@@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.gb28181.controller;
 
 import com.alibaba.fastjson2.JSONObject;
 import com.genersoft.iot.vmp.conf.DynamicTask;
+import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.conf.exception.ControllerException;
 import com.genersoft.iot.vmp.conf.security.JwtUtils;
 import com.genersoft.iot.vmp.gb28181.bean.Device;
@@ -16,6 +17,7 @@ import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeTask;
 import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
 import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
+import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService;
 import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
 import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 import com.github.pagehelper.PageInfo;
@@ -71,9 +73,15 @@ public class DeviceQuery {
 	@Autowired
 	private IDeviceService deviceService;
 
+	@Autowired
+	private UserSetting userSetting;
+
 	@Autowired
 	private DynamicTask dynamicTask;
 
+	@Autowired
+	private IRedisRpcService redisRpcService;
+
 	/**
 	 * 使用ID查询国标设备
 	 * @param deviceId 国标ID
@@ -142,30 +150,12 @@ public class DeviceQuery {
 			log.debug("设备通道信息同步API调用，deviceId：" + deviceId);
 		}
 		Device device = deviceService.getDeviceByDeviceId(deviceId);
-		boolean status = deviceService.isSyncRunning(deviceId);
+		if (!userSetting.getServerId().equals(device.getServerId())) {
-		// 已存在则返回进度
+			return redisRpcService.devicesSync(device.getServerId(), deviceId);
-		if (deviceService.isSyncRunning(deviceId)) {
-			SyncStatus channelSyncStatus = deviceService.getChannelSyncStatus(deviceId);
-			WVPResult wvpResult = new WVPResult();
-			if (channelSyncStatus.getErrorMsg() != null) {
-				wvpResult.setCode(ErrorCode.ERROR100.getCode());
-				wvpResult.setMsg(channelSyncStatus.getErrorMsg());
-			}else if (channelSyncStatus.getTotal() == null || channelSyncStatus.getTotal() == 0){
-				wvpResult.setCode(ErrorCode.SUCCESS.getCode());
-				wvpResult.setMsg("等待通道信息...");
-			}else {
-				wvpResult.setCode(ErrorCode.SUCCESS.getCode());
-				wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
-				wvpResult.setData(channelSyncStatus);
-			}
-			return wvpResult;
 		}
-		deviceService.sync(device);
 
-		WVPResult<SyncStatus> wvpResult = new WVPResult<>();
+		return deviceService.devicesSync(device);
-		wvpResult.setCode(0);
+
-		wvpResult.setMsg("开始同步");
-		return wvpResult;
 	}
 
 	/**

src/main/java/com/genersoft/iot/vmp/gb28181/service/IDeviceService.java

@@ -1,9 +1,11 @@
 package com.genersoft.iot.vmp.gb28181.service;
 
 import com.genersoft.iot.vmp.common.CommonCallback;
-import com.genersoft.iot.vmp.gb28181.bean.*;
+import com.genersoft.iot.vmp.gb28181.bean.Device;
-import com.genersoft.iot.vmp.service.bean.ErrorCallback;
+import com.genersoft.iot.vmp.gb28181.bean.SipTransactionInfo;
+import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;
 import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
+import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 import com.github.pagehelper.PageInfo;
 
 import java.util.List;
@@ -165,4 +167,5 @@ public interface IDeviceService {
 
     void subscribeMobilePosition(int id, int cycle, int interval);
 
+    WVPResult<SyncStatus> devicesSync(Device device);
 }

src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceServiceImpl.java

@@ -7,7 +7,6 @@ import com.genersoft.iot.vmp.common.enums.ChannelDataType;
 import com.genersoft.iot.vmp.conf.DynamicTask;
 import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.conf.exception.ControllerException;
-import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcResponse;
 import com.genersoft.iot.vmp.gb28181.bean.*;
 import com.genersoft.iot.vmp.gb28181.dao.DeviceChannelMapper;
 import com.genersoft.iot.vmp.gb28181.dao.DeviceMapper;
@@ -24,8 +23,6 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.respons
 import com.genersoft.iot.vmp.media.bean.MediaServer;
 import com.genersoft.iot.vmp.media.service.IMediaServerService;
 import com.genersoft.iot.vmp.service.ISendRtpServerService;
-import com.genersoft.iot.vmp.service.bean.ErrorCallback;
-import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcPlayService;
 import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.utils.DateUtil;
@@ -42,13 +39,9 @@ import org.springframework.util.Assert;
 
 import javax.sip.InvalidArgumentException;
 import javax.sip.SipException;
-import javax.sip.message.Response;
 import java.text.ParseException;
 import java.time.Instant;
 import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -602,4 +595,31 @@ public class DeviceServiceImpl implements IDeviceService {
             redisCatchStorage.updateDevice(device);
         }
     }
+
+    @Override
+    public WVPResult<SyncStatus> devicesSync(Device device) {
+
+        // 已存在则返回进度
+        if (isSyncRunning(device.getDeviceId())) {
+            SyncStatus channelSyncStatus = getChannelSyncStatus(device.getDeviceId());
+            WVPResult wvpResult = new WVPResult();
+            if (channelSyncStatus.getErrorMsg() != null) {
+                wvpResult.setCode(ErrorCode.ERROR100.getCode());
+                wvpResult.setMsg(channelSyncStatus.getErrorMsg());
+            }else if (channelSyncStatus.getTotal() == null || channelSyncStatus.getTotal() == 0){
+                wvpResult.setCode(ErrorCode.SUCCESS.getCode());
+                wvpResult.setMsg("等待通道信息...");
+            }else {
+                wvpResult.setCode(ErrorCode.SUCCESS.getCode());
+                wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
+                wvpResult.setData(channelSyncStatus);
+            }
+            return wvpResult;
+        }
+        sync(device);
+        WVPResult<SyncStatus> wvpResult = new WVPResult<>();
+        wvpResult.setCode(0);
+        wvpResult.setMsg("开始同步");
+        return wvpResult;
+    }
 }

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

@@ -4,11 +4,10 @@ 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.bean.SyncStatus;
 import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
 import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 
-import java.util.List;
-
 public interface IRedisRpcService {
 
     SendRtpInfo getSendRtpItem(String callId);
@@ -35,4 +34,6 @@ public interface IRedisRpcService {
     boolean updatePlatform(String serverId, Platform platform);
 
     void catalogEventPublish(String serverId, CatalogEvent catalogEvent);
+
+    WVPResult<SyncStatus> devicesSync(String serverId, String deviceId);
 }

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

@@ -0,0 +1,69 @@
+package com.genersoft.iot.vmp.service.redisMsg.control;
+
+import com.alibaba.fastjson2.JSONObject;
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.conf.redis.RedisRpcConfig;
+import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcMessage;
+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.Device;
+import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;
+import com.genersoft.iot.vmp.gb28181.service.IDeviceService;
+import com.genersoft.iot.vmp.service.redisMsg.dto.RedisRpcController;
+import com.genersoft.iot.vmp.service.redisMsg.dto.RedisRpcMapping;
+import com.genersoft.iot.vmp.service.redisMsg.dto.RpcController;
+import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyService;
+import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
+import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+@RedisRpcController("device")
+public class RedisRpcDeviceController extends RpcController {
+
+    @Autowired
+    private UserSetting userSetting;
+
+    @Autowired
+    private RedisTemplate<Object, Object> redisTemplate;
+
+    @Autowired
+    private IDeviceService deviceService;
+
+    @Autowired
+    private IStreamProxyService streamProxyService;
+
+
+    private void sendResponse(RedisRpcResponse response){
+        log.info("[redis-rpc] >> {}", response);
+        response.setToId(userSetting.getServerId());
+        RedisRpcMessage message = new RedisRpcMessage();
+        message.setResponse(response);
+        redisTemplate.convertAndSend(RedisRpcConfig.REDIS_REQUEST_CHANNEL_KEY, message);
+    }
+
+    /**
+     * 通道同步
+     */
+    @RedisRpcMapping("devicesSync")
+    public RedisRpcResponse devicesSync(RedisRpcRequest request) {
+        String deviceId = request.getParam().toString();
+        Device device = deviceService.getDeviceByDeviceId(deviceId);
+
+        RedisRpcResponse response = request.getResponse();
+        if (device == null || !userSetting.getServerId().equals(device.getServerId())) {
+            response.setStatusCode(ErrorCode.ERROR400.getCode());
+            response.setBody("param error");
+            return response;
+        }
+        WVPResult<SyncStatus> result = deviceService.devicesSync(device);
+        response.setStatusCode(ErrorCode.SUCCESS.getCode());
+        response.setBody(JSONObject.toJSONString(result));
+        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.bean.SyncStatus;
 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;
@@ -246,4 +247,12 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
         }
         redisRpcConfig.request(request, 10, TimeUnit.MILLISECONDS);
     }
+
+    @Override
+    public WVPResult<SyncStatus> devicesSync(String serverId, String deviceId) {
+        RedisRpcRequest request = buildRequest("device/devicesSync", deviceId);
+        request.setToId(serverId);
+        RedisRpcResponse response = redisRpcConfig.request(request, 100, TimeUnit.MILLISECONDS);
+        return JSON.parseObject(response.getBody().toString(), WVPResult.class);
+    }
 }