优化宕机后点播中设备发送bye

panlinlin
1 parent 118e4288
Showing 5 changed files with 98 additions and 3 deletions
src/main/java/com/genersoft/iot/vmp/common/StreamInfo.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java
@@ -21,6 +21,15 @@ public class StreamInfo {
     private String rtc;
     private JSONArray tracks;
  
+    public static class TransactionInfo{
+        public String callId;
+        public String localTag;
+        public String remoteTag;
+        public String branch;
+    }
+
+    private TransactionInfo transactionInfo;
+
     public String getApp() {
         return app;
     }
@@ -148,4 +157,12 @@ public class StreamInfo {
     public void setRtc(String rtc) {
         this.rtc = rtc;
     }
+
+    public TransactionInfo getTransactionInfo() {
+        return transactionInfo;
+    }
+
+    public void setTransactionInfo(TransactionInfo transactionInfo) {
+        this.transactionInfo = transactionInfo;
+    }
 }
@@ -135,6 +135,36 @@ public class SIPRequestHeaderProvider {
 		return request;
 	}
  
+	public Request createByteRequest(Device device, String channelId, String viaTag, String fromTag, String toTag, String callId) throws ParseException, InvalidArgumentException, PeerUnavailableException {
+		Request request = null;
+		//请求行
+		SipURI requestLine = sipFactory.createAddressFactory().createSipURI(channelId, device.getHostAddress());
+		// via
+		ArrayList<ViaHeader> viaHeaders = new ArrayList<ViaHeader>();
+		ViaHeader viaHeader = sipFactory.createHeaderFactory().createViaHeader(sipConfig.getSipIp(), sipConfig.getSipPort(), device.getTransport(), viaTag);
+		viaHeaders.add(viaHeader);
+		//from
+		SipURI fromSipURI = sipFactory.createAddressFactory().createSipURI(sipConfig.getSipId(),sipConfig.getSipDomain());
+		Address fromAddress = sipFactory.createAddressFactory().createAddress(fromSipURI);
+		FromHeader fromHeader = sipFactory.createHeaderFactory().createFromHeader(fromAddress, fromTag); //必须要有标记，否则无法创建会话，无法回应ack
+		//to
+		SipURI toSipURI = sipFactory.createAddressFactory().createSipURI(channelId,sipConfig.getSipDomain());
+		Address toAddress = sipFactory.createAddressFactory().createAddress(toSipURI);
+		ToHeader toHeader = sipFactory.createHeaderFactory().createToHeader(toAddress,toTag);
+
+		//Forwards
+		MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70);
+
+		//ceq
+		CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(1L, Request.BYE);
+		CallIdHeader callIdHeader = sipFactory.createHeaderFactory().createCallIdHeader(callId);
+		request = sipFactory.createMessageFactory().createRequest(requestLine, Request.BYE, callIdHeader, cSeqHeader,fromHeader, toHeader, viaHeaders, maxForwards);
+
+		Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory().createSipURI(sipConfig.getSipId(), sipConfig.getSipIp()+":"+sipConfig.getSipPort()));
+
+		return request;
+	}
+
 	public Request createSubscribeRequest(Device device, String content, String viaTag, String fromTag, String toTag, Integer expires, String event, CallIdHeader callIdHeader) throws ParseException, InvalidArgumentException, PeerUnavailableException {
 		Request request = null;
 		// sipuri
@@ -10,11 +10,14 @@ import javax.sip.header.CallIdHeader;
 import javax.sip.header.ViaHeader;
 import javax.sip.message.Request;
  
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.conf.MediaServerConfig;
 import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
 import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
+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.storager.IVideoManagerStorager;
@@ -75,6 +78,9 @@ public class SIPCommander implements ISIPCommander {
 	@Autowired
 	private ZLMRTPServerFactory zlmrtpServerFactory;
  
+	@Autowired
+	private ZLMRESTfulUtils zlmresTfulUtils;
+
 	@Value("${media.rtp.enable}")
 	private boolean rtpEnable;
  
@@ -577,13 +583,39 @@ public class SIPCommander implements ISIPCommander {
  
 		try {
 			ClientTransaction transaction = streamSession.getTransaction(deviceId, channelId);
-			// 服务重启后
+			// 服务重启后, 无法直接发送bye， 通过手动构建发送
 			if (transaction == null) {
+
 				StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
 				if (streamInfo != null) {
+					JSONObject mediaList = zlmresTfulUtils.getMediaList(streamInfo.getApp(), streamInfo.getStreamId());
+					if (mediaList != null) { // 仍在推流才发送
+						if (mediaList.getInteger("code") == 0) {
+							JSONArray data = mediaList.getJSONArray("data");
+							if (data != null && data.size() > 0) {
+								Device device = storager.queryVideoDevice(deviceId);
+								if (device != null) {
+									StreamInfo.TransactionInfo transactionInfo = streamInfo.getTransactionInfo();
+									try {
+										Request byteRequest = headerProvider.createByteRequest(device, channelId,
+												transactionInfo.branch,
+												transactionInfo.localTag,
+												transactionInfo.remoteTag,
+												transactionInfo.callId);
+										transmitRequest(device, byteRequest);
+									} catch (InvalidArgumentException e) {
+										e.printStackTrace();
+									}
+								}
+							}
+						}
+					}
 					redisCatchStorage.stopPlay(streamInfo);
 				}
-				okEvent.response(null);
+
+				if (okEvent != null) {
+					okEvent.response(null);
+				}
 				return;
 			}
  
@@ -7,6 +7,7 @@ import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.gb28181.bean.Device;
 import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
 import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
+import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
 import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
 import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
@@ -17,6 +18,7 @@ import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
 import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
 import com.genersoft.iot.vmp.service.IMediaService;
 import com.genersoft.iot.vmp.service.IPlayService;
+import gov.nist.javax.sip.stack.SIPDialog;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -24,6 +26,9 @@ import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 import org.springframework.web.context.request.async.DeferredResult;
  
+import javax.sip.ClientTransaction;
+import javax.sip.Dialog;
+import javax.sip.header.CallIdHeader;
 import javax.sip.message.Response;
 import java.util.UUID;
  
@@ -50,6 +55,9 @@ public class PlayServiceImpl implements IPlayService {
     @Autowired
     private IMediaService mediaService;
  
+    @Autowired
+    private VideoStreamSessionManager streamSession;
+
  
     @Override
     public PlayResult play(String deviceId, String channelId, ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent) {
@@ -141,7 +149,14 @@ public class PlayServiceImpl implements IPlayService {
                 deviceChannel.setStreamId(streamInfo.getStreamId());
                 storager.startPlay(deviceId, channelId, streamInfo.getStreamId());
             }
-
+            ClientTransaction transaction = streamSession.getTransaction(deviceId, channelId);
+            SIPDialog dialog = (SIPDialog)transaction.getDialog();
+            StreamInfo.TransactionInfo transactionInfo = new StreamInfo.TransactionInfo();
+            transactionInfo.callId = dialog.getCallId().getCallId();
+            transactionInfo.localTag = dialog.getLocalTag();
+            transactionInfo.remoteTag = dialog.getRemoteTag();
+            transactionInfo.branch = dialog.getFirstTransactionInt().getBranchId();
+            streamInfo.setTransactionInfo(transactionInfo);
             redisCatchStorage.startPlay(streamInfo);
             msg.setData(JSON.toJSONString(streamInfo));
             resultHolder.invokeResult(msg);
@@ -102,6 +102,7 @@ public class PlayController {
 				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());