Pre Merge pull request !34 from WangXuewen/master

README.md

@@ -121,7 +121,8 @@ https://gitee.com/pan648540858/wvp-GB28181-pro.git
 - [X] 支持录制计划, 根据设定的时间对通道进行录制. 暂不支持将录制的内容转发到国标上级
 - [X] 支持国标信令集群
 - [X] 新增支持部标808和部标1078，大量新特性不一一列表了。支持作为网关被国标上级调用部标设备
-
+- [X] 支持电子地图。支持展示通道位置，支持在地图上修改通道位置。支持了数据分层抽稀数据能力，百万级数据也可以轻松展示。提供标准的矢量瓦片图层，常见地图引擎都可以直接展示。
+- [X] 借用zlm闭源版本新能力，可以支持录像保存至s3存储，支持minio。
 
 # 闭源内容
 - [X] 国标增强版: 支持国标28181-2022协议，支持巡航轨迹查询，PTZ精准控制，存储卡格式化，设备软件升级，OSD配置，h265+aac，支持辅码流，录像倒放等。
@@ -146,12 +147,9 @@ https://gitee.com/pan648540858/wvp-GB28181-pro.git
     - 预置位控制等，
     - 可免费定制支持语音对讲、录像回放和抓拍图像。
   - [X] 支持按权限分配可以使用的通道
-  - [X] 支持电子地图。支持展示通道位置，支持在地图上修改通道位置。可扩展接入高德地图API，支持搜索位置，附近设备。
   - [X] 支持表格导出
   - [X] 拉流代理支持按照品牌拼接url。
   - [X] 播放鉴权，更加安全。
-  - [X] 此版本后续开发功能支持直接更新提供，无需二次付费。
-  - [X] 提供源码不限制部署次数和支持路数。
 
 
 # 授权协议

src/main/java/com/genersoft/iot/vmp/conf/redis/RedisRpcConfig.java

@@ -18,7 +18,6 @@ import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Component;
 
-import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.HashMap;
 import java.util.Map;
@@ -143,6 +142,10 @@ public class RedisRpcConfig implements MessageListener {
                 }
             }else {
                 if (method == null) {
+                    // 回复404结果
+                    RedisRpcResponse response = request.getResponse();
+                    response.setStatusCode(ErrorCode.ERROR404.getCode());
+                    sendResponse(response);
                     return;
                 }
                 RedisRpcResponse response = (RedisRpcResponse)method.invoke(controller, request);
@@ -150,8 +153,11 @@ public class RedisRpcConfig implements MessageListener {
                     sendResponse(response);
                 }
             }
-        }catch (InvocationTargetException | IllegalAccessException e) {
+        }catch (Exception e) {
             log.error("[redis-rpc ] 处理请求失败 ", e);
+            RedisRpcResponse response = request.getResponse();
+            response.setStatusCode(ErrorCode.ERROR100.getCode());
+            sendResponse(response);
         }
     }
 

src/main/java/com/genersoft/iot/vmp/gb28181/bean/VectorTileSource.java

@@ -2,15 +2,18 @@ package com.genersoft.iot.vmp.gb28181.bean;
 
 import lombok.Getter;
 import lombok.Setter;
+import org.jetbrains.annotations.NotNull;
 
 import java.util.ArrayList;
-import java.util.Map;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
 
 @Getter
 @Setter
