zotan/nginx-mod-rtmp
1
1
Fork 0
mirror of https://github.com/zotanmew/nginx-rtmp-module.git synced 2024-05-09 06:01:08 +02:00
Code Issues Projects Releases Wiki Activity

implemented cmd module - a layer for amf0 call dispatching

Browse source
This commit is contained in:
Roman Arutyunyan 2012-03-23 21:29:22 +04:00
parent 2c52b09841
commit d943d51975
11 changed files with 912 additions and 699 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
TODO
View file

@ -1,25 +1,13 @@
- add support of flv file writing & streaming
- add HTTP callbacks for all cmd calls
- add HTTP callbacks for all calls
- implement new broadcast module with typed callbacks
& make live/record modules register AMF0 callbacks in it
publish
play
close
set_data_frame
(implement createStream, releaseStream, deleteStream, FC* etc)
- add support for flv file streaming
- support client chunk size change
- check compilation with IPv6 enabled
- remove macros hell from ngx_rtmp_send.c
- fix broken data_frame(?)
- fix '..greeing line..' in log
- changed NGX_OK/NGX_DONE to NGX_DECLINED/NGX_OK

3
config
View file

@ -3,6 +3,7 @@ ngx_addon_name="ngx_rtmp_module"
CORE_MODULES="$CORE_MODULES
ngx_rtmp_module \
ngx_rtmp_core_module \
ngx_rtmp_cmd_module \
ngx_rtmp_access_module \
ngx_rtmp_live_module \
ngx_rtmp_record_module \
@ -10,13 +11,13 @@ CORE_MODULES="$CORE_MODULES
NGX_ADDON_SRCS="$NGX_ADDON_SRCS \
$ngx_addon_dir/ngx_rtmp.c \
$ngx_addon_dir/ngx_rtmp_app.c \
$ngx_addon_dir/ngx_rtmp_amf0.c \
$ngx_addon_dir/ngx_rtmp_send.c \
$ngx_addon_dir/ngx_rtmp_shared.c \
$ngx_addon_dir/ngx_rtmp_handler.c \
$ngx_addon_dir/ngx_rtmp_receive.c \
$ngx_addon_dir/ngx_rtmp_core_module.c \
$ngx_addon_dir/ngx_rtmp_cmd_module.c \
$ngx_addon_dir/ngx_rtmp_access_module.c \
$ngx_addon_dir/ngx_rtmp_live_module.c \
$ngx_addon_dir/ngx_rtmp_record_module.c \

11
ngx_rtmp.c
View file

