优化tcp主动方式的语音对讲

648540858

2022-12-19 fc90cd7951600ce5173f71c3e28d78e69b4db4ae

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

@@ -11,6 +11,7 @@
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.SIPSender;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
@@ -97,7 +98,7 @@
    private IMediaServerService mediaServerService;

    @Autowired
    private IMediaService mediaService;
    private ISIPCommander commander;

   @Autowired
   private ZLMRESTfulUtils zlmresTfulUtils;
@@ -385,8 +386,7 @@
                        return;
                    }
                    SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                            device.getDeviceId(), channelId,
                            mediaTransmissionTCP, platform.isRtcp());
                            device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp());

                    if (tcpActive != null) {
                        sendRtpItem.setTcpActive(tcpActive);
@@ -439,18 +439,23 @@

                        try {
                            // 超时未收到Ack应该回复bye,当前等待时间为10秒
                            dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
                                logger.info("Ack 等待超时");
                                mediaServerService.releaseSsrc(mediaServerItemInUSe.getId(), sendRtpItem.getSsrc());
                                // 回复bye
                                try {
                                    cmderFroPlatform.streamByeCmd(platform, callIdHeader.getCallId());
                                } catch (SipException | InvalidArgumentException | ParseException e) {
                                    logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
                                }
                            }, 60 * 1000);
                            responseSdpAck(request, content.toString(), platform);
                            if (userSetting.getPushStreamAfterAck()) {
                                dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
                                    logger.info("Ack 等待超时");
                                    mediaServerService.releaseSsrc(mediaServerItemInUSe.getId(), sendRtpItem.getSsrc());
                                    // 回复bye
                                    try {
                                        cmderFroPlatform.streamByeCmd(platform, callIdHeader.getCallId());
                                    } catch (SipException | InvalidArgumentException | ParseException e) {
                                        logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
                                    }
                                }, 60 * 1000);
                            }

                            SIPResponse sipResponse = responseSdpAck(request, content.toString(), platform);
                            if (!userSetting.getPushStreamAfterAck()) {
                                playService.startPushStream(sendRtpItem, sipResponse, platform, request.getCallIdHeader());
                            }
                        } catch (SipException e) {
                            e.printStackTrace();
                        } catch (InvalidArgumentException e) {
@@ -581,8 +586,7 @@
            if (streamReady) {
                // 自平台内容
                SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                        gbStream.getApp(), gbStream.getStream(), channelId,
                        mediaTransmissionTCP, platform.isRtcp());
                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());

                if (sendRtpItem == null) {
                    logger.warn("服务器端口资源不足");
@@ -621,8 +625,7 @@
            if (streamReady) {
                // 自平台内容
                SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                        gbStream.getApp(), gbStream.getStream(), channelId,
                        mediaTransmissionTCP, platform.isRtcp());
                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());

                if (sendRtpItem == null) {
                    logger.warn("服务器端口资源不足");
@@ -801,7 +804,7 @@
        // 发送redis消息
        redisGbPlayMsgListener.sendMsg(streamPushItem.getServerId(), streamPushItem.getMediaServerId(),
                streamPushItem.getApp(), streamPushItem.getStream(), addressStr, port, ssrc, requesterId,
                channelId, mediaTransmissionTCP, platform.isRtcp(), null, responseSendItemMsg -> {
                channelId, mediaTransmissionTCP, platform.isRtcp(),null, responseSendItemMsg -> {
                    SendRtpItem sendRtpItem = responseSendItemMsg.getSendRtpItem();
                    if (sendRtpItem == null || responseSendItemMsg.getMediaServerItem() == null) {
                        logger.warn("服务器端口资源不足");
@@ -880,7 +883,11 @@
        content.append("f=\r\n");

        try {
            return responseSdpAck(request, content.toString(), platform);
            SIPResponse sipResponse = responseSdpAck(request, content.toString(), platform);
            if (!userSetting.getPushStreamAfterAck()) {
                playService.startPushStream(sendRtpItem, sipResponse, platform, request.getCallIdHeader());
            }
            return sipResponse;
        } catch (SipException e) {
            e.printStackTrace();
        } catch (InvalidArgumentException e) {
@@ -970,7 +977,8 @@
                    return;
                }
                String addressStr = sdp.getOrigin().getAddress();
                logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", requesterId, addressStr, port, ssrc);
                logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}, {}", requesterId, addressStr, port, ssrc,
                        mediaTransmissionTCP ? (tcpActive? "TCP主动":"TCP被动") : "UDP");

                MediaServerItem mediaServerItem = playService.getNewMediaServerItem(device);
                if (mediaServerItem == null) {
@@ -995,15 +1003,12 @@
                    }
                    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.setPlayType(InviteStreamType.TALK);
                sendRtpItem.setCallId(callIdHeader.getCallId());
                sendRtpItem.setPlatformId(requesterId);
                sendRtpItem.setStatus(1);
@@ -1013,16 +1018,23 @@
                sendRtpItem.setUsePs(false);
                sendRtpItem.setRtcp(false);
                sendRtpItem.setOnlyAudio(true);
                sendRtpItem.setTcp(mediaTransmissionTCP);
                if (tcpActive != null) {
                    sendRtpItem.setTcpActive(tcpActive);
                }

                redisCatchStorage.updateSendRTPSever(sendRtpItem);

                Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream);
                if (streamReady) {
                    SIPResponse sipResponse = sendOk(device, sendRtpItem, sdp, request, mediaServerItem, mediaTransmissionTCP, ssrc);
                    // 添加事务信息
                    streamSession.put(device.getDeviceId(), audioBroadcastCatch.getChannelId(), request.getCallIdHeader().getCallId()
                            , stream,  sendRtpItem.getSsrc(), mediaServerItem.getId(), sipResponse, VideoStreamSessionManager.SessionType.broadcast );
                    sendOk(device, sendRtpItem, sdp, request, mediaServerItem, mediaTransmissionTCP, ssrc);
                }else {
                    logger.warn("[语音通话]， 未发现待推送的流,app={},stream={}", app, stream);
                    try {
                        responseAck(request, Response.GONE);
                    } catch (SipException | InvalidArgumentException | ParseException e) {
                        logger.error("[命令发送失败] 语音通话 回复410失败， {}", e.getMessage());
                    }
                    playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
                }
            } catch (SdpException e) {
@@ -1083,6 +1095,11 @@
            audioBroadcastCatch.setSipTransactionInfoByRequset(sipResponse);
            audioBroadcastManager.update(audioBroadcastCatch);

            // 开启发流，大华在收到200OK后就会开始建立连接
            if (!userSetting.getPushStreamAfterAck()) {
                playService.startPushStream(sendRtpItem, sipResponse, parentPlatform, request.getCallIdHeader());
            }

        } catch (SipException | InvalidArgumentException | ParseException | SdpParseException e) {
            logger.error("[命令发送失败] 语音对讲 回复200OK（SDP）: {}", e.getMessage());
        }