From 72c1b36d6d2ece497e032c8434641d6576590f9d Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期日, 03 七月 2022 00:44:36 +0800
Subject: [PATCH] 优化对讲逻辑

---
 src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java |  273 +++++++++++++++++++++++++++++++++---------------------
 1 files changed, 165 insertions(+), 108 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java
index 4297bb2..935fc42 100644
--- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java
+++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java
@@ -2,13 +2,16 @@
 
 import com.alibaba.fastjson.JSONArray;
 import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
 import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
-import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
 import com.genersoft.iot.vmp.gb28181.bean.Device;
 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.dto.MediaServerItem;
+import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult;
 import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
 import com.genersoft.iot.vmp.service.IMediaService;
@@ -31,15 +34,15 @@
 
 import com.alibaba.fastjson.JSONObject;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
-import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
 import org.springframework.web.context.request.async.DeferredResult;
 
-import java.util.Enumeration;
-import java.util.Map;
+import java.util.List;
 import java.util.UUID;
 
-import javax.sip.message.Response;
-
+/**
+ * @author lin
+ */
 @Api(tags = "鍥芥爣璁惧鐐规挱")
 @CrossOrigin
 @RestController
@@ -55,7 +58,7 @@
 	private VideoStreamSessionManager streamSession;
 
 	@Autowired
-	private IVideoManagerStorager storager;
+	private IVideoManagerStorage storager;
 
 	@Autowired
 	private IRedisCatchStorage redisCatchStorage;
@@ -72,6 +75,9 @@
 	@Autowired
 	private IMediaService mediaService;
 