@ -347,7 +347,6 @@ static ngx_int_t
ngx_rtmp_init_events(ngx_conf_t *cf, ngx_rtmp_core_main_conf_t *cmcf)
{
size_t n;
ngx_rtmp_amf0_handler_t *h;
for(n = 0; n < NGX_RTMP_MAX_EVENT; ++n) {
if (ngx_array_init(&cmcf->events[n], cf->pool, 1,
@ -363,11 +362,6 @@ ngx_rtmp_init_events(ngx_conf_t *cf, ngx_rtmp_core_main_conf_t *cmcf)
return NGX_ERROR;
}
/* this handler of connect should always be the first */
h = ngx_array_push(&cmcf->amf0);
ngx_str_set(&h->name, "connect");
h->handler = ngx_rtmp_connect;
return NGX_OK;
}
@ -411,11 +405,6 @@ ngx_rtmp_init_event_handlers(ngx_conf_t *cf, ngx_rtmp_core_main_conf_t *cmcf)
eh = ngx_array_push(&cmcf->events[NGX_RTMP_MSG_USER]);
*eh = ngx_rtmp_user_message_handler;
/* init several AMF0 callbacks */
h = ngx_array_push(&cmcf->amf0);
ngx_str_set(&h->name, "createStream");
h->handler = ngx_rtmp_create_stream;
/* init amf0 callbacks */
ngx_array_init(&cmcf->amf0_arrays, cf->pool, 1, sizeof(ngx_hash_key_t));

13
ngx_rtmp.h
View file

@ -219,8 +219,9 @@ typedef struct {
/* handler result code:
* NGX_ERROR - error
* NGX_OK - success
* NGX_DONE - success, but do not call more handlers on this event */
* NGX_OK - success, may continue
* NGX_DONE - success, input parsed, reply sent; need no
* more calls on this event */
typedef ngx_int_t (*ngx_rtmp_handler_pt)(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
@ -348,14 +349,6 @@ ngx_int_t ngx_rtmp_user_message_handler(ngx_rtmp_session_t *s,
ngx_int_t ngx_rtmp_amf0_message_handler(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
/* Standard AMF0 handlers */
ngx_int_t ngx_rtmp_connect(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in);
ngx_int_t ngx_rtmp_create_stream(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in);
ngx_int_t ngx_rtmp_amf0_default(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in);
/* Shared output buffers */
ngx_chain_t * ngx_rtmp_alloc_shared_buf(ngx_rtmp_core_srv_conf_t *cscf);

52
ngx_rtmp_access_module.c
View file

@ -6,31 +6,19 @@
#include <ngx_config.h>
#include <ngx_core.h>
#include "ngx_rtmp.h"
#include "ngx_rtmp_cmd_module.h"
#define NGX_RTMP_ACCESS_PUBLISH 0x01
#define NGX_RTMP_ACCESS_PLAY 0x02
static ngx_int_t ngx_rtmp_access_connect(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_access_publish(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_access_play(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static char * ngx_rtmp_access_rule(ngx_conf_t *cf, ngx_command_t *cmd,
void *conf);
void *conf);
static ngx_int_t ngx_rtmp_access_postconfiguration(ngx_conf_t *cf);
static void * ngx_rtmp_access_create_app_conf(ngx_conf_t *cf);
static char * ngx_rtmp_access_merge_app_conf(ngx_conf_t *cf,
void *parent, void *child);
static ngx_rtmp_amf0_handler_t ngx_rtmp_access_map[] = {
{ ngx_string("connect"), ngx_rtmp_access_connect },
{ ngx_string("publish"), ngx_rtmp_access_publish },
{ ngx_string("play"), ngx_rtmp_access_play },
};
void *parent, void *child);
typedef struct {
@ -234,7 +222,7 @@ ngx_rtmp_access(ngx_rtmp_session_t *s, ngx_uint_t flag)
if (ascf == NULL) {
ngx_log_debug0(NGX_LOG_DEBUG_HTTP, s->connection->log, 0,
"access: NULL loc conf");
"access: NULL app conf");
return NGX_ERROR;
}
@ -401,8 +389,7 @@ ngx_rtmp_access_rule(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
static ngx_int_t
ngx_rtmp_access_connect(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in)
ngx_rtmp_access_connect(ngx_rtmp_session_t *s)
{
return ngx_rtmp_access(s, NGX_RTMP_ACCESS_PUBLISH) == NGX_OK
|| ngx_rtmp_access(s, NGX_RTMP_ACCESS_PLAY) == NGX_OK
@ -413,7 +400,7 @@ ngx_rtmp_access_connect(ngx_rtmp_session_t *s,
static ngx_int_t
ngx_rtmp_access_publish(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in)
ngx_str_t *name, ngx_int_t type)
{
return ngx_rtmp_access(s, NGX_RTMP_ACCESS_PUBLISH);
}
@ -421,7 +408,7 @@ ngx_rtmp_access_publish(ngx_rtmp_session_t *s,
static ngx_int_t
ngx_rtmp_access_play(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in)
ngx_str_t *name, uint32_t start, uint32_t duration, ngx_int_t reset)
{
return ngx_rtmp_access(s, NGX_RTMP_ACCESS_PLAY);
}
@ -430,24 +417,19 @@ ngx_rtmp_access_play(ngx_rtmp_session_t *s,
static ngx_int_t
ngx_rtmp_access_postconfiguration(ngx_conf_t *cf)
{
ngx_rtmp_core_main_conf_t *cmcf;
ngx_rtmp_amf0_handler_t *ch, *bh;
size_t n, ncalls;
ngx_rtmp_cmd_main_conf_t *dmcf;
void *ch;
cmcf = ngx_rtmp_conf_get_module_main_conf(cf, ngx_rtmp_core_module);
dmcf = ngx_rtmp_conf_get_module_main_conf(cf, ngx_rtmp_cmd_module);
/* register AMF0 callbacks */
ncalls = sizeof(ngx_rtmp_access_map)
/ sizeof(ngx_rtmp_access_map[0]);
ch = ngx_array_push_n(&cmcf->amf0, ncalls);
if (ch == NULL) {
return NGX_ERROR;
}
ch = ngx_array_push(&dmcf->connect);
*(ngx_rtmp_cmd_connect_pt*)ch = ngx_rtmp_access_connect;
bh = ngx_rtmp_access_map;
for(n = 0; n < ncalls; ++n, ++ch, ++bh) {
*ch = *bh;
}
ch = ngx_array_push(&dmcf->publish);
*(ngx_rtmp_cmd_publish_pt*)ch = ngx_rtmp_access_publish;
ch = ngx_array_push(&dmcf->play);
*(ngx_rtmp_cmd_play_pt*)ch = ngx_rtmp_access_play;
return NGX_OK;
}

219
ngx_rtmp_app.c
View file

@ -1,219 +0,0 @@
/*
* Copyright (c) 2012 Roman Arutyunyan
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include "ngx_rtmp.h"
#define NGX_RTMP_FMS_VERSION "FMS/3,0,1,123"
#define NGX_RTMP_CAPABILITIES 31
ngx_int_t
ngx_rtmp_connect(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
ngx_rtmp_core_srv_conf_t *cscf;
ngx_uint_t n;
ngx_rtmp_core_app_conf_t **cacfp;
size_t len;
static double trans;
static double capabilities = NGX_RTMP_CAPABILITIES;
static struct {
u_char app[1024];
u_char flashver[1024];
u_char swf_url[1024];
u_char tc_url[1024];
double acodecs;
double vcodecs;
u_char page_url[1024];
} v;
static ngx_rtmp_amf0_elt_t in_cmd[] = {
{ NGX_RTMP_AMF0_STRING, "app", v.app, sizeof(v.app) },
{ NGX_RTMP_AMF0_STRING, "flashver", v.flashver, sizeof(v.flashver) },
{ NGX_RTMP_AMF0_STRING, "swfUrl", v.swf_url, sizeof(v.swf_url) },
{ NGX_RTMP_AMF0_STRING, "tcUrl", v.tc_url, sizeof(v.tc_url) },
{ NGX_RTMP_AMF0_NUMBER, "audioCodecs", &v.acodecs, sizeof(v.acodecs) },
{ NGX_RTMP_AMF0_NUMBER, "videoCodecs", &v.vcodecs, sizeof(v.vcodecs) },
{ NGX_RTMP_AMF0_STRING, "pageUrl", v.page_url, sizeof(v.page_url) },
};
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0, &trans, 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL, in_cmd, sizeof(in_cmd) },
};
static ngx_rtmp_amf0_elt_t out_obj[] = {
{ NGX_RTMP_AMF0_STRING, "fmsVer", NGX_RTMP_FMS_VERSION, 0 },
{ NGX_RTMP_AMF0_NUMBER, "capabilities", &capabilities, 0 },
};
static ngx_rtmp_amf0_elt_t out_inf[] = {
{ NGX_RTMP_AMF0_STRING, "level", "status", 0 },
{ NGX_RTMP_AMF0_STRING, "code", "NetConnection.Connect.Success", 0 },
{ NGX_RTMP_AMF0_STRING, "description", "Connection succeeded.", 0 },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "_result", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL, &trans, 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL, out_obj, sizeof(out_obj) },
{ NGX_RTMP_AMF0_OBJECT, NULL, out_inf, sizeof(out_inf) },
};
if (s->connected) {
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0, "duplicate connection");
return NGX_ERROR;
}
cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module);
/* parse input */
ngx_memzero(&v, sizeof(v));
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
ngx_log_debug7(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"connect: app='%s' flashver='%s' swf_url='%s' "
"tc_url='%s' page_url='%s' acodecs=%uD vcodecs=%uD",
v.app, v.flashver, v.swf_url, v.tc_url, v.page_url,
(uint32_t)v.acodecs, (uint32_t)v.vcodecs);
/* fill session parameters */
s->connected = 1;
#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_SET_STRPAR(app);
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
s->acodecs = v.acodecs;
s->vcodecs = v.vcodecs;
/* find application & set app_conf */
len = ngx_strlen(v.app);
cacfp = cscf->applications.elts;
for(n = 0; n < cscf->applications.nelts; ++n, ++cacfp) {
if ((*cacfp)->name.len == len
&& !ngx_strncmp((*cacfp)->name.data, v.app, len))
{
/* found app! */
s->app_conf = (*cacfp)->app_conf;
break;
}
}
if (s->app_conf == NULL) {
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
"application not found: '%s'", v.app);
return NGX_ERROR;
}
/* send all replies */
return ngx_rtmp_send_ack_size(s, cscf->ack_window)
|| ngx_rtmp_send_bandwidth(s, cscf->ack_window, NGX_RTMP_LIMIT_DYNAMIC)
|| ngx_rtmp_send_user_stream_begin(s, 0)
|| ngx_rtmp_send_chunk_size(s, cscf->chunk_size)
|| ngx_rtmp_send_amf0(s, h, out_elts,
sizeof(out_elts) / sizeof(out_elts[0]))
? NGX_ERROR
: NGX_OK;
/* we need NGX_OK not NGX_DONE to make access module
* work after connect has successfully finished */
}
ngx_int_t
ngx_rtmp_create_stream(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
/* support one message stream per connection */
static double stream = 1;
static double trans;
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0, &trans, sizeof(trans) },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "_result", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL, &trans, 0 },
{ NGX_RTMP_AMF0_NULL , NULL, NULL, 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL, &stream, sizeof(stream) },
};
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"createStream");
/* parse input */
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
/* send result with standard stream */
return ngx_rtmp_send_amf0(s, h, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) == NGX_OK
? NGX_DONE
: NGX_ERROR;
}
ngx_int_t
ngx_rtmp_amf0_default(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
ngx_rtmp_header_t sh;
static double trans;
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0, &trans, sizeof(trans) },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "_result", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL, &trans, 0 },
{ NGX_RTMP_AMF0_NULL , NULL, NULL, 0 },
{ NGX_RTMP_AMF0_NULL , NULL, NULL, 0 },
};
/* parse input */
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
memset(&sh, 0, sizeof(sh));
sh.csid = h->csid;
sh.type = NGX_RTMP_MSG_AMF0_CMD;
sh.msid = 0;
/* send simple _result */
return ngx_rtmp_send_amf0(s, &sh, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) == NGX_OK
? NGX_DONE
: NGX_ERROR;
}

800
ngx_rtmp_cmd_module.c Normal file
View file

@ -0,0 +1,800 @@
/*
* Copyright (c) 2012 Roman Arutyunyan
*/
#include "ngx_rtmp_cmd_module.h"
#define NGX_RTMP_FMS_VERSION "FMS/3,0,1,123"
#define NGX_RTMP_CAPABILITIES 31
#define NGX_RTMP_CMD_CSID_AMF0 5
static void * ngx_rtmp_cmd_create_main_conf(ngx_conf_t *cf);
static ngx_int_t ngx_rtmp_cmd_postconfiguration(ngx_conf_t *cf);
static ngx_int_t ngx_rtmp_cmd_connect(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_cmd_create_stream(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_cmd_publish(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_cmd_play(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_cmd_close(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_cmd_disconnect(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_cmd_default(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_command_t ngx_rtmp_cmd_commands[] = {
ngx_null_command
};
static ngx_rtmp_module_t ngx_rtmp_cmd_module_ctx = {
NULL, /* preconfiguration */
ngx_rtmp_cmd_postconfiguration, /* postconfiguration */
ngx_rtmp_cmd_create_main_conf, /* create main configuration */
NULL, /* init main configuration */
NULL, /* create server configuration */
NULL, /* merge server configuration */
NULL, /* create app configuration */
NULL /* merge app configuration */
};
ngx_module_t ngx_rtmp_cmd_module = {
NGX_MODULE_V1,
&ngx_rtmp_cmd_module_ctx, /* module context */
ngx_rtmp_cmd_commands, /* module directives */
NGX_RTMP_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
static void *
ngx_rtmp_cmd_create_main_conf(ngx_conf_t *cf)
{
ngx_rtmp_cmd_main_conf_t *cmcf;
cmcf = ngx_pcalloc(cf->pool, sizeof(ngx_rtmp_cmd_main_conf_t));
if (cmcf == NULL) {
return NULL;
}
#define NGX_RTMP_CMD_INIT_ARRAY(name) \
if (ngx_array_init(&cmcf->name, cf->pool, 1, sizeof(void *)) \
!= NGX_OK) \
{ \
return NULL; \
}
NGX_RTMP_CMD_INIT_ARRAY(connect);
NGX_RTMP_CMD_INIT_ARRAY(publish);
NGX_RTMP_CMD_INIT_ARRAY(play);
NGX_RTMP_CMD_INIT_ARRAY(close);
#undef NGX_RTMP_CMD_INIT_ARRAY
return cmcf;
}
#define NGX_RTMP_CMD_I(name) \
{ \
/* call handlers */ \
ngx_rtmp_cmd_##name##_pt *h; \
ngx_rtmp_cmd_main_conf_t *cmcf; \
ngx_uint_t n; \
\
cmcf = ngx_rtmp_get_module_main_conf(s, ngx_rtmp_cmd_module); \
h = cmcf->name.elts; \
for(n = 0; n < cmcf->name.nelts; ++n, ++h) { \
if ((*h)
#define NGX_RTMP_CMD_F \
!= NGX_OK) { \
return NGX_ERROR; \
} \
} \
}
static ngx_rtmp_amf0_handler_t ngx_rtmp_cmd_map[] = {
{ ngx_string("connect"), ngx_rtmp_cmd_connect },
{ ngx_string("createStream"), ngx_rtmp_cmd_create_stream },
{ ngx_string("publish"), ngx_rtmp_cmd_publish },
{ ngx_string("play"), ngx_rtmp_cmd_play },
{ ngx_string("close"), ngx_rtmp_cmd_close },
{ ngx_string("releaseStream"), ngx_rtmp_cmd_close },
{ ngx_string("deleteStream"), ngx_rtmp_cmd_close },
{ ngx_string("closeStream"), ngx_rtmp_cmd_close },
{ ngx_string("FCPublish"), ngx_rtmp_cmd_default },
{ ngx_string("FCSubscribe"), ngx_rtmp_cmd_default },
};
ngx_int_t
ngx_rtmp_cmd_connect(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
ngx_rtmp_core_srv_conf_t *cscf;
ngx_uint_t n;
ngx_rtmp_core_app_conf_t **cacfp;
size_t len;
static double trans;
static double capabilities = NGX_RTMP_CAPABILITIES;
static struct {
u_char app[1024];
u_char flashver[1024];
u_char swf_url[1024];
u_char tc_url[1024];
double acodecs;
double vcodecs;
u_char page_url[1024];
} v;
static ngx_rtmp_amf0_elt_t in_cmd[] = {
{ NGX_RTMP_AMF0_STRING, "app",
v.app, sizeof(v.app) },
{ NGX_RTMP_AMF0_STRING, "flashver",
v.flashver, sizeof(v.flashver) },
{ NGX_RTMP_AMF0_STRING, "swfUrl",
v.swf_url, sizeof(v.swf_url) },
{ NGX_RTMP_AMF0_STRING, "tcUrl",
v.tc_url, sizeof(v.tc_url) },
{ NGX_RTMP_AMF0_NUMBER, "audioCodecs",
&v.acodecs, sizeof(v.acodecs) },
{ NGX_RTMP_AMF0_NUMBER, "videoCodecs",
&v.vcodecs, sizeof(v.vcodecs) },
{ NGX_RTMP_AMF0_STRING, "pageUrl",
v.page_url, sizeof(v.page_url) },
};
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0,
&trans, 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL,
in_cmd, sizeof(in_cmd) },
};
static ngx_rtmp_amf0_elt_t out_obj[] = {
{ NGX_RTMP_AMF0_STRING, "fmsVer",
NGX_RTMP_FMS_VERSION, 0 },
{ NGX_RTMP_AMF0_NUMBER, "capabilities",
&capabilities, 0 },
};
static ngx_rtmp_amf0_elt_t out_inf[] = {
{ NGX_RTMP_AMF0_STRING, "level",
"status", 0 },
{ NGX_RTMP_AMF0_STRING, "code",
"NetConnection.Connect.Success",
0 },
{ NGX_RTMP_AMF0_STRING, "description",
"Connection succeeded.",
0 },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL,
"_result", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL,
&trans, 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL,
out_obj, sizeof(out_obj) },
{ NGX_RTMP_AMF0_OBJECT, NULL,
out_inf, sizeof(out_inf) },
};
if (s->connected) {
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
"connect: duplicate connection");
return NGX_ERROR;
}
cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module);
/* parse input */
ngx_memzero(&v, sizeof(v));
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
ngx_log_debug7(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"connect: app='%s' flashver='%s' swf_url='%s' "
"tc_url='%s' page_url='%s' acodecs=%uD vcodecs=%uD",
v.app, v.flashver, v.swf_url, v.tc_url, v.page_url,
(uint32_t)v.acodecs, (uint32_t)v.vcodecs);
/* fill session parameters */
s->connected = 1;
#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_SET_STRPAR(app);
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
s->acodecs = v.acodecs;
s->vcodecs = v.vcodecs;
/* find application & set app_conf */
len = ngx_strlen(v.app);
cacfp = cscf->applications.elts;
for(n = 0; n < cscf->applications.nelts; ++n, ++cacfp) {
if ((*cacfp)->name.len == len
&& !ngx_strncmp((*cacfp)->name.data, v.app, len))
{
/* found app! */
s->app_conf = (*cacfp)->app_conf;
break;
}
}
if (s->app_conf == NULL) {
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
"connect: application not found: '%s'", v.app);
return NGX_ERROR;
}
/* call handlers */
NGX_RTMP_CMD_I(connect) (s) NGX_RTMP_CMD_F;
/* send all replies */
return ngx_rtmp_send_ack_size(s, cscf->ack_window)
|| ngx_rtmp_send_bandwidth(s, cscf->ack_window, NGX_RTMP_LIMIT_DYNAMIC)
|| ngx_rtmp_send_user_stream_begin(s, 0)
|| ngx_rtmp_send_chunk_size(s, cscf->chunk_size)
|| ngx_rtmp_send_amf0(s, h, out_elts,
sizeof(out_elts) / sizeof(out_elts[0]))
? NGX_ERROR
: NGX_DONE;
}
ngx_int_t
ngx_rtmp_cmd_create_stream(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
/* support one message stream per connection */
static double stream = 1;
static double trans;
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0,
&trans, sizeof(trans) },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL,
"_result", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL,
&trans, 0 },
{ NGX_RTMP_AMF0_NULL , NULL,
NULL, 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL,
&stream, sizeof(stream) },
};
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"createStream");
/* parse input */
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
/* send result with standard stream */
return ngx_rtmp_send_amf0(s, h, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) == NGX_OK
? NGX_DONE
: NGX_ERROR;
}
ngx_int_t
ngx_rtmp_cmd_disconnect(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
/* call handlers */
NGX_RTMP_CMD_I(close) (s) NGX_RTMP_CMD_F;
return NGX_OK;
}
ngx_int_t
ngx_rtmp_cmd_close(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
ngx_rtmp_cmd_disconnect(s, h, in);
return ngx_rtmp_cmd_default(s, h, in);
}
ngx_int_t
ngx_rtmp_cmd_default(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
ngx_rtmp_header_t sh;
static double trans;
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0,
&trans, sizeof(trans) },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL,
"_result", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL,
&trans, 0 },
{ NGX_RTMP_AMF0_NULL, NULL,
NULL, 0 },
{ NGX_RTMP_AMF0_NULL, NULL,
NULL, 0 },
};
/* parse input */
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
memset(&sh, 0, sizeof(sh));
sh.csid = h->csid;
sh.type = NGX_RTMP_MSG_AMF0_CMD;
sh.msid = 0;
/* send simple _result */
return ngx_rtmp_send_amf0(s, &sh, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) == NGX_OK
? NGX_DONE
: NGX_ERROR;
}
static ngx_int_t
ngx_rtmp_cmd_publish(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
ngx_rtmp_header_t sh;
ngx_str_t stream;
ngx_int_t type;
static double trans;
static struct {
u_char name[1024];
u_char type[1024];
} v;
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0,
&trans, 0 },
{ NGX_RTMP_AMF0_NULL, 0,
NULL, 0 },
{ NGX_RTMP_AMF0_STRING, 0,
&v.name, sizeof(v.name) },
{ NGX_RTMP_AMF0_STRING, 0,
&v.type, sizeof(v.type) },
};
static ngx_rtmp_amf0_elt_t out_inf[] = {
{ NGX_RTMP_AMF0_STRING, "code",
"NetStream.Publish.Start",
0 },
{ NGX_RTMP_AMF0_STRING, "level",
"status", 0 },
{ NGX_RTMP_AMF0_STRING, "description",
"Publish succeeded.", 0 },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL,
"onStatus", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL,
&trans, 0 },
{ NGX_RTMP_AMF0_NULL , NULL,
NULL, 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL,
out_inf, sizeof(out_inf) },
};
ngx_memzero(&v, sizeof(v));
/* parse input */
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"publish: name='%s' type=%s",
v.name, v.type);
stream.len = ngx_strlen(v.name);
stream.data = ngx_palloc(s->connection->pool, stream.len);
ngx_memcpy(stream.data, v.name, stream.len);
if (ngx_strcmp(v.type, "record") == 0) {
type = NGX_RTMP_CMD_PUBLISH_RECORD;
} else if(ngx_strcmp(v.type, "append") == 0) {
type = NGX_RTMP_CMD_PUBLISH_APPEND;
} else if (ngx_strcmp(v.type, "live") == 0) {
type = NGX_RTMP_CMD_PUBLISH_LIVE;
} else {
type = 0;
}
/* call handlers */
NGX_RTMP_CMD_I(publish) (s, &stream, type) NGX_RTMP_CMD_F;
/* start stream */
if (ngx_rtmp_send_user_stream_begin(s, 1) != NGX_OK) {
return NGX_ERROR;
}
/* send onStatus reply */
memset(&sh, 0, sizeof(sh));
sh.type = NGX_RTMP_MSG_AMF0_CMD;
sh.csid = NGX_RTMP_CMD_CSID_AMF0;
sh.msid = h->msid;
if (ngx_rtmp_send_amf0(s, &sh, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) != NGX_OK)
{
return NGX_ERROR;
}
return NGX_DONE;
}
static ngx_int_t
ngx_rtmp_cmd_play(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
ngx_rtmp_header_t sh;
ngx_str_t stream;
static double trans;
static int bfalse;
static struct {
u_char name[1024];
double start;
double duration;
int reset;
} v;
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0,
&trans, 0 },
{ NGX_RTMP_AMF0_NULL, 0,
NULL, 0 },
{ NGX_RTMP_AMF0_STRING, 0,
&v.name, sizeof(v.name) },
{ NGX_RTMP_AMF0_OPTIONAL
| NGX_RTMP_AMF0_NUMBER, 0,
&v.start, 0 },
{ NGX_RTMP_AMF0_OPTIONAL
| NGX_RTMP_AMF0_NUMBER, 0,
&v.duration, 0 },
{ NGX_RTMP_AMF0_OPTIONAL
| NGX_RTMP_AMF0_BOOLEAN,0,
&v.reset, 0 }
};
static ngx_rtmp_amf0_elt_t out_inf[] = {
{ NGX_RTMP_AMF0_STRING, "code",
"NetStream.Play.Reset", 0 },
{ NGX_RTMP_AMF0_STRING, "level",
"status", 0 },
{ NGX_RTMP_AMF0_STRING, "description",
"Playing and resetting.",
0 },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL,
"onStatus", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL,
&trans, 0 },
{ NGX_RTMP_AMF0_NULL , NULL,
NULL, 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL,
out_inf, sizeof(out_inf) },
};
static ngx_rtmp_amf0_elt_t out2_inf[] = {
{ NGX_RTMP_AMF0_STRING, "code",
"NetStream.Play.Start", 0 },
{ NGX_RTMP_AMF0_STRING, "level",
"status", 0 },
{ NGX_RTMP_AMF0_STRING, "description",
"Started playing.", 0 },
};
static ngx_rtmp_amf0_elt_t out2_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL,
"onStatus", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL,
&trans, 0 },
{ NGX_RTMP_AMF0_NULL , NULL,
NULL, 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL,
out2_inf, sizeof(out2_inf) },
};
static ngx_rtmp_amf0_elt_t out3_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL,
"|RtmpSampleAccess", 0 },
{ NGX_RTMP_AMF0_BOOLEAN,NULL,
&bfalse, 0 },
{ NGX_RTMP_AMF0_BOOLEAN,NULL,
&bfalse, 0 },
};
static ngx_rtmp_amf0_elt_t out4_inf[] = {
{ NGX_RTMP_AMF0_STRING, "code",
"NetStream.Data.Start", 0 },
};
static ngx_rtmp_amf0_elt_t out4_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL,
"onStatus", 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL,
out4_inf, sizeof(out4_inf) },
};
ngx_memzero(&v, sizeof(v));
/* parse input */
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
ngx_log_debug4(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: name='%s' start=%uD duration=%uD reset=%d",
v.name, (uint32_t)v.start, (uint32_t)v.duration, v.reset);
stream.len = ngx_strlen(v.name);
stream.data = ngx_palloc(s->connection->pool, stream.len);
ngx_memcpy(stream.data, v.name, stream.len);
/* call handlers */
NGX_RTMP_CMD_I(play) (s, &stream, v.start,
v.duration, v.reset) NGX_RTMP_CMD_F;
/* start stream */
if (ngx_rtmp_send_user_stream_begin(s, 1) != NGX_OK) {
return NGX_ERROR;
}
/* send onStatus reply */
memset(&sh, 0, sizeof(sh));
sh.type = NGX_RTMP_MSG_AMF0_CMD;
sh.csid = NGX_RTMP_CMD_CSID_AMF0;
sh.msid = h->msid;
if (ngx_rtmp_send_amf0(s, &sh, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) != NGX_OK)
{
return NGX_ERROR;
}
/* send sample access meta message FIXME */
if (ngx_rtmp_send_amf0(s, &sh, out2_elts,
sizeof(out2_elts) / sizeof(out2_elts[0])) != NGX_OK)
{
return NGX_ERROR;
}
/* send data start meta message */
sh.type = NGX_RTMP_MSG_AMF0_META;
if (ngx_rtmp_send_amf0(s, &sh, out3_elts,
sizeof(out3_elts) / sizeof(out3_elts[0])) != NGX_OK)
{
return NGX_ERROR;
}
if (ngx_rtmp_send_amf0(s, &sh, out4_elts,
sizeof(out4_elts) / sizeof(out4_elts[0])) != NGX_OK)
{
return NGX_ERROR;
}
return NGX_DONE;
}
#if 0
static ngx_int_t
ngx_rtmp_live_set_data_frame(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
ngx_rtmp_live_app_conf_t *lacf;
ngx_connection_t *c;
ngx_rtmp_live_ctx_t *ctx;
ngx_rtmp_amf0_ctx_t act;
ngx_rtmp_header_t sh;
ngx_rtmp_core_srv_conf_t *cscf;
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "@setDataFrame", 0 },
};
lacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_live_module);
if (lacf == NULL || !lacf->live) {
return NGX_OK;
}
c = s->connection;
cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module);
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_live_module);
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, c->log, 0, "live: data_frame");
/* TODO: allow sending more meta packages to change live content */
if (ctx->data_frame) {
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, c->log, 0,
"duplicate data_frame");
return NGX_OK;
}
/* create full metadata chain for output */
memset(&act, 0, sizeof(act));
act.cscf = cscf;
act.alloc = ngx_rtmp_alloc_shared_buf;
act.log = c->log;
if (ngx_rtmp_amf0_write(&act, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) != NGX_OK)
{
if (act.first) {
ngx_rtmp_free_shared_bufs(cscf, act.first);
}
return NGX_ERROR;
}
if (act.first == NULL) {
return NGX_OK;
}
ctx->data_frame = act.first;
if (ngx_rtmp_append_shared_bufs(cscf, ctx->data_frame, in) == NULL) {
if (ctx->data_frame) {
ngx_rtmp_free_shared_bufs(cscf, ctx->data_frame);
}
return NGX_ERROR;
}
memset(&sh, 0, sizeof(sh));
sh.csid = NGX_RTMP_LIVE_CSID_AMF0;
sh.msid = 1;
sh.type = NGX_RTMP_MSG_AMF0_META;
ngx_rtmp_prepare_message(s, &sh, NULL, ctx->data_frame);
return NGX_OK;
}
static ngx_int_t
ngx_rtmp_live_stream_length(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in)
{
ngx_rtmp_live_app_conf_t *lacf;
ngx_rtmp_header_t sh;
static double trans;
static double length;
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0, &trans, sizeof(trans) },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "_result", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL, &trans, 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL, &length, 0 },
};
lacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_live_module);
if (lacf == NULL || !lacf->live) {
return NGX_OK;
}
/* parse input */
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
memset(&sh, 0, sizeof(sh));
sh.csid = h->csid;
sh.type = NGX_RTMP_MSG_AMF0_CMD;
sh.msid = 0;
/* send simple _result */
return ngx_rtmp_send_amf0(s, &sh, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) == NGX_OK
? NGX_DONE
: NGX_ERROR;
}
#endif
static ngx_int_t
ngx_rtmp_cmd_postconfiguration(ngx_conf_t *cf)
{
ngx_rtmp_core_main_conf_t *cmcf;
ngx_rtmp_handler_pt *h;
ngx_rtmp_amf0_handler_t *ch, *bh;
size_t n, ncalls;
cmcf = ngx_rtmp_conf_get_module_main_conf(cf, ngx_rtmp_core_module);
h = ngx_array_push(&cmcf->events[NGX_RTMP_DISCONNECT]);
*h = ngx_rtmp_cmd_disconnect;
/* register AMF0 callbacks */
ncalls = sizeof(ngx_rtmp_cmd_map) / sizeof(ngx_rtmp_cmd_map[0]);
ch = ngx_array_push_n(&cmcf->amf0, ncalls);
if (h == NULL) {
return NGX_ERROR;
}
bh = ngx_rtmp_cmd_map;
for(n = 0; n < ncalls; ++n, ++ch, ++bh) {
*ch = *bh;
}
return NGX_OK;
}

