From 9978d30aa1533c50bfaa48e44a1b0de4c157d10b Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期四, 11 四月 2024 09:36:48 +0800
Subject: [PATCH] Merge branch '2.7.0'
---
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java | 103 ++++++++++++++++++++++++++++-----------------------
1 files changed, 56 insertions(+), 47 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
index 0912f0b..7d5ba60 100755
--- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
@@ -1,7 +1,11 @@
package com.genersoft.iot.vmp.service.redisMsg;
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
+import com.genersoft.iot.vmp.media.zlm.dto.ChannelOnlineEvent;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -11,6 +15,7 @@
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
+import java.text.ParseException;
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -37,52 +42,56 @@
@Override
public void onMessage(Message message, byte[] bytes) {
-// boolean isEmpty = taskQueue.isEmpty();
-// taskQueue.offer(message);
-// if (isEmpty) {
-// taskExecutor.execute(() -> {
-// while (!taskQueue.isEmpty()) {
-// Message msg = taskQueue.poll();
-// try {
-// JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class);
-// if (steamMsgJson == null) {
-// logger.warn("[鏀跺埌redis 娴佸彉鍖朷娑堟伅瑙ｆ瀽澶辫触");
-// continue;
-// }
-// String serverId = steamMsgJson.getString("serverId");
-//
-// if (userSetting.getServerId().equals(serverId)) {
-// // 鑷繁鍙戦�佺殑娑堟伅蹇界暐鍗冲彲
-// continue;
-// }
-// logger.info("[鏀跺埌redis 娴佸彉鍖朷锛� {}", new String(message.getBody()));
-// String app = steamMsgJson.getString("app");
-// String stream = steamMsgJson.getString("stream");
-// boolean register = steamMsgJson.getBoolean("register");
-// String mediaServerId = steamMsgJson.getString("mediaServerId");
-// OnStreamChangedHookParam onStreamChangedHookParam = new OnStreamChangedHookParam();
-// onStreamChangedHookParam.setSeverId(serverId);
-// onStreamChangedHookParam.setApp(app);
-// onStreamChangedHookParam.setStream(stream);
-// onStreamChangedHookParam.setRegist(register);
-// onStreamChangedHookParam.setMediaServerId(mediaServerId);
-// onStreamChangedHookParam.setCreateStamp(System.currentTimeMillis()/1000);
-// onStreamChangedHookParam.setAliveSecond(0L);
-// onStreamChangedHookParam.setTotalReaderCount("0");
-// onStreamChangedHookParam.setOriginType(0);
-// onStreamChangedHookParam.setOriginTypeStr("0");
-// onStreamChangedHookParam.setOriginTypeStr("unknown");
-// if (register) {
-// zlmMediaListManager.addPush(onStreamChangedHookParam);
-// }else {
-// zlmMediaListManager.removeMedia(app, stream);
-// }
-// }catch (Exception e) {
-// logger.warn("[REDIS娑堟伅-娴佸彉鍖朷 鍙戠幇鏈鐞嗙殑寮傚父, \r\n{}", JSON.toJSONString(message));
-// logger.error("[REDIS娑堟伅-娴佸彉鍖朷 寮傚父鍐呭锛� ", e);
-// }
-// }
-// });
-// }
+ boolean isEmpty = taskQueue.isEmpty();
+ taskQueue.offer(message);
+ if (isEmpty) {
+ taskExecutor.execute(() -> {
+ while (!taskQueue.isEmpty()) {
+ Message msg = taskQueue.poll();
+ try {
+ JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class);
+ if (steamMsgJson == null) {
+ logger.warn("[鏀跺埌redis 娴佸彉鍖朷娑堟伅瑙ｆ瀽澶辫触");
+ continue;
+ }
+ String serverId = steamMsgJson.getString("serverId");
+
+ if (userSetting.getServerId().equals(serverId)) {
+ // 鑷繁鍙戦�佺殑娑堟伅蹇界暐鍗冲彲
+ continue;
+ }
+ logger.info("[鏀跺埌redis 娴佸彉鍖朷锛� {}", new String(message.getBody()));
+ String app = steamMsgJson.getString("app");
+ String stream = steamMsgJson.getString("stream");
+ boolean register = steamMsgJson.getBoolean("register");
+ String mediaServerId = steamMsgJson.getString("mediaServerId");
+ OnStreamChangedHookParam onStreamChangedHookParam = new OnStreamChangedHookParam();
+ onStreamChangedHookParam.setSeverId(serverId);
+ onStreamChangedHookParam.setApp(app);
+ onStreamChangedHookParam.setStream(stream);
+ onStreamChangedHookParam.setRegist(register);
+ onStreamChangedHookParam.setMediaServerId(mediaServerId);
+ onStreamChangedHookParam.setCreateStamp(System.currentTimeMillis()/1000);
+ onStreamChangedHookParam.setAliveSecond(0L);
+ onStreamChangedHookParam.setTotalReaderCount("0");
+ onStreamChangedHookParam.setOriginType(0);
+ onStreamChangedHookParam.setOriginTypeStr("0");
+ onStreamChangedHookParam.setOriginTypeStr("unknown");
+ ChannelOnlineEvent channelOnlineEventLister = zlmMediaListManager.getChannelOnlineEventLister(app, stream);
+ if ( channelOnlineEventLister != null) {
+ try {
+ channelOnlineEventLister.run(app, stream, serverId);;
+ } catch (ParseException e) {
+ logger.error("addPush: ", e);
+ }
+ zlmMediaListManager.removedChannelOnlineEventLister(app, stream);
+ }
+ }catch (Exception e) {
+ logger.warn("[REDIS娑堟伅-娴佸彉鍖朷 鍙戠幇鏈鐞嗙殑寮傚父, \r\n{}", JSON.toJSONString(message));
+ logger.error("[REDIS娑堟伅-娴佸彉鍖朷 寮傚父鍐呭锛� ", e);
+ }
+ }
+ });
+ }
}
}
--
Gitblit v1.8.0