技术背景
上篇blog，我们提到了Android平台GB28181历史视音频文件检索规范探讨及技术实现，文件检索后，GB28181平台侧，可以针对文件列表进行回放或下载操作，本文主要探讨视音频文件下载相关。
规范解读
视音频文件下载基本要求
SIP 服务器接收到媒体接收者发送的视音频文件下载请求后向媒体流发送者发送媒体文件下载命令，媒体流发送者采用RTP将视频流传输给媒体流接收者，媒体流接收者直接将视频流保存为媒体文件。
媒体流接收者可以是用户客户端或联网系统，媒体流发送者可以是媒体设备或联网系统。媒体流接收者或 SIP 服务器可通过配置查询等方式获取媒体流发送者支持的下载发送倍速，并在请求的 SDP 消息体中携带指定下载倍速。
媒体流发送者可在 Invite 请求对应的 200 0K 响应 SDP 消息体中扩展携带下载文件的大小参数，以便于媒体流接收者计算下载进度,当媒体流发送者不能提供文件大小参数时，媒体流接收者应支持根据码流中取得的时间计算下载进度。视音频文件下载宜支持媒体流保活机制。
命令流程
其中,信令 1,8,9、10,11,12 为 SIP 服务器接收到客户端的呼叫请求后通过 B2BUA 代理方式建立媒体流接受者与媒体服务器之间的媒体链接信令过程。
信令 2~7 为 SIP 服务器通过三方呼叫控制建立媒体服务器与媒体流之间的媒体链接信令过程。
信令 13~16 为媒体流发送者回放下载到文件结束向媒体接收者发送下载完成的通知消息过程。
信令 17~20 为媒体流接收者断开与媒体服务器之间的媒体链接信令过程。
信令 21~24 为 SIP 服务器断开媒体服务器与媒体流发送者之间的媒体链接信令过程。
命令流程描述如下：
- 媒体流接收者向 SIP 服务器发送Invite 消息，消息头域中携带 Subject 字段，表明点播的视频源 ID、发送方媒体流序列号、媒体流接收者 ID、接收端媒体流序列号标识等参数，SDP 消息体中s字段为“Download”代表文件下载，u字段代表下载通道 ID 和下载类型，字段代表下载时间段，可扩展 a 字段携带下载倍速参数，规定此次下载设备发流倍速，若不携带默认为1 倍速。
- SIP 服务器收到 Invite 请求后，通过三方呼叫控制建立媒体服务器和媒体流发送者之间的媒体连接。向媒体服务器发送 Invite 消息，此消息不携带 SDP 消息体。
- 媒体服务器收到 SIP 服务器的 Invite 请求后,回复 200 0K 响应,携带 SDP 消息体,消息体中描述了媒体服务器接收媒体流的 IP端口、媒体格式等内容。
- SIP 服务器收到媒体服务器返回的 200 OK响应后，向媒体流发送者发送 Invite请求，请求中携带消息 3 中媒体服务器回复的 200 OK响应消息体。s字段为“Download”代表文件下载，u字段代表下载通道 ID 和下载类型，t字段代表下载时间段，增加y字段描述 SSRC 值，f字段描述媒体参数，可扩展 a 字段携带下载倍速，将倍速参数传递给设备。
- 媒体流发送者收到 SIP 服务器的 Invite 请求后,回复 200 OK响应，携带 SDP消息体，消息体中描述了媒体流发送者发送媒体流的IP、端口、媒体格式、SSRC 字段等内容,可扩展 a 字段携带文件大小参数。
- SIP 服务器收到媒体流发送者返回的 200 OK响应后，向媒体服务器发送 ACK 请求，请求中携带消息 5 中媒体流发送者回复的 200 OK响应消息体，完成与媒体服务器的 Invite 会话建立过程。
- SIP 服务器收到媒体流发送者返回的 200 OK响应后，向媒体流发送者发送 ACK 请求，请求中不携带消息体，完成与媒体流发送者的 Invite 会话建立过程。
- 完成三方呼叫控制后，SIP 服务器通过 B2BUA 代理方式建立媒体流接收者和媒体服务器之间的媒体连接。在消息 1 中增加 SSRC 值，转发给媒体服务器。
- 媒体服务器收到 Invite 请求，回复 200 OK响应，携带 SDP 消息体，消息体中描述了媒体服务器发送媒体流的IP、端口、媒体格式、SSRC 值等内容。
- SIP 服务器将消息 9 转发给媒体流接收者，可扩展 a 字段携带文件大小参数。
- 媒体流接收者收到 200 OK响应后，回复 ACK 消息，完成与 SIP 服务器的 Invite 会话建立过程。
- SIP 服务器将消息 11 转发给媒体服务器，完成与媒体服务器的 Invite 会话建立过程。
- 媒体流发送者在文件下载结束后发送会话内 Message 消息。
- SIP 服务器收到消息 17 后转发给媒体流接收者。
- 媒体流接收者收到消息 18 后回复 200 OK响应，进行链路断开过程。
- SIP 服务器将消息 19 转发给媒体流发送者。
- 媒体流接收者向 SIP 服务器发送 BYE 消息,断开消息1、10、11建立的同媒体流接收者的Invite 会话。
- SIP服务器收到 BYE消息后回复200OK 响应,会话断开。
- SIP 服务器收到 BYE 消息后向媒体服务器发送 BYE 消息,断开消息 8,9,12 建立的同媒体服务器的 Invite 会话。
- 媒体服务器收到 BYE 消息后回复 200 OK 响应，会话断开。
- SIP 服务器向媒体服务器发送 BYE 消息，断开消息 2,3,6 建立的同媒体服务器的 Invite 会话。
- 媒体服务器收到 BYE 消息后回复 200 OK响应，会话断开。
- SIP 服务器向媒体流发送者发送 BYE 消息,断开消息 4,5,7 建立的同媒体流发送者的Invite 会话。
- 媒体流发送者收到 BYE 消息后回复 200 OK响应，会话断开。
技术实现
本文以大牛直播SDK开发的Android平台GB28181设备接入侧视音频历史文件检索和下载为例（本文侧重于下载），介绍下相关设计思路：
Android设备接入端收到国标平台侧发过来的INVITE SDP：
v=0
o=34020000001380000001 0 0 IN IP4 192.168.2.154
s=Download
u=34020000001380000001:0
c=IN IP4 192.168.2.154
t=1693796426 1693796703
m=video 30002 RTP/AVP 96 97 98
a=recvonly
a=rtpmap:96 PS/90000
a=rtpmap:97 MPEG4/90000
a=rtpmap:98 H264/90000
a=downloadspeed:4
y=1200000001上述SDP里面，s=Download表示系下载，a=downloadspeed:4 表示4倍速下载，SSRC是:1200000001(SSRC第1位为历史或实时媒体流的标识位，其中0为实时视音频，1为历史视音频)。
Android设备接入端回复：
v=0
o=34020000011310000039 0 0 IN IP4 192.168.2.212
s=Download
c=IN IP4 192.168.2.212
t=0 0
m=video 36576 RTP/AVP 96
a=rtpmap:96 PS/90000
a=filesize:15611511
a=sendonly
y=1200000001a=filesize:15611511表示录像文件大小是15611511Byte，携带文件大小参数, 便于媒体流接收者计算下载进度(a=filesize是整个媒体容器的大小和实际发送的音视频帧总字节数有一定差异)。
国标平台侧发Ack后，开始下载视音频数据，下载过程中，可以通过SIP-INFO消息和MANSRTSP协议调节下载倍速：
PLAY RTSP/1.0
CSeq: 31129
Scale: 0.25Android GB28181设备接入侧发送完音视频帧后，发送通知事件类型”121″, 表示历史媒体文件发送结束，发送会话内Message消息如下：
MediaStatus
213466963
34020000001380000001
121
接口设计
信令接口设计：
/**
* Author: daniusdk.com
*/
package com.gb.ntsignalling;
public interface GBSIPAgent {
void addDownloadListener(GBSIPAgentDownloadListener downloadListener);
void removeDownloadListener(GBSIPAgentDownloadListener removeListener);
/*
*响应Invite Download 200 OK
*/
boolean respondDownloadInviteOK(long id, String deviceId, String startTime, String stopTime, MediaSessionDescription localMediaDescription);
/*
*响应Invite Download 其他状态码
*/
boolean respondDownloadInvite(int statusCode, long id, String deviceId, String startTime, String stopTime);
/*
* 媒体流发送者在文件下载结束后发Message消息通知SIP服务器回文件已发送完成
* notifyType 必须是"121“
*/
boolean notifyDownloadMediaStatus(long id, String deviceId, String startTime, String stopTime, String notifyType);
/*
*终止Download会话
*/
void terminateDownload(long id, String deviceId, String startTime, String服务器托管网 stopTime, boolean isSendBYE);
/*
*终止所有Download会话
*/
void terminateAllDownloads(boolean isSendBYE);
}历史视音频下载listener设计：
/**
* Author: danius服务器托管网dk.com
*/
package com.gb.ntsignalling;
public interface GBSIPAgentDownloadListener {
/*
*收到s=Download的文件下载Invite
*/
void ntsOnInviteDownload(long id, String deviceId, SessionDescription sessionDescription);
/*
*发送Download invite response 异常
*/
void ntsOnDownloadInviteResponseException(long id, String deviceId, String startTime, String stopTime, int statusCode, String errorInfo);
/*
* 收到CANCEL Download INVITE请求
*/
void ntsOnCancelDownload(long id, String deviceId, String startTime, String stopTime);
/*
* 收到Ack
*/
void ntsOnAckDownload(long id, String deviceId, String startTime, String stopTime);
/*
* 更改下载速度
*/
void ntsOnDownloadMANSRTSPScaleCommand(long id, String deviceId, String startTime, String stopTime, double scale);
/*
* 收到Bye
*/
void ntsOnByeDownload(long id, String deviceId, String startTime, String stopTime);
/*
* 不是在收到BYE Message情况下， 终止Download
*/
void ntsOnTerminateDownload(long id, String deviceId, String startTime, String stopTime);
/*
* Download会话对应的对话终止, 一般不会触发这个回调,目前只有在响应了200K, 但在64*T1时间后还没收到ACK,才可能会出发
收到这个, 请做相关清理处理
*/
void ntsOnDownloadDialogTerminated(long id, String deviceId, String startTime, String stopTime);
}底层jni接口设计：
/**
* SmartPublisherJniV2.java
* Author: daniusdk.com
*/
package com.daniulive.smartpublisher;
public class SmartPublisherJniV2 {
/**
* Open publisher(启动推送实例)
*
* @param ctx: get by this.getApplicationContext()
*
* @param audio_opt:
* if 0: 不推送音频
* if 1: 推送编码前音频(PCM)
* if 2: 推送编码后音频(aac/pcma/pcmu/speex).
*
* @param video_opt:
* if 0: 不推送视频
* if 1: 推送编码前视频(NV12/I420/RGBA8888等格式)
* if 2: 推送编码后视频(AVC/HEVC)
* if 3: 层叠加模式
*
* This function must be called firstly.
*
* @return the handle of publisher instance
*/
public native long SmartPublisherOpen(Object ctx, int audio_opt, int video_opt, int width, int height);
/**
* 设置流类型
* @param type: 0:表示 live 流, 1:表示 on-demand 流, SDK默认为0(live流)
* 注意: 流类型设置当前仅对GB28181媒体流有效
* @return {0} if successful
*/
public native int SetStreamType(long handle, int type);
/**
* 投递视频 on demand包, 当前只用于GB28181推送, 注意ByteBuffer对象必须是DirectBuffer
*
* @param codec_id: 编码id, 当前支持H264和H265, 1:H264, 2:H265
*
* @param packet: 视频数据, 包格式请参考H264/H265 Annex B Byte stream format, 例如:
* 0x00000001 nal_unit 0x00000001 ...
* H264 IDR: 0x00000001 sps 0x00000001 pps 0x00000001 IDR_nal_unit .... 或 0x00000001 IDR_nal_unit ....
* H265 IDR: 0x00000001 vps 0x00000001 sps 0x00000001 pps 0x00000001 IDR_nal_unit .... 或 0x00000001 IDR_nal_unit ....
*
* @param offset: 偏移量
* @param size: packet size
* @param pts_us: 时间戳, 单位微秒
* @param is_pts_discontinuity: 是否时间戳间断，0:未间断，1:间断
* @param is_key: 是否是关键帧, 0:非关键帧, 1:关键帧
* @param codec_specific_data: 可选参数，可传null, 对于H264关键帧包, 如果packet不含sps和pps, 可传0x00000001 sps 0x00000001 pps
* ,对于H265关键帧包, 如果packet不含vps,sps和pps, 可传0x00000001 vps 0x00000001 sps 0x00000001 pps
* @param codec_specific_data_size: codec_specific_data size
* @param width: 图像宽, 可传0
* @param height: 图像高, 可传0
*
* @return {0} if successful
*/
public native int PostVideoOnDemandPacketByteBuffer(long handle, int codec_id,
ByteBuffer packet, int offset, int size, long pts_us, int is_pts_discontinuity, int is_key,
byte[] codec_specific_data, int codec_specific_data_size,
int width, int height);
/**
* 投递音频on demand包, 当前只用于GB28181推送, 注意ByteBuffer对象必须是DirectBuffer
*
* @param codec_id: 编码id, 当前支持PCMA和AAC, 65536:PCMA, 65538:AAC
* @param packet: 音频数据
* @param offset：packet偏移量
* @param size: packet size
* @param pts_us: 时间戳, 单位微秒
* @param is_pts_discontinuity: 是否时间戳间断，0:未间断，1:间断
* @param codec_specific_data: 如果是AAC的话，需要传 Audio Specific Configuration
* @param codec_specific_data_size: codec_specific_data size
* @param sample_rate: 采样率
* @param channels: 通道数
*
* @return {0} if successful
*/
public native int PostAudioOnDemandPacketByteBuffer(long handle, int codec_id,
ByteBuffer packet, int offset, int size, long pts_us, int is_pts_discontinuity,
byte[] codec_specific_data, int codec_specific_data_size,
int sample_rate, int channels);
/**
* 启动 GB28181 媒体流
*
* @return {0} if successful
*/
public native int StartGB28181MediaStream(long handle);
/**
* 停止 GB28181 媒体流
*
* @return {0} if successful
*/
public native int StopGB28181MediaStream(long handle);
/**
* 关闭推送实例，结束时必须调用close接口释放资源
*
* @return {0} if successful
*/
public native int SmartPublisherClose(long handle);
}
上次处理逻辑
RecordDownloadListenerImpl实现如下：
/**
* RecordDownloadListenerImpl.java
* Author: daniusdk.com
*/
package com.daniulive.smartpublisher;
public class RecordDownloadListenerImpl implements com.gb.ntsignalling.GBSIPAgentDownloadListener {
/*
*收到s=Download的文件下载Invite
*/
@Override
public void ntsOnInviteDownload(long id, String deviceId, SessionDescription sdp) {
if (!post_task(new OnInviteTask(this.context_, this.is_exit_, this.senders_map_, deviceId, sdp, id))) {
Log.e(TAG, "ntsOnInviteDownload post_task failed, " + RecordSender.make_print_tuple(id, deviceId, sdp.getTime().getStartTime(), sdp.getTime().getStopTime()));
// 这里不发488, 等待事务超时也可以的
GBSIPAgent agent = this.context_.get_agent();
if (agent != null)
agent.respondDownloadInvite(488, id, deviceId, sdp.getTime().getStartTime(), sdp.getTime().getStopTime());
}
}
/*
* 收到CANCEL Download INVITE请求
*/
@Override
public void ntsOnCancelDownload(long id, String deviceId, String startTime, String stopTime) {
Log.i(TAG, "ntsOnCancelDownload, " + RecordSender.make_print_tuple(id, deviceId, startTime, stopTime));
RecordSender sender = senders_map_.remove(id);
if (null == sender)
return;
StopDisposeTask task = new StopDisposeTask(sender);
if (!post_task(task))
task.run();
}
/*
* 收到Ack
*/
@Override
public void ntsOnAckDownload(long id, String deviceId, String startTime, String stopTime) {
Log.i(TAG, "ntsOnAckDownload, "+ RecordSender.make_print_tuple(id, deviceId, startTime, stopTime));
RecordSender sender = senders_map_.get(id);
if (null == sender) {
Log.e(TAG, "ntsOnAckDownload get sender is null, " + RecordSender.make_print_tuple(id, deviceId, startTime, stopTime));
GBSIPAgent agent = this.context_.get_agent();
if (agent != null)
agent.terminateDownload(id, deviceId, startTime, stopTime, false);
return;
}
StartTask task = new StartTask(sender, this.senders_map_);
if (!post_task(task))
task.run();
}
/*
* 收到Bye
*/
@Override
public void ntsOnByeDownload(long id, String deviceId, String startTime, String stopTime) {
Log.i(TAG, "ntsOnByeDownload, "+ RecordSender.make_print_tuple(id, deviceId, startTime, stopTime));
RecordSender sender = this.senders_map_.remove(id);
if (null == sender)
return;
StopDisposeTask task = new StopDisposeTask(sender);
if (!post_task(task))
task.run();
}
/*
* 更改下载速度
*/
@Override
public void ntsOnDownloadMANSRTSPScaleCommand(long id, String deviceId, String startTime, String stopTime, double scale) {
if (scale 文件发送相关处理代码如下：
/**
* RecordSender.java
* Author: daniusdk.com
*/
package com.daniulive.smartpublisher;
public class RecordSender {
public void set_speed(double speed) {
int percent_speed = (int)(speed*100);
this.percent_speed_.set(percent_speed);
}
public void set_file_description(RecordFileDescription desc) {
this.file_description_ = desc;
}
public static String make_print_tuple(long id, String device_id, String start_time, String stop_time) {
StringBuilder sb = new StringBuilder(96);
sb.append("[id:").append(id);
sb.append(", device:" + device_id);
sb.append(", t=").append(start_time).append(" ").append(start_time);
sb.append("]");
return sb.toString();
}
public boolean start() {
SendThread current_thread = thread_.get();
if (current_thread != null) {
if (current_thread.is_exit()) {
Log.e(TAG, "start, the thread already exists and has exited, return false, " + get_print_tuple());
return false;
}
Log.i(TAG, "start, the thread already exists and has exited, return true, " + get_print_tuple());
return true;
}
SendThread thread = new SendThread();
if (!thread_.compareAndSet(null, thread)) {
Log.i(TAG, "start, call compareAndSet return false, the thread already exists, return true, " + get_print_tuple());
return true;
}
try {
Log.i(TAG, "start thread, " + get_print_tuple());
thread.start();
}catch (Exception e) {
thread_.compareAndSet(thread, null);
Log.e(TAG, "start e:", e);
return false;
}
return true;
}
public void stop() {
SendThread current_thread = thread_.get();
if (current_thread != null && !current_thread.is_exit()) {
current_thread.exit();
Log.i(TAG, "stop, exit thread " + get_print_tuple());
}
}
private boolean init_native_sender(StackDisposable disposables) {
if(native_handle_ !=0) {
Log.e(TAG, "init_native_sender, native_handle_ is not 0, " + get_print_tuple());
return false;
}
if (null == this.media_info_ || !this.media_info_.is_has_track() ) {
Log.e(TAG, "init_native_sender, there is no track, " + get_print_tuple());
return false;
}
if (0 == rtp_handle_) {
Log.e(TAG, "init_native_sender, rtp_handle_ is 0, " + get_print_tuple());
return false;
}
if (null == lib_publisher_){
Log.e(TAG, "init_native_sender, lib_publisher_ is null, " + get_print_tuple());
return false;
}
Context context = this.context_.get_context();
if (null == context) {
Log.e(TAG, "init_native_sender, context is null, " + get_print_tuple());
return false;
}
long handle = lib_publisher_.SmartPublisherOpen(context, media_info_.is_has_audio_track()?2:0, media_info_.is_has_video_track()?2:0, 0, 0);
if (0 == handle) {
Log.e(TAG, "init_native_sender, call SmartPublisherOpen failed, " + get_print_tuple());
return false;
}
NativeSenderDisposable native_disposable = new NativeSenderDisposable(lib_publisher_, handle);
lib_publisher_.SetStreamType(handle, 1);
List tracks = media_info_.get_tracks();
for (MediaTrack i : tracks) {
if (i.is_video())
lib_publisher_.SetEncodedVideoCodecId(handle, i.codec_id(), i.csd_set(), i.csd_set() != null? i.csd_set().length : 0);
else if(i.is_audio())
lib_publisher_.SetEncodedAudioCodecId(handle, i.codec_id(), i.csd_set(), i.csd_set() != null? i.csd_set().length : 0);
}
lib_publisher_.SetGB28181RTPSender(handle, rtp_handle_, rtp_payload_type_, rtp_encoding_name_);
int ret = lib_publisher_.StartGB28181MediaStream(handle);
if (ret != 0) {
Log.e(TAG, "init_native_sender, call StartGB28181MediaStream failed, " + get_print_tuple());
native_disposable.dispose();
return false;
}
native_disposable.is_need_call_stop(true);
disposables.push(native_disposable);
native_handle_ = handle;
return true;
}
private boolean post_media_packet(MediaPacket packet) {
/*Log.i(TAG, "post "+ MediaTrack.get_media_type_string(packet.media_type()) + " " +
MediaTrack.get_codec_id_string(packet.codec_id()) + " packet, pts:" + out_point_3(packet.pts_us()/1000.0) +"ms, key:"
+ (packet.is_key()?1:0) + ", size:" + packet.size()); */
if (null == lib_publisher_ || 0 == native_handle_ || !packet.is_has_data())
return false;
if (packet.is_audio()) {
if (packet.is_aac()) {
if (packet.is_has_codec_specific_data_set())
return 0 == lib_publisher_.PostAudioOnDemandPacketByteBuffer(native_handle_, packet.codec_id(), packet.data(), 0, packet.size(),
packet.pts_us(), 0, packet.codec_specific_data_set(), packet.codec_specific_data_set_size(), 0, 0);
}
}else if (packet.is_video()) {
if (packet.is_avc() || packet.is_hevc()) {
return 0 == lib_publisher_.PostVideoOnDemandPacketByteBuffer(native_handle_, packet.codec_id(), packet.data(),
0, packet.size(), packet.pts_us(), 0, packet.is_key()?1:0,
packet.codec_specific_data_set(), packet.codec_specific_data_set_size(), 0, 0);
}
}
return false;
}
private void release_packets(Deque packets) {
while (!packets.isEmpty())
packets.removeFirst().release_buffer();
}
private static String out_point_3(double v) { return String.format("%.3f", v); }
public static String to_mega_bytes_string(long bytes) {
double mb = bytes/(1024*1024.0);
return out_point_3(mb);
}
private class SendThread extends Thread {
@Override
public void run() {
/***
*相关代码
**/
}
}
}总结
GB28181历史视音频文件下载，看似逻辑复杂，实际上也不简单，文件下载是在完成录像和历史视音频文件检索的基础上，分别从信令、RTP数据打包发送等角度实现，考虑到录像文件的完整性，历史视音频文件下载的设计目标是减少丢帧丢包，推荐使用RTP over TCP模式，尽管作为GB28181设备接入侧，我们尽可能按照标准规范来实现，实际对接的国标平台厂商，多少会有些差异，具体还要根据现场实际情况酌情处理。
服务器托管，北京服务器托管，服务器租用 http://www.fwqtg.net
机房租用，北京机房租用，IDC机房托管， http://www.fwqtg.net
“final” 关键字通常用于修饰类、方法或变量。当应用于类时，它表示该类是最终版本，不能被继承。当应用于方法时，它表示该方法是最终实现，不能被子类重写。 public final class MyClass { // Class implementation…