39
ngx_rtmp_cmd_module.h Normal file
View file

@ -0,0 +1,39 @@
/*
* Copyright (c) 2012 Roman Arutyunyan
*/
#ifndef _NGX_RTMP_CMD_H_INCLUDED_
#define _NGX_RTMP_CMD_H_INCLUDED_
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
#include "ngx_rtmp.h"
/* publish types */
#define NGX_RTMP_CMD_PUBLISH_RECORD 1
#define NGX_RTMP_CMD_PUBLISH_APPEND 2
#define NGX_RTMP_CMD_PUBLISH_LIVE 3
typedef ngx_int_t (*ngx_rtmp_cmd_connect_pt)(ngx_rtmp_session_t *s);
typedef ngx_int_t (*ngx_rtmp_cmd_publish_pt)(ngx_rtmp_session_t *s,
ngx_str_t *name, ngx_int_t type);
typedef ngx_int_t (*ngx_rtmp_cmd_play_pt)(ngx_rtmp_session_t *s,
ngx_str_t *name, uint32_t start, uint32_t duration, ngx_int_t reset);
typedef ngx_int_t (*ngx_rtmp_cmd_close_pt)(ngx_rtmp_session_t *s);
typedef struct {
ngx_array_t connect;
ngx_array_t publish;
ngx_array_t play;
ngx_array_t close;
} ngx_rtmp_cmd_main_conf_t;
extern ngx_module_t ngx_rtmp_cmd_module;
#endif /*_NGX_RTMP_CMD_H_INCLUDED_ */

