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

Merge branch 'push-reconnect'

Browse source
This commit is contained in:
Roman Arutyunyan 2012-07-18 13:37:17 +04:00
parent 1a2a8e2867 dbc3ac2438
commit 4e475cccb5
2 changed files with 153 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

195
ngx_rtmp_relay_module.c
View file

@ -57,15 +57,19 @@ struct ngx_rtmp_relay_ctx_s {
ngx_int_t live;
ngx_int_t start;
ngx_int_t stop;
ngx_event_t push_evt;
void *tag;
};
typedef struct {
ngx_array_t pulls; /* ngx_rtmp_relay_target_t */
ngx_array_t pushes; /* ngx_rtmp_relay_target_t */
ngx_array_t pulls; /* ngx_rtmp_relay_target_t * */
ngx_array_t pushes; /* ngx_rtmp_relay_target_t * */
ngx_log_t *log;
ngx_uint_t nbuckets;
ngx_msec_t buflen;
ngx_msec_t push_reconnect;
ngx_rtmp_relay_ctx_t **ctx;
} ngx_rtmp_relay_app_conf_t;
@ -106,6 +110,14 @@ static ngx_command_t ngx_rtmp_relay_commands[] = {
offsetof(ngx_rtmp_relay_app_conf_t, buflen),
NULL },
{ ngx_string("push_reconnect"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_RTMP_APP_CONF|NGX_CONF_TAKE1,
ngx_conf_set_msec_slot,
NGX_RTMP_APP_CONF_OFFSET,
offsetof(ngx_rtmp_relay_app_conf_t, push_reconnect),
NULL },
ngx_null_command
};
@ -149,12 +161,13 @@ ngx_rtmp_relay_create_app_conf(ngx_conf_t *cf)
}
ngx_array_init(&racf->pushes, cf->pool, 1,
sizeof(ngx_rtmp_relay_target_t));
sizeof(ngx_rtmp_relay_target_t *));
ngx_array_init(&racf->pulls, cf->pool, 1,
sizeof(ngx_rtmp_relay_target_t));
sizeof(ngx_rtmp_relay_target_t *));
racf->nbuckets = 1024;
racf->log = &cf->cycle->new_log;
racf->buflen = NGX_CONF_UNSET;
racf->push_reconnect = NGX_CONF_UNSET;
return racf;
}
@ -163,18 +176,77 @@ ngx_rtmp_relay_create_app_conf(ngx_conf_t *cf)
static char *
ngx_rtmp_relay_merge_app_conf(ngx_conf_t *cf, void *parent, void *child)
{
ngx_rtmp_relay_app_conf_t *prev = parent;
ngx_rtmp_relay_app_conf_t *conf = child;
ngx_rtmp_relay_app_conf_t *prev = parent;
ngx_rtmp_relay_app_conf_t *conf = child;
conf->ctx = ngx_pcalloc(cf->pool, sizeof(ngx_rtmp_relay_ctx_t *)
* conf->nbuckets);
ngx_conf_merge_msec_value(conf->buflen, prev->buflen, 5000);
ngx_conf_merge_msec_value(conf->push_reconnect, prev->push_reconnect,
3000);
return NGX_CONF_OK;
}
static void
ngx_rtmp_relay_reconnect(ngx_event_t *ev)
{
ngx_rtmp_session_t *s = ev->data;
ngx_rtmp_relay_app_conf_t *racf;
ngx_rtmp_relay_ctx_t *ctx, *pctx;
ngx_uint_t n;
ngx_rtmp_relay_target_t *target, **t;
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"relay: push reconnect");
racf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_relay_module);
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_relay_module);
if (ctx == NULL) {
return;
}
t = racf->pushes.elts;
for (n = 0; n < racf->pushes.nelts; ++n, ++t) {
target = *t;
if (target->name.len && (ctx->name.len != target->name.len ||
ngx_memcmp(ctx->name.data, target->name.data, ctx->name.len)))
{
continue;
}
for (pctx = ctx->play; pctx; pctx = pctx->next) {
if (pctx->tag == target) {
break;
}
}
if (pctx) {
continue;
}
if (ngx_rtmp_relay_push(s, &ctx->name, target) == NGX_OK) {
continue;
}
ngx_log_error(NGX_LOG_ERR, s->connection->log, 0,
"relay: push reconnect failed name='%V' app='%V' "
"playpath='%V' url='%V'",
&ctx->name, &target->app, &target->play_path,
&target->url.url);
if (!ctx->push_evt.timer_set) {
ngx_add_timer(&ctx->push_evt, racf->push_reconnect);
}
}
}
static ngx_int_t
ngx_rtmp_relay_get_peer(ngx_peer_connection_t *pc, void *data)
{
@ -245,6 +317,8 @@ ngx_rtmp_relay_create_remote_ctx(ngx_rtmp_session_t *s, ngx_str_t* name,
goto clear;
}
rctx->tag = target->tag;
#define NGX_RTMP_RELAY_STR_COPY(to, from) \
if (ngx_rtmp_relay_copy_str(pool, &rctx->to, &target->from) != NGX_OK) { \
goto clear; \
@ -377,6 +451,10 @@ ngx_rtmp_relay_create_local_ctx(ngx_rtmp_session_t *s, ngx_str_t *name,
}
ctx->session = s;
ctx->push_evt.data = s;
ctx->push_evt.log = s->connection->log;
ctx->push_evt.handler = ngx_rtmp_relay_reconnect;
if (ctx->publish) {
return NULL;
}
@ -477,7 +555,7 @@ static ngx_int_t
ngx_rtmp_relay_publish(ngx_rtmp_session_t *s, ngx_rtmp_publish_t *v)
{
ngx_rtmp_relay_app_conf_t *racf;
ngx_rtmp_relay_target_t *target;
ngx_rtmp_relay_target_t *target, **t;
ngx_str_t name;
size_t n;
ngx_rtmp_relay_ctx_t *ctx;
@ -495,19 +573,28 @@ ngx_rtmp_relay_publish(ngx_rtmp_session_t *s, ngx_rtmp_publish_t *v)
name.len = ngx_strlen(v->name);
name.data = v->name;
target = racf->pushes.elts;
for (n = 0; n < racf->pushes.nelts; ++n, ++target) {
if (target->name.len == 0
|| (name.len == target->name.len
&& !ngx_memcmp(name.data, target->name.data, name.len)))
t = racf->pushes.elts;
for (n = 0; n < racf->pushes.nelts; ++n, ++t) {
target = *t;
if (target->name.len && (name.len != target->name.len ||
ngx_memcmp(name.data, target->name.data, name.len)))
{
if (ngx_rtmp_relay_push(s, &name, target) != NGX_OK) {
ngx_log_error(NGX_LOG_ERR, s->connection->log, 0,
"relay: push failed name='%V' app='%V' "
"playpath='%V' url='%V'",
&name, &target->app, &target->play_path,
&target->url.url);
}
continue;
}
if (ngx_rtmp_relay_push(s, &name, target) == NGX_OK) {
continue;
}
ngx_log_error(NGX_LOG_ERR, s->connection->log, 0,
"relay: push failed name='%V' app='%V' "
"playpath='%V' url='%V'",
&name, &target->app, &target->play_path,
&target->url.url);
if (!ctx->push_evt.timer_set) {
ngx_add_timer(&ctx->push_evt, racf->push_reconnect);
}
}
@ -520,7 +607,7 @@ static ngx_int_t
ngx_rtmp_relay_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
{
ngx_rtmp_relay_app_conf_t *racf;
ngx_rtmp_relay_target_t *target;
ngx_rtmp_relay_target_t *target, **t;
ngx_str_t name;
size_t n;
ngx_rtmp_relay_ctx_t *ctx;
@ -538,21 +625,25 @@ ngx_rtmp_relay_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
name.len = ngx_strlen(v->name);
name.data = v->name;
target = racf->pulls.elts;
for (n = 0; n < racf->pulls.nelts; ++n, ++target) {
if (target->name.len == 0
|| (name.len == target->name.len
&& !ngx_memcmp(name.data, target->name.data, name.len)))
t = racf->pulls.elts;
for (n = 0; n < racf->pulls.nelts; ++n, ++t) {
target = *t;
if (target->name.len && (name.len != target->name.len ||
ngx_memcmp(name.data, target->name.data, name.len)))
{
if (ngx_rtmp_relay_pull(s, &name, target) != NGX_OK) {
ngx_log_error(NGX_LOG_ERR, s->connection->log, 0,
"relay: pull failed name='%V' app='%V' "
"playpath='%V' url='%V'",
&name, &target->app, &target->play_path,
&target->url.url);
}
break;
continue;
}
if (ngx_rtmp_relay_pull(s, &name, target) == NGX_OK) {
continue;
}
ngx_log_error(NGX_LOG_ERR, s->connection->log, 0,
"relay: pull failed name='%V' app='%V' "
"playpath='%V' url='%V'",
&name, &target->app, &target->play_path,
&target->url.url);
}
next:
@ -1127,6 +1218,8 @@ ngx_rtmp_relay_delete_stream(ngx_rtmp_session_t *s, ngx_rtmp_delete_stream_t *v)
ngx_rtmp_relay_ctx_t *ctx, **cctx;
ngx_uint_t hash;
racf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_relay_module);
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_relay_module);
if (ctx == NULL || ctx->publish == NULL) {
goto next;
@ -1145,12 +1238,11 @@ ngx_rtmp_relay_delete_stream(ngx_rtmp_session_t *s, ngx_rtmp_delete_stream_t *v)
"relay: play disconnect app='%V' name='%V'",
&ctx->app, &ctx->name);
/*TODO: add push reconnect */
/*
if (ctx->relay) {
ngx_rtmp_relay_push(ctx-publish->session,
&ctx->publish->name, &target);
}*/
/* push reconnect */
if (ctx->relay && ctx->tag && !ctx->publish->push_evt.timer_set) {
ngx_add_timer(&ctx->publish->push_evt, racf->push_reconnect);
}
#ifdef NGX_DEBUG
{
ngx_uint_t n = 0;
@ -1161,7 +1253,7 @@ ngx_rtmp_relay_delete_stream(ngx_rtmp_session_t *s, ngx_rtmp_delete_stream_t *v)
}
#endif
if (ctx->publish->play == NULL) {
if (ctx->publish->play == NULL && ctx->publish->relay) {
ngx_log_debug2(NGX_LOG_DEBUG_RTMP,
ctx->publish->session->connection->log, 0,
"relay: publish disconnect empty app='%V' name='%V'",
@ -1179,6 +1271,10 @@ ngx_rtmp_relay_delete_stream(ngx_rtmp_session_t *s, ngx_rtmp_delete_stream_t *v)
"relay: publish disconnect app='%V' name='%V'",
&ctx->app, &ctx->name);
if (ctx->push_evt.timer_set) {
ngx_del_timer(&ctx->push_evt);
}
for (cctx = &ctx->play; *cctx; cctx = &(*cctx)->next) {
(*cctx)->publish = NULL;
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, (*cctx)->session->connection->log,
@ -1188,7 +1284,6 @@ ngx_rtmp_relay_delete_stream(ngx_rtmp_session_t *s, ngx_rtmp_delete_stream_t *v)
}
ctx->publish = NULL;
racf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_relay_module);
hash = ngx_hash_key(ctx->name.data, ctx->name.len);
cctx = &racf->ctx[hash % racf->nbuckets];
for (; *cctx && *cctx != ctx; cctx = &(*cctx)->next);
@ -1206,7 +1301,7 @@ ngx_rtmp_relay_push_pull(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
{
ngx_str_t *value, v, n;
ngx_rtmp_relay_app_conf_t *racf;
ngx_rtmp_relay_target_t *target;
ngx_rtmp_relay_target_t *target, **t;
ngx_url_t *u;
ngx_uint_t i;
u_char *p;
@ -1215,11 +1310,23 @@ ngx_rtmp_relay_push_pull(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
racf = ngx_rtmp_conf_get_module_app_conf(cf, ngx_rtmp_relay_module);
target = ngx_array_push(value[0].data[3] == 'h'
t = ngx_array_push(value[0].data[3] == 'h'
? &racf->pushes /* push */
: &racf->pulls /* pull */
);
ngx_memzero(target, sizeof(ngx_rtmp_relay_target_t));
if (t == NULL) {
return NGX_CONF_ERROR;
}
target = ngx_pcalloc(cf->pool, sizeof(*target));
if (target == NULL) {
return NGX_CONF_ERROR;
}
*t = target;
target->tag = target;
u = &target->url;
u->default_port = 1935;

2
ngx_rtmp_relay_module.h
View file

@ -22,6 +22,8 @@ typedef struct {
ngx_int_t live;
ngx_int_t start;
ngx_int_t stop;
void *tag;
} ngx_rtmp_relay_target_t;