+	@Autowired
+	private IMediaServerService mediaServerService;
+
 	@ApiOperation("寮�濮嬬偣鎾�")
 	@ApiImplicitParams({
 			@ApiImplicitParam(name = "deviceId", value = "璁惧ID", dataTypeClass = String.class),
@@ -81,8 +87,10 @@
 	public DeferredResult<ResponseEntity<String>> play(@PathVariable String deviceId,
 													   @PathVariable String channelId) {
 
-		PlayResult playResult = playService.play(deviceId, channelId, null, null);
-
+		// 鑾峰彇鍙敤鐨剒lm
+		Device device = storager.queryVideoDevice(deviceId);
+		MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device);
+		PlayResult playResult = playService.play(newMediaServerItem, deviceId, channelId, null, null, null);
 
 		return playResult.getResult();
 	}
@@ -95,31 +103,32 @@
 	@GetMapping("/stop/{deviceId}/{channelId}")
 	public DeferredResult<ResponseEntity<String>> playStop(@PathVariable String deviceId, @PathVariable String channelId) {
 
-		logger.debug(String.format("璁惧棰勮/鍥炴斁鍋滄API璋冪敤锛宻treamId锛�%s/$s", deviceId, channelId ));
+		logger.debug(String.format("璁惧棰勮/鍥炴斁鍋滄API璋冪敤锛宻treamId锛�%s_%s", deviceId, channelId ));
 
-		UUID uuid = UUID.randomUUID();
-		DeferredResult<ResponseEntity<String>> result = new DeferredResult<ResponseEntity<String>>();
+		String uuid = UUID.randomUUID().toString();
+		DeferredResult<ResponseEntity<String>> result = new DeferredResult<>();
 
 		// 褰曞儚鏌ヨ浠hannelId浣滀负deviceId鏌ヨ
-		resultHolder.put(DeferredResultHolder.CALLBACK_CMD_STOP + uuid, result);
-
-		cmder.streamByeCmd(deviceId, channelId, event -> {
-			StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
-			if (streamInfo == null) {
-				RequestMessage msg = new RequestMessage();
-				msg.setId(DeferredResultHolder.CALLBACK_CMD_STOP + uuid);
-				msg.setData("鐐规挱鏈壘鍒�");
-				resultHolder.invokeResult(msg);
-				storager.stopPlay(deviceId, channelId);
-			}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("success"));
-				resultHolder.invokeResult(msg);
-			}
+		String key = DeferredResultHolder.CALLBACK_CMD_STOP + deviceId + channelId;
+		resultHolder.put(key, uuid, result);
+		StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
+		if (streamInfo == null) {
+			RequestMessage msg = new RequestMessage();
+			msg.setId(uuid);
+			msg.setKey(key);
+			msg.setData("鐐规挱鏈壘鍒�");
+			resultHolder.invokeAllResult(msg);
+			storager.stopPlay(deviceId, channelId);
+			return result;
+		}
+		cmder.streamByeCmd(deviceId, channelId, streamInfo.getStream(), null, eventResult -> {
+			redisCatchStorage.stopPlay(streamInfo);
+			storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
+			RequestMessage msgForSuccess = new RequestMessage();
+			msgForSuccess.setId(uuid);
+			msgForSuccess.setKey(key);
+			msgForSuccess.setData("success");
+			resultHolder.invokeAllResult(msgForSuccess);
 		});
 
 		if (deviceId != null || channelId != null) {
@@ -127,24 +136,27 @@
 			json.put("deviceId", deviceId);
 			json.put("channelId", channelId);
 			RequestMessage msg = new RequestMessage();
-			msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
+			msg.setId(uuid);
+			msg.setKey(key);
 			msg.setData(json.toString());
-			resultHolder.invokeResult(msg);
+			resultHolder.invokeAllResult(msg);
 		} else {
 			logger.warn("璁惧棰勮/鍥炴斁鍋滄API璋冪敤澶辫触锛�");
 			RequestMessage msg = new RequestMessage();
-			msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
+			msg.setId(uuid);
+			msg.setKey(key);
 			msg.setData("streamId null");
-			resultHolder.invokeResult(msg);
+			resultHolder.invokeAllResult(msg);
 		}
 
 		// 瓒呮椂澶勭悊
 		result.onTimeout(()->{
-			logger.warn(String.format("璁惧棰勮/鍥炴斁鍋滄瓒呮椂锛宒eviceId/channelId锛�%s/$s ", deviceId, channelId));
+			logger.warn(String.format("璁惧棰勮/鍥炴斁鍋滄瓒呮椂锛宒eviceId/channelId锛�%s_%s ", deviceId, channelId));
 			RequestMessage msg = new RequestMessage();
-			msg.setId(DeferredResultHolder.CALLBACK_CMD_STOP + uuid);
+			msg.setId(uuid);
+			msg.setKey(key);
 			msg.setData("Timeout");
-			resultHolder.invokeResult(msg);
+			resultHolder.invokeAllResult(msg);
 		});
 		return result;
 	}
@@ -160,21 +172,24 @@
 	})
 	@PostMapping("/convert/{streamId}")
 	public ResponseEntity<String> playConvert(@PathVariable String streamId) {
-		StreamInfo streamInfo = streamId.startsWith("gb_play_") ? redisCatchStorage.queryPlayByStreamId(streamId) : redisCatchStorage.queryPlaybackByStreamId(streamId);
+		StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
+		if (streamInfo == null) {
+			streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
+		}
 		if (streamInfo == null) {
 			logger.warn("瑙嗛杞爜API璋冪敤澶辫触锛�, 瑙嗛娴佸凡缁忓仠姝�!");
 			return new ResponseEntity<String>("鏈壘鍒拌棰戞祦淇℃伅, 瑙嗛娴佸彲鑳藉凡缁忓仠姝�", HttpStatus.OK);
 		}
-		JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(streamId);
+		MediaServerItem mediaInfo = mediaServerService.getOne(streamInfo.getMediaServerId());
+		JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId);
 		if (!rtpInfo.getBoolean("exist")) {
 			logger.warn("瑙嗛杞爜API璋冪敤澶辫触锛�, 瑙嗛娴佸凡鍋滄鎺ㄦ祦!");
 			return new ResponseEntity<String>("鎺ㄦ祦淇℃伅鍦ㄦ祦濯掍綋涓笉瀛樺湪, 瑙嗛娴佸彲鑳藉凡鍋滄鎺ㄦ祦", HttpStatus.OK);
 		} else {
-			ZLMServerConfig 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);
-			JSONObject jsonObject = zlmresTfulUtils.addFFmpegSource(srcUrl, dstUrl, "1000000", true, false, null);
+			JSONObject jsonObject = zlmresTfulUtils.addFFmpegSource(mediaInfo, srcUrl, dstUrl, "1000000", true, false, null);
 			logger.info(jsonObject.toJSONString());
 			JSONObject result = new JSONObject();
 			if (jsonObject != null && jsonObject.getInteger("code") == 0) {
@@ -182,7 +197,7 @@
 				JSONObject data = jsonObject.getJSONObject("data");
 				if (data != null) {
 				   	result.put("key", data.getString("key"));
-					StreamInfo streamInfoResult = mediaService.getStreamInfoByAppAndStreamWithCheck("convert", streamId);
+					StreamInfo streamInfoResult = mediaService.getStreamInfoByAppAndStreamWithCheck("convert", streamId, mediaInfo.getId());
 					result.put("data", streamInfoResult);
 				}
 			}else {
@@ -203,98 +218,140 @@
 			@ApiImplicitParam(name = "key", value = "瑙嗛娴乲ey", dataTypeClass = String.class),
 	})
 	@PostMapping("/convertStop/{key}")
