背景
在事先Android平台RTSP、RTMP转GB28181网关之前，我们已经实现了Android平台GB28181的接入，可实现Android平台采集到的音视频数据，编码后，打包按需发到GB28181服务平台。此外，拉流端，我们已经有了成熟的RTSP和RTMP拉流播放方案。
今天，我们要做的是，把外部的RTSP或RTMP流，拉过来，然后对接到GB28181国标平台，实现媒体流数据的按需播放。
和我们之前实现的轻量级RTSP服务网关模块类似，我们要做的是，实现RTSP或RTMP流，按需打包对接到GB28181服务平台。
简单来说，Android平台RTSP、RTMP转GB28181网关平台，是GB28181设备接入模块的一个扩展，由拉流端、GB28181接入端两个模块组成。
轻量级RTSP服务模块、RTSP|RTMP转GB28181网关模块和内置RTSP网关模块的区别和联系：
内置轻量级RTSP服务模块和内置RTSP网关模块，核心痛点是避免用户或者开发者单独部署RTSP或者RTMP服务，数据汇聚到内置RTSP服务，对外提供可供拉流的RTSP URL，适用于内网环境下，对并发要求不高的场景，支持H.264/H.265，支持RTSP鉴权、单播、组播模式，考虑到单个服务承载能力，我们支持同时创建多个RTSP服务，并支持获取当前RTSP服务会话连接数。
RTSP|RTMP转GB28181网关模块，实现的是音视频数据的转发，类似于RTSP|RTMP转RTMP推送模块，把本地数据源，对接到GB28181服务平台或RTMP服务平台。
三者不同点：数据来源不同
1. 内置轻量级RTSP服务模块，数据源来自摄像头、屏幕、麦克风等编码前数据，或者本地编码后的对接数据，这点和GB28181的设备接入模块类似。
2. 内置RTSP网关模块，实际上是RTSP/RTMP拉流模块 内置轻量级RTSP服务模块组合出来的。数据源来自RTSP或RTMP网络流，拉流模块完成编码后的音视频数据回调，然后，汇聚到内置轻量级RTSP服务模块。RTSP|RTMP转GB28181网关模块，和内置RTSP网关模块数据源接入一样。
技术实现
本文以之前Android平台RTSP|RTMP转发demo为例，在这个基础上，加GB28181网关扩展。
拉流端音频数据回调，拉流端获取到编码后是数据，回调上来，通过SmartPublisherPostAudioEncodedData()发送到推送模块。
代码语言：javascript复制class PlayerAudioDataCallback implements NTAudioDataCallback
{
private int audio_buffer_size = 0;
private int param_info_size = 0;
private ByteBuffer audio_buffer_ = null;
private ByteBuffer parameter_info_ = null;
@Override
public ByteBuffer getAudioByteBuffer(int size)
{
//Log.i("getAudioByteBuffer", "size: " size);
if( size < 1 )
{
return null;
}
if ( size <= audio_buffer_size && audio_buffer_ != null )
{
return audio_buffer_;
}
audio_buffer_size = size 512;
audio_buffer_size = (audio_buffer_size 0xf) & (~0xf);
audio_buffer_ = ByteBuffer.allocateDirect(audio_buffer_size);
// Log.i("getAudioByteBuffer", "size: " size " buffer_size:" audio_buffer_size);
return audio_buffer_;
}
@Override
public ByteBuffer getAudioParameterInfo(int size)
{
//Log.i("getAudioParameterInfo", "size: " size);
if(size < 1)
{
return null;
}
if ( size <= param_info_size && parameter_info_ != null )
{
return parameter_info_;
}
param_info_size = size 32;
param_info_size = (param_info_size 0xf) & (~0xf);
parameter_info_ = ByteBuffer.allocateDirect(param_info_size);
//Log.i("getAudioParameterInfo", "size: " size " buffer_size:" param_info_size);
return parameter_info_;
}
public void onAudioDataCallback(int ret, int audio_codec_id, int sample_size, int is_key_frame, long timestamp, int sample_rate, int channel, int parameter_info_size, long reserve)
{
//Log.i("onAudioDataCallback", "ret: " ret ", audio_codec_id: " audio_codec_id ", sample_size: " sample_size ", timestamp: " timestamp
// ",sample_rate:" sample_rate);
if ( audio_buffer_ == null)
return;
audio_buffer_.rewind();
if ( ret == 0 && (isPushing || isRTSPPublisherRunning || isGB28181StreamRunning)) {
libPublisher.SmartPublisherPostAudioEncodedData(publisherHandle, audio_codec_id, audio_buffer_, sample_size, is_key_frame, timestamp, parameter_info_, parameter_info_size);
}
// test
/*
byte[] test_buffer = new byte[16];
pcm_buffer_.get(test_buffer);
Log.i(TAG, "onGetPcmFrame data:" bytesToHexString(test_buffer));
*/
}
}拉流端视频数据回调，拉流端获取到编码后是数据，回调上来，通过SmartPublisherPostVideoEncodedData()发送到推送模块。
代码语言：javascript复制class PlayerVideoDataCallback implements NTVideoDataCallback
{
private int video_buffer_size = 0;
private ByteBuffer video_buffer_ = null;
@Override
public ByteBuffer getVideoByteBuffer(int size)
{
//Log.i("getVideoByteBuffer", "size: " size);
if( size < 1 )
{
return null;
}
if ( size <= video_buffer_size && video_buffer_ != null )
{
return video_buffer_;
}
video_buffer_size = size 1024;
video_buffer_size = (video_buffer_size 0xf) & (~0xf);
video_buffer_ = ByteBuffer.allocateDirect(video_buffer_size);
// Log.i("getVideoByteBuffer", "size: " size " buffer_size:" video_buffer_size);
return video_buffer_;
}
public void onVideoDataCallback(int ret, int video_codec_id, int sample_size, int is_key_frame, long timestamp, int width, int height, long presentation_timestamp)
{
//Log.i("onVideoDataCallback", "ret: " ret ", video_codec_id: " video_codec_id ", sample_size: " sample_size ", is_key_frame: " is_key_frame ", timestamp: " timestamp
// ",presentation_timestamp:" presentation_timestamp);
if ( video_buffer_ == null)
return;
video_buffer_.rewind();
if ( ret == 0 && (isPushing || isRTSPPublisherRunning || isGB28181StreamRunning) ) {
libPublisher.SmartPublisherPostVideoEncodedData(publisherHandle, video_codec_id, video_buffer_, sample_size, is_key_frame, timestamp, presentation_timestamp);
}
}
}GB28181网关模块，初始化：
代码语言：javascript复制/*
* gb28181 agent初始化
* Github: https://github.com/daniulive/SmarterStreaming
*/
private boolean initGB28181Agent() {
if ( gb28181_agent_ != null )
return true;
getLocation(myContext);
String local_ip_addr = IPAddrUtils.getIpAddress(myContext);
Log.i(TAG, "initGB28181Agent local ip addr: " local_ip_addr);
if ( local_ip_addr == null || local_ip_addr.isEmpty() ) {
Log.e(TAG, "initGB28181Agent local ip is empty");
return false;
}
gb28181_agent_ = GBSIPAgentFactory.getInstance().create();
if ( gb28181_agent_ == null ) {
Log.e(TAG, "initGB28181Agent create agent failed");
return false;
}
gb28181_agent_.addListener(this);
// 必填信息
gb28181_agent_.setLocalAddressInfo(local_ip_addr, gb28181_sip_local_port_);
gb28181_agent_.setServerParameter(gb28181_sip_server_addr_, gb28181_sip_server_port_, gb28181_sip_server_id_, gb28181_sip_server_domain_);
gb28181_agent_.setUserInfo(gb28181_sip_username_, gb28181_sip_password_);
// 可选参数
gb28181_agent_.setUserAgent(gb28181_sip_user_agent_filed_);
gb28181_agent_.setTransportProtocol(gb28181_sip_trans_protocol_==0?"UDP":"TCP");
// GB28181配置
gb28181_agent_.config(gb28181_reg_expired_, gb28181_heartbeat_interval_, gb28181_heartbeat_count_);
com.gb28181.ntsignalling.Device gb_device = new com.gb28181.ntsignalling.Device("34020000001380000001", "安卓测试设备", Build.MANUFACTURER, Build.MODEL,
"宇宙","火星1","火星", true);
if (mLongitude != null && mLatitude != null) {
com.gb28181.ntsignalling.DevicePosition device_pos = new com.gb28181.ntsignalling.DevicePosition();
device_pos.setTime(mLocationTime);
device_pos.setLongitude(mLongitude);
device_pos.setLatitude(mLatitude);
gb_device.setPosition(device_pos);
gb_device.setSupportMobilePosition(true); // 设置支持移动位置上报
}
gb28181_agent_.addDevice(gb_device);
if (!gb28181_agent_.initialize()) {
gb28181_agent_ = null;
Log.e(TAG, "initGB28181Agent gb28181_agent_.initialize failed.");
return false;
}
return true;
}GB28181网关模块按钮相关代码实现，整体和GB28181设备接入一致：
代码语言：javascript复制class ButtonGB28181AgentListener implements OnClickListener {
public void onClick(View v) {
stopGB28181Stream();
destoryRTPSender();
if (null == gb28181_agent_ ) {
if( !initGB28181Agent() )
return;
}
if (gb28181_agent_.isRunning()) {
gb28181_agent_.terminateAllPlays(true);// 目前测试下来,发送BYE之后,有些服务器会立即发送INVITE,是否发送BYE根据实际情况看
gb28181_agent_.stop();
btnGB28181Agent.setText("启动GB28181");
}
else {
if ( gb28181_agent_.start() ) {
btnGB28181Agent.setText("停止GB28181");
}
}
}
}
//停止GB28181 媒体流
private void stopGB28181Stream() {
if(!isGB28181StreamRunning)
return;
if (libPublisher != null) {
libPublisher.StopGB28181MediaStream(publisherHandle);
}
if (!isRecording && !isRTSPPublisherRunning && !isPushing) {
if (publisherHandle != 0) {
if (libPublisher != null) {
libPublisher.SmartPublisherClose(publisherHandle);
publisherHandle = 0;
}
}
}
isGB28181StreamRunning = false;
}
private void destoryRTPSender() {
if (gb28181_rtp_sender_handle_ != 0) {
libPublisher.DestoryRTPSender(gb28181_rtp_sender_handle_);
gb28181_rtp_sender_handle_ = 0;
}
}
GB28181网关模块，事件回调处理：
@Override
public void ntsRegisterOK(String dateString) {
Log.i(TAG, "ntsRegisterOK Date: " (dateString!= null? dateString : ""));
}
@Override
public void ntsRegisterTimeout() {
Log.e(TAG, "ntsRegisterTimeout");
}
@Override
public void ntsRegisterTransportError(String errorInfo) {
Log.e(TAG, "ntsRegisterTransportError error:" (errorInfo != null?errorInfo :""));
}
@Override
public void ntsOnHeartBeatException(int exceptionCount, String lastExceptionInfo) {
Log.e(TAG, "ntsOnHeartBeatException heart beat timeout count reached, count:" exceptionCount
", exception info:" (lastExceptionInfo!=null?lastExceptionInfo:""));
// 10毫秒后,停止信令, 然后重启
handler.postDelayed(new Runnable() {
@Override
public void run() {
Log.i(TAG, "gb28281_heart_beart_timeout");
stopGB28181Stream();
destoryRTPSender();
if (gb28181_agent_ != null) {
Log.i(TAG, "gb28281_heart_beart_timeout sip stop");
gb28181_agent_.stop();
Log.i(TAG, "gb28281_heart_beart_timeout sip start");
gb28181_agent_.start();
}
}
},10);
}
@Override
public void ntsOnInvitePlay(String deviceId, InvitePlaySessionDescription session_des) {
handler.postDelayed(new Runnable() {
@Override
public void run() {
Log.i(TAG,"ntsInviteReceived, device_id:" device_id_ ", is_tcp:" session_des_.isRTPOverTCP()
" rtp_port:" session_des_.getMediaPort() " ssrc:" session_des_.getSSRC()
" address_type:" session_des_.getAddressType() " address:" session_des_.getAddress());
// 可以先给信令服务器发送临时振铃响应
//sip_stack_android.respondPlayInvite(180, device_id_);
long rtp_sender_handle = libPublisher.CreateRTPSender(0);
if ( rtp_sender_handle == 0 ) {
gb28181_agent_.respondPlayInvite(488, device_id_);
Log.i(TAG, "ntsInviteReceived CreateRTPSender failed, response 488, device_id:" device_id_);
return;
}
gb28181_rtp_payload_type_ = session_des_.getPSRtpMapAttribute().getPayloadType();
libPublisher.SetRTPSenderTransportProtocol(rtp_sender_handle, session_des_.isRTPOverUDP()?0:1);
libPublisher.SetRTPSenderIPAddressType(rtp_sender_handle, session_des_.isIPv4()?0:1);
libPublisher.SetRTPSenderLocalPort(rtp_sender_handle, 0);
libPublisher.SetRTPSenderSSRC(rtp_sender_handle, session_des_.getSSRC());
libPublisher.SetRTPSenderSocketSendBuffer(rtp_sender_handle, 2*1024*1024); // 设置到2M
libPublisher.SetRTPSenderClockRate(rtp_sender_handle, session_des_.getPSRtpMapAttribute().getClockRate());
libPublisher.SetRTPSenderDestination(rtp_sender_handle, session_des_.getAddress(), session_des_.getMediaPort());
if ( libPublisher.InitRTPSender(rtp_sender_handle) != 0 ) {
gb28181_agent_.respondPlayInvite(488, device_id_);
libPublisher.DestoryRTPSender(rtp_sender_handle);
return;
}
int local_port = libPublisher.GetRTPSenderLocalPort(rtp_sender_handle);
if (local_port == 0) {
gb28181_agent_.respondPlayInvite(488, device_id_);
libPublisher.DestoryRTPSender(rtp_sender_handle);
return;
}
Log.i(TAG,"get local_port:" local_port);
String local_ip_addr = IPAddrUtils.getIpAddress(myContext);
gb28181_agent_.respondPlayInviteOK(device_id_,local_ip_addr, local_port);
gb28181_rtp_sender_handle_ = rtp_sender_handle;
}
private String device_id_;
private InvitePlaySessionDescription session_des_;
public Runnable set(String device_id, InvitePlaySessionDescription session_des) {
this.device_id_ = device_id;
this.session_des_ = session_des;
return this;
}
}.set(deviceId, session_des),0);
}
@Override
public void ntsOnCancelPlay(String deviceId) {
// 这里取消Play会话
handler.postDelayed(new Runnable() {
@Override
public void run() {
Log.i(TAG, "ntsOnCancelPlay, deviceId=" device_id_);
destoryRTPSender();
}
private String device_id_;
public Runnable set(String device_id) {
this.device_id_ = device_id;
return this;
}
}.set(deviceId),0);
}
@Override
public void ntsOnAckPlay(String deviceId) {
handler.postDelayed(new Runnable() {
@Override
public void run() {
Log.i(TAG,"ntsOnACKPlay, device_id:" device_id_);
if (!isRecording && !isRTSPPublisherRunning && !isPushing) {
OpenPushHandle();
}
libPublisher.SetGB28181RTPSender(publisherHandle, gb28181_rtp_sender_handle_, gb28181_rtp_payload_type_);
int startRet = libPublisher.StartGB28181MediaStream(publisherHandle);
if (startRet != 0) {
if (!isRecording && !isRTSPPublisherRunning && !isPushing) {
if (publisherHandle != 0) {
libPublisher.SmartPublisherClose(publisherHandle);
publisherHandle = 0;
}
}
destoryRTPSender();
Log.e(TAG, "Failed to start GB28181 service..");
return;
}
isGB28181StreamRunning = true;
}
private String device_id_;
public Runnable set(String device_id) {
this.device_id_ = device_id;
return this;
}
}.set(deviceId),0);
}
@Override
public void ntsOnPlayInviteResponseException(String deviceId, int statusCode, String errorInfo) {
// 这里要释放掉响应的资源
Log.i(TAG, "ntsOnPlayInviteResponseException, deviceId=" deviceId " statusCode=" statusCode
" errorInfo:" errorInfo);
handler.postDelayed(new Runnable() {
@Override
public void run() {
Log.i(TAG, "ntsOnPlayInviteResponseException, deviceId=" device_id_);
destoryRTPSender();
}
private String device_id_;
public Runnable set(String device_id) {
this.device_id_ = device_id;
return this;
}
}.set(deviceId),0);
}
@Override
public void ntsOnByePlay(String deviceId) {
handler.postDelayed(new Runnable() {
@Override
public void run() {
Log.i(TAG, "ntsOnByePlay, stop GB28181 media stream, deviceId=" device_id_);
stopGB28181Stream();
destoryRTPSender();
}
private String device_id_;
public Runnable set(String device_id) {
this.device_id_ = device_id;
return this;
}
}.set(deviceId),0);
}
@Override
public void ntsOnPlayDialogTerminated(String deviceId) {
/*
Play会话对应的对话终止, 一般不会出发这个回调,目前只有在响应了200K, 但在64*T1时间后还没收到ACK,才可能会出发
收到这个请做相关清理处理
*/
handler.postDelayed(new Runnable() {
@Override
public void run() {
Log.i(TAG, "ntsOnPlayDialogTerminated, deviceId=" device_id_);
stopGB28181Stream();
destoryRTPSender();
}
private String device_id_;
public Runnable set(String device_id) {
this.device_id_ = device_id;
return this;
}
}.set(deviceId),0);
}
@Override
public void ntsOnDevicePositionRequest(String deviceId, int interval) {
handler.postDelayed(new Runnable() {
@Override
public void run() {
getLocation(myContext);
Log.v(TAG, "ntsOnDevicePositionRequest, deviceId:" this.device_id_ ", Longitude:" mLongitude
", Latitude:" mLatitude ", Time:" mLocationTime);
if (mLongitude != null && mLatitude != null) {
com.gb28181.ntsignalling.DevicePosition device_pos = new com.gb28181.ntsignalling.DevicePosition();
device_pos.setTime(mLocationTime);
device_pos.setLongitude(mLongitude);
device_pos.setLatitude(mLatitude);
if (gb28181_agent_ != null ) {
gb28181_agent_.updateDevicePosition(device_id_, device_pos);
}
}
}
private String device_id_;
private int interval_;
public Runnable set(String device_id, int interval) {
this.device_id_ = device_id;
this.interval_ = interval;
return this;
}
}.set(deviceId, interval),0);
}OnDestroy()，停掉GB28181 stream，释放资源。
代码语言：javascript复制@Override
protected void onDestroy() {
Log.i(TAG, "Run into activity destory ");
if (gb28181_agent_ != null ) {
if (gb28181_agent_.isRunning()) {
gb28181_agent_.terminateAllPlays(false);
gb28181_agent_.stop();
}
Log.i(TAG, " gb28181_agent_.unInitialize ");
gb28181_agent_.unInitialize();
Log.i(TAG, " gb28181_agent_.unInitialize--");
gb28181_agent_ = null;
}
stopGB28181Stream();
destoryRTPSender();
super.onDestroy();
finish();
System.exit(0);
}以上是大概流程，感兴趣的开发者，可酌情参考。