391
ngx_rtmp_live_module.c
View file

@ -6,10 +6,10 @@
#include <ngx_config.h>
#include <ngx_core.h>
#include "ngx_rtmp.h"
#include "ngx_rtmp_cmd_module.h"
/* Standard stream ids for liveing */
#define NGX_RTMP_LIVE_CSID_AMF0 5
/* Chunk stream ids for output */
#define NGX_RTMP_LIVE_CSID_AUDIO 6
#define NGX_RTMP_LIVE_CSID_VIDEO 7
@ -20,27 +20,6 @@ static char * ngx_rtmp_live_merge_app_conf(ngx_conf_t *cf,
void *parent, void *child);
static ngx_int_t ngx_rtmp_live_publish(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_live_play(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_live_set_data_frame(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_live_stream_length(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_rtmp_amf0_handler_t ngx_rtmp_live_map[] = {
{ ngx_string("publish"), ngx_rtmp_live_publish },
{ ngx_string("play"), ngx_rtmp_live_play },
{ ngx_string("-@setDataFrame"), ngx_rtmp_live_set_data_frame },
{ ngx_string("getStreamLength"), ngx_rtmp_live_stream_length },
{ ngx_string("releaseStream"), ngx_rtmp_amf0_default },
{ ngx_string("FCPublish"), ngx_rtmp_amf0_default },
{ ngx_string("FCSubscribe"), ngx_rtmp_amf0_default },
};
typedef struct ngx_rtmp_live_ctx_s ngx_rtmp_live_ctx_t;
@ -112,7 +91,6 @@ ngx_module_t ngx_rtmp_live_module = {
#define NGX_RTMP_LIVE_PUBLISHING 0x01
#define NGX_RTMP_LIVE_PLAYING 0x02
#define NGX_RTMP_LIVE_KEYFRAME 0x04
#define NGX_RTMP_LIVE_DATA_FRAME 0x08
struct ngx_rtmp_live_ctx_s {
@ -216,8 +194,7 @@ ngx_rtmp_live_join(ngx_rtmp_session_t *s, ngx_str_t *stream,
static ngx_int_t
ngx_rtmp_live_leave(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
ngx_rtmp_live_close(ngx_rtmp_session_t *s)
{
ngx_connection_t *c;
ngx_rtmp_live_ctx_t *ctx, **hctx;
@ -274,7 +251,6 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_rtmp_session_t *ss;
ngx_rtmp_header_t sh;
ngx_uint_t priority;
ngx_int_t rc;
int keyframe;
c = s->connection;
@ -351,24 +327,6 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ss = cctx->session;
/* if we have metadata check if the subscriber
* has already received one */
if (ctx->data_frame
&& !(cctx->flags & NGX_RTMP_LIVE_DATA_FRAME))
{
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"live: sending data_frame");
rc = ngx_rtmp_send_message(ss, ctx->data_frame, 0);
if (rc == NGX_ERROR) {
continue;
}
if (rc == NGX_OK) {
cctx->flags |= NGX_RTMP_LIVE_DATA_FRAME;
}
}
/* waiting for a keyframe? */
if (lacf->wait_key_frame
&& sh.type == NGX_RTMP_MSG_VIDEO
@ -395,39 +353,10 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
static ngx_int_t
ngx_rtmp_live_publish(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
ngx_rtmp_live_publish(ngx_rtmp_session_t *s, ngx_str_t *name,
ngx_int_t type)
{
ngx_rtmp_live_app_conf_t *lacf;
ngx_rtmp_header_t sh;
ngx_str_t stream;
static double trans;
static struct {
u_char name[1024];
u_char type[1024];
} v;
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0, &trans, 0 },
{ NGX_RTMP_AMF0_NULL, 0, NULL, 0 },
{ NGX_RTMP_AMF0_STRING, 0, &v.name, sizeof(v.name) },
{ NGX_RTMP_AMF0_STRING, 0, &v.type, sizeof(v.type) },
};
static ngx_rtmp_amf0_elt_t out_inf[] = {
{ NGX_RTMP_AMF0_STRING, "code", "NetStream.Publish.Start", 0 },
{ NGX_RTMP_AMF0_STRING, "level", "status", 0 },
{ NGX_RTMP_AMF0_STRING, "description", "Publish succeeded.", 0 },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "onStatus", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL, &trans, 0 },
{ NGX_RTMP_AMF0_NULL , NULL, NULL, 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL, out_inf, sizeof(out_inf) },
};
lacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_live_module);
@ -435,118 +364,22 @@ ngx_rtmp_live_publish(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
return NGX_OK;
}
ngx_memzero(&v, sizeof(v));
/* parse input */
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"live: publish: name='%s' type=%s",
v.name, v.type);
"live: publish: name='%V' type=%d",
name, type);
/* join stream as publisher */
stream.len = ngx_strlen(v.name);
stream.data = ngx_palloc(s->connection->pool, stream.len);
ngx_memcpy(stream.data, v.name, stream.len);
ngx_rtmp_live_join(s, &stream, NGX_RTMP_LIVE_PUBLISHING);
/* TODO: we can probably make any use of v.type: live/record/append */
/* start stream */
if (ngx_rtmp_send_user_stream_begin(s, 1) != NGX_OK) {
return NGX_ERROR;
}
/* send onStatus reply */
memset(&sh, 0, sizeof(sh));
sh.type = NGX_RTMP_MSG_AMF0_CMD;
sh.csid = NGX_RTMP_LIVE_CSID_AMF0;
sh.msid = h->msid;
if (ngx_rtmp_send_amf0(s, &sh, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) != NGX_OK)
{
return NGX_ERROR;
}
ngx_rtmp_live_join(s, name, NGX_RTMP_LIVE_PUBLISHING);
return NGX_OK;
}
static ngx_int_t
ngx_rtmp_live_play(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
ngx_rtmp_live_play(ngx_rtmp_session_t *s, ngx_str_t *name,
uint32_t start, uint32_t duration, ngx_int_t reset)
{
ngx_rtmp_live_app_conf_t *lacf;
ngx_rtmp_header_t sh;
ngx_str_t stream;
static double trans;
static int bfalse;
static struct {
u_char name[1024];
double start;
double duration;
int flush;
} v;
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0, &trans, 0 },
{ NGX_RTMP_AMF0_NULL, 0, NULL, 0 },
{ NGX_RTMP_AMF0_STRING, 0, &v.name, sizeof(v.name) },
{ NGX_RTMP_AMF0_OPTIONAL
| NGX_RTMP_AMF0_NUMBER, 0, &v.start, 0 },
{ NGX_RTMP_AMF0_OPTIONAL
| NGX_RTMP_AMF0_NUMBER, 0, &v.duration, 0 },
{ NGX_RTMP_AMF0_OPTIONAL
| NGX_RTMP_AMF0_BOOLEAN,0, &v.flush, 0 }
};
static ngx_rtmp_amf0_elt_t out_inf[] = {
{ NGX_RTMP_AMF0_STRING, "code", "NetStream.Play.Reset", 0 },
{ NGX_RTMP_AMF0_STRING, "level", "status", 0 },
{ NGX_RTMP_AMF0_STRING, "description", "Playing and resetting.", 0 },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "onStatus", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL, &trans, 0 },
{ NGX_RTMP_AMF0_NULL , NULL, NULL, 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL, out_inf, sizeof(out_inf) },
};
static ngx_rtmp_amf0_elt_t out2_inf[] = {
{ NGX_RTMP_AMF0_STRING, "code", "NetStream.Play.Start", 0 },
{ NGX_RTMP_AMF0_STRING, "level", "status", 0 },
{ NGX_RTMP_AMF0_STRING, "description", "Started playing.", 0 },
};
static ngx_rtmp_amf0_elt_t out2_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "onStatus", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL, &trans, 0 },
{ NGX_RTMP_AMF0_NULL , NULL, NULL, 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL, out2_inf, sizeof(out2_inf) },
};
static ngx_rtmp_amf0_elt_t out3_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "|RtmpSampleAccess", 0 },
{ NGX_RTMP_AMF0_BOOLEAN,NULL, &bfalse, 0 },
{ NGX_RTMP_AMF0_BOOLEAN,NULL, &bfalse, 0 },
};
static ngx_rtmp_amf0_elt_t out4_inf[] = {
{ NGX_RTMP_AMF0_STRING, "code", "NetStream.Data.Start", 0 },
};
static ngx_rtmp_amf0_elt_t out4_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "onStatus", 0 },
{ NGX_RTMP_AMF0_OBJECT, NULL, out4_inf, sizeof(out4_inf) },
};
lacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_live_module);
@ -554,219 +387,45 @@ ngx_rtmp_live_play(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
return NGX_OK;
}
ngx_memzero(&v, sizeof(v));
/* parse input */
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
ngx_log_debug4(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"live: play: name='%s' start=%uD duration=%uD flush=%d",
v.name, (uint32_t)v.start, (uint32_t)v.duration, v.flush);
"live: play: name='%V' start=%uD duration=%uD reset=%d",
name, start, duration, reset);
/* join stream as player */
stream.len = ngx_strlen(v.name);
stream.data = ngx_palloc(s->connection->pool, stream.len);
ngx_memcpy(stream.data, v.name, stream.len);
ngx_rtmp_live_join(s, &stream, NGX_RTMP_LIVE_PLAYING);
/* start stream */
if (ngx_rtmp_send_user_stream_begin(s, 1) != NGX_OK) {
return NGX_ERROR;
}
/* send onStatus reply */
memset(&sh, 0, sizeof(sh));
sh.type = NGX_RTMP_MSG_AMF0_CMD;
sh.csid = NGX_RTMP_LIVE_CSID_AMF0;
sh.msid = h->msid;
if (ngx_rtmp_send_amf0(s, &sh, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) != NGX_OK)
{
return NGX_ERROR;
}
/* send sample access meta message FIXME */
if (ngx_rtmp_send_amf0(s, &sh, out2_elts,
sizeof(out2_elts) / sizeof(out2_elts[0])) != NGX_OK)
{
return NGX_ERROR;
}
/* send data start meta message */
sh.type = NGX_RTMP_MSG_AMF0_META;
if (ngx_rtmp_send_amf0(s, &sh, out3_elts,
sizeof(out3_elts) / sizeof(out3_elts[0])) != NGX_OK)
{
return NGX_ERROR;
}
if (ngx_rtmp_send_amf0(s, &sh, out4_elts,
sizeof(out4_elts) / sizeof(out4_elts[0])) != NGX_OK)
{
return NGX_ERROR;
}
return NGX_DONE;
}
static ngx_int_t
ngx_rtmp_live_set_data_frame(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
{
ngx_rtmp_live_app_conf_t *lacf;
ngx_connection_t *c;
ngx_rtmp_live_ctx_t *ctx;
ngx_rtmp_amf0_ctx_t act;
ngx_rtmp_header_t sh;
ngx_rtmp_core_srv_conf_t *cscf;
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "@setDataFrame", 0 },
};
lacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_live_module);
if (lacf == NULL || !lacf->live) {
return NGX_OK;
}
c = s->connection;
cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module);
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_live_module);
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, c->log, 0, "live: data_frame");
/* TODO: allow sending more meta packages to change live content */
if (ctx->data_frame) {
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, c->log, 0,
"duplicate data_frame");
return NGX_OK;
}
/* create full metadata chain for output */
memset(&act, 0, sizeof(act));
act.cscf = cscf;
act.alloc = ngx_rtmp_alloc_shared_buf;
act.log = c->log;
if (ngx_rtmp_amf0_write(&act, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) != NGX_OK)
{
if (act.first) {
ngx_rtmp_free_shared_bufs(cscf, act.first);
}
return NGX_ERROR;
}
if (act.first == NULL) {
return NGX_OK;
}
ctx->data_frame = act.first;
if (ngx_rtmp_append_shared_bufs(cscf, ctx->data_frame, in) == NULL) {
if (ctx->data_frame) {
ngx_rtmp_free_shared_bufs(cscf, ctx->data_frame);
}
return NGX_ERROR;
}
memset(&sh, 0, sizeof(sh));
sh.csid = NGX_RTMP_LIVE_CSID_AMF0;
sh.msid = 1;
sh.type = NGX_RTMP_MSG_AMF0_META;
ngx_rtmp_prepare_message(s, &sh, NULL, ctx->data_frame);
ngx_rtmp_live_join(s, name, NGX_RTMP_LIVE_PLAYING);
return NGX_OK;
}
static ngx_int_t
ngx_rtmp_live_stream_length(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in)
{
ngx_rtmp_live_app_conf_t *lacf;
ngx_rtmp_header_t sh;
static double trans;
static double length;
static ngx_rtmp_amf0_elt_t in_elts[] = {
{ NGX_RTMP_AMF0_NUMBER, 0, &trans, sizeof(trans) },
};
static ngx_rtmp_amf0_elt_t out_elts[] = {
{ NGX_RTMP_AMF0_STRING, NULL, "_result", 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL, &trans, 0 },
{ NGX_RTMP_AMF0_NUMBER, NULL, &length, 0 },
};
lacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_live_module);
if (lacf == NULL || !lacf->live) {
return NGX_OK;
}
/* parse input */
if (ngx_rtmp_receive_amf0(s, in, in_elts,
sizeof(in_elts) / sizeof(in_elts[0])))
{
return NGX_ERROR;
}
memset(&sh, 0, sizeof(sh));
sh.csid = h->csid;
sh.type = NGX_RTMP_MSG_AMF0_CMD;
sh.msid = 0;
/* send simple _result */
return ngx_rtmp_send_amf0(s, &sh, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) == NGX_OK
? NGX_DONE
: NGX_ERROR;
}
static ngx_int_t
ngx_rtmp_live_postconfiguration(ngx_conf_t *cf)
{
ngx_rtmp_core_main_conf_t *cmcf;
ngx_rtmp_cmd_main_conf_t *dmcf;
ngx_rtmp_handler_pt *h;
ngx_rtmp_amf0_handler_t *ch, *bh;
size_t n, ncalls;
void *ch;
/* register raw event handlers */
cmcf = ngx_rtmp_conf_get_module_main_conf(cf, ngx_rtmp_core_module);
/* register event handlers */
h = ngx_array_push(&cmcf->events[NGX_RTMP_MSG_AUDIO]);
*h = ngx_rtmp_live_av;
h = ngx_array_push(&cmcf->events[NGX_RTMP_MSG_VIDEO]);
*h = ngx_rtmp_live_av;
h = ngx_array_push(&cmcf->events[NGX_RTMP_DISCONNECT]);
*h = ngx_rtmp_live_leave;
/* register command handlers */
dmcf = ngx_rtmp_conf_get_module_main_conf(cf, ngx_rtmp_cmd_module);
/* register AMF0 callbacks */
ncalls = sizeof(ngx_rtmp_live_map)
/ sizeof(ngx_rtmp_live_map[0]);
ch = ngx_array_push_n(&cmcf->amf0, ncalls);
if (h == NULL) {
return NGX_ERROR;
}
ch = ngx_array_push(&dmcf->publish);
*(ngx_rtmp_cmd_publish_pt*)ch = ngx_rtmp_live_publish;
bh = ngx_rtmp_live_map;
for(n = 0; n < ncalls; ++n, ++ch, ++bh) {
*ch = *bh;
}
ch = ngx_array_push(&dmcf->play);
*(ngx_rtmp_cmd_play_pt*)ch = ngx_rtmp_live_play;
ch = ngx_array_push(&dmcf->close);
*(ngx_rtmp_cmd_close_pt*)ch = ngx_rtmp_live_close;
return NGX_OK;
}

