优化语音广播流程

648540858

2022-05-10 a9bdb0a706e10d2dffb50ae5a8086dd744bbd976

优化语音广播流程

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java

@@ -4,9 +4,8 @@
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamType;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
@@ -82,6 +81,9 @@
    @Autowired
    private ISIPCommanderForPlatform commanderForPlatform;

    @Autowired
    private AudioBroadcastManager audioBroadcastManager;


    /**   
     * 处理  ACK请求
@@ -122,6 +124,13 @@
            if (jsonObject == null) {
                logger.error("RTP推流失败: 请检查ZLM服务");
            } else if (jsonObject.getInteger("code") == 0) {
                if (sendRtpItem.isOnlyAudio()) {
                    AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
                    audioBroadcastCatch.setStatus(AudioBroadcastCatchStatus.Ok);
                    audioBroadcastCatch.setDialog((SIPDialog) evt.getDialog());
                    audioBroadcastCatch.setRequest((SIPRequest) evt.getRequest());
                    audioBroadcastManager.update(audioBroadcastCatch);
                }
                logger.info("RTP推流成功[ {}/{} ]，{}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port"));
            } else {
                logger.error("RTP推流失败: {}, 参数：{}",jsonObject.getString("msg"),JSONObject.toJSON(param));

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java

@@ -91,7 +91,7 @@
            if (dialog.getState().equals(DialogState.TERMINATED)) {
                String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser();
                String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
                SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, channelId, null, callIdHeader.getCallId());
                SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, null, null, callIdHeader.getCallId());
                logger.info("收到bye, [{}/{}]", platformGbId, channelId);
                if (sendRtpItem != null){
                    String streamId = sendRtpItem.getStreamId();
@@ -103,15 +103,15 @@
                    logger.info("收到bye:停止向上级推流：" + streamId);
                    MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
                    zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
                    redisCatchStorage.deleteSendRTPServer(platformGbId, channelId, callIdHeader.getCallId(), null);
                    redisCatchStorage.deleteSendRTPServer(platformGbId, sendRtpItem.getChannelId(), callIdHeader.getCallId(), null);
                    int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
                    if (totalReaderCount <= 0) {
                        logger.info("收到bye: {} 无其它观看者，通知设备停止推流", streamId);
                        if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) {
                            cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId, streamId, null);
                            cmder.streamByeCmd(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId(), streamId, null);
                        }
                        if (sendRtpItem.isOnlyAudio()) {
                            playService.stopAudioBroadcast(sendRtpItem.getDeviceId(), channelId);
                            playService.stopAudioBroadcast(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
                        }
                        if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
                            MessageForPushChannel messageForPushChannel = new MessageForPushChannel();

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java

@@ -713,6 +713,7 @@
                String waiteStreamTimeoutTaskKey = "waite-stream-" + device.getDeviceId() + audioBroadcastCatch.getChannelId();
                dynamicTask.startDelay(waiteStreamTimeoutTaskKey, ()->{
                    logger.info("等待推流超时: {}/{}", app, stream);
                    subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
                    playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
                    // 发送bye
                    try {
@@ -728,16 +729,17 @@

                subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
                        (MediaServerItem mediaServerItemInUse, JSONObject json)->{
                    logger.info("收到语音对讲推流");
                    try {
                            sendRtpItem.setStatus(2);
                            dynamicTask.stop(waiteStreamTimeoutTaskKey);
                            redisCatchStorage.updateSendRTPSever(sendRtpItem);
                            StringBuffer content = new StringBuffer(200);
                            content.append("v=0\r\n");
                            content.append("o="+ audioBroadcastCatch.getChannelId() +" 0 0 IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
                        content.append("o="+ config.getId() +" "+ sdp.getOrigin().getSessionId() +" " + sdp.getOrigin().getSessionVersion()  + " IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
                            content.append("s=Play\r\n");
                            content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
                            content.append("t=0 0\r\n");
                            content.append("m=video "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
                        content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
                            content.append("a=sendonly\r\n");
                            content.append("a=rtpmap:8 PCMA/8000\r\n");
                            content.append("y="+ finalSsrc + "\r\n");
@@ -747,14 +749,20 @@
                            parentPlatform.setServerIP(device.getIp());
                            parentPlatform.setServerPort(device.getPort());
                            parentPlatform.setServerGBId(device.getDeviceId());
                            try {

                                responseSdpAck(evt, content.toString(), parentPlatform);
                        Dialog dialog = evt.getDialog();
                        audioBroadcastCatch.setDialog((SIPDialog) dialog);
                        audioBroadcastCatch.setRequest((SIPRequest) request);
                        audioBroadcastManager.update(audioBroadcastCatch);
                            } catch (SipException e) {
                                throw new RuntimeException(e);
                            } catch (InvalidArgumentException e) {
                                throw new RuntimeException(e);
                            } catch (ParseException e) {
                                throw new RuntimeException(e);
                    } catch (SdpParseException e) {
                        throw new RuntimeException(e);
                            }
                        });
            }

 src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java

@@ -675,9 +675,8 @@
        AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(deviceId, channelId);
        if (audioBroadcastCatch != null) {
            audioBroadcastManager.del(deviceId, audioBroadcastCatch.getChannelId());
        }
        try {
            SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(deviceId, channelId, null, null);
                SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(deviceId, audioBroadcastCatch.getChannelId(), null, null);
            if (sendRtpItem != null) {
                redisCatchStorage.deleteSendRTPServer(deviceId, sendRtpItem.getChannelId(), null, null);
                MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
@@ -695,6 +694,8 @@
        } catch (ParseException e) {
            throw new RuntimeException(e);
        }
        }


    }
}