diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/controller/DeviceQuery.java b/src/main/java/com/genersoft/iot/vmp/gb28181/controller/DeviceQuery.java index 30720b363..fdcf5b4d4 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/controller/DeviceQuery.java +++ b/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 (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; + if (!userSetting.getServerId().equals(device.getServerId())) { + return redisRpcService.devicesSync(device.getServerId(), deviceId); } - deviceService.sync(device); - WVPResult wvpResult = new WVPResult<>(); - wvpResult.setCode(0); - wvpResult.setMsg("开始同步"); - return wvpResult; + return deviceService.devicesSync(device); + } /** diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/service/IDeviceService.java b/src/main/java/com/genersoft/iot/vmp/gb28181/service/IDeviceService.java index d18b9423e..f25c7d830 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/service/IDeviceService.java +++ b/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.service.bean.ErrorCallback; +import com.genersoft.iot.vmp.gb28181.bean.Device; +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 devicesSync(Device device); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceServiceImpl.java index a75abf148..7fc6d69cb 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceServiceImpl.java +++ b/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 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 wvpResult = new WVPResult<>(); + wvpResult.setCode(0); + wvpResult.setMsg("开始同步"); + return wvpResult; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java index ba372a184..580313760 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java +++ b/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 devicesSync(String serverId, String deviceId); } diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcDeviceController.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcDeviceController.java new file mode 100644 index 000000000..573056ee1 --- /dev/null +++ b/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 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 result = deviceService.devicesSync(device); + response.setStatusCode(ErrorCode.SUCCESS.getCode()); + response.setBody(JSONObject.toJSONString(result)); + return response; + } + +} diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java index 0744ef890..b02ef36df 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java +++ b/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 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); + } }