-public class VectorTileSource {
+public class VectorTileSource implements Delayed {
 
     /**
      * 抽稀的图层数据
@@ -22,6 +25,8 @@ public class VectorTileSource {
      */
     private List<CommonGBChannel> channelList = new ArrayList<>();
 
+    private String id;
+
     /**
      * 创建时间， 大于6小时后删除
      */
@@ -30,4 +35,14 @@ public class VectorTileSource {
     public VectorTileSource() {
         this.time = System.currentTimeMillis();
     }
+
+    @Override
+    public long getDelay(@NotNull TimeUnit unit) {
+        return unit.convert(time + 6 * 60 * 60 * 1000 - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public int compareTo(@NotNull Delayed o) {
+        return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
+    }
 }

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

@@ -8,7 +8,7 @@ import com.genersoft.iot.vmp.gb28181.bean.*;
 import com.genersoft.iot.vmp.gb28181.controller.bean.*;
 import com.genersoft.iot.vmp.gb28181.service.IGbChannelPlayService;
 import com.genersoft.iot.vmp.gb28181.service.IGbChannelService;
-import com.genersoft.iot.vmp.gb28181.utils.VectorTileUtils;
+import com.genersoft.iot.vmp.gb28181.utils.VectorTileCatch;
 import com.genersoft.iot.vmp.service.bean.ErrorCallback;
 import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
 import com.genersoft.iot.vmp.utils.DateUtil;
@@ -59,6 +59,9 @@ public class ChannelController {
     @Autowired
     private UserSetting userSetting;
 
+    @Autowired
+    private VectorTileCatch vectorTileCatch;
+
 
     @Operation(summary = "查询通道信息", security = @SecurityRequirement(name = JwtUtils.HEADER))
     @Parameter(name = "id", description = "通道的数据库自增Id", required = true)
@@ -525,7 +528,7 @@ public class ChannelController {
     @Parameter(name = "id", description = "抽稀ID", required = true)
     @GetMapping("/map/thin/clear")
     public void clearThin(String id){
-        VectorTileUtils.INSTANCE.remove(id);
+        vectorTileCatch.remove(id);
     }
 
     @Operation(summary = "保存的抽稀结果", security = @SecurityRequirement(name = JwtUtils.HEADER))
@@ -574,7 +577,7 @@ public class ChannelController {
             thinId = "DEFAULT";
         }
         String catchKey = z + "_" + x + "_" + y + "_" + geoCoordSys.toUpperCase();
-        byte[] mvt = VectorTileUtils.INSTANCE.getVectorTile(thinId, catchKey);
+        byte[] mvt = vectorTileCatch.getVectorTile(thinId, catchKey);
         HttpHeaders headers = new HttpHeaders();
         headers.setContentType(MediaType.parseMediaType("application/x-protobuf"));
         if (mvt == null) {

src/main/java/com/genersoft/iot/vmp/gb28181/dao/provider/ChannelProvider.java

@@ -925,7 +925,8 @@ public class ChannelProvider {
     public String queryAllWithPosition(Map<String, Object> params ){
         return BASE_SQL + " where channel_type = 0 " +
                 " AND coalesce(gb_longitude, longitude) > 0" +
-                " AND coalesce(gb_latitude,  latitude) > 0";
+                " AND coalesce(gb_latitude,  latitude) > 0 " +
+                " ORDER BY map_level";
     }
 
     public String queryListInExtent(Map<String, Object> params ){

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

@@ -17,7 +17,7 @@ import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
 import com.genersoft.iot.vmp.gb28181.event.channel.ChannelEvent;
 import com.genersoft.iot.vmp.gb28181.service.IGbChannelService;
 import com.genersoft.iot.vmp.gb28181.service.IPlatformChannelService;
-import com.genersoft.iot.vmp.gb28181.utils.VectorTileUtils;
+import com.genersoft.iot.vmp.gb28181.utils.VectorTileCatch;
 import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
 import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
 import com.genersoft.iot.vmp.utils.Coordtransform;
@@ -34,6 +34,7 @@ import org.locationtech.jts.geom.GeometryFactory;
 import org.locationtech.jts.geom.Point;
 import org.springframework.beans.BeanWrapperImpl;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -42,11 +43,12 @@ import org.springframework.util.ObjectUtils;
 
 import java.time.Duration;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
 
 @Slf4j
 @Service
-public class GbChannelServiceImpl implements IGbChannelService {
+public class GbChannelServiceImpl implements IGbChannelService, CommandLineRunner {
 
     @Autowired
     private EventPublisher eventPublisher;
@@ -72,8 +74,38 @@ public class GbChannelServiceImpl implements IGbChannelService {
     @Autowired
     private RedisTemplate<Object, Object> redisTemplate;
 
+    @Autowired
+    private VectorTileCatch vectorTileCatch;
+
     private final GeometryFactory geometryFactory = new GeometryFactory();
 
+
+    @Override
+    public void run(String... args) throws Exception {
+        // 启动时重新发布抽稀图层
+        List<CommonGBChannel> channelList = commonGBChannelMapper.queryAllWithPosition();
+        Map<Integer, List<CommonGBChannel>> zoomCameraMap = new ConcurrentHashMap<>();
+
+        channelList.stream().forEach(commonGBChannel ->  {
+            if (commonGBChannel.getMapLevel() == null) {
+                return;
+            }
+            List<CommonGBChannel> channelListForZoom = zoomCameraMap.computeIfAbsent(commonGBChannel.getMapLevel(), k -> new ArrayList<>());
+            channelListForZoom.add(commonGBChannel);
+        });
+
+        String id = "DEFAULT";
+        List<CommonGBChannel> beforeData = new ArrayList<>();
+        for (Integer zoom : zoomCameraMap.keySet()) {
+            beforeData.addAll(zoomCameraMap.get(zoom));
+            log.info("[抽稀-发布mvt矢量瓦片] ID：{}，当前层级： {}, ", id, zoom);
+            // 按照 z/x/y 数据组织数据， 矢量数据暂时保存在内存中
+            // 按照范围生成 x y范围，
+            saveTile(id, zoom, "WGS84", beforeData);
+            saveTile(id, zoom, "GCJ02", beforeData);
+        }
+    }
+
     @Override
     public CommonGBChannel queryByDeviceId(String gbDeviceId) {
         List<CommonGBChannel> commonGBChannels = commonGBChannelMapper.queryByDeviceId(gbDeviceId);
@@ -1069,25 +1101,25 @@ public class GbChannelServiceImpl implements IGbChannelService {
                     saveProcess(id, process.get(), "抽稀图层： " + zoom);
                 }
 
-                // 抽稀完成, 对数据生成mvt矢量瓦片
+                // 抽稀完成, 对数据发布mvt矢量瓦片
                 List<CommonGBChannel> beforeData = new ArrayList<>();
-                for (Integer key : zoomCameraMap.keySet()) {
-                    beforeData.addAll(zoomCameraMap.get(key));
-                    log.info("[抽稀-生成mvt矢量瓦片] ID：{}，当前层级： {}", id, key);
+                for (Integer zoom : zoomCameraMap.keySet()) {
+                    beforeData.addAll(zoomCameraMap.get(zoom));
+                    log.info("[抽稀-发布mvt矢量瓦片] ID：{}，当前层级： {}", id, zoom);
                     // 按照 z/x/y 数据组织数据， 矢量数据暂时保存在内存中
                     // 按照范围生成 x y范围，
-                    saveTile(id, key, "WGS84", beforeData);
-                    saveTile(id, key, "GCJ02", beforeData);
+                    saveTile(id, zoom, "WGS84", beforeData);
+                    saveTile(id, zoom, "GCJ02", beforeData);
                     process.updateAndGet(v -> (v + 0.5 / zoomParam.size()));
-                    saveProcess(id, process.get(), "发布矢量瓦片： " + key);
+                    saveProcess(id, process.get(), "发布矢量瓦片： " + zoom);
                 }
                 // 记录原始数据，未保存做准备
-                VectorTileUtils.INSTANCE.addSource(id, new ArrayList<>(useCameraMap.values()));
+                vectorTileCatch.addSource(id, new ArrayList<>(useCameraMap.values()));
 
                 log.info("[抽稀完成] ID：{}, 耗时： {}ms", id, (System.currentTimeMillis() - time));
                 saveProcess(id, 1, "抽稀完成");
             } catch (Exception e) {
-
+                log.info("[抽稀] 失败 ID：{}", id, e);
             }
 
         }, 1);
@@ -1121,7 +1153,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
             encoder.addFeature("points", beanMap, pointGeom);
         });
         encoderMap.forEach((key, encoder) -> {
-            VectorTileUtils.INSTANCE.addVectorTile(id, key, encoder.encode());
+            vectorTileCatch.addVectorTile(id, key, encoder.encode());
         });
     }
 
@@ -1141,7 +1173,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
     @Transactional
     public void saveThin(String id) {
         commonGBChannelMapper.resetLevel();
-        List<CommonGBChannel> channelList = VectorTileUtils.INSTANCE.getChannelList(id);
+        List<CommonGBChannel> channelList = vectorTileCatch.getChannelList(id);
         if (channelList != null && !channelList.isEmpty()) {
             int limitCount = 1000;
             if (channelList.size() > limitCount) {
@@ -1156,6 +1188,6 @@ public class GbChannelServiceImpl implements IGbChannelService {
                 commonGBChannelMapper.saveLevel(channelList);
             }
         }
-        VectorTileUtils.INSTANCE.save(id);
+        vectorTileCatch.save(id);
     }
 }

src/main/java/com/genersoft/iot/vmp/gb28181/utils/VectorTileCatch.java

@@ -2,21 +2,30 @@ package com.genersoft.iot.vmp.gb28181.utils;
 
 import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
 import com.genersoft.iot.vmp.gb28181.bean.VectorTileSource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
 import org.springframework.util.ConcurrentReferenceHashMap;
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.TimeUnit;
 
-public enum VectorTileUtils {
-    INSTANCE;
+@Slf4j
+@Component
+public class VectorTileCatch {
 
-    private Map<String, VectorTileSource> vectorTileMap = new ConcurrentReferenceHashMap<>();
+    private final Map<String, VectorTileSource> vectorTileMap = new ConcurrentReferenceHashMap<>();
+    private final DelayQueue<VectorTileSource> delayQueue = new DelayQueue<>();
 
     public void addVectorTile(String id, String key, byte[] content) {
         VectorTileSource vectorTileSource = vectorTileMap.get(id);
         if (vectorTileSource == null) {
             vectorTileSource = new VectorTileSource();
+            vectorTileSource.setId(id);
             vectorTileMap.put(id, vectorTileSource);
+            delayQueue.offer(vectorTileSource);
         }
 
         vectorTileSource.getVectorTileMap().put(key, content);
@@ -33,13 +42,19 @@ public enum VectorTileUtils {
         VectorTileSource vectorTileSource = vectorTileMap.get(id);
         if (vectorTileSource == null) {
             vectorTileSource = new VectorTileSource();
+            vectorTileSource.setId(id);
             vectorTileMap.put(id, vectorTileSource);
+            delayQueue.offer(vectorTileSource);
         }
         vectorTileMap.get(id).getChannelList().addAll(channelList);
     }
 
 
     public void remove(String id) {
+        VectorTileSource vectorTileSource = vectorTileMap.get(id);
+        if (vectorTileSource != null) {
+            delayQueue.remove(vectorTileSource);
+        }
         vectorTileMap.remove(id);
     }
 
@@ -59,8 +74,22 @@ public enum VectorTileUtils {
             return;
         }
         vectorTileMap.remove(id);
+        delayQueue.remove(vectorTileSource);
         vectorTileMap.put("DEFAULT", vectorTileSource);
     }
 
 
+    // 缓存数据过期检查
+    @Scheduled(fixedDelay = 30, timeUnit = TimeUnit.MINUTES)
+    public void expirationCheck(){
+        while (!delayQueue.isEmpty()) {
+            try {
+                VectorTileSource vectorTileSource = delayQueue.take();
+                vectorTileMap.remove(vectorTileSource.getId());
+            } catch (InterruptedException e) {
+                log.error("[清理过期的抽稀数据] ", e);
+            }
+        }
+    }
+
 }

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

@@ -189,13 +189,19 @@ public class RedisRpcStreamPushController extends RpcController {
             response.setBody("param error");
             return response;
         }
-        streamPushPlayService.start(id, (code, msg, data) -> {
-            if (code == ErrorCode.SUCCESS.getCode()) {
-                response.setStatusCode(ErrorCode.SUCCESS.getCode());
-                response.setBody(JSONObject.toJSONString(data));
-                sendResponse(response);
-            }
-        }, null, null);
+        try {
+            streamPushPlayService.start(id, (code, msg, data) -> {
+                if (code == ErrorCode.SUCCESS.getCode()) {
+                    response.setStatusCode(ErrorCode.SUCCESS.getCode());
+                    response.setBody(JSONObject.toJSONString(data));
+                    sendResponse(response);
+                }
+            }, null, null);
+        }catch (IllegalArgumentException e) {
+            response.setStatusCode(ErrorCode.ERROR100.getCode());
+            response.setBody(e.getMessage());
+            sendResponse(response);
+        }
         return null;
     }
 

src/main/java/com/genersoft/iot/vmp/web/custom/CameraChannelController.java

@@ -14,7 +14,6 @@ import com.genersoft.iot.vmp.service.bean.CloudRecordItem;
 import com.genersoft.iot.vmp.service.bean.ErrorCallback;
 import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
-import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyService;
 import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
 import com.genersoft.iot.vmp.streamPush.service.IStreamPushPlayService;
 import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;