标签：define opp dia port sample 握手 nstat event 网络文件
在握手结束后，即进入该函数中做进一步处理。
void ngx_rtmp_cycle(ngx_rtmp_session_t *s)
{
ngx_connection_t *c;
c = s->connection;
/* 重新设置读/写事件的回调函数 */
c->read->handler = ngx_rtmp_recv;
c->write->handler = ngx_rtmp_send;
/* 初始化该会话的 ping_evt 事件，当网络出现问题时，
* 就会尝试调用该事件的回调函数 ngx_rtmp_ping */
s->ping_evt.data = c;
s->ping_evt.log = c->log;
s->ping_evt.handler = ngx_rtmp_ping;
/* 将 ping_evt 添加到定时器中 */
ngx_rtmp_reset_ping(s);
/* 开始接收客户端发来的rtmp数据，若接收不到客户端的数据时，则将读事件添加
* 到epoll中，并设置回调函数为ngx_rtmp_recv方法，当再次监听到客户端发来
* 数据时会再次调用该方法进行处理 */
ngx_rtmp_recv(c->read);
}void ngx_rtmp_reset_ping(ngx_rtmp_session_t *s)
{
ngx_rtmp_core_srv_conf_t *cscf;
/* 获取该 server{} 下 ngx_rtmp_core_module 模块的 srv 级别的配置结构体 */
cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module);
if (cscf->ping == 0) {
return;
}
s->ping_active = 0;
s->ping_reset = 0;
/* 将 ping_evt 添加到定时器中，定时时间为 cscf->ping，
* 若配置文件中没有配置有，则默认时间为 60000 ms */
ngx_add_timer(&s->ping_evt, cscf->ping);
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"ping: wait %Mms", cscf->ping);
}static void ngx_rtmp_recv(ngx_event_t *rev)
{
ngx_int_t n;
ngx_connection_t *c;
ngx_rtmp_session_t *s;
ngx_rtmp_core_srv_conf_t *cscf;
ngx_rtmp_header_t *h;
ngx_rtmp_stream_t *st, *st0;
ngx_chain_t *in, *head;
ngx_buf_t *b;
u_char *p, *pp, *old_pos;
size_t size, fsize, old_size;
uint8_t fmt, ext;
uint32_t csid, timestamp;
c = rev->data;
s = c->data;
b = NULL;
old_pos = NULL;
old_size = 0;
cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module);
/* 判断该连接是否已被销毁 */
if (c->destroyed) {
return;
}
for( ;; ) {
st = &s->in_streams[s->in_csid];
/* allocate new buffer */
if (st->in == NULL) {
/* 为类型为 ngx_chain_t 的结构体指针 st->in 分配内存 */
st->in = ngx_rtmp_alloc_in_buf(s);
if (st->in == NULL) {
ngx_log_error(NGX_LOG_INFO, c->log, 0,
"in buf alloc failed");
ngx_rtmp_finalize_session(s);
return;
}
}
h = &st->hdr;
in = st->in;
b = in->buf;
if (old_size) {
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, c->log, 0,
"reusing formerly read data: %d", old_size);
/* 对上一次接收到的数据按一个 chunk size 进行切分之后，
* 余下的多余的数据再次循环时进行重新使用 */
b->pos = b->start;
b->last = ngx_movemem(b->pos, old_pos, old_size);
if (s->in_chunk_size_changing) {
ngx_rtmp_finalize_set_chunk_size(s);
}
} else {
if (old_pos) {
b->pos = b->last = b->start;
}
/* 这里调用的回调函数为 ngx_unix_recv 方法接收数据 */
n = c->recv(c, b->last, b->end - b->last);
if (n == NGX_ERROR || n == 0) {
ngx_rtmp_finalize_session(s);
return;
}
if (n == NGX_AGAIN) {
if (ngx_handle_read_event(c->read, 0) != NGX_OK) {
ngx_rtmp_finalize_session(s);
}
return;
}
s->ping_reset = 1;
ngx_rtmp_update_bandwidth(&ngx_rtmp_bw_in, n);
/* 更新缓存区的数据指针 */
b->last += n;
s->in_bytes += n;
/* 当缓存区中数据满溢时 */
if (s->in_bytes >= 0xf0000000) {
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, c->log, 0,
"resetting byte counter");
s->in_bytes = 0;
s->in_last_ack = 0;
}
if (s->ack_size && s->in_bytes - s->in_last_ack >= s->ack_size) {
s->in_last_ack = s->in_bytes;
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, c->log, 0,
"sending RTMP ACK(%uD)", s->in_bytes);
if (ngx_rtmp_send_ack(s, s->in_bytes)) {
ngx_rtmp_finalize_session(s);
return;
}
}
}
old_pos = NULL;
old_size = 0;
/* parse headers */
if (b->pos == b->start) {
/* 最初的时候: b->pos == b->start == b->last */
/**
* Chunk format:
* +----------------+------------------+--------------------+-------------+
* | Basic Header | Message Header | Extended Timestamp | Chunk Data |
* +----------------+------------------+--------------------+-------------+
*/
p = b->pos;
/* Basic Header:基本头信息
*
* 包含chunk stream ID（流通道id）和chunk type（即fmt），chunk stream id一般被
* 简写为CSID，用来唯一标识一个特定的流通道，chunk type决定了后面Message Header的格式。
* Basic Header的长度可能是1，2，或3个字节，其中chunk type的长度是固定的（占2位，单位
* 是bit），Basic Header的长度取决于CSID的大小，在足够存储这两个字段的前提下最好用尽量
* 少的字节从而减少由于引入Header增加的数据量。
*
* RTMP协议支持用户自定义[3,65599]之间的CSID，0,1,2由协议保留表示特殊信息。0代表Basic
* Header总共要占用2个字节，CSID在[64,319]之间; 1代表占用3个字节，CSID在[64,65599]之
* 间; 2 代表该 chunk 是控制信息和一些命令信息。
*
* chunk type的长度是固定2 bit，因此CSID的长度是(6=8-2)、(14=16-2)、(22=24-2)中的一
* 个。当Basic Header为1个字节时，CSID占6bit，6bit最多可以表示64个数，因此在这种情况
* 下CSID在[0,63]之间，其中用户可自定义的范围为[3,63]。
*
* Basic Header：1 byte
* 0 1 2 3 4 5 6 7
* +-+-+-+-+-+-+-+-+
* |fmt| cs id |
* +-+-+-+-+-+-+-+-+
*
* Basic Header: 2 byte , csid == 0
* CSID占14bit，此时协议将于chunk type所在字节的其他bit都置为0，
* 剩下的一个字节表示CSID - 64，这样共有8个bit来存储CSID，8bit可以表示[0,255]个数，因此
* 这种情况下CSID在[64,319]，其中319=255+64。
* 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* |fmt| 0 | cs id - 64 |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
*
* Basic Header: 3 bytes , csid == 1
* CSID占22bit，此时协议将第一个字节的[2,8]bit置1，余下的16个bit表示CSID - 64，这样共有
* 16个bit来存储CSID，16bit可以表示[0,65535]共 65536 个数，因此这种情况下 CSID 在
* [64,65599]，其中65599=65535+64，需要注意的是，Basic Header是采用小端存储的方式，越往
* 后的字节数量级越高，因此通过3个字节的每一个bit的值来计算CSID时，应该是:
* <第三个字节的值> * 256 + <第二个字节的值> + 64.
* 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* |fmt| 1 | cs id - 64 |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
*/
/* chunk basic header */
fmt = (*p >> 6) & 0x03;
csid = *p++ & 0x3f;
/* csid 为 0 表示 Basic Header 占 2 bytes */
if (csid == 0) {
if (b->last - p < 1)
continue;
csid = 64;
csid += *(uint8_t*)p++;
}
/* csid 为 1 表示 Basic Header 占 3 bytes */
else if (csid == 1) {
if (b->last - p < 2)
continue;
csid = 64;
csid += *(uint8_t*)p++;
csid += (uint32_t)256 * (*(uint8_t*)p++);
}
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, c->log, 0,
"RTMP bheader fmt=%d csid=%D",
(int)fmt, csid);
if (csid >= (uint32_t)cscf->max_streams) {
ngx_log_error(NGX_LOG_INFO, c->log, 0,
"RTMP in chunk stream too big: %D >= %D",
csid, cscf->max_streams);
ngx_rtmp_finalize_session(s);
return;
}
/* link orphan */
if (s->in_csid == 0) {
/* unlink from stream #0 */
st->in = st->in->next;
/* link to new stream */
s->in_csid = csid;
st = &s->in_streams[csid];
if (st->in == NULL) {
in->next = in;
} else {
in->next = st->in->next;
st->in->next = in;
}
st->in = in;
h = &st->hdr;
h->csid = csid;
}
/* Message Header:
*
* 包含了要发送的实际信息（可能是完整的，也可能是一部分）的描述信息。Message Header的格式
* 和长度取决于Basic Header的chunk type，即fmt，共有四种不同的格式。其中第一种格式可以表
* 示其他三种表示的所有数据，但由于其他三种格式是基于对之前chunk的差量化的表示，因此可以
* 更简洁地表示相同的数据，实际使用的时候还是应该采用尽量少的字节表示相同意义的数据。下面
* 按字节从多到少的顺序分别介绍这四种格式的 Message Header。
*
* 下面是 Message Header 四种消息头格式。
*
* 一、Chunk Type(fmt) = 0：11 bytes
* 0 1 2 3
* 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* | timestamp |message length |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* | message length (coutinue) |message type id| msg stream id |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* | msg stream id |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* type=0时Message Header占用11个字节，其他三种能表示的数据它都能表示，但在chunk stream
* 的开始第一个chunk和头信息中的时间戳后退（即值与上一个chunk相比减小，通常在回退播放的
* 时候会出现这种情况）的时候必须采用这种格式。
* - timestamp（时间戳）：占用3个字节，因此它最多能表示到16777215=0xFFFFFF=2^24-1，当它
* 的值超过这个最大值时，这三个字节都置为1，这样实际的timestamp会转存到 Extended
* Timestamp 字段中，接收端在判断timestamp字段24个位都为1时就会去Extended Timestamp
* 中解析实际的时间戳。
* - message length（消息数据长度）：占用3个字节，表示实际发送的消息的数据如音频帧、视频
* 帧等数据的长度，单位是字节。注意这里是Message的长度，也就是chunk属于的Message的总长
* 度，而不是chunk本身data的长度。
* - message type id(消息的类型id)：1个字节，表示实际发送的数据的类型，如8代表音频数据，
* 9代表视频数据。
* - message stream id(消息的流id)：4个字节，表示该chunk所在的流的ID，和Basic Header
* 的CSID一样，它采用小端存储方式。
*
* 二、Chunk Type(fmt) = 1：7 bytes
* 0 1 2 3
* 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* | timestamp delta |message length |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* | message length (coutinue) |message type id|
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* type为1时占用7个字节，省去了表示message stream id的4个字节，表示此chunk和上一次发的
* chunk所在的流相同，如果在发送端和对端有一个流链接的时候可以尽量采取这种格式。
* - timestamp delta：3 bytes，这里和type=0时不同，存储的是和上一个chunk的时间差。类似
* 上面提到的timestamp，当它的值超过3个字节所能表示的最大值时，三个字节都置为1，实际
* 的时间戳差值就会转存到Extended Timestamp字段中，接收端在判断timestamp delta字段24
* 个bit都为1时就会去Extended Timestamp 中解析实际的与上次时间戳的差值。
* - 其他字段与上面的解释相同.
*
* 三、Chunk Type(fmt) = 2：3 bytes
* 0 1 2
* 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* | timestamp delta |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* type为2时占用3个字节，相对于type=1格式又省去了表示消息长度的3个字节和表示消息
* 类型的1个字节，表示此chunk和上一次发送的chunk所在的流、消息的长度和消息的类型都相同。
* 余下的这三个字节表示timestamp delta，使用同type=1。
*
*
* 四、Chunk Type(fmt) = 3: 0 byte
* type=3时，为0字节，表示这个chunk的Message Header和上一个是完全相同的。当它跟在type=0
* 的chunk后面时，表示和前一个chunk的时间戳都是相同。什么时候连时间戳都是相同呢？就是
* 一个Message拆分成多个chunk，这个chunk和上一个chunk同属于一个Message。而当它跟在
* type = 1或 type = 2 的chunk后面时的chunk后面时，表示和前一个chunk的时间戳的差是相同
* 的。比如第一个 chunk 的 type = 0，timestamp = 100，第二个 chunk 的 type = 2，
* timestamp delta = 20，表示时间戳为 100 + 20 = 120，第三个 chunk 的 type = 3，
* 表示 timestamp delta = 20 时间戳为 120 + 20 = 140。
*/
ext = st->ext;
timestamp = st->dtime;
if (fmt <= 2 ) {
if (b->last - p < 3)
continue;
/* timestamp:
* big-endian 3b -> little-endian 4b */
pp = (u_char*)×tamp;
pp[2] = *p++;
pp[1] = *p++;
pp[0] = *p++;
pp[3] = 0;
ext = (timestamp == 0x00ffffff);
if (fmt <= 1) {
if (b->last - p < 4)
continue;
/* size:
* big-endian 3b -> little-endian 4b
* type:
* 1b -> 1b*/
pp = (u_char*)&h->mlen;
pp[2] = *p++;
pp[1] = *p++;
pp[0] = *p++;
pp[3] = 0;
/* message type id */
h->type = *(uint8_t*)p++;
if (fmt == 0) {
if (b->last - p < 4)
continue;
/* stream:
* little-endian 4b -> little-endian 4b */
pp = (u_char*)&h->msid;
pp[0] = *p++;
pp[1] = *p++;
pp[2] = *p++;
pp[3] = *p++;
}
}
}
/*
* Extended Timestamp(扩展时间戳)：
* 在 chunk 中会有时间戳 timestamp 和时间戳差 timestamp delta，
* 并且它们不会同时存在，只有这两者之一大于3字节能表示的最大数值
* 0xFFFFFF＝16777215时，才会用这个字段来表示真正的时间戳，否则
* 这个字段为 0。扩展时间戳占 4 个字节，能表示的最大数值就是
* 0xFFFFFFFF＝4294967295。当扩展时间戳启用时，timestamp字段或者
* timestamp delta要全置为1，而不是减去时间戳或者时间戳差的值。
*/
/* extended header */
if (ext) {
if (b->last - p < 4)
continue;
pp = (u_char*)×tamp;
pp[3] = *p++;
pp[2] = *p++;
pp[1] = *p++;
pp[0] = *p++;
}
if (st->len == 0) {
/* Messages with type=3 should
* never have ext timestamp field
* according to standard.
* However that‘s not always the case
* in real life */
st->ext = (ext && cscf->publish_time_fix);
if (fmt) {
st->dtime = timestamp;
} else {
h->timestamp = timestamp;
st->dtime = 0;
}
}
ngx_log_debug8(NGX_LOG_DEBUG_RTMP, c->log, 0,
"RTMP mheader fmt=%d %s (%d) "
"time=%uD+%uD mlen=%D len=%D msid=%D",
(int)fmt, ngx_rtmp_message_type(h->type), (int)h->type,
h->timestamp, st->dtime, h->mlen, st->len, h->msid);
/* header done */
/* 更新缓存区pos指针的位置，更新后pos应指向rtmp trunk的实际数据 */
b->pos = p;
if (h->mlen > cscf->max_message) {
ngx_log_error(NGX_LOG_INFO, c->log, 0,
"too big message: %uz", cscf->max_message);
ngx_rtmp_finalize_session(s);
return;
}
}
/* b->last：指向本次recv到的数据的尾部
* b->pos: 指向rtmp trunk的实际data起始
* 因此，size为本次接收到的实际数据的大小
*/
size = b->last - b->pos;
/* h->mlen：该RTMP message的长度
* st->len：该流的长度
*/
fsize = h->mlen - st->len;
if (size < ngx_min(fsize, s->in_chunk_size))
continue;
/* buffer is ready */
/* 本次所要接收的 rtmp message 的大小大于 s->in_shunk_size 块的大小
* 因此要进行切分 */
if (fsize > s->in_chunk_size) {
/* collect fragmented chunks */
st->len += s->in_chunk_size;
b->last = b->pos + s->in_chunk_size;
old_pos = b->last;
/* 切分一个块后余下的数据大小 */
old_size = size - s->in_chunk_size;
} else {
/* 完整接收一个 rtmp message 后，进行处理 */
/* handle! */
head = st->in->next;
st->in->next = NULL;
b->last = b->pos + fsize;
old_pos = b->last;
old_size = size - fsize;
st->len = 0;
h->timestamp += st->dtime;
/* 根据该 rtmp 消息的类型调用相应的回调函数进行处理 */
if (ngx_rtmp_receive_message(s, h, head) != NGX_OK) {
ngx_rtmp_finalize_session(s);
return;
}
if (s->in_chunk_size_changing) {
/* copy old data to a new buffer */
if (!old_size) {
ngx_rtmp_finalize_set_chunk_size(s);
}
} else {
/* add used bufs to stream #0 */
st0 = &s->in_streams[0];
st->in->next = st0->in;
st0->in = head;
st->in = NULL;
}
}
s->in_csid = 0;
}
}ngx_int_t ngx_rtmp_receive_message(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in)
{
ngx_rtmp_core_main_conf_t *cmcf;
ngx_array_t *evhs;
size_t n;
ngx_rtmp_handler_pt *evh;
cmcf = ngx_rtmp_get_module_main_conf(s, ngx_rtmp_core_module);
#ifdef NGX_DEBUG
{
int nbufs;
ngx_chain_t *ch;
for(nbufs = 1, ch = in;
ch->next;
ch = ch->next, ++nbufs);
ngx_log_debug7(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"RTMP recv %s (%d) csid=%D timestamp=%D "
"mlen=%D msid=%D nbufs=%d",
ngx_rtmp_message_type(h->type), (int)h->type,
h->csid, h->timestamp, h->mlen, h->msid, nbufs);
}
#endif
if (h->type > NGX_RTMP_MSG_MAX) {
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"unexpected RTMP message type: %d", (int)h->type);
return NGX_OK;
}
/* 根据 message type 取出事件 */
evhs = &cmcf->events[h->type];
evh = evhs->elts;
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"nhandlers: %d", evhs->nelts);
for(n = 0; n < evhs->nelts; ++n, ++evh) {
if (!evh) {
continue;
}
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"calling handler %d", n);
/* 若接收到的是 amf 类型的事件，则对应的处理函数为 ngx_rtmp_amf_message_handler；
* 若接收到的是 standard protocol 类型事件，则对应的处理函数为
* ngx_rtmp_protocol_message_handler；
* 若接收到的是 user protocol 事件，则对应的处理函数为 ngx_rtmp_user_message_handler；
*/
switch ((*evh)(s, h, in)) {
case NGX_ERROR:
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"handler %d failed", n);
return NGX_ERROR;
case NGX_DONE:
return NGX_OK;
}
}
return NGX_OK;
}
如上图，接收到客户端发来的第一个 RTMP message 的类型为 20，即 NGX_RTMP_MSG_AMF_CMD，因此将会调用之前由
ngx_rtmp_init_event_handlers() 方法对该类型设置的回调方法 ngx_rtmp_amf_message_handler。并且，由上图可
知，客户端发送的是 connect 连接，因此，在 ngx_rtmp_amf_message_handler 主要做的就是：
ngx_int_t ngx_rtmp_amf_message_handler(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in)
{
ngx_rtmp_amf_ctx_t act;
ngx_rtmp_core_main_conf_t *cmcf;
ngx_array_t *ch;
ngx_rtmp_handler_pt *ph;
size_t len, n;
static u_char func[128];
static ngx_rtmp_amf_elt_t elts[] = {
{ NGX_RTMP_AMF_STRING,
ngx_null_string,
func, sizeof(func) },
};
/* AMF command names come with string type, but shared object names
* come without type */
if (h->type == NGX_RTMP_MSG_AMF_SHARED ||
h->type == NGX_RTMP_MSG_AMF3_SHARED)
{
elts[0].type |= NGX_RTMP_AMF_TYPELESS;
} else {
elts[0].type &= ~NGX_RTMP_AMF_TYPELESS;
}
if ((h->type == NGX_RTMP_MSG_AMF3_SHARED ||
h->type == NGX_RTMP_MSG_AMF3_META ||
h->type == NGX_RTMP_MSG_AMF3_CMD)
&& in->buf->last > in->buf->pos)
{
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"AMF3 prefix: %ui", (ngx_int_t)*in->buf->pos);
++in->buf->pos;
}
cmcf = ngx_rtmp_get_module_main_conf(s, ngx_rtmp_core_module);
/* read AMF func name & transaction id */
ngx_memzero(&act, sizeof(act));
act.link = in;
act.log = s->connection->log;
memset(func, 0, sizeof(func));
/* 读取接收到的 amf 数据，获取 elts 中指定类型的 amf 数据，并保存在 elts->data 中 */
if (ngx_rtmp_amf_read(&act, elts,
sizeof(elts) / sizeof(elts[0])) != NGX_OK)
{
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"AMF cmd failed");
return NGX_ERROR;
}
/* skip name */
in = act.link;
/* 跳过已经读取的数据 */
in->buf->pos += act.offset;
len = ngx_strlen(func);
/* 根据读取到的 func，在 amf_hash 指向的 hash 表中查找，若能找到，
* 则返回对应的 ngx_array_t * 指针数组，该数组中保存着该 func 中
* 保存的内容的相关处理函数，这些处理函数是各 RTMP 模块在
* postconfiguration 方法中保存到 cmcf->amf (ngx_array_t)数组中，
* 然后在 ngx_rtmp_init_event_handlers 方法中存到 amf_hash 哈希表中，
* 以便快速查找 */
ch = ngx_hash_find(&cmcf->amf_hash,
ngx_hash_strlow(func, func, len), func, len);
/* 若 ch 不为 NULL 且 ch->nelts 大于 0，表明有该类 ch 的处理函数 */
if (ch && ch->nelts) {
ph = ch->elts;
for (n = 0; n < ch->nelts; ++n, ++ph) {
ngx_log_debug3(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"AMF func ‘%s‘ passed to handler %d/%d",
func, n, ch->nelts);
switch ((*ph)(s, h, in)) {
case NGX_ERROR:
return NGX_ERROR;
case NGX_DONE:
return NGX_OK;
}
}
} else {
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"AMF cmd ‘%s‘ no handler", func);
}
return NGX_OK;
}在 ngx_rtmp_cmd_module.c 文件，即 ngx_rtmp_cmd_module 中，可以找到如下：
static ngx_rtmp_amf_handler_t ngx_rtmp_cmd_map[] = {
{ ngx_string("connect"), ngx_rtmp_cmd_connect_init },
{ ngx_string("createStream"), ngx_rtmp_cmd_create_stream_init },
{ ngx_string("closeStream"), ngx_rtmp_cmd_close_stream_init },
{ ngx_string("deleteStream"), ngx_rtmp_cmd_delete_stream_init },
{ ngx_string("publish"), ngx_rtmp_cmd_publish_init },
{ ngx_string("play"), ngx_rtmp_cmd_play_init },
{ ngx_string("play2"), ngx_rtmp_cmd_play2_init },
{ ngx_string("seek"), ngx_rtmp_cmd_seek_init },
{ ngx_string("pause"), ngx_rtmp_cmd_pause_init },
{ ngx_string("pauseraw"), ngx_rtmp_cmd_pause_init },
};上面即 ngx_rtmp_cmd_module 模块在接收到左边的命令，如 connect 时，就会调用右边的处理函数。因此，下面将
会调用 ngx_rtmp_cmd_connect_init 方法。
static ngx_int_t ngx_rtmp_cmd_connect_init(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
size_t len;
static ngx_rtmp_connect_t v;
/* 从这里可以看出，对比图1，这里是指定要获取的 amf 类型的值 */
static ngx_rtmp_amf_elt_t in_cmd[] = {
{ NGX_RTMP_AMF_STRING,
ngx_string("app"),
v.app, sizeof(v.app) },
{ NGX_RTMP_AMF_STRING,
ngx_string("flashVer"),
v.flashver, sizeof(v.flashver) },
{ NGX_RTMP_AMF_STRING,
ngx_string("swfUrl"),
v.swf_url, sizeof(v.swf_url) },
{ NGX_RTMP_AMF_STRING,
ngx_string("tcUrl"),
v.tc_url, sizeof(v.tc_url) },
{ NGX_RTMP_AMF_NUMBER,
ngx_string("audioCodecs"),
&v.acodecs, sizeof(v.acodecs) },
{ NGX_RTMP_AMF_NUMBER,
ngx_string("videoCodecs"),
&v.vcodecs, sizeof(v.vcodecs) },
{ NGX_RTMP_AMF_STRING,
ngx_string("pageUrl"),
v.page_url, sizeof(v.page_url) },
{ NGX_RTMP_AMF_NUMBER,
ngx_string("objectEncoding"),
&v.object_encoding, 0},
};
static ngx_rtmp_amf_elt_t in_elts[] = {
{ NGX_RTMP_AMF_NUMBER,
ngx_null_string,
&v.trans, 0 },
{ NGX_RTMP_AMF_OBJECT,
ngx_null_string,
in_cmd, sizeof(in_cmd) },
};
ngx_memzero(&v, sizeof(v));
/* 从 in 中提取 in_elts 中指定的数据，并保存在 in_elts 中的相应位置 */
if (ngx_rtmp_receive_amf(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
len = ngx_strlen(v.app);
if (len > 10 && !ngx_memcmp(v.app + len - 10, "/_definst_", 10)) {
v.app[len - 10] = 0;
} else if (len && v.app[len - 1] == ‘/‘) {
v.app[len - 1] = 0;
}
ngx_rtmp_cmd_fill_args(v.app, v.args);
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
"connect: app=‘%s‘ args=‘%s‘ flashver=‘%s‘ swf_url=‘%s‘ "
"tc_url=‘%s‘ page_url=‘%s‘ acodecs=%uD vcodecs=%uD "
"object_encoding=%ui",
v.app, v.args, v.flashver, v.swf_url, v.tc_url, v.page_url,
(uint32_t)v.acodecs, (uint32_t)v.vcodecs,
(ngx_int_t)v.object_encoding);
/* ngx_rtmp_connect 和 各个 RTMP 模块的 next_connect 构成了一个单链表，
* 该链表的元素都是各 RTMP 模块想要在 connect 阶段进行的操作，初始化该链表
* 是在 postconfiguration 方法中进行的 */
return ngx_rtmp_connect(s, &v);
}对于 connect，主要有以下的 RTMP 模块添加了相应的操作：(下面是根据调用顺序排列的)
ngx_int_t ngx_rtmp_receive_amf(ngx_rtmp_session_t *s, ngx_chain_t *in,
ngx_rtmp_amf_elt_t *elts, size_t nelts)
{
ngx_rtmp_amf_ctx_t act;
/* 构建一个 amf 的上下文结构体 */
ngx_memzero(&act, sizeof(act));
act.link = in;
act.log = s->connection->log;
/* 从 amf数据 中提取所需的数据 */
return ngx_rtmp_amf_read(&act, elts, nelts);
}static ngx_int_t ngx_rtmp_notify_connect(ngx_rtmp_session_t *s, ngx_rtmp_connect_t *v)
{
ngx_rtmp_notify_srv_conf_t *nscf;
ngx_rtmp_netcall_init_t ci;
ngx_url_t *url;
/* 检测是否置位了 auto_pushed 或者 relay，由于是点播，因此该模块相当于什么也没有做 */
if (s->auto_pushed || s->relay) {
goto next;
}
/* 暂不分析 */
...
next:
return next_connect(s, v);
}static ngx_int_t ngx_rtmp_cmd_connect(ngx_rtmp_session_t *s, ngx_rtmp_connect_t *v)
{
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
"rtmp cmd: connect");
ngx_rtmp_core_srv_conf_t *cscf;
ngx_rtmp_core_app_conf_t **cacfp;
ngx_uint_t n;
ngx_rtmp_header_t h;
u_char *p;
static double trans;
static double capabilities = NGX_RTMP_CAPABILITIES;
static double object_encoding = 0;
static ngx_rtmp_amf_elt_t out_obj[] = {
{ NGX_RTMP_AMF_STRING,
ngx_string("fmsVer"),
NGX_RTMP_FMS_VERSION, 0 },
{ NGX_RTMP_AMF_NUMBER,
ngx_string("capabilities"),
&capabilities, 0 },
};
static ngx_rtmp_amf_elt_t out_inf[] = {
{ NGX_RTMP_AMF_STRING,
ngx_string("level"),
"status", 0 },
{ NGX_RTMP_AMF_STRING,
ngx_string("code"),
"NetConnection.Connect.Success", 0 },
{ NGX_RTMP_AMF_STRING,
ngx_string("description"),
"Connection succeeded.", 0 },
{ NGX_RTMP_AMF_NUMBER,
ngx_string("objectEncoding"),
&object_encoding, 0 }
};
/* 从内容可以看出，这里是针对客户端发来的 connect，服务器将要发给客户端的响应 */
static ngx_rtmp_amf_elt_t out_elts[] = {
{ NGX_RTMP_AMF_STRING,
ngx_null_string,
"_result", 0 },
{ NGX_RTMP_AMF_NUMBER,
ngx_null_string,
&trans, 0 },
{ NGX_RTMP_AMF_OBJECT,
ngx_null_string,
out_obj, sizeof(out_obj) },
{ NGX_RTMP_AMF_OBJECT,
ngx_null_string,
out_inf, sizeof(out_inf) },
};
/* 检测当前会话是否已经连接了，若是，则表明重复接收到了客户端的 connect，
* 因此返回 ERROR */
if (s->connected) {
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
"connect: duplicate connection");
return NGX_ERROR;
}
/* 获取该 server{} 下的配置结构体 */
cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module);
trans = v->trans;
/* fill session parameters */
s->connected = 1;
/* 初始化 RTMP 头 */
ngx_memzero(&h, sizeof(h));
h.csid = NGX_RTMP_CSID_AMF_INI;
h.type = NGX_RTMP_MSG_AMF_CMD;
#define NGX_RTMP_SET_STRPAR(name) s->name.len = ngx_strlen(v->name); s->name.data = ngx_palloc(s->connection->pool, s->name.len); ngx_memcpy(s->name.data, v->name, s->name.len)
/* 将这些参数设置到 ngx_rtmp_session_t 会话结构体 s 中 */
NGX_RTMP_SET_STRPAR(app);
NGX_RTMP_SET_STRPAR(args);
NGX_RTMP_SET_STRPAR(flashver);
NGX_RTMP_SET_STRPAR(swf_url);
NGX_RTMP_SET_STRPAR(tc_url);
NGX_RTMP_SET_STRPAR(page_url);
#undef NGX_RTMP_SET_STRPAR
p = ngx_strlchr(s->app.data, s->app.data + s->app.len, ‘?‘);
if (p) {
s->app.len = (p - s->app.data);
}
s->acodecs = (uint32_t) v->acodecs;
s->vcodecs = (uint32_t) v->vcodecs;
/* 遍历当前 server{} 下的所有 application{}，找到与客户端发来的 connect 命令中
* 要求连接的 app，然后把该 app 对应的 application{} 下所属的 ngx_rtmp_conf_ctx_t
* 的 app_conf 指针数组赋给当前会话 s->app_conf */
/* find application & set app_conf */
cacfp = cscf->applications.elts;
for(n = 0; n < cscf->applications.nelts; ++n, ++cacfp) {
if ((*cacfp)->name.len == s->app.len &&
ngx_strncmp((*cacfp)->name.data, s->app.data, s->app.len) == 0)
{
/* found app! */
s->app_conf = (*cacfp)->app_conf;
break;
}
}
/* 若是没有找到，则表示没有客户端所要请求的 application，因此返回 ERROR */
if (s->app_conf == NULL) {
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
"connect: application not found: ‘%V‘", &s->app);
return NGX_ERROR;
}
object_encoding = v->object_encoding;
/* 这里依次发送 ack_size、bandwidth、chunk_size、amf 给客户端 */
return ngx_rtmp_send_ack_size(s, cscf->ack_window) != NGX_OK ||
ngx_rtmp_send_bandwidth(s, cscf->ack_window,
NGX_RTMP_LIMIT_DYNAMIC) != NGX_OK ||
ngx_rtmp_send_chunk_size(s, cscf->chunk_size) != NGX_OK ||
ngx_rtmp_send_amf(s, &h, out_elts,
sizeof(out_elts) / sizeof(out_elts[0]))
!= NGX_OK ? NGX_ERROR : NGX_OK;
}
rtmp 服务端发送完上面的几个 rtmp 包后，若客户端没有立刻发送 rtmp 包，则将读事件添加到 epoll 中，
然后监听客户端发来的第二个 rtmp 包。
其实，rtmp 服务器在接收到 "createStream" 之前，还接收到了客户端一个 NGX_RTMP_MSG_ACK_SIZE(5) 的包,
接收到该包后，仅是调用 ngx_rtmp_protocol_message_handler 方法将包中的实际数据取出来，赋给
s->ack_size，如下：
ngx_int_t ngx_rtmp_protocol_message_handler(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in)
{
ngx_buf_t *b;
u_char *p;
uint32_t val;
uint8_t limit;
b = in->buf;
...
p = (u_char *)&val;
p[0] = b->pos[3];
p[1] = b->pos[2];
p[2] = b->pos[1];
p[3] = b->pos[0];
switch (h->type)
{
...
case NGX_RTMP_MSG_ACK_SIZE:
/* receive window size =val */
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"receive ack_size=%uD", val);
s->ack_size = val;
break;
...
}
return NGX_OK;
}
接收到客户端发来的该 "createStream" amf 命令后，与上面分析的 "connect" 命令类似，调用到
ngx_rtmp_amf_message_handler 方法，然后在 amf_hash 中查找是否有 "createStream" 命令的回调方法，
这里找到的回调方法为 ngx_rtmp_cmd_create_stream_init。
static ngx_int_t
ngx_rtmp_cmd_create_stream_init(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
static ngx_rtmp_create_stream_t v;
static ngx_rtmp_amf_elt_t in_elts[] = {
{ NGX_RTMP_AMF_NUMBER,
ngx_null_string,
&v.trans, sizeof(v.trans) },
};
/* 从 amf 数据中读取 in_elts 中指定的数据 */
if (ngx_rtmp_receive_amf(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0, "createStream");
/* 同理，这里开始调用 ngx_rtmp_create_stream 构建的链表中的方法 */
return ngx_rtmp_create_stream(s, &v);
}在 nginx-rtmp 源码中，可知，仅有 ngx_rtmp_cmd_module 模块设置了该命令的方法。
static ngx_int_t
ngx_rtmp_cmd_create_stream(ngx_rtmp_session_t *s, ngx_rtmp_create_stream_t *v)
{
/* support one message stream per connection */
static double stream;
static double trans;
ngx_rtmp_header_t h;
static ngx_rtmp_amf_elt_t out_elts[] = {
{ NGX_RTMP_AMF_STRING,
ngx_null_string,
"_result", 0 },
{ NGX_RTMP_AMF_NUMBER,
ngx_null_string,
&trans, 0 },
{ NGX_RTMP_AMF_NULL,
ngx_null_string,
NULL, 0 },
{ NGX_RTMP_AMF_NUMBER,
ngx_null_string,
&stream, sizeof(stream) },
};
trans = v->trans;
stream = NGX_RTMP_MSID;
ngx_memzero(&h, sizeof(h));
h.csid = NGX_RTMP_CSID_AMF_INI;
h.type = NGX_RTMP_MSG_AMF_CMD;
/* 发送该 amf 数据 */
return ngx_rtmp_send_amf(s, &h, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) == NGX_OK ?
NGX_DONE : NGX_ERROR;
}
然后，服务器又再次等待客户端发来的 rtmp 包。此次，接收 client 发来的到 rtmp 包为 类型为 amf_cmd(20)
的 "play" 命令.
从接收到的这个 rtmp 包可以看出其实是一个 rtmp message 中包含两个 chunk，第一个 chunk 为 amf_cmd(20)，
这里为 "play"，第二个 chunk 为 user control message(4)。
static ngx_int_t
ngx_rtmp_cmd_play_init(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
static ngx_rtmp_play_t v;
static ngx_rtmp_amf_elt_t in_elts[] = {
/* transaction is always 0 */
{ NGX_RTMP_AMF_NUMBER,
ngx_null_string,
NULL, 0 },
{ NGX_RTMP_AMF_NULL,
ngx_null_string,
NULL, 0 },
{ NGX_RTMP_AMF_STRING,
ngx_null_string,
&v.name, sizeof(v.name) },
{ NGX_RTMP_AMF_OPTIONAL | NGX_RTMP_AMF_NUMBER,
ngx_null_string,
&v.start, 0 },
{ NGX_RTMP_AMF_OPTIONAL | NGX_RTMP_AMF_NUMBER,
ngx_null_string,
&v.duration, 0 },
{ NGX_RTMP_AMF_OPTIONAL | NGX_RTMP_AMF_BOOLEAN,
ngx_null_string,
&v.reset, 0 }
};
ngx_memzero(&v, sizeof(v));
/* 从接收到的 amf 数据中提取指定数据 */
if (ngx_rtmp_receive_amf(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
ngx_rtmp_cmd_fill_args(v.name, v.args);
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
"play: name=‘%s‘ args=‘%s‘ start=%i duration=%i "
"reset=%i silent=%i",
v.name, v.args, (ngx_int_t) v.start,
(ngx_int_t) v.duration, (ngx_int_t) v.reset,
(ngx_int_t) v.silent);
return ngx_rtmp_play(s, &v);
}ngx_rtmp_play 构成的链表主要有以下几个 RTMP 模块添加有回调方法：(调用顺序排序)
从 nginx-rtmp 的源码可知，主要就是调用 ngx_rtmp_play_module 模块设置的回调方法 ngx_rtmp_play_play，
而其他模块的都没有做什么，因为没有在配置文件中配置有相关命令来启动它们。
在看 ngx_rtmp_play_play 方法前，先看 ngx_rtmp_play_module 模块的配置项：
static ngx_command_t ngx_rtmp_play_commands[] = {
{ ngx_string("play"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_RTMP_APP_CONF|NGX_CONF_1MORE,
ngx_rtmp_play_url,
NGX_RTMP_APP_CONF_OFFSET,
0,
NULL },
{ ngx_string("play_temp_path"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_RTMP_APP_CONF|NGX_CONF_TAKE1,
ngx_conf_set_str_slot,
NGX_RTMP_APP_CONF_OFFSET,
offsetof(ngx_rtmp_play_app_conf_t, temp_path),
NULL },
{ ngx_string("play_local_path"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_RTMP_APP_CONF|NGX_CONF_TAKE1,
ngx_conf_set_str_slot,
NGX_RTMP_APP_CONF_OFFSET,
offsetof(ngx_rtmp_play_app_conf_t, local_path),
NULL },
ngx_null_command
};可见，第一项为 "play"，回顾点播业务的 nginx.conf 中，就配置了该命令。因此先看该模块是如何解析 play 配置项的：
static char *
ngx_rtmp_play_url(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
{
ngx_rtmp_play_app_conf_t *pacf = conf;
ngx_rtmp_play_entry_t *pe, **ppe;
ngx_str_t url;
ngx_url_t *u;
size_t add, n;
ngx_str_t *value;
/* 初始化 pacf->entries 数组 */
if (pacf->entries.nalloc == 0 &&
ngx_array_init(&pacf->entries, cf->pool, 1, sizeof(void *)) != NGX_OK)
{
return NGX_CONF_ERROR;
}
value = cf->args->elts;
for (n = 1; n < cf->args->nelts; ++n) {
ppe = ngx_array_push(&pacf->entries);
if (ppe == NULL) {
return NGX_CONF_ERROR;
}
pe = ngx_pcalloc(cf->pool, sizeof(ngx_rtmp_play_entry_t));
if (pe == NULL) {
return NGX_CONF_ERROR;
}
*ppe = pe;
/* 如果 "play" 配置项后的值，即 url 不为 "http://" 开始的，即为本地文件 */
if (ngx_strncasecmp(value[n].data, (u_char *) "http://", 7)) {
/* local file */
pe->root = ngx_palloc(cf->pool, sizeof(ngx_str_t));
if (pe->root == NULL) {
return NGX_CONF_ERROR;
}
*pe->root = value[n];
continue;
}
/* 否则，为网络文件 */
/* http case */
url = value[n];
add = sizeof("http://") - 1;
url.data += add;
url.len -= add;
u = ngx_pcalloc(cf->pool, sizeof(ngx_url_t));
if (u == NULL) {
return NGX_CONF_ERROR;
}
u->url.len = url.len;
u->url.data = url.data;
u->default_port = 80;
u->uri_part = 1;
if (ngx_parse_url(cf->pool, u) != NGX_OK) {
if (u->err) {
ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
"%s in url \"%V\"", u->err, &u->url);
}
return NGX_CONF_ERROR;
}
pe->url = u;
}
return NGX_CONF_OK;
}该函数主要就是提取 "play" 配置项后的值，判断是本地文件还是网络文件，根据相应值构造 ngx_rtmp_play_entry_t，并将该指向该
结构体的指针放入到 pacf->entries 数组中.
下面继续看 ngx_rtmp_play_play 函数。
static ngx_int_t
ngx_rtmp_play_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
{
ngx_rtmp_play_main_conf_t *pmcf;
ngx_rtmp_play_app_conf_t *pacf;
ngx_rtmp_play_ctx_t *ctx;
u_char *p;
ngx_rtmp_play_fmt_t *fmt, **pfmt;
ngx_str_t *pfx, *sfx;
ngx_str_t name;
ngx_uint_t n;
pmcf = ngx_rtmp_get_module_main_conf(s, ngx_rtmp_play_module);
pacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_play_module);
if (pacf == NULL || pacf->entries.nelts == 0) {
goto next;
}
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
"play: play name=‘%s‘ timestamp=%i",
v->name, (ngx_int_t) v->start);
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
if (ctx && ctx->file.fd != NGX_INVALID_FILE) {
ngx_log_error(NGX_LOG_ERR, s->connection->log, 0,
"play: already playing");
goto next;
}
/* check for double-dot in v->name;
* we should not move out of play directory */
for (p = v->name; *p; ++p) {
if (ngx_path_separator(p[0]) &&
p[1] == ‘.‘ && p[2] == ‘.‘ &&
ngx_path_separator(p[3]))
{
ngx_log_error(NGX_LOG_ERR, s->connection->log, 0,
"play: bad name ‘%s‘", v->name);
return NGX_ERROR;
}
}
if (ctx == NULL) {
/* 分配 ngx_rtmp_play_module 模块的上下文结构体，并将其放入到全局数组 s->ctx 中 */
ctx = ngx_palloc(s->connection->pool, sizeof(ngx_rtmp_play_ctx_t));
ngx_rtmp_set_ctx(s, ctx, ngx_rtmp_play_module);
}
ngx_memzero(ctx, sizeof(*ctx));
ctx->session = s;
ctx->aindex = ngx_rtmp_play_parse_index(‘a‘, v->args);
ctx->vindex = ngx_rtmp_play_parse_index(‘v‘, v->args);
ctx->file.log = s->connection->log;
ngx_memcpy(ctx->name, v->name, NGX_RTMP_MAX_NAME);
name.len = ngx_strlen(v->name);
name.data = v->name;
/* pmcf->fmts 数组是在 postconfiguration 方法中初始化好的，
* 在 nginx-rtmp 中，该数组仅有两项，分别是 flv 和 mp4 */
pfmt = pmcf->fmts.elts;
for (n = 0; n < pmcf->fmts.nelts; ++n, ++pfmt) {
fmt = *pfmt;
pfx = &fmt->pfx; // 封装格式的前缀
sfx = &fmt->sfx; // 封装格式的后缀
if (pfx->len == 0 && ctx->fmt == NULL) {
ctx->fmt = fmt;
}
if (pfx->len && name.len >= pfx->len &&
ngx_strncasecmp(pfx->data, name.data, pfx->len) == 0)
{
ctx->pfx_size = pfx->len;
ctx->fmt = fmt;
break;
}
/* 比较客户端请求播放的文件的后缀与服务器支持的文件后缀是否一致 */
if (name.len >= sfx->len &&
ngx_strncasecmp(sfx->data, name.data + name.len - sfx->len,
sfx->len) == 0)
{
ctx->fmt = fmt;
}
}
if (ctx->fmt == NULL) {
ngx_log_error(NGX_LOG_ERR, s->connection->log, 0,
"play: fmt not found");
goto next;
}
ctx->file.fd = NGX_INVALID_FILE;
ctx->nentry = NGX_CONF_UNSET_UINT;
ctx->post_seek = NGX_CONF_UNSET_UINT;
sfx = &ctx->fmt->sfx;
if (name.len < sfx->len ||
ngx_strncasecmp(sfx->data, name.data + name.len - sfx->len,
sfx->len))
{
ctx->sfx = *sfx;
}
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: fmt=%V", &ctx->fmt->name);
return ngx_rtmp_play_next_entry(s, v);
next:
return next_play(s, v);
}static ngx_int_t
ngx_rtmp_play_next_entry(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
{
ngx_rtmp_play_app_conf_t *pacf;
ngx_rtmp_play_ctx_t *ctx;
ngx_rtmp_play_entry_t *pe;
u_char *p;
static u_char path[NGX_MAX_PATH + 1];
pacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_play_module);
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
for ( ;; ) {
/* 若文件描述符不为 -1 ，则关闭它 */
if (ctx->file.fd != NGX_INVALID_FILE) {
ngx_close_file(ctx->file.fd);
ctx->file.fd = NGX_INVALID_FILE;
}
if (ctx->file_id) {
ngx_rtmp_play_cleanup_local_file(s);
}
ctx->nentry = (ctx->nentry == NGX_CONF_UNSET_UINT ?
0 : ctx->nentry + 1);
if (ctx->nentry >= pacf->entries.nelts) {
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: all entries failed");
ngx_rtmp_send_status(s, "NetStream.Play.StreamNotFound", "error",
"Video on demand stream not found");
break;
}
/* 根据 ctx->nentry 获取当前要播放的路径，该路径保存在 ngx_rtmp_play_entry_t，
* 若播放的是本地文件，则路径保存在成员 root 中，若播放的是网络文件，则路径
* 保存在 url 中 */
pe = ngx_rtmp_play_get_current_entry(s);
ngx_log_debug4(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: trying %s entry %ui/%uz ‘%V‘",
pe->url ? "remote" : "local",
ctx->nentry + 1, pacf->entries.nelts,
pe->url ? &pe->url->url : pe->root);
/* open remote */
if (pe->url) {
return ngx_rtmp_play_open_remote(s, v);
}
/* open local */
p = ngx_snprintf(path, NGX_MAX_PATH, "%V/%s%V",
pe->root, v->name + ctx->pfx_size, &ctx->sfx);
*p = 0;
/* 打开要播放的本地文件 */
ctx->file.fd = ngx_open_file(path, NGX_FILE_RDONLY, NGX_FILE_OPEN,
NGX_FILE_DEFAULT_ACCESS);
if (ctx->file.fd == NGX_INVALID_FILE) {
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, ngx_errno,
"play: error opening file ‘%s‘", path);
continue;
}
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: open local file ‘%s‘", path);
if (ngx_rtmp_play_open(s, v->start) != NGX_OK) {
return NGX_ERROR;
}
break;
}
return next_play(s, v);
}static ngx_int_t
ngx_rtmp_play_open(ngx_rtmp_session_t *s, double start)
{
ngx_rtmp_play_ctx_t *ctx;
ngx_event_t *e;
ngx_uint_t timestamp;
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
if (ctx->file.fd == NGX_INVALID_FILE) {
return NGX_ERROR;
}
/* 向客户端发送 消息类型为NGX_RTMP_MSG_USER(4) 的 stream_begin 消息, 如下图[9] */
if (ngx_rtmp_send_stream_begin(s, NGX_RTMP_MSID) != NGX_OK) {
return NGX_ERROR;
}
/* 然后接着发送状态消息，如下图[10] */
if (ngx_rtmp_send_status(s, "NetStream.Play.Start", "status",
"Start video on demand")
!= NGX_OK)
{
return NGX_ERROR;
}
if (ngx_rtmp_play_join(s) != NGX_OK) {
return NGX_ERROR;
}
/* 初始化该 send_evt 事件 */
e = &ctx->send_evt;
e->data = s;
/* 初始化发送 指定播放文件数据的 回调函数 */
e->handler = ngx_rtmp_play_send;
e->log = s->connection->log;
/* 发送 recored 命令，如下图[11] */
ngx_rtmp_send_recorded(s, 1);
/* 发送 sample access，如下图[12] */
if (ngx_rtmp_send_sample_access(s) != NGX_OK) {
return NGX_ERROR;
}
/* 调用相应封装格式的 init 函数 */
if (ngx_rtmp_play_do_init(s) != NGX_OK) {
return NGX_ERROR;
}
/* 计算时间戳，有下一步 seek 可知，其实就是计算发送 媒体文件的起始位置 */
timestamp = ctx->post_seek != NGX_CONF_UNSET_UINT ? ctx->post_seek :
(start < 0 ? 0 : (ngx_uint_t) start);
/* 同理，调用相应封装格式的 seek 函数，将文件指针 seek 到 timstamp 位置（相对文件起始） */
if (ngx_rtmp_play_do_seek(s, timestamp) != NGX_OK) {
return NGX_ERROR;
}
if (ngx_rtmp_play_do_start(s) != NGX_OK) {
return NGX_ERROR;
}
/* 置位 opend 标志位，表示文件已经打开了 */
ctx->opened = 1;
return NGX_OK;
}static ngx_int_t
ngx_rtmp_play_join(ngx_rtmp_session_t *s)
{
ngx_rtmp_play_ctx_t *ctx, **pctx;
ngx_rtmp_play_app_conf_t *pacf;
ngx_uint_t h;
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: join");
pacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_play_module);
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
if (ctx == NULL || ctx->joined) {
return NGX_ERROR;
}
h = ngx_hash_key(ctx->name, ngx_strlen(ctx->name));
pctx = &pacf->ctx[h % pacf->nbuckets];
while (*pctx) {
if (!ngx_strncmp((*pctx)->name, ctx->name, NGX_RTMP_MAX_NAME)) {
break;
}
pctx = &(*pctx)->next;
}
ctx->next = *pctx;
*pctx = ctx;
ctx->joined = 1;
return NGX_OK;
}该函数主要根据 ctx->name 通过 hash 的 key 方法，得到映射槽的位置 h，即 pacf->ctx[h]，然后将其插入到
ngx_rtmp_play_module 的上下文结构体 ctx 的 ctx->next 构建的链表中.
static ngx_int_t
ngx_rtmp_play_do_init(ngx_rtmp_session_t *s)
{
ngx_rtmp_play_ctx_t *ctx;
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
if (ctx == NULL) {
return NGX_ERROR;
}
/* 根据 fmt 调用相应封装格式的 init 函数，这里只支持两种封装格式，flv 和 mp4 */
if (ctx->fmt && ctx->fmt->init &&
ctx->fmt->init(s, &ctx->file, ctx->aindex, ctx->vindex) != NGX_OK)
{
return NGX_ERROR;
}
return NGX_OK;
}static ngx_int_t
ngx_rtmp_play_do_seek(ngx_rtmp_session_t *s, ngx_uint_t timestamp)
{
ngx_rtmp_play_ctx_t *ctx;
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
if (ctx == NULL) {
return NGX_ERROR;
}
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: seek timestamp=%ui", timestamp);
/* 调用相应封装的 seek 函数 */
if (ctx->fmt && ctx->fmt->seek &&
ctx->fmt->seek(s, &ctx->file, timestamp) != NGX_OK)
{
return NGX_ERROR;
}
/* 若播放标志位已经置位，则将 ctx->send_evt 事件放入到 延迟队列 ngx_posted_events 中，
* 正常来说应该还没有置位，仅在 ngx_rtmp_play_do_stark 中将其置 1 */
if (ctx->playing) {
ngx_post_event((&ctx->send_evt), &ngx_posted_events);
}
return NGX_OK;
}static ngx_int_t
ngx_rtmp_play_do_start(ngx_rtmp_session_t *s)
{
ngx_rtmp_play_ctx_t *ctx;
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
if (ctx == NULL) {
return NGX_ERROR;
}
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: start");
/* 调用相应封装的 start 函数 */
if (ctx->fmt && ctx->fmt->start &&
ctx->fmt->start(s, &ctx->file) != NGX_OK)
{
return NGX_ERROR;
}
/* 将 ctx->send_evt 事件放入到 ngx_posted_events 延迟队列中 */
ngx_post_event((&ctx->send_evt), &ngx_posted_events);
/* 置位 playing 标志位，表示开始发送文件 */
ctx->playing = 1;
return NGX_OK;
}
从 ngx_rtmp_play_open 函数返回后，会一路返回到 ngx_rtmp_recv 函数中，然后解析完接收到的下一个客户端
发送来的 user control message(4) 消息（在开始分析 play 时说过）后，就一路返回到
ngx_process_events_and_timers 函数中，即如下：
void
ngx_process_events_and_timers(ngx_cycle_t *cycle)
{
...
(void) ngx_process_events(cycle, timer, flags);
/* 返回到这里，然后接着往下执行 */
...
/* 将会执行到这里，由上面的分析可知，已经将 send_evt 事件放入到了
* ngx_posted_events 延迟队列中，因此会取出该事件，并执行 */
ngx_event_process_posted(cycle, &ngx_posted_events);
}send_evt 事件的回调方法为 ngx_rtmp_play_send 函数。
点播的最后一个调用就是在该函数中循环的，直到全部发送文件完成后，才返回.
static void
ngx_rtmp_play_send(ngx_event_t *e)
{
ngx_rtmp_session_t *s = e->data;
ngx_rtmp_play_ctx_t *ctx;
ngx_int_t rc;
ngx_uint_t ts;
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
if (ctx == NULL || ctx->fmt == NULL || ctx->fmt->send == NULL) {
return;
}
ts = 0;
/* 调用相应封装的 send 函数 */
rc = ctx->fmt->send(s, &ctx->file, &ts);
if (rc > 0) {
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: send schedule %i", rc);
ngx_add_timer(e, rc);
return;
}
if (rc == NGX_AGAIN) {
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: send buffer full");
ngx_post_event(e, &s->posted_dry_events);
return;
}
if (rc == NGX_OK) {
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: send restart");
/* 每发送完成一次，就将其放入到 ngx_posted_events 延迟队列中,
* 然后由下一次 ngx_process_events_and_timers 函数中的循环
* 调用到它 */
ngx_post_event(e, &ngx_posted_events);
return;
}
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: send done");
/* 发送 stream_eof */
ngx_rtmp_send_stream_eof(s, NGX_RTMP_MSID);
ngx_rtmp_send_play_status(s, "NetStream.Play.Complete", "status", ts, 0);
ngx_rtmp_send_status(s, "NetStream.Play.Stop", "status", "Stopped");
}
致此，点播分析完毕。
标签：define opp dia port sample 握手 nstat event 网络文件
原文地址：https://www.cnblogs.com/jimodetiantang/p/8980265.html
