合并开源主线

648540858

2023-04-04 b4048fbe80dba8e7756ae557a15ab60b4f80a44b

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

@@ -8,6 +8,7 @@
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.session.SsrcConfig;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.SIPSender;
@@ -202,7 +203,7 @@

                MediaServerItem mediaServerItem = null;
                StreamPushItem streamPushItem = null;
                StreamProxyItem proxyByAppAndStream =null;
                StreamProxyItem proxyByAppAndStream = null;
                // 不是通道可能是直播流
                if (channel != null && gbStream == null) {
                    // 通道存在，发100，TRYING
@@ -275,7 +276,7 @@
                    }
                    return;
                } else {
                    logger.info("通道不存在，返回404");
                    logger.info("通道不存在，返回404: {}", channelId);
                    try {
                        // 通道不存在，发404，资源不存在
                        responseAck(request, Response.NOT_FOUND);
@@ -360,7 +361,7 @@
                    return;
                }
                String username = sdp.getOrigin().getUsername();
                String addressStr = sdp.getOrigin().getAddress();
                String addressStr = sdp.getConnection().getAddress();

                logger.info("[上级点播]用户：{}， 通道：{}, 地址：{}:{}， ssrc：{}", username, channelId, addressStr, port, ssrc);
                Device device = null;
@@ -457,12 +458,8 @@
                            if (!userSetting.getPushStreamAfterAck()) {
                                playService.startPushStream(sendRtpItem, sipResponse, platform, request.getCallIdHeader());
                            }
                        } catch (SipException e) {
                            e.printStackTrace();
                        } catch (InvalidArgumentException e) {
                            e.printStackTrace();
                        } catch (ParseException e) {
                            e.printStackTrace();
                        } catch (SipException | InvalidArgumentException | ParseException e) {
                            logger.error("[命令发送失败] 国标级联 回复SdpAck", e);
                        }
                    };
                    SipSubscribe.Event errorEvent = ((event) -> {
@@ -471,14 +468,14 @@
                            Response response = getMessageFactory().createResponse(event.statusCode, evt.getRequest());
                            sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response);
                        } catch (ParseException | SipException  e) {
                            e.printStackTrace();
                            logger.error("未处理的异常 ", e);
                        }
                    });
                    sendRtpItem.setApp("rtp");
                    if ("Playback".equalsIgnoreCase(sessionName)) {
                        sendRtpItem.setPlayType(InviteStreamType.PLAYBACK);
                        SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, device.isSsrcCheck(), true);
                        sendRtpItem.setStreamId(ssrcInfo.getStream());
                        sendRtpItem.setStream(ssrcInfo.getStream());
                        // 写入redis， 超时时回复
                        redisCatchStorage.updateSendRTPSever(sendRtpItem);
                        playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start),
@@ -523,7 +520,7 @@
                            }
                            SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, device.isSsrcCheck(), false);
                            logger.info(JSONObject.toJSONString(ssrcInfo));
                            sendRtpItem.setStreamId(ssrcInfo.getStream());
                            sendRtpItem.setStream(ssrcInfo.getStream());
                            sendRtpItem.setSsrc(ssrc.equals(ssrcDefault) ? ssrcInfo.getSsrc() : ssrc);

                            // 写入redis， 超时时回复
@@ -533,16 +530,25 @@
                                redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), finalChannelId, callIdHeader.getCallId(), null);
                            });
                        } else {
                            sendRtpItem.setStreamId(playTransaction.getStream());
                            sendRtpItem.setStream(playTransaction.getStream());
                            // 写入redis， 超时时回复
                            redisCatchStorage.updateSendRTPSever(sendRtpItem);
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put("app", sendRtpItem.getApp());
                            jsonObject.put("stream", sendRtpItem.getStreamId());
                            jsonObject.put("stream", sendRtpItem.getStream());
                            hookEvent.response(mediaServerItem, jsonObject);
                        }
                    }
                } else if (gbStream != null) {
                    if(ssrc.equals(ssrcDefault))
                    {
                        SsrcConfig ssrcConfig = mediaServerItem.getSsrcConfig();
                        if(ssrcConfig != null)
                        {
                            ssrc = ssrcConfig.getPlaySsrc();
                            ssrcConfig.releaseSsrc(ssrc);
                        }
                    }
                    if("push".equals(gbStream.getStreamType())) {
                        if (streamPushItem != null && streamPushItem.isPushIng()) {
                            // 推流状态
@@ -572,7 +578,7 @@
        } catch (SdpParseException e) {
            logger.error("sdp解析错误", e);
        } catch (SdpException e) {
            e.printStackTrace();
            logger.error("未处理的异常 ", e);
        }
    }

@@ -727,11 +733,11 @@
                    mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
                    responseAck(request, Response.REQUEST_TIMEOUT); // 超时
                } catch (SipException e) {
                    e.printStackTrace();
                    logger.error("未处理的异常 ", e);
                } catch (InvalidArgumentException e) {
                    e.printStackTrace();
                    logger.error("未处理的异常 ", e);
                } catch (ParseException e) {
                    e.printStackTrace();
                    logger.error("未处理的异常 ", e);
                }
            }, userSetting.getPlatformPlayTimeout());
            // 添加监听