63
ngx_rtmp_record_module.c
View file

@ -6,6 +6,7 @@
#include <ngx_config.h>
#include <ngx_core.h>
#include "ngx_rtmp.h"
#include "ngx_rtmp_cmd_module.h"
static ngx_int_t ngx_rtmp_record_postconfiguration(ngx_conf_t *cf);
@ -14,20 +15,6 @@ static char * ngx_rtmp_record_merge_app_conf(ngx_conf_t *cf,
void *parent, void *child);
static ngx_int_t ngx_rtmp_record_start(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_record_end(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_rtmp_amf0_handler_t ngx_rtmp_record_map[] = {
{ ngx_string("publish"), ngx_rtmp_record_start },
{ ngx_string("releaseStream"), ngx_rtmp_record_end },
{ ngx_string("deleteStream"), ngx_rtmp_record_end },
{ ngx_string("FCPublish"), ngx_rtmp_amf0_default },
};
typedef struct {
ngx_str_t root;
size_t max_size;
@ -71,7 +58,7 @@ ngx_module_t ngx_rtmp_record_module = {
&ngx_rtmp_record_module_ctx, /* module context */
ngx_rtmp_record_commands, /* module directives */
NGX_RTMP_MODULE, /* module type */
NULL, /* init master */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
@ -137,7 +124,7 @@ ngx_rtmp_record_write_header(ngx_file_t *file)
0x00,
0x00,
0x00,
0x00 /* PreviousTagSize0 */
0x00 /* PreviousTagSize0 (not actually a header) */
};
return ngx_write_file(file, flv_header, sizeof(flv_header), 0) == NGX_ERROR
@ -147,8 +134,8 @@ ngx_rtmp_record_write_header(ngx_file_t *file)
static ngx_int_t
ngx_rtmp_record_start(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
ngx_rtmp_record_publish(ngx_rtmp_session_t *s,
ngx_str_t *name, ngx_int_t type)
{
ngx_rtmp_record_app_conf_t *racf;
ngx_rtmp_record_ctx_t *ctx;
@ -195,12 +182,15 @@ ngx_rtmp_record_start(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ctx->last = *tod;
/* TODO: use stream name here */
/* TODO: can use 'name' here
* but need to check for bad symbols first
* it comes right from user */
ctx->path.len = ngx_snprintf(ctx->path.data, NGX_MAX_PATH,
"%V/rec-%T.%M.%d.flv", &racf->root, tod->sec, tod->msec,
ctx->counter) - ctx->path.data;
ctx->path.data[ctx->path.len] = 0;
/* open file */
ngx_memzero(&ctx->file, sizeof(ctx->file));
ctx->file.log = s->connection->log;
ctx->file.fd = ngx_open_file(ctx->path.data, NGX_FILE_WRONLY,
@ -223,8 +213,7 @@ ngx_rtmp_record_start(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
static ngx_int_t
ngx_rtmp_record_end(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t *in)
ngx_rtmp_record_close(ngx_rtmp_session_t *s)
{
ngx_rtmp_record_ctx_t *ctx;
ngx_err_t err;
@ -305,8 +294,11 @@ ngx_rtmp_record_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
return NGX_ERROR;
}
/* write tag body;
* ngx_write_chain allocates a lot.
/* write tag body
* FIXME: NGINX
* ngx_write_chain seems to fit best
* but it suffers from uncontrollable
* allocations.
* we're left with plain writing */
for(; in; in = in->next) {
if (in->buf->pos == in->buf->last) {
@ -339,7 +331,7 @@ ngx_rtmp_record_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
{
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"record: closed on size limit");
ngx_rtmp_record_end(s, h, in);
ngx_rtmp_record_close(s);
}
return NGX_OK;
@ -350,9 +342,9 @@ static ngx_int_t
ngx_rtmp_record_postconfiguration(ngx_conf_t *cf)
{
ngx_rtmp_core_main_conf_t *cmcf;
ngx_rtmp_cmd_main_conf_t *dmcf;
ngx_rtmp_handler_pt *h;
ngx_rtmp_amf0_handler_t *ch, *bh;
size_t n, ncalls;
void *ch;
cmcf = ngx_rtmp_conf_get_module_main_conf(cf, ngx_rtmp_core_module);
@ -363,21 +355,14 @@ ngx_rtmp_record_postconfiguration(ngx_conf_t *cf)
h = ngx_array_push(&cmcf->events[NGX_RTMP_MSG_VIDEO]);
*h = ngx_rtmp_record_av;
h = ngx_array_push(&cmcf->events[NGX_RTMP_DISCONNECT]);
*h = ngx_rtmp_record_end;
/* register command handlers */
dmcf = ngx_rtmp_conf_get_module_main_conf(cf, ngx_rtmp_cmd_module);
/* register AMF0 callbacks */
ncalls = sizeof(ngx_rtmp_record_map)
/ sizeof(ngx_rtmp_record_map[0]);
ch = ngx_array_push_n(&cmcf->amf0, ncalls);
if (h == NULL) {
return NGX_ERROR;
}
ch = ngx_array_push(&dmcf->publish);
*(ngx_rtmp_cmd_publish_pt*)ch = ngx_rtmp_record_publish;
bh = ngx_rtmp_record_map;
for(n = 0; n < ncalls; ++n, ++ch, ++bh) {
*ch = *bh;
}
ch = ngx_array_push(&dmcf->close);
*(ngx_rtmp_cmd_close_pt*)ch = ngx_rtmp_record_close;
return NGX_OK;
}

4
test/nginx.conf
View file

@ -49,13 +49,9 @@ http {
server {
# testing framework
# based on flowplayer (http://flowplayer.org)
listen 8080;
location / {
# deny 192.168.33.33;
root /home/rarutyunyan/nginx-rtmp-module/test/www;
}
}