合并主线

648540858

2022-06-27 1d1f1748521930515fdab58e767c410d26108959

合并主线

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java

@@ -4,7 +4,6 @@
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
@@ -26,12 +25,13 @@
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;


import com.genersoft.iot.vmp.utils.DateUtil;
import javax.sip.*;
import javax.sip.address.SipURI;
import javax.sip.header.*;
import javax.sip.message.Request;
import java.lang.reflect.Field;
import java.net.InetAddress;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.HashSet;

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

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

import com.alibaba.fastjson.JSON;
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.*;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
@@ -22,7 +20,6 @@
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.stack.SIPDialog;
import com.genersoft.iot.vmp.utils.SerializeUtils;
import org.ehcache.shadow.org.terracotta.offheapstore.storage.IntegerStorageEngine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
@@ -37,7 +34,6 @@
import javax.sip.header.CallIdHeader;
import javax.sip.header.FromHeader;
import javax.sip.header.HeaderAddress;
import javax.sip.header.ToHeader;
import java.text.ParseException;
import java.util.*;

@@ -99,28 +95,28 @@
    @Override
    public void process(RequestEvent evt) {
        Dialog dialog = evt.getDialog();
        CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME);
        CallIdHeader callIdHeader = (CallIdHeader) evt.getRequest().getHeader(CallIdHeader.NAME);
        if (dialog == null) {
            return;
        }
        if (dialog.getState()== DialogState.CONFIRMED) {
        if (dialog.getState() == DialogState.CONFIRMED) {
            String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser();
            logger.info("ACK请求： platformGbId->{}", platformGbId);
            ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformGbId);
            // 取消设置的超时任务
            dynamicTask.stop(callIdHeader.getCallId());
//            String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
            SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, null, null, callIdHeader.getCallId());
            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]，开始使用{}向上级推流 {}/{}->{}:{}({})", sendRtpItem.isTcp() ? "TCP" : "UDP",
                    sendRtpItem.getApp(), sendRtpItem.getStreamId(),
                    sendRtpItem.getIp() ,sendRtpItem.getPort(),
                    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("vhost", "__defaultVhost__");
            param.put("app", sendRtpItem.getApp());
            param.put("stream", sendRtpItem.getStreamId());
            param.put("ssrc", sendRtpItem.getSsrc());
            param.put("src_port", sendRtpItem.getLocalPort());
            param.put("pt", sendRtpItem.getPt());
@@ -129,9 +125,9 @@
            JSONObject jsonObject;
            if (sendRtpItem.isTcpActive()) {
                jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);
            }else {
            } else {
                param.put("is_udp", is_Udp);
                param.put("dst_url",sendRtpItem.getIp());
                param.put("dst_url", sendRtpItem.getIp());
                param.put("dst_port", sendRtpItem.getPort());
                jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
            }
