使用异步接口, 更好的并发, 对hook使用订阅机制

替换前段播放器, 支持h265的播放
放弃循环获取编码信息,
This commit is contained in:
648540858
2020-12-08 18:11:02 +08:00
parent 5a4859f05c
commit f75b3e6cda
20 changed files with 681 additions and 359 deletions

View File

@@ -4,7 +4,10 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.MediaServerConfig;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.vmanager.service.IPlayService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -22,6 +25,10 @@ import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import org.springframework.web.context.request.async.DeferredResult;
import java.text.DecimalFormat;
import java.util.UUID;
@CrossOrigin
@RestController
@@ -39,95 +46,56 @@ public class PlayController {
@Autowired
private ZLMRESTfulUtils zlmresTfulUtils;
@Value("${media.closeWaitRTPInfo}")
private boolean closeWaitRTPInfo;
@Autowired
private DeferredResultHolder resultHolder;
@Autowired
private IPlayService playService;
@GetMapping("/play/{deviceId}/{channelId}")
public ResponseEntity<String> play(@PathVariable String deviceId, @PathVariable String channelId,
Integer getEncoding) {
public DeferredResult<ResponseEntity<String>> play(@PathVariable String deviceId,
@PathVariable String channelId) {
if (getEncoding == null) getEncoding = 0;
getEncoding = closeWaitRTPInfo ? 0 : getEncoding;
Device device = storager.queryVideoDevice(deviceId);
StreamInfo streamInfo = storager.queryPlayByDevice(deviceId, channelId);
UUID uuid = UUID.randomUUID();
DeferredResult<ResponseEntity<String>> result = new DeferredResult<ResponseEntity<String>>();
// 超时处理
result.onTimeout(()->{
RequestMessage msg = new RequestMessage();
msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
msg.setData("Timeout");
resultHolder.invokeResult(msg);
});
// 录像查询以channelId作为deviceId查询
resultHolder.put(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid, result);
if (streamInfo == null) {
streamInfo = cmder.playStreamCmd(device, channelId);
// TODO playStreamCmd 超时处理
cmder.playStreamCmd(device, channelId, (JSONObject response) -> {
logger.info("收到订阅消息: " + response.toJSONString());
playService.onPublishHandlerForPlay(response, deviceId, channelId, uuid.toString());
});
} else {
String streamId = String.format("%08x", Integer.parseInt(streamInfo.getSsrc())).toUpperCase();
JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(streamId);
if (rtpInfo.getBoolean("exist")) {
return new ResponseEntity<String>(JSON.toJSONString(streamInfo), HttpStatus.OK);
RequestMessage msg = new RequestMessage();
msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
msg.setData(JSON.toJSONString(streamInfo));
resultHolder.invokeResult(msg);
} else {
storager.stopPlay(streamInfo);
streamInfo = cmder.playStreamCmd(device, channelId);
// TODO playStreamCmd 超时处理
cmder.playStreamCmd(device, channelId, (JSONObject response) -> {
logger.info("收到订阅消息: " + response.toJSONString());
playService.onPublishHandlerForPlay(response, deviceId, channelId, uuid.toString());
});
}
}
String streamId = String.format("%08x", Integer.parseInt(streamInfo.getSsrc())).toUpperCase();
// 等待推流, TODO 默认超时30s
boolean lockFlag = true;
boolean rtpPushed = false;
long startTime = System.currentTimeMillis();
JSONObject rtpInfo = null;
if (getEncoding == 1) {
while (lockFlag) {
try {
if (System.currentTimeMillis() - startTime > 60 * 1000) {
storager.stopPlay(streamInfo);
logger.info("播放等待超时");
return new ResponseEntity<String>("timeout", HttpStatus.OK);
} else {
streamInfo = storager.queryPlayByDevice(deviceId, channelId);
if (!rtpPushed) {
logger.info("查询RTP推流信息...");
rtpInfo = zlmresTfulUtils.getRtpInfo(streamId);
}
if (rtpInfo != null && rtpInfo.getBoolean("exist") && streamInfo != null
&& streamInfo.getFlv() != null) {
logger.info("查询流编码信息:" + streamInfo.getFlv());
rtpPushed = true;
Thread.sleep(2000);
JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo("rtp", "rtmp", streamId);
if (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online")) {
lockFlag = false;
logger.info("流编码信息已获取");
JSONArray tracks = mediaInfo.getJSONArray("tracks");
logger.info(tracks.toJSONString());
streamInfo.setTracks(tracks);
storager.startPlay(streamInfo);
} else {
logger.info("流编码信息未获取2秒后重试...");
}
} else {
Thread.sleep(2000);
continue;
}
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} else {
String flv = storager.getMediaInfo().getWanIp() + ":" + storager.getMediaInfo().getHttpPort() + "/rtp/"
+ streamId + ".flv";
streamInfo.setFlv("http://" + flv);
streamInfo.setWs_flv("ws://" + flv);
storager.startPlay(streamInfo);
}
if (logger.isDebugEnabled()) {
logger.debug(String.format("设备预览 API调用deviceId%s channelId%s", deviceId, channelId));
logger.debug("设备预览 API调用ssrc" + streamInfo.getSsrc() + ",ZLMedia streamId:"
+ Integer.toHexString(Integer.parseInt(streamInfo.getSsrc())));
}
if (streamInfo != null) {
return new ResponseEntity<String>(JSON.toJSONString(streamInfo), HttpStatus.OK);
} else {
logger.warn("设备预览API调用失败");
return new ResponseEntity<String>(HttpStatus.INTERNAL_SERVER_ERROR);
}
return result;
}
@PostMapping("/play/{ssrc}/stop")
@@ -180,10 +148,20 @@ public class PlayController {
result.put("code", 0);
JSONObject data = jsonObject.getJSONObject("data");
if (data != null) {
result.put("key", data.getString("key"));
result.put("rtmp", dstUrl);
result.put("flv", String.format("http://%s:%s/convert/%s.flv", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), streamId));
result.put("ws_flv", String.format("ws://%s:%s/convert/%s.flv", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), streamId));
result.put("key", data.getString("key"));
StreamInfo streamInfoResult = new StreamInfo();
streamInfoResult.setRtmp(dstUrl);
streamInfoResult.setRtsp(String.format("rtsp://%s:%s/convert/%s", mediaInfo.getWanIp(), mediaInfo.getRtspPort(), streamId));
streamInfoResult.setStreamId(streamId);
streamInfoResult.setFlv(String.format("http://%s:%s/convert/%s.flv", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), streamId));
streamInfoResult.setWs_flv(String.format("ws://%s:%s/convert/%s.flv", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), streamId));
streamInfoResult.setHls(String.format("http://%s:%s/convert/%s/hls.m3u8", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), streamId));
streamInfoResult.setWs_hls(String.format("ws://%s:%s/convert/%s/hls.m3u8", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), streamId));
streamInfoResult.setFmp4(String.format("http://%s:%s/convert/%s.live.mp4", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), streamId));
streamInfoResult.setWs_fmp4(String.format("ws://%s:%s/convert/%s.live.mp4", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), streamId));
streamInfoResult.setTs(String.format("http://%s:%s/convert/%s.live.ts", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), streamId));
streamInfoResult.setWs_ts(String.format("ws://%s:%s/convert/%s.live.ts", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), streamId));
result.put("data", streamInfoResult);
}
}else {
result.put("code", 1);