package com.genersoft.iot.vmp.service.redisMsg; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.connection.MessageListener; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import java.util.concurrent.ConcurrentLinkedQueue; /** * 接收其他wvp发送流变化通知 * @author lin */ @Component public class RedisStreamMsgListener implements MessageListener { private final static Logger logger = LoggerFactory.getLogger(RedisStreamMsgListener.class); @Autowired private UserSetting userSetting; @Autowired private ZLMMediaListManager zlmMediaListManager; private ConcurrentLinkedQueue taskQueue = new ConcurrentLinkedQueue<>(); @Qualifier("taskExecutor") @Autowired private ThreadPoolTaskExecutor taskExecutor; @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); // } // } // }); // } } }