From 89a9ab4534f10a224f70e546db838423e84a1965 Mon Sep 17 00:00:00 2001
From: 64850858 <648540858@qq.com>
Date: 星期五, 16 七月 2021 16:34:51 +0800
Subject: [PATCH] 添加zlm集群支持
---
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java | 76 ++++++++++++++++++++++++++++----------
1 files changed, 56 insertions(+), 20 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
index 172d980..85a4f48 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
@@ -14,6 +14,9 @@
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
+import com.genersoft.iot.vmp.media.zlm.dto.IMediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
@@ -60,6 +63,9 @@
private IMediaService mediaService;
@Autowired
+ private IMediaServerService mediaServerService;
+
+ @Autowired
private VideoStreamSessionManager streamSession;
@Autowired
@@ -67,8 +73,18 @@
@Override
- public PlayResult play(String deviceId, String channelId, ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent) {
+ public PlayResult play(IMediaServerItem mediaServerItem, String deviceId, String channelId, ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent) {
PlayResult playResult = new PlayResult();
+ if (mediaServerItem == null) {
+ RequestMessage msg = new RequestMessage();
+ msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + playResult.getUuid());
+ WVPResult wvpResult = new WVPResult();
+ wvpResult.setCode(-1);
+ wvpResult.setMsg("鏈壘鍒板彲鐢ㄧ殑zlm");
+ msg.setData(wvpResult);
+ resultHolder.invokeResult(msg);
+ return playResult;
+ }
Device device = storager.queryVideoDevice(deviceId);
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
playResult.setDevice(device);
@@ -82,7 +98,7 @@
result.onTimeout(()->{
logger.warn(String.format("璁惧鐐规挱瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId));
// 閲婃斁rtpserver
- cmder.closeRTPServer(playResult.getDevice(), channelId);
+ mediaServerService.closeRTPServer(playResult.getDevice(), channelId);
RequestMessage msg = new RequestMessage();
msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + playResult.getUuid());
WVPResult wvpResult = new WVPResult();
@@ -115,9 +131,10 @@
WVPResult wvpResult = (WVPResult)responseEntity.getBody();
if (wvpResult.getCode() == 0) {
StreamInfo streamInfoForSuccess = (StreamInfo)wvpResult.getData();
+ IMediaServerItem mediaInfo = mediaServerService.getOne(streamInfoForSuccess.getMediaServerId());
String streamUrl = streamInfoForSuccess.getFmp4();
// 璇锋眰鎴浘
- zlmresTfulUtils.getSnap(streamUrl, 15, 1, path, fileName);
+ zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName);
}
}
} catch (FileNotFoundException e) {
@@ -126,17 +143,17 @@
});
if (streamInfo == null) {
// 鍙戦�佺偣鎾秷鎭�
- cmder.playStreamCmd(device, channelId, (JSONObject response) -> {
+ cmder.playStreamCmd(mediaServerItem, device, channelId, (IMediaServerItem mediaServerItemInUse, JSONObject response) -> {
logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString());
- onPublishHandlerForPlay(response, deviceId, channelId, uuid.toString());
+ onPublishHandlerForPlay(mediaServerItemInUse, response, deviceId, channelId, uuid.toString());
if (hookEvent != null) {
- hookEvent.response(response);
+ hookEvent.response(mediaServerItem, response);
}
- }, event -> {
+ }, (event) -> {
RequestMessage msg = new RequestMessage();
msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
Response response = event.getResponse();
- cmder.closeRTPServer(playResult.getDevice(), channelId);
+ mediaServerService.closeRTPServer(playResult.getDevice(), channelId);
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
wvpResult.setMsg(String.format("鐐规挱澶辫触锛� 閿欒鐮侊細 %s, %s", response.getStatusCode(), response.getReasonPhrase()));
@@ -158,7 +175,10 @@
resultHolder.invokeResult(msg);
return playResult;
}
- JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(streamId);
+ String mediaServerId = streamInfo.getMediaServerId();
+ IMediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
+
+ JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId);
if (rtpInfo != null && rtpInfo.getBoolean("exist")) {
RequestMessage msg = new RequestMessage();
msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
@@ -171,16 +191,16 @@
resultHolder.invokeResult(msg);
if (hookEvent != null) {
- hookEvent.response(JSONObject.parseObject(JSON.toJSONString(streamInfo)));
+ hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
}
} else {
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
- cmder.playStreamCmd(device, channelId, (JSONObject response) -> {
+ cmder.playStreamCmd(mediaServerItem, device, channelId, (IMediaServerItem mediaServerItemInuse, JSONObject response) -> {
logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString());
- onPublishHandlerForPlay(response, deviceId, channelId, uuid.toString());
- }, event -> {
- cmder.closeRTPServer(playResult.getDevice(), channelId);
+ onPublishHandlerForPlay(mediaServerItemInuse, response, deviceId, channelId, uuid.toString());
+ }, (event) -> {
+ mediaServerService.closeRTPServer(playResult.getDevice(), channelId);
RequestMessage msg = new RequestMessage();
msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
Response response = event.getResponse();
@@ -198,10 +218,10 @@
}
@Override
- public void onPublishHandlerForPlay(JSONObject resonse, String deviceId, String channelId, String uuid) {
+ public void onPublishHandlerForPlay(IMediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid) {
RequestMessage msg = new RequestMessage();
msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
- StreamInfo streamInfo = onPublishHandler(resonse, deviceId, channelId, uuid);
+ StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId, uuid);
if (streamInfo != null) {
DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
if (deviceChannel != null) {
@@ -234,10 +254,26 @@
}
@Override
- public void onPublishHandlerForPlayBack(JSONObject resonse, String deviceId, String channelId, String uuid) {
+ public IMediaServerItem getNewMediaServerItem(Device device) {
+ if (device == null) return null;
+ String mediaServerId = device.getMediaServerId();
+ IMediaServerItem mediaServerItem = null;
+ if (mediaServerId == null) {
+ mediaServerItem = mediaServerService.getMediaServerForMinimumLoad();
+ }else {
+ mediaServerItem = mediaServerService.getOne(mediaServerId);
+ }
+ if (mediaServerItem == null) {
+ logger.warn("鐐规挱鏃舵湭鎵惧埌鍙娇鐢ㄧ殑ZLM...");
+ }
+ return mediaServerItem;
+ }
+
+ @Override
+ public void onPublishHandlerForPlayBack(IMediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid) {
RequestMessage msg = new RequestMessage();
msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
- StreamInfo streamInfo = onPublishHandler(resonse, deviceId, channelId, uuid);
+ StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId, uuid);
if (streamInfo != null) {
redisCatchStorage.startPlayback(streamInfo);
msg.setData(JSON.toJSONString(streamInfo));
@@ -249,10 +285,10 @@
}
}
- public StreamInfo onPublishHandler(JSONObject resonse, String deviceId, String channelId, String uuid) {
+ public StreamInfo onPublishHandler(IMediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid) {
String streamId = resonse.getString("stream");
JSONArray tracks = resonse.getJSONArray("tracks");
- StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream("rtp", streamId, tracks);
+ StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem,"rtp", streamId, tracks);
streamInfo.setDeviceID(deviceId);
streamInfo.setChannelId(channelId);
return streamInfo;
--
Gitblit v1.8.0