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 0912f0b7..7d5ba609 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.data.redis.connection.MessageListener; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; +import java.text.ParseException; import java.util.concurrent.ConcurrentLinkedQueue; @@ -37,52 +42,56 @@ public class RedisStreamMsgListener implements MessageListener { @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); + } + } + }); + } } }