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<Message> 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);
|
// }
|
// }
|
// });
|
// }
|
}
|
}
|
