Merge pull request #1137 from xiaoQQya/develop

bugfix: 修复推流鉴权、报警推送等问题
2023-12-11 10:22:26 +08:00
parent 7a52db4e89 a30ff222ec
commit e73d00060e
13 changed files with 569 additions and 570 deletions
--- a/src/main/java/com/genersoft/iot/vmp/conf/security/WebSecurityConfig.java
+++ b/src/main/java/com/genersoft/iot/vmp/conf/security/WebSecurityConfig.java
@@ -1,12 +1,12 @@
 package com.genersoft.iot.vmp.conf.security;

 import com.genersoft.iot.vmp.conf.UserSetting;
-import org.springframework.core.annotation.Order;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.annotation.Order;
 import org.springframework.security.authentication.AuthenticationManager;
 import org.springframework.security.authentication.dao.DaoAuthenticationProvider;
 import org.springframework.security.config.annotation.authentication.builders.AuthenticationManagerBuilder;
@@ -28,6 +28,7 @@ import java.util.Arrays;

 /**
 * 配置Spring Security
+ *
 * @author lin
 */
@Configuration
@@ -75,6 +76,7 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {
            matchers.add("/js/**");
            matchers.add("/api/device/query/snap/**");
            matchers.add("/record_proxy/*/**");
+            matchers.add("/api/emit");
            matchers.addAll(userSetting.getInterfaceAuthenticationExcludes());
            // 可以直接访问的静态数据
            web.ignoring().antMatchers(matchers.toArray(new String[0]));
@@ -83,6 +85,7 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {

    /**
     * 配置认证方式
+     *
     * @param auth
     * @throws Exception
     */
@@ -111,7 +114,7 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {
                .authorizeRequests()
                .requestMatchers(CorsUtils::isPreFlightRequest).permitAll()
                .antMatchers(userSetting.getInterfaceAuthenticationExcludes().toArray(new String[0])).permitAll()
-                .antMatchers("/api/user/login","/index/hook/**","/zlm_Proxy/FhTuMYqB2HeCuNOb/record/t/1/2023-03-25/16:35:07-16:35:16-9353.mp4").permitAll()
+                .antMatchers("/api/user/login", "/index/hook/**").permitAll()
                .anyRequest().authenticated()
                // 异常处理器
                .and()
@@ -124,7 +127,7 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {

    }

-    CorsConfigurationSource configurationSource(){
+    CorsConfigurationSource configurationSource() {
        // 配置跨域
        CorsConfiguration corsConfiguration = new CorsConfiguration();
        corsConfiguration.setAllowedHeaders(Arrays.asList("*"));
@@ -135,7 +138,7 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {
        corsConfiguration.setExposedHeaders(Arrays.asList(JwtUtils.getHeader()));

        UrlBasedCorsConfigurationSource url = new UrlBasedCorsConfigurationSource();
-        url.registerCorsConfiguration("/**",corsConfiguration);
+        url.registerCorsConfiguration("/**", corsConfiguration);
        return url;
    }

--- a/src/main/java/com/genersoft/iot/vmp/gb28181/event/alarm/AlarmEventListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/event/alarm/AlarmEventListener.java
@@ -1,55 +1,68 @@
 package com.genersoft.iot.vmp.gb28181.event.alarm;

-import org.springframework.context.ApplicationListener;
-import org.springframework.stereotype.Component;
-import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
-import java.io.IOException;
-import java.util.Hashtable;
-import java.util.Iterator;
-import java.util.Map;
-
+import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationListener;
+import org.springframework.stereotype.Component;
+
+import java.io.PrintWriter;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;

 /**
- * @description: 报警事件监听
- * @author: lawrencehj
- * @data: 2021-01-20
+ * 报警事件监听器.
+ *
+ * @author lawrencehj
+ * @author <a href="mailto:xiaoQQya@126.com">xiaoQQya</a>
+ * @since 2021/01/20
 */
-
@Component
 public class AlarmEventListener implements ApplicationListener<AlarmEvent> {

-    private final static Logger logger = LoggerFactory.getLogger(AlarmEventListener.class);
+    private static final Logger logger = LoggerFactory.getLogger(AlarmEventListener.class);

-    private static Map<String, SseEmitter> sseEmitters = new Hashtable<>();
+    private static final Map<String, PrintWriter> SSE_CACHE = new ConcurrentHashMap<>();

-    public void addSseEmitters(String browserId, SseEmitter sseEmitter) {
-        sseEmitters.put(browserId, sseEmitter);
+    public void addSseEmitter(String browserId, PrintWriter writer) {
+        SSE_CACHE.put(browserId, writer);
+        logger.info("SSE 在线数量: {}", SSE_CACHE.size());
+    }
+
+    public void removeSseEmitter(String browserId, PrintWriter writer) {
+        SSE_CACHE.remove(browserId, writer);
+        logger.info("SSE 在线数量: {}", SSE_CACHE.size());
    }

    @Override
-    public void onApplicationEvent(AlarmEvent event) {
+    public void onApplicationEvent(@NotNull AlarmEvent event) {
        if (logger.isDebugEnabled()) {
-            logger.debug("设备报警事件触发，deviceId：" + event.getAlarmInfo().getDeviceId() + ", "
-                    + event.getAlarmInfo().getAlarmDescription());
+            logger.debug("设备报警事件触发, deviceId: {}, {}", event.getAlarmInfo().getDeviceId(), event.getAlarmInfo().getAlarmDescription());
        }
-        String msg = "<strong>设备编码：</strong> <i>" + event.getAlarmInfo().getDeviceId() + "</i>"
-                    + "<br><strong>报警描述：</strong> <i>" + event.getAlarmInfo().getAlarmDescription() + "</i>"
-                    + "<br><strong>报警时间：</strong> <i>" + event.getAlarmInfo().getAlarmTime() + "</i>"
-                    + "<br><strong>报警位置：</strong> <i>" + event.getAlarmInfo().getLongitude() + "</i>"
-                    + ", <i>" + event.getAlarmInfo().getLatitude() + "</i>";

-        for (Iterator<Map.Entry<String, SseEmitter>> it = sseEmitters.entrySet().iterator(); it.hasNext();) {
-            Map.Entry<String, SseEmitter> emitter = it.next();
-            logger.info("推送到SSE连接，浏览器ID: " + emitter.getKey());
+        String msg = "<strong>设备编号：</strong> <i>" + event.getAlarmInfo().getDeviceId() + "</i>"
+                + "<br><strong>通道编号：</strong> <i>" + event.getAlarmInfo().getChannelId() + "</i>"
+                + "<br><strong>报警描述：</strong> <i>" + event.getAlarmInfo().getAlarmDescription() + "</i>"
+                + "<br><strong>报警时间：</strong> <i>" + event.getAlarmInfo().getAlarmTime() + "</i>";
+
+        for (Iterator<Map.Entry<String, PrintWriter>> it = SSE_CACHE.entrySet().iterator(); it.hasNext(); ) {
+            Map.Entry<String, PrintWriter> response = it.next();
+            logger.info("推送到 SSE 连接, 浏览器 ID: {}", response.getKey());
            try {
-                emitter.getValue().send(msg);
-            } catch (IOException | IllegalStateException e) {
-                if (logger.isDebugEnabled()) {
-                    logger.debug("SSE连接已关闭");
+                PrintWriter writer = response.getValue();
+
+                if (writer.checkError()) {
+                    it.remove();
+                    continue;
                }
-                // 移除已关闭的连接
+
+                String sseMsg = "event:message\n" +
+                        "data:" + msg + "\n" +
+                        "\n";
+                writer.write(sseMsg);
+                writer.flush();
+            } catch (Exception e) {
                it.remove();
            }
        }
--- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
@@ -244,7 +244,6 @@ public class ZLMHttpHookListener {


        HookResultForOnPublish result = HookResultForOnPublish.SUCCESS();
-        result.setEnable_audio(true);
        taskExecutor.execute(() -> {
            ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_publish, json);
            if (subscribe != null) {
@@ -262,29 +261,36 @@ public class ZLMHttpHookListener {
        } else {
            result.setEnable_mp4(userSetting.isRecordPushLive());
        }
-        // 替换流地址
-        if ("rtp".equals(param.getApp()) && !mediaInfo.isRtpEnable()) {
-            String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));;
+
+        // 国标流
+        if ("rtp".equals(param.getApp())) {
+            String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));
            InviteInfo inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);
-            if (inviteInfo != null) {
+
+            // 单端口模式下修改流 ID
+            if (!mediaInfo.isRtpEnable() && inviteInfo != null) {
                result.setStream_replace(inviteInfo.getStream());
                logger.info("[ZLM HOOK]推流鉴权 stream: {} 替换为 {}", param.getStream(), inviteInfo.getStream());
            }
-        }
-        List<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream());
-        if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {
-            String deviceId = ssrcTransactionForAll.get(0).getDeviceId();
-            String channelId = ssrcTransactionForAll.get(0).getChannelId();
-            DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
-            if (deviceChannel != null) {
-                result.setEnable_audio(deviceChannel.isHasAudio());
-            }
-            // 如果是录像下载就设置视频间隔十秒
-            if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.DOWNLOAD) {
-                result.setMp4_max_second(10);
-                result.setEnable_mp4(true);
+
+            // 设置音频信息及录制信息
+            List<SsrcTransaction> ssrcTransactionForAll = (inviteInfo == null ? null :
+                    sessionManager.getSsrcTransactionForAll(inviteInfo.getDeviceId(), inviteInfo.getChannelId(), null, null));
+            if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {
+                String deviceId = ssrcTransactionForAll.get(0).getDeviceId();
+                String channelId = ssrcTransactionForAll.get(0).getChannelId();
+                DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
+                if (deviceChannel != null) {
+                    result.setEnable_audio(deviceChannel.isHasAudio());
+                }
+                // 如果是录像下载就设置视频间隔十秒
+                if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.DOWNLOAD) {
+                    result.setMp4_max_second(10);
+                    result.setEnable_mp4(true);
+                }
            }
        }
+
        if (mediaInfo.getRecordAssistPort() > 0 && userSetting.getRecordPath() == null) {
            logger.info("推流时发现尚未设置录像路径，从assist服务中读取");
            JSONObject info = assistRESTfulUtils.getInfo(mediaInfo, null);
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
@@ -569,7 +569,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
        Map<String, Object> param = new HashMap<>();
        param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline
        if (mediaServerItem.getRtspPort() != 0) {
-            param.put("ffmpeg.snap", "%s -rtsp_transport tcp -i %s -y -f mjpeg -t 0.001 %s");
+            param.put("ffmpeg.snap", "%s -rtsp_transport tcp -i %s -y -f mjpeg -frames:v 1 %s");
        }
        param.put("hook.enable","1");
        param.put("hook.on_flow_report","");
--- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/SseController/SseController.java
+++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/SseController/SseController.java
@@ -1,37 +0,0 @@
-package com.genersoft.iot.vmp.vmanager.gb28181.SseController;
-
-import com.genersoft.iot.vmp.gb28181.event.alarm.AlarmEventListener;
-
-import io.swagger.v3.oas.annotations.tags.Tag;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Controller;
-import org.springframework.web.bind.annotation.CrossOrigin;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RequestParam;
-import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
-
-/**
- * @description: SSE推送
- * @author: lawrencehj
- * @data: 2021-01-20
- */
-@Tag(name  = "SSE推送")
-
-@Controller
-@RequestMapping("/api")
-public class SseController {
-    @Autowired
-    AlarmEventListener alarmEventListener;
-
-    @GetMapping("/emit")
-    public SseEmitter emit(@RequestParam String browserId) {
-        final SseEmitter sseEmitter = new SseEmitter(0L);
-        try {
-            alarmEventListener.addSseEmitters(browserId, sseEmitter);
-        }catch (Exception e){
-            sseEmitter.completeWithError(e);
-        }
-        return sseEmitter;
-    }
-}
--- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/sse/SseController.java
+++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/sse/SseController.java
@@ -0,0 +1,55 @@
+package com.genersoft.iot.vmp.vmanager.gb28181.sse;
+
+import com.genersoft.iot.vmp.gb28181.event.alarm.AlarmEventListener;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.annotation.Resource;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.io.PrintWriter;
+
+
+/**
+ * SSE 推送.
+ *
+ * @author lawrencehj
+ * @author <a href="mailto:xiaoQQya@126.com">xiaoQQya</a>
+ * @since 2021/01/20
+ */
+@Tag(name = "SSE 推送")
+@RestController
+@RequestMapping("/api")
+public class SseController {
+
+    @Resource
+    private AlarmEventListener alarmEventListener;
+
+    /**
+     * SSE 推送.
+     *
+     * @param response  响应
+     * @param browserId 浏览器ID
+     * @throws IOException IOEXCEPTION
+     * @author <a href="mailto:xiaoQQya@126.com">xiaoQQya</a>
+     * @since 2023/11/06
+     */
+    @GetMapping("/emit")
+    public void emit(HttpServletResponse response, @RequestParam String browserId) throws IOException, InterruptedException {
+        response.setContentType("text/event-stream");
+        response.setCharacterEncoding("utf-8");
+
+        PrintWriter writer = response.getWriter();
+        alarmEventListener.addSseEmitter(browserId, writer);
+
+        while (!writer.checkError()) {
+            Thread.sleep(1000);
+            writer.write(":keep alive\n\n");
+            writer.flush();
+        }
+        alarmEventListener.removeSseEmitter(browserId, writer);
+    }
+}