648540858
2023-06-29 c014a90cc6a294dfc2aac740be87e75f44193a29
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
 
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
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;
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.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
 
import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.header.CallIdHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
 
/**
 * SIP命令类型: BYE请求
 */
@Component
public class ByeRequestProcessor extends SIPRequestProcessorParent implements InitializingBean, ISIPRequestProcessor {
 
    private final Logger logger = LoggerFactory.getLogger(ByeRequestProcessor.class);
    private final String method = "BYE";
 
    @Autowired
    private ISIPCommander cmder;
 
    @Autowired
    private ISIPCommanderForPlatform commanderForPlatform;
 
    @Autowired
    private IRedisCatchStorage redisCatchStorage;
 
    @Autowired
    private IInviteStreamService inviteStreamService;
 
    @Autowired
    private IPlatformService platformService;
 
    @Autowired
    private IDeviceService deviceService;
 
    @Autowired
    private AudioBroadcastManager audioBroadcastManager;
 
    @Autowired
    private IDeviceChannelService channelService;
 
    @Autowired
    private IVideoManagerStorage storager;
 
    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;
 
    @Autowired
    private SSRCFactory ssrcFactory;
 
    @Autowired
    private IMediaServerService mediaServerService;
 
    @Autowired
    private SIPProcessorObserver sipProcessorObserver;
 
    @Autowired
    private VideoStreamSessionManager streamSession;
 
    @Autowired
    private IPlayService playService;
 
    @Autowired
    private UserSetting userSetting;
 
    @Override
    public void afterPropertiesSet() throws Exception {
        // 添加消息处理的订阅
        sipProcessorObserver.addRequestProcessor(method, this);
    }
 
    /**
     * 处理BYE请求
     * @param evt
     */
    @Override
    public void process(RequestEvent evt) {
        SIPRequest request = (SIPRequest) evt.getRequest();
        try {
            responseAck(request, Response.OK);
        } catch (SipException | InvalidArgumentException | ParseException e) {
            logger.error("[回复BYE信息失败],{}", e.getMessage());
        }
        CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME);
        SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(null, null, null, callIdHeader.getCallId());
 
        if (sendRtpItem != null){
            logger.info("[收到bye] 来自平台{}, 停止通道:{}", sendRtpItem.getPlatformId(), sendRtpItem.getChannelId());
            String streamId = sendRtpItem.getStream();
            Map<String, Object> param = new HashMap<>();
            param.put("vhost","__defaultVhost__");
            param.put("app",sendRtpItem.getApp());
            param.put("stream",streamId);
            param.put("ssrc",sendRtpItem.getSsrc());
            logger.info("[收到bye] 停止向上级推流:{}", streamId);
            MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
            redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(),
                    callIdHeader.getCallId(), null);
            zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
            if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
                ParentPlatform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
                if (platform != null) {
                    MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
                            sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),
                            sendRtpItem.getPlatformId(), platform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
                    messageForPushChannel.setPlatFormIndex(platform.getId());
                    redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
                }else {
                    logger.info("[上级平台停止观看] 未找到平台{}的信息,发送redis消息失败", sendRtpItem.getPlatformId());
                }
            }
 
            int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
            if (totalReaderCount <= 0) {
                logger.info("[收到bye] {} 无其它观看者,通知设备停止推流", streamId);
                if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) {
                    Device device = deviceService.getDevice(sendRtpItem.getDeviceId());
                    if (device == null) {
                        logger.info("[收到bye] {} 通知设备停止推流时未找到设备信息", streamId);
                    }
                    try {
                        logger.warn("[停止点播] {}/{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
                        cmder.streamByeCmd(device, sendRtpItem.getChannelId(), streamId, null);
                    } catch (InvalidArgumentException | ParseException | SipException |
                             SsrcTransactionNotFoundException e) {
                        logger.error("[收到bye] {} 无其它观看者,通知设备停止推流, 发送BYE失败 {}",streamId, e.getMessage());
                    }
                }
            }
        }
 
 
 
            // 可能是设备发送的停止
            SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(null, null, callIdHeader.getCallId(), null);
            if (ssrcTransaction == null) {
                logger.info("[收到bye] 但是无法获取推流信息和发流信息,忽略此请求");
                logger.info(request.toString());
                return;
            }
            logger.info("[收到bye] 来自设备:{}, 通道已停止推流: {}", ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
 
            Device device = deviceService.getDevice(ssrcTransaction.getDeviceId());
            if (device == null) {
                logger.info("[收到bye] 未找到设备:{} ", ssrcTransaction.getDeviceId());
                return;
            }
            DeviceChannel channel = channelService.getOne(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
            if (channel == null) {
                logger.info("[收到bye] 未找到通道,设备:{}, 通道:{}", ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
                return;
            }
            storager.stopPlay(device.getDeviceId(), channel.getChannelId());
            InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channel.getChannelId());
            if (inviteInfo != null) {
                inviteStreamService.removeInviteInfo(inviteInfo);
                if (inviteInfo.getStreamInfo() != null) {
                    mediaServerService.closeRTPServer(inviteInfo.getStreamInfo().getMediaServerId(), inviteInfo.getStreamInfo().getStream());
                }
            }
            // 释放ssrc
            MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransaction.getMediaServerId());
            if (mediaServerItem != null) {
                mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransaction.getSsrc());
            }
            streamSession.remove(device.getDeviceId(), channel.getChannelId(), ssrcTransaction.getStream());
            if (ssrcTransaction.getType() == InviteSessionType.BROADCAST) {
                // 查找来源的对讲设备,发送停止
                Device sourceDevice = storager.queryVideoDeviceByPlatformIdAndChannelId(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
                if (sourceDevice != null) {
                    playService.stopAudioBroadcast(sourceDevice.getDeviceId(), channel.getChannelId());
                }
            }
            AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(ssrcTransaction.getDeviceId(), channel.getChannelId());
            if (audioBroadcastCatch != null) {
                // 来自上级平台的停止对讲
                logger.info("[停止对讲] 来自上级,平台:{}, 通道:{}", ssrcTransaction.getDeviceId(), channel.getChannelId());
                audioBroadcastManager.del(ssrcTransaction.getDeviceId(), channel.getChannelId());
            }
    }
}