@@ -750,11 +756,11 @@
                        try {
                            responseAck(request, Response.BUSY_HERE);
                        } catch (SipException e) {
                            e.printStackTrace();
                            logger.error("未处理的异常 ", e);
                        } catch (InvalidArgumentException e) {
                            e.printStackTrace();
                            logger.error("未处理的异常 ", e);
                        } catch (ParseException e) {
                            e.printStackTrace();
                            logger.error("未处理的异常 ", e);
                        }
                        return;
                    }
@@ -812,11 +818,11 @@
                        try {
                            responseAck(request, Response.BUSY_HERE);
                        } catch (SipException e) {
                            e.printStackTrace();
                            logger.error("未处理的异常 ", e);
                        } catch (InvalidArgumentException e) {
                            e.printStackTrace();
                            logger.error("未处理的异常 ", e);
                        } catch (ParseException e) {
                            e.printStackTrace();
                            logger.error("未处理的异常 ", e);
                        }
                        return;
                    }
@@ -869,7 +875,13 @@
        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 96\r\n");
        // 非严格模式端口不统一, 增加兼容性，修改为一个不为0的端口
        int localPort = sendRtpItem.getLocalPort();
        if(localPort == 0)
        {
            localPort = new Random().nextInt(65535) + 1;
        }
        content.append("m=video " + localPort + " RTP/AVP 96\r\n");
        content.append("a=sendonly\r\n");
        content.append("a=rtpmap:96 PS/90000\r\n");
        if (sendRtpItem.isTcp()) {
@@ -890,21 +902,47 @@
            }
            return sipResponse;
        } catch (SipException e) {
            e.printStackTrace();
            logger.error("未处理的异常 ", e);
        } catch (InvalidArgumentException e) {
            e.printStackTrace();
            logger.error("未处理的异常 ", e);
        } catch (ParseException e) {
            e.printStackTrace();
            logger.error("未处理的异常 ", e);
        }
        return null;
    }

    public void inviteFromDeviceHandle(SIPRequest request, String requesterId, String channelId) {

        String realChannelId = null;

        // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
        Device device = redisCatchStorage.getDevice(requesterId);
        AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(requesterId, channelId);
        if (audioBroadcastCatch == null) {
        // 判断requesterId是设备还是通道
        if (device == null) {
            device = storager.queryVideoDeviceByChannelId(requesterId);
            realChannelId = requesterId;
        }
        if (device == null) {
            // 检查channelID是否可用
            device = redisCatchStorage.getDevice(channelId);
            if (device == null) {
                device = storager.queryVideoDeviceByChannelId(channelId);
                realChannelId = channelId;
            }
        }

        if (device == null) {
            logger.warn("来自设备的Invite请求，无法从请求信息中确定所属设备，已忽略，requesterId： {}/{}", requesterId, channelId);
            try {
                responseAck(request, Response.FORBIDDEN);
            } catch (SipException | InvalidArgumentException | ParseException e) {
                logger.error("[命令发送失败] 来自设备的Invite请求，无法从请求信息中确定所属设备 FORBIDDEN: {}", e.getMessage());
            }
            return;
        }

        AudioBroadcastCatch broadcastCatch = audioBroadcastManager.get(device.getDeviceId(), realChannelId);
        if (broadcastCatch == null) {
            logger.warn("来自设备的Invite请求非语音广播，已忽略，requesterId： {}/{}", requesterId, channelId);
            try {
                responseAck(request, Response.FORBIDDEN);
@@ -915,13 +953,13 @@
        }
        if (device != null) {
            logger.info("收到设备" + requesterId + "的语音广播Invite请求");
            String key = VideoManagerConstants.BROADCAST_WAITE_INVITE +  device.getDeviceId() + audioBroadcastCatch.getChannelId();
            String key = VideoManagerConstants.BROADCAST_WAITE_INVITE +  device.getDeviceId() + broadcastCatch.getChannelId();
            dynamicTask.stop(key);
            try {
                responseAck(request, Response.TRYING);
            } catch (SipException | InvalidArgumentException | ParseException e) {
                logger.error("[命令发送失败] invite BAD_REQUEST: {}", e.getMessage());
                playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
                playService.stopAudioBroadcast(device.getDeviceId(), broadcastCatch.getChannelId());
                return;
            }
            String contentString = new String(request.getRawContent());
@@ -977,7 +1015,7 @@
                        responseAck(request, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
                    } catch (SipException | InvalidArgumentException | ParseException e) {
                        logger.error("[命令发送失败] invite 不支持的媒体格式: {}", e.getMessage());
                        playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
                        playService.stopAudioBroadcast(device.getDeviceId(), broadcastCatch.getChannelId());
                        return;
                    }
                    return;
@@ -986,19 +1024,22 @@
                logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}, {}", requesterId, addressStr, port, ssrc,
                        mediaTransmissionTCP ? (tcpActive? "TCP主动":"TCP被动") : "UDP");

                MediaServerItem mediaServerItem = playService.getNewMediaServerItem(device);
                MediaServerItem mediaServerItem = broadcastCatch.getMediaServerItem();
                if (mediaServerItem == null) {
                    logger.warn("未找到可用的zlm");
                    logger.warn("未找到语音喊话使用的zlm");
                    try {
                        responseAck(request, Response.BUSY_HERE);
                    } catch (SipException | InvalidArgumentException | ParseException e) {
                        logger.error("[命令发送失败] invite 未找到可用的zlm: {}", e.getMessage());
                        playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
                        playService.stopAudioBroadcast(device.getDeviceId(), broadcastCatch.getChannelId());
                    }
                    return;
                }
                logger.info("设备{}请求语音流， 收流地址：{}:{}，ssrc：{}, {}, 对讲方式：{}", requesterId, addressStr, port, ssrc,
                        mediaTransmissionTCP ? (tcpActive? "TCP主动":"TCP被动") : "UDP", sdp.getSessionName().getValue());

                SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                        device.getDeviceId(), audioBroadcastCatch.getChannelId(),
                        device.getDeviceId(), broadcastCatch.getChannelId(),
                        mediaTransmissionTCP, false);

                if (sendRtpItem == null) {
@@ -1007,22 +1048,20 @@
                        responseAck(request, Response.BUSY_HERE);
                    } catch (SipException | InvalidArgumentException | ParseException e) {
                        logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage());
                        playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
                        playService.stopAudioBroadcast(device.getDeviceId(), broadcastCatch.getChannelId());
                        return;
                    }
                    return;
                }

                String app = "broadcast";
                String stream = device.getDeviceId() + "_" + audioBroadcastCatch.getChannelId();

                CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
                sendRtpItem.setPlayType(InviteStreamType.TALK);
                sendRtpItem.setPlayType(InviteStreamType.BROADCAST);
                sendRtpItem.setCallId(callIdHeader.getCallId());
                sendRtpItem.setPlatformId(requesterId);
                sendRtpItem.setStatus(1);
                sendRtpItem.setApp(app);
                sendRtpItem.setStreamId(stream);
                sendRtpItem.setApp(broadcastCatch.getApp());
                sendRtpItem.setStream(broadcastCatch.getStream());
                sendRtpItem.setPt(8);
                sendRtpItem.setUsePs(false);
                sendRtpItem.setRtcp(false);