@@ -149,36 +145,37 @@
                    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"));
                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));
                logger.error("RTP推流失败: {}, 参数：{}", jsonObject.getString("msg"), JSONObject.toJSON(param));
                if (sendRtpItem.isOnlyAudio()) {
                    // 语音对讲
                    try {
                        cmder.streamByeCmd((SIPDialog) evt.getDialog(), (SIPRequest)evt.getRequest(), null);
                        cmder.streamByeCmd((SIPDialog) evt.getDialog(), (SIPRequest) evt.getRequest(), null);
                    } catch (SipException e) {
                        throw new RuntimeException(e);
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }else {
                } else {
                    // 向上级平台
                    commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId());
                }
            if (mediaInfo == null) {
                RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance(
                        sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(),
                        sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(),
                        sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio());
                redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, jsonObject->{
                    startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader);
                });
            }else {
                JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
                startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader);
                if (mediaInfo == null) {
                    RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance(
                            sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(),
                            sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(),
                            sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio());
                    redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, json -> {
                        startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, json, param, callIdHeader);
                    });
                } else {
                    JSONObject startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
                    startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, startSendRtpStreamResult, param, callIdHeader);
                }


            }


        }
    }
    private void startSendRtpStreamHand(RequestEvent evt, SendRtpItem sendRtpItem, ParentPlatform parentPlatform,

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

@@ -19,15 +19,14 @@
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
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.StreamPushItem;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItemLite;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.IStreamPushService;
@@ -58,8 +57,6 @@
import javax.sip.message.Response;
import java.text.ParseException;
import java.time.Instant;
import java.util.Date;
import java.util.List;
import java.util.Vector;

/**
@@ -108,9 +105,6 @@

    @Autowired
    private ZLMRESTfulUtils zlmresTfulUtils;

    @Autowired
    private IMediaServerService mediaServerService;

    @Autowired
    private SIPProcessorObserver sipProcessorObserver;
@@ -171,7 +165,7 @@
            // 查询请求是否来自上级平台\设备
            ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
            if (platform == null) {
                inviteFromDeviceHandle(evt, requesterId);
                inviteFromDeviceHandle(evt, requesterId, channelId);
            } else {
                // 查询平台下是否有该通道
                DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId);
@@ -724,10 +718,16 @@
        }
    }

    public void inviteFromDeviceHandle(RequestEvent evt, String requesterId) throws InvalidArgumentException, ParseException, SipException, SdpException {
    public void inviteFromDeviceHandle(RequestEvent evt, String requesterId, String channelId1) throws InvalidArgumentException, ParseException, SipException, SdpException {

        // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
        Device device = redisCatchStorage.getDevice(requesterId);
        AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(requesterId, channelId1);
        if (audioBroadcastCatch == null) {
            logger.warn("来自设备的Invite请求非语音广播，已忽略");
            responseAck(evt, Response.FORBIDDEN);
            return;
        }
        Request request = evt.getRequest();
        if (device != null) {
            logger.info("收到设备" + requesterId + "的语音广播Invite请求");
@@ -740,7 +740,7 @@
            int ssrcIndex = contentString.indexOf("y=");
            if (ssrcIndex > 0) {
                substring = contentString.substring(0, ssrcIndex);
                ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
                ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12).trim();
            }
            ssrcIndex = substring.indexOf("f=");
            if (ssrcIndex > 0) {
@@ -750,13 +750,13 @@

            //  获取支持的格式
            Vector mediaDescriptions = sdp.getMediaDescriptions(true);

            // 查看是否支持PS 负载96
            int port = -1;
            //boolean recvonly = false;
            boolean mediaTransmissionTCP = false;
            Boolean tcpActive = null;
            for (int i = 0; i < mediaDescriptions.size(); i++) {
                MediaDescription mediaDescription = (MediaDescription) mediaDescriptions.get(i);
                MediaDescription mediaDescription = (MediaDescription)mediaDescriptions.get(i);
                Media media = mediaDescription.getMedia();

                Vector mediaFormats = media.getMediaFormats(false);
@@ -784,233 +784,248 @@
                responseAck(evt, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
                return;
            }
            String username = sdp.getOrigin().getUsername();
            String addressStr = sdp.getOrigin().getAddress();
            logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc);
            logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", requesterId, addressStr, port, ssrc);

            MediaServerItem mediaServerItem = playService.getNewMediaServerItem(device);
            if (mediaServerItem == null) {
                logger.warn("未找到可用的zlm");
                responseAck(evt, Response.BUSY_HERE);
                return;
            }
            SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                    device.getDeviceId(), audioBroadcastCatch.getChannelId(),
                    mediaTransmissionTCP);
            if (sendRtpItem == null) {
                logger.warn("服务器端口资源不足");
                responseAck(evt, Response.BUSY_HERE);
                return;
            }
            sendRtpItem.setTcp(mediaTransmissionTCP);
            if (tcpActive != null) {
                sendRtpItem.setTcpActive(tcpActive);
            }
            String app = "broadcast";
            String stream = device.getDeviceId() + "_" + audioBroadcastCatch.getChannelId();

            CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
            sendRtpItem.setPlayType(InviteStreamType.PLAY);
            sendRtpItem.setCallId(callIdHeader.getCallId());
            sendRtpItem.setPlatformId(requesterId);
            sendRtpItem.setStatus(1);
            sendRtpItem.setApp(app);
            sendRtpItem.setStreamId(stream);
            sendRtpItem.setPt(8);
            sendRtpItem.setUsePs(false);
            sendRtpItem.setOnlyAudio(true);
            redisCatchStorage.updateSendRTPSever(sendRtpItem);

            // hook监听等待设备推流上来
            // 添加订阅
            JSONObject subscribeKey = new JSONObject();
            subscribeKey.put("app", app);
            subscribeKey.put("stream", stream);
            subscribeKey.put("regist", true);
            subscribeKey.put("schema", "rtmp");
            subscribeKey.put("mediaServerId", mediaServerItem.getId());
            String finalSsrc = ssrc;
            // 流已经存在时直接推流
            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);
                StringBuffer content = new StringBuffer(200);
                content.append("v=0\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");
                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=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");

                ParentPlatform parentPlatform = new ParentPlatform();
                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);
                }
            }else {
                // 流不存在时监听流上线
                // 设置等待推流的超时; 默认20s
                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 {
                        responseAck(evt, Response.BUSY_HERE);
                    } catch (SipException e) {
                        throw new RuntimeException(e);
                    } catch (InvalidArgumentException e) {
                        throw new RuntimeException(e);
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }, 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);
                                StringBuffer content = new StringBuffer(200);
                                content.append("v=0\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");
                                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=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");

                                ParentPlatform parentPlatform = new ParentPlatform();
                                parentPlatform.setServerIP(device.getIp());
                                parentPlatform.setServerPort(device.getPort());
                                parentPlatform.setServerGBId(device.getDeviceId());

                                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);
                            }
                        });
            }
            String key = DeferredResultHolder.CALLBACK_CMD_BROADCAST + device.getDeviceId();
            WVPResult<AudioBroadcastResult> wvpResult = new WVPResult<>();
            wvpResult.setCode(0);
            wvpResult.setMsg("success");
            AudioBroadcastResult audioBroadcastResult = new AudioBroadcastResult();
            audioBroadcastResult.setApp(app);
            audioBroadcastResult.setStream(stream);
            audioBroadcastResult.setMediaServerItem(new MediaServerItemLite(mediaServerItem));
            audioBroadcastResult.setCodec("G.711");
            wvpResult.setData(audioBroadcastResult);
            RequestMessage requestMessage = new RequestMessage();
            requestMessage.setKey(key);
            requestMessage.setData(wvpResult);
            resultHolder.invokeAllResult(requestMessage);
        } else {
            logger.warn("来自无效设备/平台的请求");
            responseAck(evt, Response.BAD_REQUEST);
        }
    }
            CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
            sendRtpItem.setPlayType(InviteStreamType.PLAY);
            sendRtpItem.setCallId(callIdHeader.getCallId());
            sendRtpItem.setPlatformId(requesterId);
            sendRtpItem.setStatus(1);
            sendRtpItem.setApp(app);
            sendRtpItem.setStreamId(stream);
            sendRtpItem.setPt(8);
            sendRtpItem.setUsePs(false);
            sendRtpItem.setOnlyAudio(true);
            redisCatchStorage.updateSendRTPSever(sendRtpItem);

            // hook监听等待设备推流上来
            // 添加订阅
            JSONObject subscribeKey = new JSONObject();
            subscribeKey.put("app", app);
            subscribeKey.put("stream", stream);
            subscribeKey.put("regist", true);
            subscribeKey.put("schema", "rtmp");
            subscribeKey.put("mediaServerId", mediaServerItem.getId());
            String finalSsrc = ssrc;
            // 流已经存在时直接推流
            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);
                StringBuffer content = new StringBuffer(200);
                content.append("v=0\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");
                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=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");

                ParentPlatform parentPlatform = new ParentPlatform();
                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);
                }
            }else {
                // 流不存在时监听流上线
                // 设置等待推流的超时; 默认20s
                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 {
                        responseAck(evt, Response.BUSY_HERE);
                    } catch (SipException e) {
                        throw new RuntimeException(e);
                    } catch (InvalidArgumentException e) {
                        throw new RuntimeException(e);
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }, 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);
                        StringBuffer content = new StringBuffer(200);
                        content.append("v=0\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");
                        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=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");

                        ParentPlatform parentPlatform = new ParentPlatform();
                        parentPlatform.setServerIP(device.getIp());
                        parentPlatform.setServerPort(device.getPort());
                        parentPlatform.setServerGBId(device.getDeviceId());

                        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);
                    }
                });
            }
            String key = DeferredResultHolder.CALLBACK_CMD_BROADCAST + device.getDeviceId();
            WVPResult<AudioBroadcastResult> wvpResult = new WVPResult<>();
            wvpResult.setCode(0);
            wvpResult.setMsg("success");
            AudioBroadcastResult audioBroadcastResult = new AudioBroadcastResult();
            audioBroadcastResult.setApp(app);
            audioBroadcastResult.setStream(stream);
            audioBroadcastResult.setMediaServerItem(new MediaServerItemLite(mediaServerItem));
            audioBroadcastResult.setCodec("G.711");
            wvpResult.setData(audioBroadcastResult);
            RequestMessage requestMessage = new RequestMessage();
            requestMessage.setKey(key);
            requestMessage.setData(wvpResult);
            resultHolder.invokeAllResult(requestMessage);
        } else {
            logger.warn("来自无效设备/平台的请求");
            responseAck(evt, Response.BAD_REQUEST);
        }
    }
}

 src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java

@@ -1,7 +1,6 @@
package com.genersoft.iot.vmp.service;

import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;

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

@@ -6,7 +6,6 @@
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.session.SsrcConfig;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
@@ -34,11 +33,8 @@
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.StringUtils;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 媒体服务器节点管理

 web_src/package-lock.json

@@ -50,7 +50,7 @@
        "postcss-url": "^7.2.1",
        "rimraf": "^2.6.0",
        "semver": "^5.3.0",
        "shelljs": "^0.7.6",
        "shelljs": "^0.8.5",
        "uglifyjs-webpack-plugin": "^1.1.1",
        "url-loader": "^0.5.8",
        "vue-loader": "^13.3.0",
@@ -11655,9 +11655,9 @@
      }
    },
    "node_modules/shelljs": {
      "version": "0.7.8",
      "resolved": "https://registry.npm.taobao.org/shelljs/download/shelljs-0.7.8.tgz",
      "integrity": "sha1-3svPh0sNHl+3LhSxZKloMEjprLM=",
      "version": "0.8.5",
      "resolved": "https://registry.npmmirror.com/shelljs/-/shelljs-0.8.5.tgz",
      "integrity": "sha512-TiwcRcrkhHvbrZbnRcFYMLl30Dfov3HKqzp5tO5b4pt6G/SezKcYhmDg15zXVBswHmctSAQKznqNW2LO5tTDow==",
      "dev": true,
      "dependencies": {
        "glob": "^7.0.0",
@@ -11668,8 +11668,7 @@
        "shjs": "bin/shjs"
      },
      "engines": {
        "iojs": "*",
        "node": ">=0.11.0"
        "node": ">=4"
      }
    },
    "node_modules/shellwords": {
@@ -24156,9 +24155,9 @@
      "dev": true
    },
    "shelljs": {
      "version": "0.7.8",
      "resolved": "https://registry.npm.taobao.org/shelljs/download/shelljs-0.7.8.tgz",
      "integrity": "sha1-3svPh0sNHl+3LhSxZKloMEjprLM=",
      "version": "0.8.5",
      "resolved": "https://registry.npmmirror.com/shelljs/-/shelljs-0.8.5.tgz",
      "integrity": "sha512-TiwcRcrkhHvbrZbnRcFYMLl30Dfov3HKqzp5tO5b4pt6G/SezKcYhmDg15zXVBswHmctSAQKznqNW2LO5tTDow==",
      "dev": true,
      "requires": {
        "glob": "^7.0.0",