XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加设备删除接口，只允许删除离线设备；增加视频停止播放接口

Browse Source
This commit is contained in:
songww
2020-05-13 20:38:32 +08:00
parent ca5139929b
commit 6ecd801c23
22 changed files with 266 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
View File

@@ -117,7 +117,7 @@ public class SipLayer implements SipListener, Runnable {
@Override
public void processRequest(RequestEvent evt) {
ISIPRequestProcessor processor = processorFactory.createRequestProcessor(evt);
processor.process(evt, this, getServerTransaction(evt));
processor.process(evt, this);
}
@Override
@@ -200,7 +200,7 @@ public class SipLayer implements SipListener, Runnable {
}
private ServerTransaction getServerTransaction(RequestEvent evt) {
public ServerTransaction getServerTransaction(RequestEvent evt) {
Request request = evt.getRequest();
ServerTransaction serverTransaction = evt.getServerTransaction();
// 判断TCP还是UDP

2
src/main/java/com/genersoft/iot/vmp/gb28181/utils/SsrcUtil.java → src/main/java/com/genersoft/iot/vmp/gb28181/session/SsrcUtil.java
View File

@@ -1,4 +1,4 @@
package com.genersoft.iot.vmp.gb28181.utils;
package com.genersoft.iot.vmp.gb28181.session;
import java.util.ArrayList;
import java.util.List;

41
src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java Normal file
View File

@@ -0,0 +1,41 @@
package com.genersoft.iot.vmp.gb28181.session;
import java.util.concurrent.ConcurrentHashMap;
import javax.sip.ClientTransaction;
import org.springframework.stereotype.Component;
/**
* @Description:视频流session管理器，管理视频预览、预览回放的通信句柄
* @author: songww
* @date: 2020年5月13日 下午4:03:02
*/
@Component
public class VideoStreamSessionManager {
private ConcurrentHashMap<String, ClientTransaction> sessionMap = new ConcurrentHashMap<>();
public String createPlaySsrc(){
String ssrc = SsrcUtil.getPlaySsrc();
return ssrc;
}
public String createPlayBackSsrc(){
String ssrc = SsrcUtil.getPlayBackSsrc();
return ssrc;
}
public void put(String ssrc,ClientTransaction transaction){
sessionMap.put(ssrc, transaction);
}
public ClientTransaction get(String ssrc){
return sessionMap.get(ssrc);
}
public void remove(String ssrc) {
sessionMap.remove(ssrc);
SsrcUtil.releaseSsrc(ssrc);
}
}

7
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
View File

@@ -81,6 +81,13 @@ public interface ISIPCommander {
*/
public String playbackStreamCmd(Device device,String channelId, String startTime, String endTime);
/**
* 视频流停止
*
* @param ssrc ssrc
*/
public void streamByeCmd(String ssrc);
/**
* 语音广播
*

57
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java
View File

@@ -46,14 +46,15 @@ public class SIPRequestHeaderProvider {
ArrayList<ViaHeader> viaHeaders = new ArrayList<ViaHeader>();
ViaHeader viaHeader = layer.getHeaderFactory().createViaHeader(sipConfig.getSipIp(), sipConfig.getSipPort(),
device.getTransport(), viaTag);
viaHeader.setRPort();
viaHeaders.add(viaHeader);
// from
SipURI fromSipURI = layer.getAddressFactory().createSipURI(device.getDeviceId(),
SipURI fromSipURI = layer.getAddressFactory().createSipURI(sipConfig.getSipId(),
sipConfig.getSipIp() + ":" + sipConfig.getSipPort());
Address fromAddress = layer.getAddressFactory().createAddress(fromSipURI);
FromHeader fromHeader = layer.getHeaderFactory().createFromHeader(fromAddress, fromTag);
// to
SipURI toSipURI = layer.getAddressFactory().createSipURI(device.getDeviceId(), host.getAddress());
SipURI toSipURI = layer.getAddressFactory().createSipURI(device.getDeviceId(), sipConfig.getSipDomain());
Address toAddress = layer.getAddressFactory().createAddress(toSipURI);
ToHeader toHeader = layer.getHeaderFactory().createToHeader(toAddress, toTag);
// callid
@@ -71,6 +72,49 @@ public class SIPRequestHeaderProvider {
return request;
}
// public Request createInviteRequest(Device device, String content, String viaTag, String fromTag, String toTag) throws ParseException, InvalidArgumentException {
// Request request = null;
// Host host = device.getHost();
// //请求行
// SipURI requestLine = layer.getAddressFactory().createSipURI(device.getDeviceId(), host.getAddress());
// //via
// ArrayList<ViaHeader> viaHeaders = new ArrayList<ViaHeader>();
// ViaHeader viaHeader = layer.getHeaderFactory().createViaHeader(sipConfig.getSipIp(), sipConfig.getSipPort(), device.getTransport(), viaTag);
// viaHeader.setRPort();
// viaHeaders.add(viaHeader);
// //from
// SipURI fromSipURI = layer.getAddressFactory().createSipURI(device.getDeviceId(),sipConfig.getSipIp()+":"+sipConfig.getSipPort());
// Address fromAddress = layer.getAddressFactory().createAddress(fromSipURI);
// FromHeader fromHeader = layer.getHeaderFactory().createFromHeader(fromAddress, fromTag); //必须要有标记，否则无法创建会话，无法回应ack
// //to
// SipURI toSipURI = layer.getAddressFactory().createSipURI(device.getDeviceId(),host.getAddress());
// Address toAddress = layer.getAddressFactory().createAddress(toSipURI);
// ToHeader toHeader = layer.getHeaderFactory().createToHeader(toAddress,null);
//
// //callid
// CallIdHeader callIdHeader = null;
// if(device.getTransport().equals("TCP")) {
// callIdHeader = layer.getTcpSipProvider().getNewCallId();
// }
// if(device.getTransport().equals("UDP")) {
// callIdHeader = layer.getUdpSipProvider().getNewCallId();
// }
//
// //Forwards
// MaxForwardsHeader maxForwards = layer.getHeaderFactory().createMaxForwardsHeader(70);
//
// //ceq
// CSeqHeader cSeqHeader = layer.getHeaderFactory().createCSeqHeader(1L, Request.INVITE);
// request = layer.getMessageFactory().createRequest(requestLine, Request.INVITE, callIdHeader, cSeqHeader,fromHeader, toHeader, viaHeaders, maxForwards);
//
// Address concatAddress = layer.getAddressFactory().createAddress(layer.getAddressFactory().createSipURI(sipConfig.getSipId(), sipConfig.getSipIp()+":"+sipConfig.getSipPort()));
// request.addHeader(layer.getHeaderFactory().createContactHeader(concatAddress));
//
// ContentTypeHeader contentTypeHeader = layer.getHeaderFactory().createContentTypeHeader("Application", "SDP");
// request.setContent(content, contentTypeHeader);
// return request;
// }
public Request createInviteRequest(Device device, String content, String viaTag, String fromTag, String toTag) throws ParseException, InvalidArgumentException {
Request request = null;
Host host = device.getHost();
@@ -82,11 +126,11 @@ public class SIPRequestHeaderProvider {
viaHeader.setRPort();
viaHeaders.add(viaHeader);
//from
SipURI fromSipURI = layer.getAddressFactory().createSipURI(device.getDeviceId(),sipConfig.getSipIp()+":"+sipConfig.getSipPort());
SipURI fromSipURI = layer.getAddressFactory().createSipURI(sipConfig.getSipId(),sipConfig.getSipDomain());
Address fromAddress = layer.getAddressFactory().createAddress(fromSipURI);
FromHeader fromHeader = layer.getHeaderFactory().createFromHeader(fromAddress, fromTag); //必须要有标记，否则无法创建会话，无法回应ack
//to
SipURI toSipURI = layer.getAddressFactory().createSipURI(device.getDeviceId(),host.getAddress());
SipURI toSipURI = layer.getAddressFactory().createSipURI(device.getDeviceId(),sipConfig.getSipDomain());
Address toAddress = layer.getAddressFactory().createAddress(toSipURI);
ToHeader toHeader = layer.getHeaderFactory().createToHeader(toAddress,null);
@@ -101,9 +145,14 @@ public class SIPRequestHeaderProvider {
//Forwards
MaxForwardsHeader maxForwards = layer.getHeaderFactory().createMaxForwardsHeader(70);
//ceq
CSeqHeader cSeqHeader = layer.getHeaderFactory().createCSeqHeader(1L, Request.INVITE);
request = layer.getMessageFactory().createRequest(requestLine, Request.INVITE, callIdHeader, cSeqHeader,fromHeader, toHeader, viaHeaders, maxForwards);
Address concatAddress = layer.getAddressFactory().createAddress(layer.getAddressFactory().createSipURI(sipConfig.getSipId(), sipConfig.getSipIp()+":"+sipConfig.getSipPort()));
request.addHeader(layer.getHeaderFactory().createContactHeader(concatAddress));
ContentTypeHeader contentTypeHeader = layer.getHeaderFactory().createContentTypeHeader("Application", "SDP");
request.setContent(content, contentTypeHeader);
return request;

67
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
View File

@@ -3,8 +3,11 @@ package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl;
import java.text.ParseException;
import javax.sip.ClientTransaction;
import javax.sip.Dialog;
import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
import javax.sip.TransactionDoesNotExistException;
import javax.sip.header.ViaHeader;
import javax.sip.message.Request;
import org.springframework.beans.factory.annotation.Autowired;
@@ -13,10 +16,10 @@ import org.springframework.stereotype.Component;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.gb28181.SipLayer;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderProvider;
import com.genersoft.iot.vmp.gb28181.utils.DateUtil;
import com.genersoft.iot.vmp.gb28181.utils.SsrcUtil;
/**
* @Description:设备能力接口，用于定义设备的控制、查询能力
@@ -35,6 +38,9 @@ public class SIPCommander implements ISIPCommander {
@Autowired
private SipLayer sipLayer;
@Autowired
private VideoStreamSessionManager streamSession;
/**
* 云台方向放控制，使用配置文件中的默认镜头移动速度
*
@@ -135,11 +141,11 @@ public class SIPCommander implements ISIPCommander {
public String playStreamCmd(Device device, String channelId) {
try {
String ssrc = SsrcUtil.getPlaySsrc();
String ssrc = streamSession.createPlaySsrc();
//
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+channelId+" 0 0 IN IP4 "+sipConfig.getSipIp()+"\r\n");
content.append("o="+channelId+" 0 0 IN IP4 "+sipConfig.getMediaIp()+"\r\n");
content.append("s=Play\r\n");
content.append("c=IN IP4 "+sipConfig.getMediaIp()+"\r\n");
content.append("t=0 0\r\n");
@@ -161,7 +167,8 @@ public class SIPCommander implements ISIPCommander {
Request request = headerProvider.createInviteRequest(device, content.toString(), null, "live", null);
transmitRequest(device, request);
ClientTransaction transaction = transmitRequest(device, request);
streamSession.put(ssrc, transaction);
return ssrc;
} catch ( SipException | ParseException | InvalidArgumentException e) {
e.printStackTrace();
@@ -181,11 +188,11 @@ public class SIPCommander implements ISIPCommander {
public String playbackStreamCmd(Device device, String channelId, String startTime, String endTime) {
try {
String ssrc = SsrcUtil.getPlayBackSsrc();
String ssrc = streamSession.createPlayBackSsrc();
//
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+device.getDeviceId()+" 0 0 IN IP4 "+sipConfig.getSipIp()+"\r\n");
content.append("o="+device.getDeviceId()+" 0 0 IN IP4 "+sipConfig.getMediaIp()+"\r\n");
content.append("s=Playback\r\n");
content.append("u="+channelId+":3\r\n");
content.append("c=IN IP4 "+sipConfig.getMediaIp()+"\r\n");
@@ -208,13 +215,50 @@ public class SIPCommander implements ISIPCommander {
Request request = headerProvider.createInviteRequest(device, content.toString(), null, "live", null);
transmitRequest(device, request);
ClientTransaction transaction = transmitRequest(device, request);
streamSession.put(ssrc, transaction);
return ssrc;
} catch ( SipException | ParseException | InvalidArgumentException e) {
e.printStackTrace();
return null;
}
}
/**
* 视频流停止
*
* @param device 视频设备
* @param channelId 预览通道
*/
@Override
public void streamByeCmd(String ssrc) {
try {
ClientTransaction transaction = streamSession.get(ssrc);
if (transaction == null) {
return;
}
Dialog dialog = transaction.getDialog();
if (dialog == null) {
return;
}
Request byeRequest = dialog.createRequest(Request.BYE);
ViaHeader viaHeader = (ViaHeader) byeRequest.getHeader(ViaHeader.NAME);
String protocol = viaHeader.getTransport();
ClientTransaction clientTransaction = null;
if("TCP".equals(protocol)) {
clientTransaction = sipLayer.getTcpSipProvider().getNewClientTransaction(byeRequest);
} else if("UDP".equals(protocol)) {
clientTransaction = sipLayer.getUdpSipProvider().getNewClientTransaction(byeRequest);
}
dialog.sendRequest(clientTransaction);
} catch (TransactionDoesNotExistException e) {
e.printStackTrace();
} catch (SipException e) {
e.printStackTrace();
}
}
/**
* 语音广播
@@ -435,16 +479,15 @@ public class SIPCommander implements ISIPCommander {
return false;
}
private void transmitRequest(Device device, Request request) throws SipException {
private ClientTransaction transmitRequest(Device device, Request request) throws SipException {
ClientTransaction clientTransaction = null;
if(device.getTransport().equals("TCP")) {
if("TCP".equals(device.getTransport())) {
clientTransaction = sipLayer.getTcpSipProvider().getNewClientTransaction(request);
//sipLayer.getTcpSipProvider().sendRequest(request);
} else if(device.getTransport().equals("UDP")) {
} else if("UDP".equals(device.getTransport())) {
clientTransaction = sipLayer.getUdpSipProvider().getNewClientTransaction(request);
//sipLayer.getUdpSipProvider().sendRequest(request);
}
clientTransaction.sendRequest();
return clientTransaction;
}
}

3
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/ISIPRequestProcessor.java
View File

@@ -1,7 +1,6 @@
package com.genersoft.iot.vmp.gb28181.transmit.request;
import javax.sip.RequestEvent;
import javax.sip.ServerTransaction;
import com.genersoft.iot.vmp.gb28181.SipLayer;
@@ -12,6 +11,6 @@ import com.genersoft.iot.vmp.gb28181.SipLayer;
*/
public interface ISIPRequestProcessor {
public void process(RequestEvent evt, SipLayer layer, ServerTransaction transaction);
public void process(RequestEvent evt, SipLayer layer);
}

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/AckRequestProcessor.java
View File

@@ -31,7 +31,7 @@ public class AckRequestProcessor implements ISIPRequestProcessor {
* @param config
*/
@Override
public void process(RequestEvent evt, SipLayer layer, ServerTransaction transaction) {
public void process(RequestEvent evt, SipLayer layer) {
Request request = evt.getRequest();
Dialog dialog = evt.getDialog();
try {

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/ByeRequestProcessor.java
View File

@@ -25,7 +25,7 @@ public class ByeRequestProcessor implements ISIPRequestProcessor {
* @param config
*/
@Override
public void process(RequestEvent evt, SipLayer layer, ServerTransaction transaction) {
public void process(RequestEvent evt, SipLayer layer) {
// TODO Auto-generated method stub
}

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/CancelRequestProcessor.java
View File

@@ -25,7 +25,7 @@ public class CancelRequestProcessor implements ISIPRequestProcessor {
* @param config
*/
@Override
public void process(RequestEvent evt, SipLayer layer, ServerTransaction transaction) {
public void process(RequestEvent evt, SipLayer layer) {
// TODO Auto-generated method stub
}

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java
View File

@@ -23,7 +23,7 @@ public class InviteRequestProcessor implements ISIPRequestProcessor {
* 请求消息
*/
@Override
public void process(RequestEvent evt, SipLayer layer, ServerTransaction transaction) {
public void process(RequestEvent evt, SipLayer layer) {
// TODO Auto-generated method stub
// Request request = requestEvent.getRequest();
//

4
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java
View File

@@ -93,10 +93,10 @@ public class MessageRequestProcessor implements ISIPRequestProcessor {
* @param transaction
*/
@Override
public void process(RequestEvent evt, SipLayer layer, ServerTransaction transaction) {
public void process(RequestEvent evt, SipLayer layer) {
this.layer = layer;
this.transaction = transaction;
this.transaction = layer.getServerTransaction(evt);
Request request = evt.getRequest();
SAXReader reader = new SAXReader();

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/OtherRequestProcessor.java
View File

@@ -25,7 +25,7 @@ public class OtherRequestProcessor implements ISIPRequestProcessor {
* @param config
*/
@Override
public void process(RequestEvent evt, SipLayer layer, ServerTransaction transaction) {
public void process(RequestEvent evt, SipLayer layer) {
System.out.println("no support the method! Method:" + evt.getRequest().getMethod());
}

4
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/RegisterRequestProcessor.java
View File

@@ -63,7 +63,7 @@ public class RegisterRequestProcessor implements ISIPRequestProcessor {
* 请求消息
*/
@Override
public void process(RequestEvent evt, SipLayer layer, ServerTransaction transaction) {
public void process(RequestEvent evt, SipLayer layer) {
try {
System.out.println("收到注册请求，开始处理");
Request request = evt.getRequest();
@@ -141,7 +141,7 @@ public class RegisterRequestProcessor implements ISIPRequestProcessor {
device.setTransport(isTcp ? "TCP" : "UDP");
}
}
transaction.sendResponse(response);
layer.getServerTransaction(evt).sendResponse(response);
// 注册成功
// 保存到redis
// 下发catelog查询目录

4
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/SubscribeRequestProcessor.java
View File

@@ -32,7 +32,7 @@ public class SubscribeRequestProcessor implements ISIPRequestProcessor {
* @param config
*/
@Override
public void process(RequestEvent evt, SipLayer layer, ServerTransaction transaction) {
public void process(RequestEvent evt, SipLayer layer) {
Request request = evt.getRequest();
try {
@@ -43,7 +43,7 @@ public class SubscribeRequestProcessor implements ISIPRequestProcessor {
response.setExpires(expireHeader);
}
System.out.println("response : " + response.toString());
ServerTransaction transaction = layer.getServerTransaction(evt);
if (transaction != null) {
transaction.sendResponse(response);
transaction.terminate();

46
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/response/impl/InviteResponseProcessor.java
View File

@@ -50,31 +50,33 @@ public class InviteResponseProcessor implements ISIPResponseProcessor {
//成功响应
//下发ack
if(statusCode == Response.OK){
ClientTransaction clientTransaction = evt.getClientTransaction();
if(clientTransaction == null){
logger.error("回复ACK时，clientTransaction为null >>> {}",response);
return;
}
Dialog clientDialog = clientTransaction.getDialog();
CSeqHeader clientCSeqHeader = (CSeqHeader) response.getHeader(CSeqHeader.NAME);
long cseqId = clientCSeqHeader.getSeqNumber();
/*
createAck函数，创建的ackRequest，会采用Invite响应的200OK，中的contact字段中的地址，作为目标地址。
有的终端传上来的可能还是内网地址，会造成ack发送不出去。接受不到音视频流
所以在此处统一替换地址。和响应消息的Via头中的地址保持一致。
*/
Request ackRequest = clientDialog.createAck(cseqId);
SipURI requestURI = (SipURI) ackRequest.getRequestURI();
ViaHeader viaHeader = (ViaHeader) response.getHeader(ViaHeader.NAME);
requestURI.setHost(viaHeader.getHost());
requestURI.setPort(viaHeader.getPort());
clientDialog.sendAck(ackRequest);
// ClientTransaction clientTransaction = evt.getClientTransaction();
// if(clientTransaction == null){
// logger.error("回复ACK时，clientTransaction为null >>> {}",response);
// return;
// }
// Dialog clientDialog = clientTransaction.getDialog();
//
// CSeqHeader clientCSeqHeader = (CSeqHeader) response.getHeader(CSeqHeader.NAME);
// long cseqId = clientCSeqHeader.getSeqNumber();
// /*
// createAck函数，创建的ackRequest，会采用Invite响应的200OK，中的contact字段中的地址，作为目标地址。
// 有的终端传上来的可能还是内网地址，会造成ack发送不出去。接受不到音视频流
// 所以在此处统一替换地址。和响应消息的Via头中的地址保持一致。
// */
// Request ackRequest = clientDialog.createAck(cseqId);
// SipURI requestURI = (SipURI) ackRequest.getRequestURI();
// ViaHeader viaHeader = (ViaHeader) response.getHeader(ViaHeader.NAME);
// requestURI.setHost(viaHeader.getHost());
// requestURI.setPort(viaHeader.getPort());
// clientDialog.sendAck(ackRequest);
Dialog dialog = evt.getDialog();
Request reqAck =dialog.createAck(1L);
dialog.sendAck(reqAck);
}
} catch (InvalidArgumentException | SipException e) {
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
}
}