-	public ResponseEntity<String> playConvertStop(@PathVariable String key) {
-
-		JSONObject jsonObject = zlmresTfulUtils.delFFmpegSource(key);
-		logger.info(jsonObject.toJSONString());
+	public ResponseEntity<String> playConvertStop(@PathVariable String key, String mediaServerId) {
 		JSONObject result = new JSONObject();
-		if (jsonObject != null && jsonObject.getInteger("code") == 0) {
-			result.put("code", 0);
-			JSONObject data = jsonObject.getJSONObject("data");
-			if (data != null && data.getBoolean("flag")) {
-				result.put("code", "0");
-				result.put("msg", "success");
-			}else {
-
-			}
-		}else {
-			result.put("code", 1);
-			result.put("msg", "delFFmpegSource fail");
+		if (mediaServerId == null) {
+			result.put("code", 400);
+			result.put("msg", "mediaServerId is null");
+			return new ResponseEntity<String>( result.toJSONString(), HttpStatus.BAD_REQUEST);
 		}
-		return new ResponseEntity<String>( result.toJSONString(), HttpStatus.OK);
+		MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
+		if (mediaInfo == null) {
+			result.put("code", 0);
+			result.put("msg", "浣跨敤鐨勬祦濯掍綋宸茬粡鍋滄杩愯");
+			return new ResponseEntity<String>( result.toJSONString(), HttpStatus.OK);
+		}else {
+			JSONObject jsonObject = zlmresTfulUtils.delFFmpegSource(mediaInfo, key);
+			logger.info(jsonObject.toJSONString());
+			if (jsonObject != null && jsonObject.getInteger("code") == 0) {
+				result.put("code", 0);
+				JSONObject data = jsonObject.getJSONObject("data");
+				if (data != null && data.getBoolean("flag")) {
+					result.put("code", "0");
+					result.put("msg", "success");
+				}else {
+
+				}
+			}else {
+				result.put("code", 1);
+				result.put("msg", "delFFmpegSource fail");
+			}
+			return new ResponseEntity<String>( result.toJSONString(), HttpStatus.OK);
+		}
+
+
 	}
 
 	@ApiOperation("璇煶骞挎挱鍛戒护")
 	@ApiImplicitParams({
 			@ApiImplicitParam(name = "deviceId", value = "璁惧Id", dataTypeClass = String.class),
+			@ApiImplicitParam(name = "channelId", value = "閫氶亾Id", dataTypeClass = String.class),
+			@ApiImplicitParam(name = "timeout", value = "鎺ㄦ祦瓒呮椂鏃堕棿(绉�)", dataTypeClass = Integer.class),
 	})
-    @GetMapping("/broadcast/{deviceId}")
-    @PostMapping("/broadcast/{deviceId}")
-    public DeferredResult<ResponseEntity<String>> broadcastApi(@PathVariable String deviceId) {
+    @GetMapping("/broadcast/{deviceId}/{channelId}")
+    @PostMapping("/broadcast/{deviceId}/{channelId}")
+    public DeferredResult<WVPResult<AudioBroadcastResult>> broadcastApi(@PathVariable String deviceId, @PathVariable String channelId,  Integer timeout) {
         if (logger.isDebugEnabled()) {
             logger.debug("璇煶骞挎挱API璋冪敤");
         }
         Device device = storager.queryVideoDevice(deviceId);
-		DeferredResult<ResponseEntity<String>> result = new DeferredResult<ResponseEntity<String>>(3 * 1000L);
-        if (device == null) {
-			resultHolder.put(DeferredResultHolder.CALLBACK_CMD_BROADCAST + deviceId, result);
-			RequestMessage msg = new RequestMessage();
-			msg.setId(DeferredResultHolder.CALLBACK_CMD_BROADCAST + deviceId);
-			JSONObject json = new JSONObject();
-			json.put("DeviceID", deviceId);
-			json.put("CmdType", "Broadcast");
-			json.put("Result", "Failed");
-			json.put("Description", "Device 涓嶅瓨鍦�");
-			msg.setData(json);
-			resultHolder.invokeResult(msg);
-			return result;
+		if (device == null) {
+			WVPResult<AudioBroadcastResult> result = new WVPResult<>();
+			result.setCode(-1);
+			result.setMsg("鏈壘鍒拌澶囷細 " + deviceId);
+			DeferredResult<WVPResult<AudioBroadcastResult>> deferredResult = new DeferredResult<>();
+			deferredResult.setResult(result);
+			return deferredResult;
 		}
-		cmder.audioBroadcastCmd(device, event -> {
-			Response response = event.getResponse();
-			RequestMessage msg = new RequestMessage();
-			msg.setId(DeferredResultHolder.CALLBACK_CMD_BROADCAST + deviceId);
-			JSONObject json = new JSONObject();
-			json.put("DeviceID", deviceId);
-			json.put("CmdType", "Broadcast");
-			json.put("Result", "Failed");
-			json.put("Description", String.format("璇煶骞挎挱鎿嶄綔澶辫触锛岄敊璇爜锛� %s, %s", response.getStatusCode(), response.getReasonPhrase()));
-			msg.setData(json);
-			resultHolder.invokeResult(msg);
-		});
+		if (channelId == null) {
+			WVPResult<AudioBroadcastResult> result = new WVPResult<>();
+			result.setCode(-1);
+			result.setMsg("鏈壘鍒伴�氶亾锛� " + channelId);
+			DeferredResult<WVPResult<AudioBroadcastResult>> deferredResult = new DeferredResult<>();
+			deferredResult.setResult(result);
+			return deferredResult;
+		}
 
-		result.onTimeout(() -> {
-			logger.warn(String.format("璇煶骞挎挱鎿嶄綔瓒呮椂, 璁惧鏈繑鍥炲簲绛旀寚浠�"));
-			RequestMessage msg = new RequestMessage();
-			msg.setId(DeferredResultHolder.CALLBACK_CMD_BROADCAST + deviceId);
-			JSONObject json = new JSONObject();
-			json.put("DeviceID", deviceId);
-			json.put("CmdType", "Broadcast");
-			json.put("Result", "Failed");
-			json.put("Error", "Timeout. Device did not response to broadcast command.");
-			msg.setData(json);
-			resultHolder.invokeResult(msg);
+		String key = DeferredResultHolder.CALLBACK_CMD_BROADCAST + deviceId;
+		if (resultHolder.exist(key, null)) {
+			WVPResult<AudioBroadcastResult> wvpResult = new WVPResult<>();
+			wvpResult.setCode(-1);
+			wvpResult.setMsg("璁惧浣跨敤涓�");
+			DeferredResult<WVPResult<AudioBroadcastResult>> deferredResult = new DeferredResult<>();
+			deferredResult.setResult(wvpResult);
+			return deferredResult;
+		}
+		if (timeout == null){
+			timeout = 30;
+		}
+		DeferredResult<WVPResult<AudioBroadcastResult>> result = new DeferredResult<>(timeout.longValue()*1000 + 2000);
+		String uuid  = UUID.randomUUID().toString();
+		result.onTimeout(()->{
+			WVPResult<AudioBroadcastResult> wvpResult = new WVPResult<>();
+			wvpResult.setCode(-1);
+			wvpResult.setMsg("璇锋眰瓒呮椂");
+			RequestMessage requestMessage = new RequestMessage();
+			requestMessage.setKey(key);
+			requestMessage.setData(wvpResult);
+			resultHolder.invokeAllResult(requestMessage);
 		});
-		resultHolder.put(DeferredResultHolder.CALLBACK_CMD_BROADCAST + deviceId, result);
+		playService.audioBroadcast(device, channelId, timeout, (msg)->{
+			WVPResult<AudioBroadcastResult> wvpResult = new WVPResult<>();
+			wvpResult.setCode(-1);
+			wvpResult.setMsg(msg);
+			RequestMessage requestMessage = new RequestMessage();
+			requestMessage.setKey(key);
+			requestMessage.setData(wvpResult);
+			resultHolder.invokeAllResult(requestMessage);
+		});
+		resultHolder.put(key, uuid, result);
+
 		return result;
+	}
+
+
+	@ApiOperation("鍋滄璇煶骞挎挱")
+	@ApiImplicitParams({
+			@ApiImplicitParam(name = "deviceId", value = "璁惧Id", dataTypeClass = String.class),
+			@ApiImplicitParam(name = "channelId", value = "閫氶亾Id", dataTypeClass = String.class),
+	})
+	@GetMapping("/broadcast/stop/{deviceId}/{channelId}")
+	@PostMapping("/broadcast/stop/{deviceId}/{channelId}")
+	public WVPResult<String> stopBroadcastA(@PathVariable String deviceId, @PathVariable String channelId) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("鍋滄璇煶骞挎挱API璋冪敤");
+		}
+		playService.stopAudioBroadcast(deviceId, channelId);
+		return new WVPResult<>(0, "success", null);
 	}
 
 	@ApiOperation("鑾峰彇鎵�鏈夌殑ssrc")
 	@GetMapping("/ssrc")
-	public WVPResult<JSONObject> getSSRC() {
+	public WVPResult<JSONObject> getSsrc() {
 		if (logger.isDebugEnabled()) {
 			logger.debug("鑾峰彇鎵�鏈夌殑ssrc");
 		}
 		JSONArray objects = new JSONArray();
-		for(Map.Entry<String, String> entry: streamSession.getSsrcMap().entrySet()) {
-			System.out.println("Key = " + entry.getKey() + ", Value = " + entry.getValue());
+		List<SsrcTransaction> allSsrc = streamSession.getAllSsrc();
+		for (SsrcTransaction transaction : allSsrc) {
 			JSONObject jsonObject = new JSONObject();
-			String[] keyArray = entry.getKey().split("_");
-			jsonObject.put("deviceId", keyArray[0]);
-			jsonObject.put("channelId", keyArray[1]);
-			jsonObject.put("ssrc", entry.getValue());
-			jsonObject.put("streamId", streamSession.getStreamIdMap().get(entry.getKey()));
+			jsonObject.put("deviceId", transaction.getDeviceId());
+			jsonObject.put("channelId", transaction.getChannelId());
+			jsonObject.put("ssrc", transaction.getSsrc());
+			jsonObject.put("streamId", transaction.getStream());
 			objects.add(jsonObject);
 		}
+
 		WVPResult<JSONObject> result = new WVPResult<>();
 		result.setCode(0);
 		result.setMsg("success");

--
Gitblit v1.8.0