package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.gb28181.SipLayer; import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.transmit.SIPSender; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider; import com.genersoft.iot.vmp.gb28181.utils.SipUtils; import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo; import com.genersoft.iot.vmp.utils.DateUtil; import gov.nist.javax.sip.message.MessageFactoryImpl; import gov.nist.javax.sip.message.SIPRequest; import gov.nist.javax.sip.message.SIPResponse; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.DependsOn; import org.springframework.lang.Nullable; import org.springframework.stereotype.Component; import org.springframework.util.ObjectUtils; import javax.sip.InvalidArgumentException; import javax.sip.ResponseEvent; import javax.sip.SipException; import javax.sip.header.CallIdHeader; import javax.sip.header.WWWAuthenticateHeader; import javax.sip.message.Request; import java.text.ParseException; import java.util.ArrayList; import java.util.List; @Component @DependsOn("sipLayer") public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform { private final Logger logger = LoggerFactory.getLogger(SIPCommanderFroPlatform.class); @Autowired private SIPRequestHeaderPlarformProvider headerProviderPlatformProvider; @Autowired private IRedisCatchStorage redisCatchStorage; @Autowired private IMediaServerService mediaServerService; @Autowired private SipSubscribe sipSubscribe; @Autowired private ZLMRTPServerFactory zlmrtpServerFactory; @Autowired private SipLayer sipLayer; @Autowired private SIPSender sipSender; @Autowired private ZlmHttpHookSubscribe subscribe; @Autowired private UserSetting userSetting; @Autowired private VideoStreamSessionManager streamSession; @Autowired private DynamicTask dynamicTask; @Override public void register(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) throws InvalidArgumentException, ParseException, SipException { register(parentPlatform, null, null, errorEvent, okEvent, false, true); } @Override public void unregister(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) throws InvalidArgumentException, ParseException, SipException { register(parentPlatform, null, null, errorEvent, okEvent, false, false); } @Override public void register(ParentPlatform parentPlatform, @Nullable String callId, @Nullable WWWAuthenticateHeader www, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent, boolean registerAgain, boolean isRegister) throws SipException, InvalidArgumentException, ParseException { Request request; if (!registerAgain ) { CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport()); request = headerProviderPlatformProvider.createRegisterRequest(parentPlatform, redisCatchStorage.getCSEQ(), SipUtils.getNewFromTag(), SipUtils.getNewViaTag(), callIdHeader, isRegister); // 将 callid 写入缓存, 等注册成功可以更新状态 String callIdFromHeader = callIdHeader.getCallId(); redisCatchStorage.updatePlatformRegisterInfo(callIdFromHeader, PlatformRegisterInfo.getInstance(parentPlatform.getServerGBId(), isRegister)); sipSubscribe.addErrorSubscribe(callIdHeader.getCallId(), (event)->{ if (event != null) { logger.info("向上级平台 [ {} ] 注册发生错误: {} ", parentPlatform.getServerGBId(), event.msg); } redisCatchStorage.delPlatformRegisterInfo(callIdFromHeader); if (errorEvent != null ) { errorEvent.response(event); } }); }else { CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport()); request = headerProviderPlatformProvider.createRegisterRequest(parentPlatform, SipUtils.getNewFromTag(), null, callId, www, callIdHeader, isRegister); } sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, null, okEvent); } @Override public String keepalive(ParentPlatform parentPlatform,SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) throws SipException, InvalidArgumentException, ParseException { String characterSet = parentPlatform.getCharacterSet(); StringBuffer keepaliveXml = new StringBuffer(200); keepaliveXml.append("\r\n") .append("\r\n") .append("Keepalive\r\n") .append("" + (int)((Math.random()*9+1)*100000) + "\r\n") .append("" + parentPlatform.getDeviceGBId() + "\r\n") .append("OK\r\n") .append("\r\n"); CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport()); Request request = headerProviderPlatformProvider.createMessageRequest( parentPlatform, keepaliveXml.toString(), SipUtils.getNewFromTag(), SipUtils.getNewViaTag(), callIdHeader); sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, errorEvent, okEvent); return callIdHeader.getCallId(); } /** * 向上级回复通道信息 * @param channel 通道信息 * @param parentPlatform 平台信息 */ @Override public void catalogQuery(DeviceChannel channel, ParentPlatform parentPlatform, String sn, String fromTag, int size) throws SipException, InvalidArgumentException, ParseException { if ( parentPlatform ==null) { return ; } List channels = new ArrayList<>(); if (channel != null) { channels.add(channel); } String catalogXml = getCatalogXml(channels, sn, parentPlatform, size); // callid CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport()); Request request = headerProviderPlatformProvider.createMessageRequest(parentPlatform, catalogXml.toString(), fromTag, SipUtils.getNewViaTag(), callIdHeader); sipSender.transmitRequest(parentPlatform.getDeviceIp(), request); } @Override public void catalogQuery(List channels, ParentPlatform parentPlatform, String sn, String fromTag) throws InvalidArgumentException, ParseException, SipException { if ( parentPlatform ==null) { return ; } sendCatalogResponse(channels, parentPlatform, sn, fromTag, 0, true); } private String getCatalogXml(List channels, String sn, ParentPlatform parentPlatform, int size) { String characterSet = parentPlatform.getCharacterSet(); StringBuffer catalogXml = new StringBuffer(600); catalogXml.append("\r\n") .append("\r\n") .append("Catalog\r\n") .append("" +sn + "\r\n") .append("" + parentPlatform.getDeviceGBId() + "\r\n") .append("" + size + "\r\n") .append("\r\n"); if (channels.size() > 0) { for (DeviceChannel channel : channels) { if (parentPlatform.getServerGBId().equals(channel.getParentId())) { channel.setParentId(parentPlatform.getDeviceGBId()); } catalogXml.append("\r\n"); // 行政区划分组只需要这两项就可以 catalogXml.append("" + channel.getChannelId() + "\r\n"); catalogXml.append("" + channel.getName() + "\r\n"); if (channel.getParentId() != null) { // 业务分组加上这一项即可,提高兼容性, catalogXml.append("" + channel.getParentId() + "\r\n"); // catalogXml.append("" + parentPlatform.getDeviceGBId() + "/" + channel.getParentId() + "\r\n"); } if (channel.getChannelId().length() == 20 && Integer.parseInt(channel.getChannelId().substring(10, 13)) == 216) { // 虚拟组织增加BusinessGroupID字段 catalogXml.append("" + channel.getParentId() + "\r\n"); } if (!channel.getChannelId().equals(parentPlatform.getDeviceGBId())) { catalogXml.append("" + channel.getParental() + "\r\n"); if (channel.getParental() == 0) { catalogXml.append("" + (channel.getStatus() == 0 ? "OFF" : "ON") + "\r\n"); } } if (channel.getParental() == 0) { // 通道项 catalogXml.append("" + channel.getManufacture() + "\r\n"); catalogXml.append("" + channel.getSecrecy() + "\r\n"); catalogXml.append("" + channel.getRegisterWay() + "\r\n"); String civilCode = channel.getCivilCode() == null?parentPlatform.getAdministrativeDivision() : channel.getCivilCode(); if (channel.getChannelType() != 2) { // 业务分组/虚拟组织/行政区划 不设置以下属性 catalogXml.append("" + channel.getModel() + "\r\n"); catalogXml.append("" + parentPlatform.getDeviceGBId()+ "\r\n"); catalogXml.append("" + civilCode + "\r\n"); if (channel.getAddress() == null) { catalogXml.append("
\r\n"); }else { catalogXml.append("
" + channel.getAddress() + "
\r\n"); } catalogXml.append("" + channel.getBlock() + "\r\n"); catalogXml.append("" + channel.getSafetyWay() + "\r\n"); catalogXml.append("" + channel.getCertNum() + "\r\n"); catalogXml.append("" + channel.getCertifiable() + "\r\n"); catalogXml.append("" + channel.getErrCode() + "\r\n"); catalogXml.append("" + channel.getEndTime() + "\r\n"); catalogXml.append("" + channel.getSecrecy() + "\r\n"); catalogXml.append("" + channel.getIpAddress() + "\r\n"); catalogXml.append("" + channel.getPort() + "\r\n"); catalogXml.append("" + channel.getPort() + "\r\n"); catalogXml.append("" + (channel.getStatus() == 1?"ON":"OFF") + "\r\n"); catalogXml.append("" + (channel.getLongitudeWgs84() != 0? channel.getLongitudeWgs84():channel.getLongitude()) + "\r\n"); catalogXml.append("" + (channel.getLatitudeWgs84() != 0? channel.getLatitudeWgs84():channel.getLatitude()) + "\r\n"); } } catalogXml.append("
\r\n"); } } catalogXml.append("
\r\n"); catalogXml.append("
\r\n"); return catalogXml.toString(); } private void sendCatalogResponse(List channels, ParentPlatform parentPlatform, String sn, String fromTag, int index, boolean sendAfterResponse) throws SipException, InvalidArgumentException, ParseException { if (index >= channels.size()) { return; } List deviceChannels; if (index + parentPlatform.getCatalogGroup() < channels.size()) { deviceChannels = channels.subList(index, index + parentPlatform.getCatalogGroup()); }else { deviceChannels = channels.subList(index, channels.size()); } String catalogXml = getCatalogXml(deviceChannels, sn, parentPlatform, channels.size()); // callid CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport()); SIPRequest request = (SIPRequest)headerProviderPlatformProvider.createMessageRequest(parentPlatform, catalogXml, fromTag, SipUtils.getNewViaTag(), callIdHeader); String timeoutTaskKey = "catalog_task_" + parentPlatform.getServerGBId() + sn; String callId = request.getCallIdHeader().getCallId(); if (sendAfterResponse) { // 默认按照收到200回复后发送下一条, 如果超时收不到回复,就以30毫秒的间隔直接发送。 dynamicTask.startDelay(timeoutTaskKey, ()->{ sipSubscribe.removeOkSubscribe(callId); int indexNext = index + parentPlatform.getCatalogGroup(); try { sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext, false); } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage()); } }, 3000); sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, eventResult -> { logger.error("[目录推送失败] 国标级联 platform : {}, code: {}, msg: {}, 停止发送", parentPlatform.getServerGBId(), eventResult.statusCode, eventResult.msg); dynamicTask.stop(timeoutTaskKey); }, eventResult -> { dynamicTask.stop(timeoutTaskKey); int indexNext = index + parentPlatform.getCatalogGroup(); try { sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext, true); } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage()); } }); }else { sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, eventResult -> { logger.error("[目录推送失败] 国标级联 platform : {}, code: {}, msg: {}, 停止发送", parentPlatform.getServerGBId(), eventResult.statusCode, eventResult.msg); dynamicTask.stop(timeoutTaskKey); }, null); dynamicTask.startDelay(timeoutTaskKey, ()->{ int indexNext = index + parentPlatform.getCatalogGroup(); try { sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext, false); } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage()); } }, 30); } } /** * 向上级回复DeviceInfo查询信息 * @param parentPlatform 平台信息 * @param sn * @param fromTag * @return */ @Override public void deviceInfoResponse(ParentPlatform parentPlatform,Device device, String sn, String fromTag) throws SipException, InvalidArgumentException, ParseException { if (parentPlatform == null) { return; } String characterSet = parentPlatform.getCharacterSet(); StringBuffer deviceInfoXml = new StringBuffer(600); deviceInfoXml.append("\r\n"); deviceInfoXml.append("\r\n"); deviceInfoXml.append("DeviceInfo\r\n"); deviceInfoXml.append("" +sn + "\r\n"); deviceInfoXml.append("" + device.getDeviceId() + "\r\n"); deviceInfoXml.append("" + device.getName() + "\r\n"); deviceInfoXml.append("" + device.getManufacturer() + "\r\n"); deviceInfoXml.append("" + device.getModel() + "\r\n"); deviceInfoXml.append("" + device.getFirmware() + "\r\n"); deviceInfoXml.append("OK\r\n"); deviceInfoXml.append("\r\n"); CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport()); Request request = headerProviderPlatformProvider.createMessageRequest(parentPlatform, deviceInfoXml.toString(), fromTag, SipUtils.getNewViaTag(), callIdHeader); sipSender.transmitRequest(parentPlatform.getDeviceIp(), request); } /** * 向上级回复DeviceStatus查询信息 * @param parentPlatform 平台信息 * @param sn * @param fromTag * @return */ @Override public void deviceStatusResponse(ParentPlatform parentPlatform,String channelId, String sn, String fromTag,int status) throws SipException, InvalidArgumentException, ParseException { if (parentPlatform == null) { return ; } String statusStr = (status==1)?"ONLINE":"OFFLINE"; String characterSet = parentPlatform.getCharacterSet(); StringBuffer deviceStatusXml = new StringBuffer(600); deviceStatusXml.append("\r\n") .append("\r\n") .append("DeviceStatus\r\n") .append("" +sn + "\r\n") .append("" + channelId + "\r\n") .append("OK\r\n") .append(""+statusStr+"\r\n") .append("OK\r\n") .append("\r\n"); CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport()); Request request = headerProviderPlatformProvider.createMessageRequest(parentPlatform, deviceStatusXml.toString(), fromTag, SipUtils.getNewViaTag(), callIdHeader); sipSender.transmitRequest(parentPlatform.getDeviceIp(), request); } @Override public void sendNotifyMobilePosition(ParentPlatform parentPlatform, GPSMsgInfo gpsMsgInfo, SubscribeInfo subscribeInfo) throws InvalidArgumentException, ParseException, NoSuchFieldException, SipException, IllegalAccessException { if (parentPlatform == null) { return; } if (logger.isDebugEnabled()) { logger.debug("[发送 移动位置订阅] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat()); } String characterSet = parentPlatform.getCharacterSet(); StringBuffer deviceStatusXml = new StringBuffer(600); deviceStatusXml.append("\r\n") .append("\r\n") .append("MobilePosition\r\n") .append("" + (int)((Math.random()*9+1)*100000) + "\r\n") .append("" + gpsMsgInfo.getId() + "\r\n") .append("\r\n") .append("" + gpsMsgInfo.getLng() + "\r\n") .append("" + gpsMsgInfo.getLat() + "\r\n") .append("" + gpsMsgInfo.getSpeed() + "\r\n") .append("" + gpsMsgInfo.getDirection() + "\r\n") .append("" + gpsMsgInfo.getAltitude() + "\r\n") .append("\r\n"); sendNotify(parentPlatform, deviceStatusXml.toString(), subscribeInfo, eventResult -> { logger.error("发送NOTIFY通知消息失败。错误:{} {}", eventResult.statusCode, eventResult.msg); }, null); } @Override public void sendAlarmMessage(ParentPlatform parentPlatform, DeviceAlarm deviceAlarm) throws SipException, InvalidArgumentException, ParseException { if (parentPlatform == null) { return; } logger.info("[发送报警通知] {}/{}->{},{}: {}", parentPlatform.getServerGBId(), deviceAlarm.getChannelId(), deviceAlarm.getLongitude(), deviceAlarm.getLatitude(), JSON.toJSONString(deviceAlarm)); String characterSet = parentPlatform.getCharacterSet(); StringBuffer deviceStatusXml = new StringBuffer(600); deviceStatusXml.append("\r\n") .append("\r\n") .append("Alarm\r\n") .append("" + (int)((Math.random()*9+1)*100000) + "\r\n") .append("" + deviceAlarm.getChannelId() + "\r\n") .append("" + deviceAlarm.getAlarmPriority() + "\r\n") .append("" + deviceAlarm.getAlarmMethod() + "\r\n") .append("" + deviceAlarm.getAlarmTime() + "\r\n") .append("" + deviceAlarm.getAlarmDescription() + "\r\n") .append("" + deviceAlarm.getLongitude() + "\r\n") .append("" + deviceAlarm.getLatitude() + "\r\n") .append("\r\n") .append("" + deviceAlarm.getAlarmType() + "\r\n") .append("\r\n") .append("\r\n"); CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport()); Request request = headerProviderPlatformProvider.createMessageRequest(parentPlatform, deviceStatusXml.toString(), SipUtils.getNewFromTag(), SipUtils.getNewViaTag(), callIdHeader); sipSender.transmitRequest(parentPlatform.getDeviceIp(), request); } @Override public void sendNotifyForCatalogAddOrUpdate(String type, ParentPlatform parentPlatform, List deviceChannels, SubscribeInfo subscribeInfo, Integer index) throws InvalidArgumentException, ParseException, NoSuchFieldException, SipException, IllegalAccessException { if (parentPlatform == null || deviceChannels == null || deviceChannels.size() == 0 || subscribeInfo == null) { return; } if (index == null) { index = 0; } if (index >= deviceChannels.size()) { return; } List channels; if (index + parentPlatform.getCatalogGroup() < deviceChannels.size()) { channels = deviceChannels.subList(index, index + parentPlatform.getCatalogGroup()); }else { channels = deviceChannels.subList(index, deviceChannels.size()); } Integer finalIndex = index; String catalogXmlContent = getCatalogXmlContentForCatalogAddOrUpdate(parentPlatform, channels, deviceChannels.size(), type, subscribeInfo); sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> { logger.error("发送NOTIFY通知消息失败。错误:{} {}", eventResult.statusCode, eventResult.msg); }, (eventResult -> { try { sendNotifyForCatalogAddOrUpdate(type, parentPlatform, deviceChannels, subscribeInfo, finalIndex + parentPlatform.getCatalogGroup()); } catch (InvalidArgumentException | ParseException | NoSuchFieldException | SipException | IllegalAccessException e) { logger.error("[命令发送失败] 国标级联 NOTIFY通知: {}", e.getMessage()); } })); } private void sendNotify(ParentPlatform parentPlatform, String catalogXmlContent, SubscribeInfo subscribeInfo, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent ) throws SipException, ParseException, InvalidArgumentException { MessageFactoryImpl messageFactory = (MessageFactoryImpl) sipLayer.getSipFactory().createMessageFactory(); String characterSet = parentPlatform.getCharacterSet(); // 设置编码, 防止中文乱码 messageFactory.setDefaultContentEncodingCharset(characterSet); SIPRequest notifyRequest = headerProviderPlatformProvider.createNotifyRequest(parentPlatform, catalogXmlContent, subscribeInfo); sipSender.transmitRequest(parentPlatform.getDeviceIp(), notifyRequest); } private String getCatalogXmlContentForCatalogAddOrUpdate(ParentPlatform parentPlatform, List channels, int sumNum, String type, SubscribeInfo subscribeInfo) { StringBuffer catalogXml = new StringBuffer(600); String characterSet = parentPlatform.getCharacterSet(); catalogXml.append("\r\n") .append("\r\n") .append("Catalog\r\n") .append("" + (int) ((Math.random() * 9 + 1) * 100000) + "\r\n") .append("" + parentPlatform.getDeviceGBId() + "\r\n") .append("1\r\n") .append("\r\n"); if (channels.size() > 0) { for (DeviceChannel channel : channels) { if (parentPlatform.getServerGBId().equals(channel.getParentId())) { channel.setParentId(parentPlatform.getDeviceGBId()); } catalogXml.append("\r\n"); // 行政区划分组只需要这两项就可以 catalogXml.append("" + channel.getChannelId() + "\r\n"); catalogXml.append("" + channel.getName() + "\r\n"); if (channel.getParentId() != null) { // 业务分组加上这一项即可,提高兼容性, catalogXml.append("" + channel.getParentId() + "\r\n"); } if (channel.getChannelId().length() == 20 && Integer.parseInt(channel.getChannelId().substring(10, 13)) == 216) { // 虚拟组织增加BusinessGroupID字段 catalogXml.append("" + channel.getParentId() + "\r\n"); } catalogXml.append("" + channel.getParental() + "\r\n"); if (channel.getParental() == 0) { // 通道项 catalogXml.append("" + channel.getManufacture() + "\r\n") .append("" + channel.getSecrecy() + "\r\n") .append("" + channel.getRegisterWay() + "\r\n") .append("" + (channel.getStatus() == 0 ? "OFF" : "ON") + "\r\n"); if (channel.getChannelType() != 2) { // 业务分组/虚拟组织/行政区划 不设置以下属性 catalogXml.append("" + channel.getModel() + "\r\n") .append(" " + channel.getOwner()+ "\r\n") .append("" + channel.getCivilCode() + "\r\n") .append("
" + channel.getAddress() + "
\r\n"); } if (!"presence".equals(subscribeInfo.getEventType())) { catalogXml.append("" + type + "\r\n"); } } catalogXml.append("
\r\n"); } } catalogXml.append("
\r\n") .append("
\r\n"); return catalogXml.toString(); } @Override public void sendNotifyForCatalogOther(String type, ParentPlatform parentPlatform, List deviceChannels, SubscribeInfo subscribeInfo, Integer index) throws InvalidArgumentException, ParseException, NoSuchFieldException, SipException, IllegalAccessException { if (parentPlatform == null || deviceChannels == null || deviceChannels.size() == 0 || subscribeInfo == null) { logger.warn("[缺少必要参数]"); return; } if (index == null) { index = 0; } if (index >= deviceChannels.size()) { return; } List channels; if (index + parentPlatform.getCatalogGroup() < deviceChannels.size()) { channels = deviceChannels.subList(index, index + parentPlatform.getCatalogGroup()); }else { channels = deviceChannels.subList(index, deviceChannels.size()); } Integer finalIndex = index; String catalogXmlContent = getCatalogXmlContentForCatalogOther(parentPlatform, channels, type); sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> { logger.error("发送NOTIFY通知消息失败。错误:{} {}", eventResult.statusCode, eventResult.msg); }, eventResult -> { try { sendNotifyForCatalogOther(type, parentPlatform, deviceChannels, subscribeInfo, finalIndex + parentPlatform.getCatalogGroup()); } catch (InvalidArgumentException | ParseException | NoSuchFieldException | SipException | IllegalAccessException e) { logger.error("[命令发送失败] 国标级联 NOTIFY通知: {}", e.getMessage()); } }); } private String getCatalogXmlContentForCatalogOther(ParentPlatform parentPlatform, List channels, String type) { String characterSet = parentPlatform.getCharacterSet(); StringBuffer catalogXml = new StringBuffer(600); catalogXml.append("\r\n") .append("\r\n") .append("Catalog\r\n") .append("" + (int) ((Math.random() * 9 + 1) * 100000) + "\r\n") .append("" + parentPlatform.getDeviceGBId() + "\r\n") .append("1\r\n") .append("\r\n"); if (channels.size() > 0) { for (DeviceChannel channel : channels) { if (parentPlatform.getServerGBId().equals(channel.getParentId())) { channel.setParentId(parentPlatform.getDeviceGBId()); } catalogXml.append("\r\n") .append("" + channel.getChannelId() + "\r\n") .append("" + type + "\r\n") .append("\r\n"); } } catalogXml.append("\r\n") .append("\r\n"); return catalogXml.toString(); } @Override public void recordInfo(DeviceChannel deviceChannel, ParentPlatform parentPlatform, String fromTag, RecordInfo recordInfo) throws SipException, InvalidArgumentException, ParseException { if ( parentPlatform ==null) { return ; } String characterSet = parentPlatform.getCharacterSet(); StringBuffer recordXml = new StringBuffer(600); recordXml.append("\r\n") .append("\r\n") .append("RecordInfo\r\n") .append("" +recordInfo.getSn() + "\r\n") .append("" + recordInfo.getDeviceId() + "\r\n") .append("" + recordInfo.getSumNum() + "\r\n"); if (recordInfo.getRecordList() == null ) { recordXml.append("\r\n"); }else { recordXml.append("\r\n"); if (recordInfo.getRecordList().size() > 0) { for (RecordItem recordItem : recordInfo.getRecordList()) { recordXml.append("\r\n"); if (deviceChannel != null) { recordXml.append("" + recordItem.getDeviceId() + "\r\n") .append("" + recordItem.getName() + "\r\n") .append("" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getStartTime()) + "\r\n") .append("" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getEndTime()) + "\r\n") .append("" + recordItem.getSecrecy() + "\r\n") .append("" + recordItem.getType() + "\r\n"); if (!ObjectUtils.isEmpty(recordItem.getFileSize())) { recordXml.append("" + recordItem.getFileSize() + "\r\n"); } if (!ObjectUtils.isEmpty(recordItem.getFilePath())) { recordXml.append("" + recordItem.getFilePath() + "\r\n"); } } recordXml.append("\r\n"); } } } recordXml.append("\r\n") .append("\r\n"); // callid CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport()); Request request = headerProviderPlatformProvider.createMessageRequest(parentPlatform, recordXml.toString(), fromTag, SipUtils.getNewViaTag(), callIdHeader); sipSender.transmitRequest(parentPlatform.getDeviceIp(), request); } @Override public void sendMediaStatusNotify(ParentPlatform parentPlatform, SendRtpItem sendRtpItem) throws SipException, InvalidArgumentException, ParseException { if (sendRtpItem == null || parentPlatform == null) { return; } String characterSet = parentPlatform.getCharacterSet(); StringBuffer mediaStatusXml = new StringBuffer(200); mediaStatusXml.append("\r\n") .append("\r\n") .append("MediaStatus\r\n") .append("" + (int)((Math.random()*9+1)*100000) + "\r\n") .append("" + sendRtpItem.getChannelId() + "\r\n") .append("121\r\n") .append("\r\n"); SIPRequest messageRequest = (SIPRequest)headerProviderPlatformProvider.createMessageRequest(parentPlatform, mediaStatusXml.toString(), sendRtpItem); sipSender.transmitRequest(parentPlatform.getDeviceIp(),messageRequest); } @Override public void streamByeCmd(ParentPlatform platform, String callId) throws SipException, InvalidArgumentException, ParseException { if (platform == null) { return; } SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platform.getServerGBId(), null, null, callId); if (sendRtpItem != null) { streamByeCmd(platform, sendRtpItem); } } @Override public synchronized void streamByeCmd(ParentPlatform platform, SendRtpItem sendRtpItem) throws SipException, InvalidArgumentException, ParseException { if (sendRtpItem == null ) { logger.info("[向上级发送BYE], sendRtpItem 为NULL"); return; } if (platform == null) { logger.info("[向上级发送BYE], platform 为NULL"); return; } logger.info("[向上级发送BYE], {}/{}", platform.getServerGBId(), sendRtpItem.getChannelId()); String mediaServerId = sendRtpItem.getMediaServerId(); MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); if (mediaServerItem != null) { mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc()); zlmrtpServerFactory.closeRtpServer(mediaServerItem, sendRtpItem.getStream()); } SIPRequest byeRequest = headerProviderPlatformProvider.createByeRequest(platform, sendRtpItem); if (byeRequest == null) { logger.warn("[向上级发送bye]:无法创建 byeRequest"); } sipSender.transmitRequest(platform.getDeviceIp(),byeRequest); } @Override public void streamByeCmd(ParentPlatform platform, String channelId, String stream, String callId, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException, SsrcTransactionNotFoundException { SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(platform.getServerGBId(), channelId, callId, stream); if (ssrcTransaction == null) { throw new SsrcTransactionNotFoundException(platform.getServerGBId(), channelId, callId, stream); } mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc()); mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream()); streamSession.remove(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getStream()); Request byteRequest = headerProviderPlatformProvider.createByteRequest(platform, channelId, ssrcTransaction.getSipTransactionInfo()); sipSender.transmitRequest(sipLayer.getLocalIp(platform.getDeviceIp()), byteRequest, null, okEvent); } @Override public void broadcastResultCmd(ParentPlatform platform, DeviceChannel deviceChannel, String sn, boolean result, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException { if (platform == null || deviceChannel == null) { return; } String characterSet = platform.getCharacterSet(); StringBuffer mediaStatusXml = new StringBuffer(200); mediaStatusXml.append("\r\n"); mediaStatusXml.append("\r\n"); mediaStatusXml.append("Broadcast\r\n"); mediaStatusXml.append("" + sn + "\r\n"); mediaStatusXml.append("" + deviceChannel.getChannelId() + "\r\n"); mediaStatusXml.append("" + (result?"OK":"ERROR") + "\r\n"); mediaStatusXml.append("\r\n"); CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(platform.getDeviceIp(), platform.getTransport()); SIPRequest messageRequest = (SIPRequest)headerProviderPlatformProvider.createMessageRequest(platform, mediaStatusXml.toString(), SipUtils.getNewFromTag(), SipUtils.getNewViaTag(), callIdHeader); sipSender.transmitRequest(platform.getDeviceIp(),messageRequest, errorEvent, okEvent); } @Override public void broadcastInviteCmd(ParentPlatform platform, String channelId, MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, ZlmHttpHookSubscribe.Event event, SipSubscribe.Event okEvent, SipSubscribe.Event errorEvent) throws ParseException, SipException, InvalidArgumentException { String stream = ssrcInfo.getStream(); if (platform == null) { return; } logger.info("{} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort()); HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId()); subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json) -> { if (event != null) { event.response(mediaServerItemInUse, json); subscribe.removeSubscribe(hookSubscribe); } }); String sdpIp = mediaServerItem.getSdpIp(); StringBuffer content = new StringBuffer(200); content.append("v=0\r\n"); content.append("o=" + channelId + " 0 0 IN IP4 " + sdpIp + "\r\n"); content.append("s=Play\r\n"); content.append("c=IN IP4 " + sdpIp + "\r\n"); content.append("t=0 0\r\n"); if ("TCP-PASSIVE".equalsIgnoreCase(userSetting.getBroadcastForPlatform())) { content.append("m=audio " + ssrcInfo.getPort() + " TCP/RTP/AVP 8 96\r\n"); } else if ("TCP-ACTIVE".equalsIgnoreCase(userSetting.getBroadcastForPlatform())) { content.append("m=audio " + ssrcInfo.getPort() + " TCP/RTP/AVP 8 96\r\n"); } else if ("UDP".equalsIgnoreCase(userSetting.getBroadcastForPlatform())) { content.append("m=audio " + ssrcInfo.getPort() + " RTP/AVP 8 96\r\n"); } content.append("a=recvonly\r\n"); content.append("a=rtpmap:8 PCMA/8000\r\n"); content.append("a=rtpmap:96 PS/90000\r\n"); if ("TCP-PASSIVE".equalsIgnoreCase(userSetting.getBroadcastForPlatform())) { content.append("a=setup:passive\r\n"); content.append("a=connection:new\r\n"); }else if ("TCP-ACTIVE".equalsIgnoreCase(userSetting.getBroadcastForPlatform())) { content.append("a=setup:active\r\n"); content.append("a=connection:new\r\n"); } content.append("y=" + ssrcInfo.getSsrc() + "\r\n");//ssrc CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(sipLayer.getLocalIp(platform.getDeviceIp()), platform.getTransport()); Request request = headerProviderPlatformProvider.createInviteRequest(platform, channelId, content.toString(), SipUtils.getNewViaTag(), SipUtils.getNewFromTag(), ssrcInfo.getSsrc(), callIdHeader); sipSender.transmitRequest(sipLayer.getLocalIp(platform.getDeviceIp()), request, (e -> { streamSession.remove(platform.getServerGBId(), channelId, ssrcInfo.getStream()); mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); subscribe.removeSubscribe(hookSubscribe); errorEvent.response(e); }), e -> { ResponseEvent responseEvent = (ResponseEvent) e.event; SIPResponse response = (SIPResponse) responseEvent.getResponse(); streamSession.put(platform.getServerGBId(), channelId, callIdHeader.getCallId(), stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.broadcast); okEvent.response(e); }); } }