本文分享RTMP推流到SRS流媒体服务器消息处理的相关流程。
SRS和客户端是怎么交换消息的？各个消息有什么作用？握手成功后，SRS和客户端进行消息交换，对应wiresharek这部分截图：
客户端发送connect连接请求到SRS服务端
connect命令，⽤于客户端向服务器发送连接请求。
wiresharek截图：
connect消息解析
Chunk Stream Id：接收端根据相同的chunk stream id拼装出message， 这⾥是3，对应ffmpeg的枚举RTMP_SYSTEM_CHANNEL。
Type ID （即是message type id） ⽐如8⾳频，9视频，20命令消息，这⾥是20命令消息，对应ffmpeg的枚举RTMP_PT_INVOKE。
Stream ID为0。
RTMP body是⼀个connect 命令消息，这些信息是以AMF格式发送的，消息的结构如下：
第三个字段中的Command Object中会涉及到很多键值对，使⽤时可以参考协议的官⽅⽂档。消息的回应有两种，_result表示接受连接，_error表示连接失败。
以下是连接命令对象中使⽤的健值对的描述：
connect消息代码分析
1. 客户端生成connect消息发送到SRS服务器
对应FFmpeg代码在：
gen_connect(s, rt)
static int gen_connect(URLContext *s, RTMPContext *rt)
//部分代码如下：
ff_amf_write_string(&p, "connect");
ff_amf_write_number(&p, ++rt->nb_invokes);
ff_amf_write_object_start(&p);
ff_amf_write_field_name(&p, "app");
ff_amf_write_string2(&p, rt->app, rt->auth_params);2. SRS服务端接收connect消息并解析
对应SRS代码在：
rtmp->connect_app(req)) != srs_success)
srs_error_t SrsRtmpServer::connect_app(SrsRequest* req) {
srs_error_t err = srs_success;
SrsCommonMessage* msg = NULL;
SrsConnectAppPacket* pkt = NULL;
if ((err = expect_message<SrsConnectAppPacket>(&msg, &pkt)) != srs_success) { //解析connect消息，从流式数据->chunk->message->packet
return srs_error_wrap(err, "expect connect app response");
}
SrsAutoFree(SrsCommonMessage, msg);
SrsAutoFree(SrsConnectAppPacket, pkt);
SrsAmf0Any* prop = NULL;
if ((prop = pkt->command_object->ensure_property_string("tcUrl")) == NULL) {
return srs_error_new(ERROR_RTMP_REQ_CONNECT, "invalid request without tcUrl");
}
req->tcUrl = prop->to_str();
if ((prop = pkt->command_object->ensure_property_string("pageUrl")) != NULL) {
req->pageUrl = prop->to_str();
}
if ((prop = pkt->command_object->ensure_property_string("swfUrl")) != NULL) {
req->swfUrl = prop->to_str();
}
if ((prop = pkt->command_object->ensure_property_number("objectEncoding")) != NULL) {
req->objectEncoding = prop->to_number();
}
if (pkt->args) {
srs_freep(req->args);
req->args = pkt->args->copy()->to_object();
}
srs_discovery_tc_url(req->tcUrl, req->schema, req->host, req->vhost, req->app, req->stream, req->port, req->param);
req->strip();
return err;
}具体解析为packet代码为：
srs_error_t SrsProtocol::do_decode_message(SrsMessageHeader& header, SrsBuffer* stream, SrsPacket** ppacket)
{
...
// decode command object.
if (command == RTMP_AMF0_COMMAND_CONNECT) { //第一个接收到的message为"connect"消息
*ppacket = packet = new SrsConnectAppPacket();
return packet->decode(stream);
}
...
}gdb显示调用栈为：
(gdb) bt
#0 SrsProtocol::do_decode_message (this=0xac1280, header=..., stream=0xadb560, ppacket=0xadb550) at src/protocol/srs_rtmp_stack.cpp:700
#1 0x00000000004655ec in SrsProtocol::decode_message (this=0xac1280, msg=0xade990, ppacket=0xadb5e0) at src/protocol/srs_rtmp_stack.cpp:430
#2 0x000000000047f8c0 in SrsProtocol::expect_message<SrsConnectAppPacket> (this=0xac1280, pmsg=0xadb670, ppacket=0xadb678) at src/protocol/srs_rtmp_stack.hpp:335
#3 0x000000000047dc7d in SrsRtmpServer::expect_message<SrsConnectAppPacket> (this=0xac1250, pmsg=0xadb670, ppacket=0xadb678) at src/protocol/srs_rtmp_stack.hpp:776
#4 0x000000000046df7b in SrsRtmpServer::connect_app (this=0xac1250, req=0xacab30) at src/protocol/srs_rtmp_stack.cpp:2310
#5 0x00000000004d3594 in SrsRtmpConn::do_cycle (this=0xac0f10) at src/app/srs_app_rtmp_conn.cpp:170
#6 0x00000000004d1d99 in SrsConnection::cycle (this=0xac0f88) at src/app/srs_app_conn.cpp:171
#7 0x000000000050ab08 in SrsSTCoroutine::cycle (this=0xac11f0) at src/app/srs_app_st.cpp:198
#8 0x000000000050ab7d in SrsSTCoroutine::pfn (arg=0xac11f0) at src/app/srs_app_st.cpp:213
#9 0x00000000005bed1a in _st_thread_main () at sched.c:337
#10 0x00000000005bf492 in st_thread_create (start=0x5be696 <_st_vp_schedule+170>, arg=0x700000001, joinable=1, stk_size=1) at sched.c:616
Backtrace stopped: previous frame inner to this frame (corrupt stack?)SRS服务器发送Window Acknowledgement Size消息给客户端，默认2500000
Window Acknowledgement Size消息解析
wiresharek截图：
Type Id：0x5 对应ffmepg的RTMP_PT_WINDOW_ACK_SIZE
Window Acknowledgement Size⽤于设置窗⼝确认⼤⼩，⽐如这⾥是服务器发送给客户端的，当客户端收到每次该size就要Acknowledgement (0x3)。
对于ffmpeg，在接收到Window Acknowledgement Size的⼀半后发送确认包（Acknowledgement），以确保对等⽅可以继续发送⽽不等待确认。
注意点：
- 客户端作为推流端时，⼀般即使没有收到服务器的ack，客户端也不会停⽌码流的推送。
- 当客户端作为拉流端时，⼀般即使拉流端不回应ack，服务器也不会停⽌码流的发送。
- 但彼此如果作为接收⽅时，收到1/2Windows size的数据后对会ack对⽅。
Window Acknowledgement Size消息代码分析
1. SRS服务器发送Window Acknowledgement Size消息给客户端
rtmp->set_window_ack_size(out_ack_size)
srs_error_t SrsRtmpServer::set_window_ack_size(int ack_size){
srs_error_t err = srs_success;
SrsSetWindowAckSizePacket* pkt = new SrsSetWindowAckSizePacket();
pkt->ackowledgement_window_size = ack_size;
if ((err = protocol->send_and_free_packet(pkt, 0)) != srs_success) {
return srs_error_wrap(err, "send ack");
}
return err;
}2. FFmpeg客户端处理SRS服务器发送过来的
static int handle_window_ack_size(URLContext *s, RTMPPacket *pkt)
{
RTMPContext *rt = s->priv_data;
if (pkt->size < 4) {
av_log(s, AV_LOG_ERROR,
"Too short window acknowledgement size packet (%d)\n",
pkt->size);
return AVERROR_INVALIDDATA;
}
rt->receive_report_size = AV_RB32(pkt->data);
if (rt->receive_report_size <= 0) {
av_log(s, AV_LOG_ERROR, "Incorrect window acknowledgement size %d\n",
rt->receive_report_size);
return AVERROR_INVALIDDATA;
}
av_log(s, AV_LOG_DEBUG, "Window acknowledgement size = %d\n", rt->receive_report_size);
// Send an Acknowledgement packet after receiving half the maximum
// size, to make sure the peer can keep on sending without waiting
// for acknowledgements.
rt->receive_report_size >>= 1;
return 0;
}srs服务器发送Peer Bandiwdth消息给客户端，默认2500000
Peer Bandiwdth消息解析
wiresharek截图：
客户端或服务器端发送此消息更新对端的输出带宽，作用是限制对端的输出带宽。和Window Acknowledgement Size相⽐，重点是更新。
Type ID：0x6 对应ffmpeg的RTMP_PT_SET_PEER_BW
如果消息中的Window ACK Size与上⼀次发送给发送端的size不同的话要回馈⼀个Window Acknowledgement Size的控制消息：
- Hard(Limit Type＝0)：接收端应该将Window Ack Size设置为消息中的值
- Soft(Limit Type=1)：接收端可以将Window Ack Size设为消息中的值，也可以保存原来的值（前提是原来的Size⼩与该控制消息中的Window Ack Size）
- Dynamic(Limit Type=2)：如果上次的Set Peer Bandwidth消息中的Limit Type为0，本次也按Hard处理，否则忽略本消息，不去设置Window Ack Size
Peer Bandiwdth消息代码分析
SRS服务器发送 Peer Bandiwdth消息给客户端。
srs_error_t SrsRtmpServer::set_peer_bandwidth(int bandwidth, int type)
{
srs_error_t err = srs_success;
SrsSetPeerBandwidthPacket* pkt = new SrsSetPeerBandwidthPacket();
pkt->bandwidth = bandwidth;
pkt->type = type;
if ((err = protocol->send_and_free_packet(pkt, 0)) != srs_success) {
return srs_error_wrap(err, "send set peer bandwidth");
}
return err;
}srs服务器发送Set Chunk Size消息给客户端
Set Chunk Size消息解析
wiresharek截图：
Set Chunk Size(Message Type ID=1)：设置chunk中Data字段所能承载的最⼤字节数，默认为128B，通信过程中可以通过发送该消息来设置chunk Size的⼤⼩（不得⼩于128B），⽽且通信双⽅会各⾃维护⼀个chunkSize，两端的chunkSize是独⽴的。
- ⽐如当A想向B发送⼀个200B的Message，但默认的chunkSize是128B，因此就要将该消息拆分为Data分别为128B和72B的两个chunk发送，如果此时先发送⼀个设置chunkSize为256B的消息，再发送Data为200B的chunk，本地不再划分Message，B接受到Set Chunk Size的协议控制消息时会调整的接受的chunk的Data的⼤⼩，也不⽤再将两个chunk组成为⼀个Message。
- 在实际中⼀般会把chunk size设置的很⼤，有的会设置为4096，FFMPEG推流的时候设置的是 60*1000，这样设置的好处是避免了频繁的拆包组包，占⽤过多的CPU。
以下为代表Set Chunk Size消息的chunk的Data：
其中第⼀位必须为0，chunk Size占31个位，最⼤可代表2147483647＝0x7FFFFFFF＝2 -1，但实际上所有⼤于16777215=0xFFFFFF的值都⽤不上，因为chunk size不能⼤于Message的⻓度，表示Message的⻓度字段是⽤3个字节表示的，最⼤只能为0xFFFFFF。
Set Chunk Size消息代码分析
SRS服务器向客户端发送Set Chunk Size消息：
srs_error_t SrsRtmpServer::set_chunk_size(int chunk_size)
{
srs_error_t err = srs_success;
SrsSetChunkSizePacket* pkt = new SrsSetChunkSizePacket();
pkt->chunk_size = chunk_size;
if ((err = protocol->send_and_free_packet(pkt, 0)) != srs_success) {
return srs_error_wrap(err, "send set chunk size");
}
return err;
}FFmpeg客户端处理SRS服务器发送过来的Set Chunk Size消息：
static int handle_chunk_size(URLContext *s, RTMPPacket *pkt)
{
RTMPContext *rt = s->priv_data;
int ret;
if (pkt->size < 4) {
av_log(s, AV_LOG_ERROR,
"Too short chunk size change packet (%d)\n",
pkt->size);
return AVERROR_INVALIDDATA;
}
if (!rt->is_input) {
/* Send the same chunk size change packet back to the server,
* setting the outgoing chunk size to the same as the incoming one. */
if ((ret = ff_rtmp_packet_write(rt->stream, pkt, rt->out_chunk_size,
&rt->prev_pkt[1], &rt->nb_prev_pkt[1])) < 0)
return ret;
rt->out_chunk_size = AV_RB32(pkt->data);
}
rt->in_chunk_size = AV_RB32(pkt->data);
if (rt->in_chunk_size <= 0) {
av_log(s, AV_LOG_ERROR, "Incorrect chunk size %d\n",
rt->in_chunk_size);
return AVERROR_INVALIDDATA;
}
av_log(s, AV_LOG_DEBUG, "New incoming chunk size = %d\n",
rt->in_chunk_size);
return 0;
}SRS服务器发送response_connect_app消息给客户端
SRS服务器是会发送response_connect_app消息给客户端用来响应客户端发送的connect消息，但这部分wiresharek是没有抓取到此包的。
通过tcpdump命令查看包信息：
sudo tcpdump -i any port 1935 -XX and dst 36.112.32.2 #不是公网地址是有返回result消息的，只是wiresharek没有抓取到此包。
response_connect_app消息代码分析
SRS服务器向客户端发送esponse_connect_app消息：
srs_error_t SrsRtmpServer::response_connect_app(SrsRequest *req, const char* server_ip)
{
srs_error_t err = srs_success;
SrsConnectAppResPacket* pkt = new SrsConnectAppResPacket();
// @remark For windows, there must be a space between const string and macro.
pkt->props->set("fmsVer", SrsAmf0Any::str("FMS/" RTMP_SIG_FMS_VER));
pkt->props->set("capabilities", SrsAmf0Any::number(127));
pkt->props->set("mode", SrsAmf0Any::number(1));
pkt->info->set(StatusLevel, SrsAmf0Any::str(StatusLevelStatus));
pkt->info->set(StatusCode, SrsAmf0Any::str(StatusCodeConnectSuccess));
pkt->info->set(StatusDescription, SrsAmf0Any::str("Connection succeeded"));
pkt->info->set("objectEncoding", SrsAmf0Any::number(req->objectEncoding));
SrsAmf0EcmaArray* data = SrsAmf0Any::ecma_array();
pkt->info->set("data", data);
data->set("version", SrsAmf0Any::str(RTMP_SIG_FMS_VER));
data->set("srs_sig", SrsAmf0Any::str(RTMP_SIG_SRS_KEY));
data->set("srs_server", SrsAmf0Any::str(RTMP_SIG_SRS_SERVER));
data->set("srs_license", SrsAmf0Any::str(RTMP_SIG_SRS_LICENSE));
data->set("srs_url", SrsAmf0Any::str(RTMP_SIG_SRS_URL));
data->set("srs_version", SrsAmf0Any::str(RTMP_SIG_SRS_VERSION));
if (server_ip) {
data->set("srs_server_ip", SrsAmf0Any::str(server_ip));
}
// for edge to directly get the id of client.
data->set("srs_pid", SrsAmf0Any::number(getpid()));
data->set("srs_id", SrsAmf0Any::number(_srs_context->get_id()));
if ((err = protocol->send_and_free_packet(pkt, 0)) != srs_success) {
return srs_error_wrap(err, "send connect app response");
}
return err;
}客户端发送Set Chunk Size消息给SRS服务器
wiresharek截图：
客户端也会发送Set Chunk Size消息给SRS服务器，值为60000。
客户端发送releaseStream、FCPublish、CreateStream和checkbw消息给SRS服务器
客户端还需要发送 releaseStream , FCPublish，CreateStream和checkbw消息给SRS服务器，具体可以参考ffmpeg的handle_invoke_result函数和gen_check_bw函数。
handle_invoke_result函数部分代码：
static int handle_invoke_result(URLContext *s, RTMPPacket *pkt)
{
...
if (!strcmp(tracked_method, "connect")) {
if (!rt->is_input) { //输出
if ((ret = gen_release_stream(s, rt)) < 0) //生成releaseStream发送给服务器
goto fail;
if ((ret = gen_fcpublish_stream(s, rt)) < 0)//生成FCPublish发送给服务器
goto fail;
} else { //输入
if ((ret = gen_window_ack_size(s, rt)) < 0)
goto fail;
}
if ((ret = gen_create_stream(s, rt)) < 0)//生成createStream发送给服务器
goto fail;
...
} else if (!strcmp(tracked_method, "createStream")) {
...
}
fail:
av_free(tracked_method);
return ret;
}客户端发送releaseStream消息给srs服务器
对应的值liveStream是要处理的流，作用是生成’releaseStream’调用并发送到服务器，让服务器为媒体流释放一些通道。
wiresharek截图：
releaseStream消息代码分析
客户端生成releaseStream消息发送给SRS服务端，代码：
static int gen_release_stream(URLContext *s, RTMPContext *rt)
{
RTMPPacket pkt;
uint8_t *p;
int ret;
if ((ret = ff_rtmp_packet_create(&pkt, RTMP_SYSTEM_CHANNEL, RTMP_PT_INVOKE,
0, 29 + strlen(rt->playpath))) < 0)
return ret;
av_log(s, AV_LOG_DEBUG, "Releasing stream...\n");
p = pkt.data;
ff_amf_write_string(&p, "releaseStream");
ff_amf_write_number(&p, ++rt->nb_invokes);
ff_amf_write_null(&p);
ff_amf_write_string(&p, rt->playpath);
return rtmp_send_packet(rt, &pkt, 1);
}SRS服务端接收到后会解析成对应的packet，然后返回_result消息。如下wiresharek所示：
SRS服务器解析releaseStream消息代码：
// SrsProtocol::do_decode_message函数下
} else if (command == RTMP_AMF0_COMMAND_RELEASE_STREAM) {
*ppacket = packet = new SrsFMLEStartPacket(); // 处理"releaseStream"消息，rtmp推流时是这个包
return packet->decode(stream);
}调用栈：是从SrsRtmpServer::identify_client函数进去的
Breakpoint 3, SrsProtocol::do_decode_message (this=0xac1280, header=..., stream=0xadb540, ppacket=0xadb530) at src/protocol/srs_rtmp_stack.cpp:712
712 *ppacket = packet = new SrsFMLEStartPacket();
(gdb) bt
#0 SrsProtocol::do_decode_message (this=0xac1280, header=..., stream=0xadb540, ppacket=0xadb530) at src/protocol/srs_rtmp_stack.cpp:712
#1 0x00000000004655ec in SrsProtocol::decode_message (this=0xac1280, msg=0xadce30, ppacket=0xadb5c8) at src/protocol/srs_rtmp_stack.cpp:430
#2 0x00000000004702bb in SrsRtmpServer::identify_client (this=0xac1250, stream_id=1, type=@0xacab08: SrsRtmpConnUnknown, stream_name="", duration=@0xacac88: -1) at src/protocol/srs_rtmp_stack.cpp:2521
#3 0x00000000004d519a in SrsRtmpConn::stream_service_cycle (this=0xac0f10) at src/app/srs_app_rtmp_conn.cpp:440
#4 0x00000000004d4ddf in SrsRtmpConn::service_cycle (this=0xac0f10) at src/app/srs_app_rtmp_conn.cpp:388
#5 0x00000000004d3ba7 in SrsRtmpConn::do_cycle (this=0xac0f10) at src/app/srs_app_rtmp_conn.cpp:209
#6 0x00000000004d1d99 in SrsConnection::cycle (this=0xac0f88) at src/app/srs_app_conn.cpp:171
#7 0x000000000050ab08 in SrsSTCoroutine::cycle (this=0xac11f0) at src/app/srs_app_st.cpp:198
#8 0x000000000050ab7d in SrsSTCoroutine::pfn (arg=0xac11f0) at src/app/srs_app_st.cpp:213
#9 0x00000000005bed1a in _st_thread_main () at sched.c:337
#10 0x00000000005bf492 in st_thread_create (start=0x5be696 <_st_vp_schedule+170>, arg=0x700000001, joinable=1, stk_size=1) at sched.c:616
Backtrace stopped: previous frame inner to this frame (corrupt stack?)根据packet类型返回对应响应，ReleaseStream/PublishStream/FCPublish/FCUnpublish都是用SrsRtmpServer::identify_fmle_publish_client函数返回_result消息。
srs_error_t SrsRtmpServer::identify_fmle_publish_client(SrsFMLEStartPacket* req, SrsRtmpConnType& type, string& stream_name)
{
srs_error_t err = srs_success;
type = SrsRtmpConnFMLEPublish;
stream_name = req->stream_name;
// releaseStream response
if (true) {
SrsFMLEStartResPacket* pkt = new SrsFMLEStartResPacket(req->transaction_id);
if ((err = protocol->send_and_free_packet(pkt, 0)) != srs_success) {
return srs_error_wrap(err, "send releaseStream response");
}
}
return err;
}客户端发送FCPublish消息给srs服务器
FCPublish消息作用是使服务器为接收媒体流做好准备。
wiresharek截图：
FCPublish消息代码解析
客户端生成FCPublish消息发送给SRS服务器：
static int gen_fcpublish_stream(URLContext *s, RTMPContext *rt)
{
RTMPPacket pkt;
uint8_t *p;
int ret;
if ((ret = ff_rtmp_packet_create(&pkt, RTMP_SYSTEM_CHANNEL, RTMP_PT_INVOKE,
0, 25 + strlen(rt->playpath))) < 0)
return ret;
av_log(s, AV_LOG_DEBUG, "FCPublish stream...\n");
p = pkt.data;
ff_amf_write_string(&p, "FCPublish");
ff_amf_write_number(&p, ++rt->nb_invokes);
ff_amf_write_null(&p);
ff_amf_write_string(&p, rt->playpath);
return rtmp_send_packet(rt, &pkt, 1);
}SRS服务器解析FCPublish消息：
} else if (command == RTMP_AMF0_COMMAND_FC_PUBLISH) {
*ppacket = packet = new SrsFMLEStartPacket(); //第四个接收到的message为"FCPublish"
return packet->decode(stream);
}调用栈：
FCPublish、createStream、publish、onFCPublish、onStatus都是通过SrsRtmpServer::start_fmle_publish函数进行处理：
Breakpoint 4, SrsProtocol::do_decode_message (this=0xac1280, header=..., stream=0xadb480, ppacket=0xadb470) at src/protocol/srs_rtmp_stack.cpp:715
715 *ppacket = packet = new SrsFMLEStartPacket();
(gdb) bt
#0 SrsProtocol::do_decode_message (this=0xac1280, header=..., stream=0xadb480, ppacket=0xadb470) at src/protocol/srs_rtmp_stack.cpp:715
#1 0x00000000004655ec in SrsProtocol::decode_message (this=0xac1280, msg=0xadeca0, ppacket=0xadb500) at src/protocol/srs_rtmp_stack.cpp:430
#2 0x000000000047fd40 in SrsProtocol::expect_message<SrsFMLEStartPacket> (this=0xac1280, pmsg=0xadb588, ppacket=0xadb590) at src/protocol/srs_rtmp_stack.hpp:335
#3 0x000000000047dfd9 in SrsRtmpServer::expect_message<SrsFMLEStartPacket> (this=0xac1250, pmsg=0xadb588, ppacket=0xadb590) at src/protocol/srs_rtmp_stack.hpp:776
#4 0x000000000047190c in SrsRtmpServer::start_fmle_publish (this=0xac1250, stream_id=1) at src/protocol/srs_rtmp_stack.cpp:2714
#5 0x00000000004d5e81 in SrsRtmpConn::stream_service_cycle (this=0xac0f10) at src/app/srs_app_rtmp_conn.cpp:530
#6 0x00000000004d4ddf in SrsRtmpConn::service_cycle (this=0xac0f10) at src/app/srs_app_rtmp_conn.cpp:388
#7 0x00000000004d3ba7 in SrsRtmpConn::do_cycle (this=0xac0f10) at src/app/srs_app_rtmp_conn.cpp:209
#8 0x00000000004d1d99 in SrsConnection::cycle (this=0xac0f88) at src/app/srs_app_conn.cpp:171
#9 0x000000000050ab08 in SrsSTCoroutine::cycle (this=0xac11f0) at src/app/srs_app_st.cpp:198
#10 0x000000000050ab7d in SrsSTCoroutine::pfn (arg=0xac11f0) at src/app/srs_app_st.cpp:213
#11 0x00000000005bed1a in _st_thread_main () at sched.c:337
#12 0x00000000005bf492 in st_thread_create (start=0x5be696 <_st_vp_schedule+170>, arg=0x700000001, joinable=1, stk_size=1) at sched.c:616
Backtrace stopped: previous frame inner to this frame (corrupt stack?)SRS服务器发送_result消息给客户端进行响应FCPublish消息，见如上代码：SrsRtmpServer::identify_fmle_publish_client。
客户端发送CreateStream消息给srs服务器
wiresharek截图：
Create Stream：创建传递具体信息的通道，从⽽可以在这个流中传递具体信息，传输信息单元为Chunk。
当发送完createStream消息之后，解析服务器返回的消息会得到⼀个stream ID，这个ID也就是以后和服务器通信的 message stream ID，⼀般返回的是1，不固定。
CreateStream消息代码解析
客户端生成CreateStream消息发送给SRS服务器：
static int gen_create_stream(URLContext *s, RTMPContext *rt)
{
RTMPPacket pkt;
uint8_t *p;
int ret;
av_log(s, AV_LOG_DEBUG, "Creating stream...\n");
if ((ret = ff_rtmp_packet_create(&pkt, RTMP_SYSTEM_CHANNEL, RTMP_PT_INVOKE,
0, 25)) < 0)
return ret;
p = pkt.data;
ff_amf_write_string(&p, "createStream");
ff_amf_write_number(&p, ++rt->nb_invokes);
ff_amf_write_null(&p);
return rtmp_send_packet(rt, &pkt, 1);
}SRS服务器解析CreateStream消息：
} else if (command == RTMP_AMF0_COMMAND_CREATE_STREAM) {
*ppacket = packet = new SrsCreateStreamPacket(); //第五个接收到的message为"createStream"消息
return packet->decode(stream);
}SRS服务器发送_result消息给客户端进行响应CreateStream消息，见如上代码：SrsRtmpServer::identify_fmle_publish_client。
客户端发送checkbw消息给srs服务器
客户端生成检查带宽消息发送给服务器。
wiresharek截图：
checkbw消息代码解析
客户端生成checkbw消息发送给SRS服务器：
static int gen_check_bw(URLContext *s, RTMPContext *rt)
{
RTMPPacket pkt;
uint8_t *p;
int ret;
if ((ret = ff_rtmp_packet_create(&pkt, RTMP_SYSTEM_CHANNEL, RTMP_PT_INVOKE,
0, 21)) < 0)
return ret;
p = pkt.data;
ff_amf_write_string(&p, "_checkbw");
ff_amf_write_number(&p, ++rt->nb_invokes);
ff_amf_write_null(&p);
return rtmp_send_packet(rt, &pkt, 1);
}客户端向SRS服务器发送publish消息以及SRS服务器向客户端发送onFCPublish消息和onStatue消息
publish消息解析
推流客户端使用publish消息向SRS服务器端发布一个命名的流，发布之后，任意客户端都可以通过该名称请求视频、音频和数据。
wiresharek截图：
publish消息代码解析
客户端生成publish消息并发送到SRS服务器：
static int gen_publish(URLContext *s, RTMPContext *rt)
{
RTMPPacket pkt;
uint8_t *p;
int ret;
av_log(s, AV_LOG_DEBUG, "Sending publish command for '%s'\n", rt->playpath);
if ((ret = ff_rtmp_packet_create(&pkt, RTMP_SOURCE_CHANNEL, RTMP_PT_INVOKE,
0, 30 + strlen(rt->playpath))) < 0)
return ret;
pkt.extra = rt->stream_id;
p = pkt.data;
ff_amf_write_string(&p, "publish");
ff_amf_write_number(&p, ++rt->nb_invokes);
ff_amf_write_null(&p);
ff_amf_write_string(&p, rt->playpath);
ff_amf_write_string(&p, "live");
return rtmp_send_packet(rt, &pkt, 1);
}SRS服务器解析publish消息：
} else if (command == RTMP_AMF0_COMMAND_PUBLISH) { //message为"publish"
*ppacket = packet = new SrsPublishPacket();
return packet->decode(stream);
}调用栈：
(gdb) bt
#0 SrsProtocol::do_decode_message (this=0xac1280, header=..., stream=0xadb480, ppacket=0xadb470) at src/protocol/srs_rtmp_stack.cpp:718
#1 0x00000000004655ec in SrsProtocol::decode_message (this=0xac1280, msg=0xaded60, ppacket=0xadb500) at src/protocol/srs_rtmp_stack.cpp:430
#2 0x00000000004801c0 in SrsProtocol::expect_message<SrsPublishPacket> (this=0xac1280, pmsg=0xadb588, ppacket=0xadb590) at src/protocol/srs_rtmp_stack.hpp:335
#3 0x000000000047e35b in SrsRtmpServer::expect_message<SrsPublishPacket> (this=0xac1250, pmsg=0xadb588, ppacket=0xadb590) at src/protocol/srs_rtmp_stack.hpp:776
#4 0x0000000000471c42 in SrsRtmpServer::start_fmle_publish (this=0xac1250, stream_id=1) at src/protocol/srs_rtmp_stack.cpp:2757
#5 0x00000000004d5e81 in SrsRtmpConn::stream_service_cycle (this=0xac0f10) at src/app/srs_app_rtmp_conn.cpp:530
#6 0x00000000004d4ddf in SrsRtmpConn::service_cycle (this=0xac0f10) at src/app/srs_app_rtmp_conn.cpp:388
#7 0x00000000004d3ba7 in SrsRtmpConn::do_cycle (this=0xac0f10) at src/app/srs_app_rtmp_conn.cpp:209
#8 0x00000000004d1d99 in SrsConnection::cycle (this=0xac0f88) at src/app/srs_app_conn.cpp:171
#9 0x000000000050ab08 in SrsSTCoroutine::cycle (this=0xac11f0) at src/app/srs_app_st.cpp:198
#10 0x000000000050ab7d in SrsSTCoroutine::pfn (arg=0xac11f0) at src/app/srs_app_st.cpp:213
#11 0x00000000005bed1a in _st_thread_main () at sched.c:337
#12 0x00000000005bf492 in st_thread_create (start=0x5be696 <_st_vp_schedule+170>, arg=0x700000001, joinable=1, stk_size=1) at sched.c:616
Backtrace stopped: previous frame inner to this frame (corrupt stack?)解析成功后SRS服务器会生成onFCPublish消息返回给客户端。
onFCPublish消息解析
wiresharek截图：
onFCPublish消息是回应publish消息。
onFCPublish消息代码解析
SRS服务器生成onFCPublish消息并发送到客户端：
// publish response onFCPublish(NetStream.Publish.Start)
if (true) { //客户端发送publish消息，服务端返回onFCPublish消息。
SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
pkt->command_name = RTMP_AMF0_COMMAND_ON_FC_PUBLISH;
pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodePublishStart));
pkt->data->set(StatusDescription, SrsAmf0Any::str("Started publishing stream."));
if ((err = protocol->send_and_free_packet(pkt, stream_id)) != srs_success) { //onFCPublish
return srs_error_wrap(err, "send NetStream.Publish.Start");
}
}onStatue消息解析
SRS服务器还会生成onStatus消息向客户端发送，描述的状态内容中code为NetStream.Publish.Start，description为Start publishing，目的就是告诉推流客户端，现在可以推流了。
onStatue消息代码解析
SRS服务器生成onStatue消息并发送到客户端：
// publish response onStatus(NetStream.Publish.Start)
if (true) { //服务器发送onStatus消息给客户端，wiresharek无法解析
SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
pkt->data->set(StatusLevel, SrsAmf0Any::str(StatusLevelStatus));
pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodePublishStart));
pkt->data->set(StatusDescription, SrsAmf0Any::str("Started publishing stream."));
pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID));
if ((err = protocol->send_and_free_packet(pkt, stream_id)) != srs_success) {
return srs_error_wrap(err, "send NetStream.Publish.Start");
}
}原文链接：https://www.yuque.com/wahaha-0yfyj/mnfloz/gm3skg
版权声明：本文内容转自互联网，本文观点仅代表作者本人。本站仅提供信息存储空间服务，所有权归原作者所有。如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至1393616908@qq.com 举报，一经查实，本站将立刻删除。
