优化语音广播的TCP主动模式

648540858

2022-05-18 9c555b56b70ed05c1b13f0a26098b702d7364839

优化语音广播的TCP主动模式

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

@@ -107,29 +107,41 @@
            SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, null, null, callIdHeader.getCallId());
            String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
            MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
            logger.info("收到ACK，开始向上级推流 rtp/{}", sendRtpItem.getStreamId());
            logger.info("[收到ACK]，开始使用{}向上级推流 {}/{}->{}:{}({})", sendRtpItem.isTcp() ? "TCP" : "UDP",
                    sendRtpItem.getApp(), sendRtpItem.getStreamId(),
                    sendRtpItem.getIp() ,sendRtpItem.getPort(),
                    sendRtpItem.getSsrc());
            Map<String, Object> param = new HashMap<>();
            param.put("vhost","__defaultVhost__");
            param.put("app",sendRtpItem.getApp());
            param.put("stream",sendRtpItem.getStreamId());
            param.put("ssrc", sendRtpItem.getSsrc());
            param.put("dst_url",sendRtpItem.getIp());
            param.put("dst_port", sendRtpItem.getPort());
            param.put("is_udp", is_Udp);
            param.put("src_port", sendRtpItem.getLocalPort());
            param.put("pt", sendRtpItem.getPt());
            param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
            param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
            JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
            JSONObject jsonObject;
            if (sendRtpItem.isTcpActive()) {
                jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);
            }else {
                param.put("is_udp", is_Udp);
                param.put("dst_url",sendRtpItem.getIp());
                param.put("dst_port", sendRtpItem.getPort());
                jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
            }

            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);
                    String waiteStreamTimeoutTaskKey = "waite-stream-" + audioBroadcastCatch.getDeviceId() + audioBroadcastCatch.getChannelId();
                    dynamicTask.stop(waiteStreamTimeoutTaskKey);
                }
                logger.info("RTP推流成功[ {}/{} ]，{}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port"));
            } else {

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

@@ -1,5 +1,7 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig;
@@ -20,7 +22,9 @@
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItemLite;
import com.genersoft.iot.vmp.service.IMediaServerService;
@@ -89,6 +93,9 @@

    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;

    @Autowired
    private ZLMRESTfulUtils zlmresTfulUtils;

    @Autowired
    private IMediaServerService mediaServerService;
@@ -674,7 +681,19 @@
            subscribeKey.put("mediaServerId", mediaServerItem.getId());
            String finalSsrc = ssrc;
            // 流已经存在时直接推流
            if (zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream)) {
            JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtsp", stream);
            JSONArray tracks = mediaInfo.getJSONArray("tracks");
            Integer codecId = null;
            if (tracks != null && tracks.size() > 0) {
                for (int i = 0; i < tracks.size(); i++) {
                    MediaItem.MediaTrack track = JSON.toJavaObject((JSON)tracks.get(i),MediaItem.MediaTrack.class);
                    if (track.getCodecType() == 1) {
                        codecId = track.getCodecId();
                        break;
                    }
                }
            }
            if ((mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online"))) {
                logger.info("发现已经在推流");
                sendRtpItem.setStatus(2);
                redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -684,9 +703,40 @@
                content.append("s=Play\r\n");
                content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
                content.append("t=0 0\r\n");
                if (codecId == null) {
                    if (mediaTransmissionTCP) {
                        content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
                    }else {
                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");
                }else {
                    if (codecId == 4) {
                        if (mediaTransmissionTCP) {
                            content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n");
                        }else {
                            content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n");
                        }
                        content.append("a=rtpmap:0 PCMU/8000\r\n");
                    }else {
                        if (mediaTransmissionTCP) {
                            content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
                        }else {
                            content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
                        }
                        content.append("a=rtpmap:8 PCMA/8000\r\n");
                    }
                }
                if (sendRtpItem.isTcp()) {
                    content.append("a=connection:new\r\n");
                    if (!sendRtpItem.isTcpActive()) {
                        content.append("a=setup:active\r\n");
                    }else {
                        content.append("a=setup:passive\r\n");
                    }
                }
                content.append("a=sendonly\r\n");
                content.append("y="+ finalSsrc + "\r\n");
                content.append("f=v/////a/1/8/1\r\n");

@@ -727,9 +777,22 @@
                    }
                }, 20*1000);

                boolean finalMediaTransmissionTCP = mediaTransmissionTCP;
                subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
                        (MediaServerItem mediaServerItemInUse, JSONObject json)->{
                    logger.info("收到语音对讲推流");
                    MediaItem mediaItem = JSON.toJavaObject(json, MediaItem.class);
                    Integer audioCodecId = null;
                    if (mediaItem.getTracks() != null && mediaItem.getTracks().size() > 0) {
                        for (int i = 0; i < mediaItem.getTracks().size(); i++) {
                            MediaItem.MediaTrack mediaTrack = mediaItem.getTracks().get(i);
                            if (mediaTrack.getCodecType() == 1) {
                                audioCodecId = mediaTrack.getCodecId();
                                break;
                            }
                        }
                    }

                    try {
                        sendRtpItem.setStatus(2);
                        redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -739,9 +802,40 @@
                        content.append("s=Play\r\n");
                        content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
                        content.append("t=0 0\r\n");
                        if (audioCodecId == null) {
                            if (finalMediaTransmissionTCP) {
                                content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
                            }else {
                        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");
                        }else {
                            if (audioCodecId == 4) {
                                if (finalMediaTransmissionTCP) {
                                    content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n");
                                }else {
                                    content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n");
                                }
                                content.append("a=rtpmap:0 PCMU/8000\r\n");
                            }else {
                                if (finalMediaTransmissionTCP) {
                                    content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
                                }else {
                                    content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
                                }
                                content.append("a=rtpmap:8 PCMA/8000\r\n");
                            }
                        }
                        content.append("a=sendonly\r\n");
                        if (sendRtpItem.isTcp()) {
                            content.append("a=connection:new\r\n");
                            if (!sendRtpItem.isTcpActive()) {
                                content.append("a=setup:active\r\n");
                            }else {
                                content.append("a=setup:passive\r\n");
                            }
                        }
                        content.append("y="+ finalSsrc + "\r\n");
                        content.append("f=v/////a/1/8/1\r\n");


 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java

@@ -253,6 +253,10 @@
        return sendPost(mediaServerItem, "startSendRtp",param, null);
    }

    public JSONObject startSendRtpPassive(MediaServerItem mediaServerItem, Map<String, Object> param) {
        return sendPost(mediaServerItem, "startSendRtpPassive",param, null);
    }

    public JSONObject stopSendRtp(MediaServerItem mediaServerItem, Map<String, Object> param) {
        return sendPost(mediaServerItem, "stopSendRtp",param, null);
    }

 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java

@@ -250,6 +250,13 @@
    }

    /**
     * 调用zlm RESTFUL API —— startSendRtpPassive
     */
    public JSONObject startSendRtpPassive(MediaServerItem mediaServerItem, Map<String, Object>param) {
        return zlmresTfulUtils.startSendRtpPassive(mediaServerItem, param);
    }

    /**
     * 查询待转推的流是否就绪
     */
    public Boolean isRtpReady(MediaServerItem mediaServerItem, String streamId) {