rtmp接收的时候在通过分析chunk,组合成消息,然后在判断收到完整的消息后,就调用ngx_rtmp_fire_event来处理码流,链接,AMF等消息。使用的是类似http的动态数组处理链的方法。
只是http是分为11个阶段,每个阶段为处理回调函数的动态数组,然后依次调用每个阶段的动态数组的回调函数指针运行,
而rtmp是每个消息就只有一个阶段,每个阶段的处理为动态数组的回调。
typedef ngx_int_t (*ngx_rtmp_handler_pt)(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
ngx_int_t
ngx_rtmp_fire_event(ngx_rtmp_session_t *s, ngx_uint_t evt,//指明命令的类型
ngx_rtmp_header_t *h, ngx_chain_t *in)
{
ngx_rtmp_core_main_conf_t *cmcf;
ngx_array_t *ch;
ngx_rtmp_handler_pt *hh;
size_t n;
cmcf = ngx_rtmp_get_module_main_conf(s, ngx_rtmp_core_module);//即在ngx_rtmp_core_main_conf_t中找回调
ch = &cmcf->events[evt];
hh = ch->elts;
for(n = 0; n < ch->nelts; ++n, ++hh) {
if (*hh && (*hh)(s, h, in) != NGX_OK) {
return NGX_ERROR;
}
}
return NGX_OK;
}
对于AMF命令的处理使用的是类似http的filter的方法。
首先AMF的命令是通过类型http的的动态数组回调的方法处理,
这样AMF的类型的所有命令都是通过ngx_rtmp_amf_message_handler来进行处理的,
在ngx_rtmp_amf_message_handler函数中根据具体的amf命令名,找到对应的回调指针数组,即又使用了一次动态回调数组。
即找ngx_rtmp_core_main_conf_t中amf_hash这个hash表
看到ngx_rtmp_core_main_conf_t的amf_hash是通过amf_arrays初始化的,而amf_arrays是通过amf来初始化的,
amf的初始化是在:
最后可以看到使用的为ngx_rtmp_cmd_map
比如对于publish命令其处理为调用ngx_rtmp_cmd_publish_init,而在ngx_rtmp_cmd_publish_init函数中使用了类型http filter的机制。
对于publish命令,有全局的ngx_rtmp_publish,
ngx_rtmp_publish最开始初始化的地方在
注意其是没有调用next_publish的,
这是最先初始化,也是最后调用的,这个应该是装饰模式
在每个模块中有局部全局的next_publish
其他模块的处理如下:
这样对于publish命令,先调用ngx_rtmp_cmd_publish_init命令,然后其调用全局的ngx_rtmp_publish
然后调用每个添加的publish处理,然后调用到next_publish,最后调用到最先设置的
所以amf的命令的处理经过了两次动态回调函数数组,一次装饰模式的回调函数处理链。
typedef struct {
ngx_array_t servers; /* ngx_rtmp_core_srv_conf_t */
ngx_array_t listen; /* ngx_rtmp_listen_t */
ngx_array_t events[NGX_RTMP_MAX_EVENT];
ngx_event_t hot_deploy_checker;
ngx_hash_t amf_hash;
ngx_array_t amf_arrays;
ngx_array_t amf;//设置amf命令处理的回调
} ngx_rtmp_core_main_conf_t;
两次动态回调函数数组的过程都是和ngx_rtmp_core_main_conf_t相关。
第一次是找events
第二次是找amf
amf_hash是使用amf来建立的。