From e4918d29f39c3da1a3c036221c57ce97834cfff4 Mon Sep 17 00:00:00 2001 From: Lawrence <1934378145@qq.com> Date: 星期四, 21 一月 2021 20:58:11 +0800 Subject: [PATCH] 添加浏览器ID,确保SSE可同时推送到不同的前端 --- src/main/java/com/genersoft/iot/vmp/vmanager/play/PlayController.java | 119 +++++++++++++++++++++++++++++------------------------------ 1 files changed, 58 insertions(+), 61 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/play/PlayController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/play/PlayController.java index eba40bb..9b9a69b 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/play/PlayController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/play/PlayController.java @@ -4,10 +4,13 @@ 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.bean.DeviceChannel; 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.media.zlm.ZLMRTPServerFactory; +import com.genersoft.iot.vmp.storager.IRedisCatchStorage; +import com.genersoft.iot.vmp.vmanager.play.bean.PlayResult; import com.genersoft.iot.vmp.vmanager.service.IPlayService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -46,6 +49,9 @@ private IVideoManagerStorager storager; @Autowired + private IRedisCatchStorage redisCatchStorage; + + @Autowired private ZLMRESTfulUtils zlmresTfulUtils; @Autowired @@ -59,83 +65,74 @@ @PathVariable String channelId) { - Device device = storager.queryVideoDevice(deviceId); - StreamInfo streamInfo = storager.queryPlayByDevice(deviceId, channelId); + PlayResult playResult = playService.play(deviceId, channelId, null, null); + + // 瓒呮椂澶勭悊 + playResult.getResult().onTimeout(()->{ + logger.warn(String.format("璁惧鐐规挱瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId)); + // 閲婃斁rtpserver + cmder.closeRTPServer(playResult.getDevice(), channelId); + RequestMessage msg = new RequestMessage(); + msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + playResult.getUuid()); + msg.setData("Timeout"); + resultHolder.invokeResult(msg); + }); + return playResult.getResult(); + } + + @PostMapping("/play/{streamId}/stop") + public DeferredResult<ResponseEntity<String>> playStop(@PathVariable String streamId) { + + logger.debug(String.format("璁惧棰勮/鍥炴斁鍋滄API璋冪敤锛宻treamId锛�%s", streamId)); UUID uuid = UUID.randomUUID(); DeferredResult<ResponseEntity<String>> result = new DeferredResult<ResponseEntity<String>>(); // 褰曞儚鏌ヨ浠hannelId浣滀负deviceId鏌ヨ - resultHolder.put(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid, result); + resultHolder.put(DeferredResultHolder.CALLBACK_CMD_STOP + uuid, result); - if (streamInfo == null) { - // 鍙戦�佺偣鎾秷鎭� - cmder.playStreamCmd(device, channelId, (JSONObject response) -> { - logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString()); - playService.onPublishHandlerForPlay(response, deviceId, channelId, uuid.toString()); - }, event -> { + cmder.streamByeCmd(streamId, event -> { + StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId); + if (streamInfo == null) { RequestMessage msg = new RequestMessage(); msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid); + msg.setData("streamId not found"); + resultHolder.invokeResult(msg); + }else { + redisCatchStorage.stopPlay(streamInfo); + storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); + RequestMessage msg = new RequestMessage(); + msg.setId(DeferredResultHolder.CALLBACK_CMD_STOP + uuid); Response response = event.getResponse(); - msg.setData(String.format("鐐规挱澶辫触锛� 閿欒鐮侊細 %s, %s", response.getStatusCode(), response.getReasonPhrase())); + msg.setData(String.format("success")); resultHolder.invokeResult(msg); - }); - } else { - String streamId = streamInfo.getStreamId(); - JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(streamId); - if (rtpInfo.getBoolean("exist")) { - RequestMessage msg = new RequestMessage(); - msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid); - msg.setData(JSON.toJSONString(streamInfo)); - resultHolder.invokeResult(msg); - } else { - storager.stopPlay(streamInfo); - cmder.playStreamCmd(device, channelId, (JSONObject response) -> { - logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString()); - playService.onPublishHandlerForPlay(response, deviceId, channelId, uuid.toString()); - }, event -> { - RequestMessage msg = new RequestMessage(); - msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid); - Response response = event.getResponse(); - msg.setData(String.format("鐐规挱澶辫触锛� 閿欒鐮侊細 %s, %s", response.getStatusCode(), response.getReasonPhrase())); - resultHolder.invokeResult(msg); - }); } - } - - // 瓒呮椂澶勭悊 - result.onTimeout(()->{ - logger.warn(String.format("璁惧鐐规挱瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId)); - // 閲婃斁rtpserver - cmder.closeRTPServer(device, channelId); - RequestMessage msg = new RequestMessage(); - msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid); - msg.setData("Timeout"); - resultHolder.invokeResult(msg); }); - return result; - } - - @PostMapping("/play/{streamId}/stop") - public ResponseEntity<String> playStop(@PathVariable String streamId) { - - cmder.streamByeCmd(streamId); - StreamInfo streamInfo = storager.queryPlayByStreamId(streamId); - if (streamInfo == null) - return new ResponseEntity<String>("streamId not found", HttpStatus.OK); - storager.stopPlay(streamInfo); - if (logger.isDebugEnabled()) { - logger.debug(String.format("璁惧棰勮鍋滄API璋冪敤锛宻treamId锛�%s", streamId)); - } if (streamId != null) { JSONObject json = new JSONObject(); json.put("streamId", streamId); - return new ResponseEntity<String>(json.toString(), HttpStatus.OK); + RequestMessage msg = new RequestMessage(); + msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid); + msg.setData(json.toString()); + resultHolder.invokeResult(msg); } else { - logger.warn("璁惧棰勮鍋滄API璋冪敤澶辫触锛�"); - return new ResponseEntity<String>(HttpStatus.INTERNAL_SERVER_ERROR); + logger.warn("璁惧棰勮/鍥炴斁鍋滄API璋冪敤澶辫触锛�"); + RequestMessage msg = new RequestMessage(); + msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid); + msg.setData("streamId null"); + resultHolder.invokeResult(msg); } + + // 瓒呮椂澶勭悊 + result.onTimeout(()->{ + logger.warn(String.format("璁惧棰勮/鍥炴斁鍋滄瓒呮椂锛宻treamId锛�%s ", streamId)); + RequestMessage msg = new RequestMessage(); + msg.setId(DeferredResultHolder.CALLBACK_CMD_STOP + uuid); + msg.setData("Timeout"); + resultHolder.invokeResult(msg); + }); + return result; } /** @@ -145,7 +142,7 @@ */ @PostMapping("/play/{streamId}/convert") public ResponseEntity<String> playConvert(@PathVariable String streamId) { - StreamInfo streamInfo = storager.queryPlayByStreamId(streamId); + StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId); if (streamInfo == null) { logger.warn("瑙嗛杞爜API璋冪敤澶辫触锛�, 瑙嗛娴佸凡缁忓仠姝�!"); return new ResponseEntity<String>("鏈壘鍒拌棰戞祦淇℃伅, 瑙嗛娴佸彲鑳藉凡缁忓仠姝�", HttpStatus.OK); @@ -155,7 +152,7 @@ logger.warn("瑙嗛杞爜API璋冪敤澶辫触锛�, 瑙嗛娴佸凡鍋滄鎺ㄦ祦!"); return new ResponseEntity<String>("鎺ㄦ祦淇℃伅鍦ㄦ祦濯掍綋涓笉瀛樺湪, 瑙嗛娴佸彲鑳藉凡鍋滄鎺ㄦ祦", HttpStatus.OK); } else { - MediaServerConfig mediaInfo = storager.getMediaInfo(); + MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo(); String dstUrl = String.format("rtmp://%s:%s/convert/%s", "127.0.0.1", mediaInfo.getRtmpPort(), streamId ); String srcUrl = String.format("rtsp://%s:%s/rtp/%s", "127.0.0.1", mediaInfo.getRtspPort(), streamId); -- Gitblit v1.8.0