@@ -1034,22 +1073,22 @@

                redisCatchStorage.updateSendRTPSever(sendRtpItem);

                Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream);
                Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, broadcastCatch.getApp(), broadcastCatch.getStream());
                if (streamReady) {
                    sendOk(device, sendRtpItem, sdp, request, mediaServerItem, mediaTransmissionTCP, ssrc);
                }else {
                    logger.warn("[语音通话]， 未发现待推送的流,app={},stream={}", app, stream);
                    logger.warn("[语音通话]， 未发现待推送的流,app={},stream={}", broadcastCatch.getApp(), broadcastCatch.getStream());
                    try {
                        responseAck(request, Response.GONE);
                    } catch (SipException | InvalidArgumentException | ParseException e) {
                        logger.error("[命令发送失败] 语音通话 回复410失败， {}", e.getMessage());
                        return;
                    }
                    playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
                    playService.stopAudioBroadcast(device.getDeviceId(), broadcastCatch.getChannelId());
                }
            } catch (SdpException e) {
                logger.error("[SDP解析异常]", e);
                playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
                playService.stopAudioBroadcast(device.getDeviceId(), broadcastCatch.getChannelId());
            }
        } else {
            logger.warn("来自无效设备/平台的请求");
@@ -1112,7 +1151,7 @@
            }

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