From bfb7df5f6a5320bd059256b182973be18724a9e7 Mon Sep 17 00:00:00 2001 From: Roman Arutyunyan Date: Wed, 11 Jul 2012 12:58:07 +0400 Subject: [PATCH 1/3] implemented stock push reconnect --- ngx_rtmp_relay_module.c | 176 +++++++++++++++++++++++++++++++++------- 1 file changed, 145 insertions(+), 31 deletions(-) diff --git a/ngx_rtmp_relay_module.c b/ngx_rtmp_relay_module.c index 51be0dc..486bf01 100644 --- a/ngx_rtmp_relay_module.c +++ b/ngx_rtmp_relay_module.c @@ -57,6 +57,11 @@ struct ngx_rtmp_relay_ctx_s { ngx_int_t live; ngx_int_t start; ngx_int_t stop; + + /* push-specific */ + ngx_event_t push_evt; + ngx_uint_t *push_map; + ngx_uint_t *push_id; }; @@ -66,6 +71,7 @@ typedef struct { 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 +112,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 }; @@ -155,6 +169,7 @@ ngx_rtmp_relay_create_app_conf(ngx_conf_t *cf) racf->nbuckets = 1024; racf->log = &cf->cycle->new_log; racf->buflen = NGX_CONF_UNSET; + racf->push_reconnect = NGX_CONF_UNSET; return racf; } @@ -163,18 +178,64 @@ 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; + ngx_uint_t n; + ngx_rtmp_relay_target_t *target; + + 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 || ctx->push_map == NULL) { + return; + } + + target = racf->pushes.elts; + for (n = 0; n < racf->pushes.nelts; ++n, ++target) { + if (ctx->push_map[n] == 0) { + continue; + } + + if (ngx_rtmp_relay_push(s, &ctx->name, target) == NGX_OK) { + ctx->push_map[n] = 0; + 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) { @@ -224,6 +285,8 @@ ngx_rtmp_relay_create_remote_ctx(ngx_rtmp_session_t *s, ngx_str_t* name, ngx_int_t rc; ngx_str_t v, *uri; u_char *first, *last, *p; + ngx_rtmp_relay_target_t *t; + ngx_uint_t n; racf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_relay_module); @@ -239,6 +302,19 @@ ngx_rtmp_relay_create_remote_ctx(ngx_rtmp_session_t *s, ngx_str_t* name, goto clear; } + /* stock push target? save id for reconnect */ + t = racf->pushes.elts; + for (n = 0; n < racf->pushes.nelts; ++n, ++t) { + if (t == target) { + rctx->push_id = ngx_palloc(pool, sizeof(ngx_uint_t)); + if (rctx->push_id == NULL) { + goto clear; + } + *rctx->push_id = n; + break; + } + } + if (ngx_rtmp_relay_copy_str(pool, &rctx->name, name) != NGX_OK || ngx_rtmp_relay_copy_str(pool, &rctx->url, &target->url.url) != NGX_OK) { @@ -366,6 +442,9 @@ ngx_rtmp_relay_create_local_ctx(ngx_rtmp_session_t *s, ngx_str_t *name, ngx_rtmp_relay_target_t *target) { ngx_rtmp_relay_ctx_t *ctx; + ngx_rtmp_relay_app_conf_t *racf; + + 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) { @@ -377,6 +456,18 @@ 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->push_map == NULL) { + ctx->push_map = ngx_pcalloc(s->connection->pool, sizeof(ngx_uint_t) + * racf->pushes.nelts); + if (ctx->push_map == NULL) { + return NULL; + } + } + if (ctx->publish) { return NULL; } @@ -497,17 +588,31 @@ ngx_rtmp_relay_publish(ngx_rtmp_session_t *s, ngx_rtmp_publish_t *v) 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))) + 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; + } + + ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_relay_module); + if (ctx == NULL || ctx->push_map == NULL) { + return NGX_ERROR; + } + + ctx->push_map[n] = 1; + + 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); } } @@ -540,19 +645,21 @@ ngx_rtmp_relay_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v) 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))) + 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 +1234,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 +1254,14 @@ 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->publish->push_map && ctx->push_id) { + ctx->publish->push_map[*ctx->push_id] = 1; + if (!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 +1272,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 +1290,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 +1303,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); From e21ba2dc2c9d4f92c4d561490c2666be25437adf Mon Sep 17 00:00:00 2001 From: Roman Arutyunyan Date: Fri, 13 Jul 2012 20:08:28 +0400 Subject: [PATCH 2/3] reimplemented push reconnect using target tags --- ngx_rtmp_relay_module.c | 122 +++++++++++++++++++--------------------- ngx_rtmp_relay_module.h | 2 + 2 files changed, 61 insertions(+), 63 deletions(-) diff --git a/ngx_rtmp_relay_module.c b/ngx_rtmp_relay_module.c index 486bf01..0fe9998 100644 --- a/ngx_rtmp_relay_module.c +++ b/ngx_rtmp_relay_module.c @@ -58,16 +58,14 @@ struct ngx_rtmp_relay_ctx_s { ngx_int_t start; ngx_int_t stop; - /* push-specific */ ngx_event_t push_evt; - ngx_uint_t *push_map; - ngx_uint_t *push_id; + 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; @@ -163,9 +161,9 @@ 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; @@ -198,9 +196,9 @@ 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; + ngx_rtmp_relay_ctx_t *ctx, *pctx; ngx_uint_t n; - ngx_rtmp_relay_target_t *target; + ngx_rtmp_relay_target_t *target, **t; ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, "relay: push reconnect"); @@ -208,18 +206,31 @@ ngx_rtmp_relay_reconnect(ngx_event_t *ev) 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->push_map == NULL) { + if (ctx == NULL) { return; } - target = racf->pushes.elts; - for (n = 0; n < racf->pushes.nelts; ++n, ++target) { - if (ctx->push_map[n] == 0) { + 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) { - ctx->push_map[n] = 0; continue; } @@ -285,8 +296,6 @@ ngx_rtmp_relay_create_remote_ctx(ngx_rtmp_session_t *s, ngx_str_t* name, ngx_int_t rc; ngx_str_t v, *uri; u_char *first, *last, *p; - ngx_rtmp_relay_target_t *t; - ngx_uint_t n; racf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_relay_module); @@ -302,25 +311,14 @@ ngx_rtmp_relay_create_remote_ctx(ngx_rtmp_session_t *s, ngx_str_t* name, goto clear; } - /* stock push target? save id for reconnect */ - t = racf->pushes.elts; - for (n = 0; n < racf->pushes.nelts; ++n, ++t) { - if (t == target) { - rctx->push_id = ngx_palloc(pool, sizeof(ngx_uint_t)); - if (rctx->push_id == NULL) { - goto clear; - } - *rctx->push_id = n; - break; - } - } - if (ngx_rtmp_relay_copy_str(pool, &rctx->name, name) != NGX_OK || ngx_rtmp_relay_copy_str(pool, &rctx->url, &target->url.url) != NGX_OK) { 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; \ @@ -460,14 +458,6 @@ ngx_rtmp_relay_create_local_ctx(ngx_rtmp_session_t *s, ngx_str_t *name, ctx->push_evt.log = s->connection->log; ctx->push_evt.handler = ngx_rtmp_relay_reconnect; - if (ctx->push_map == NULL) { - ctx->push_map = ngx_pcalloc(s->connection->pool, sizeof(ngx_uint_t) - * racf->pushes.nelts); - if (ctx->push_map == NULL) { - return NULL; - } - } - if (ctx->publish) { return NULL; } @@ -568,7 +558,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; @@ -586,10 +576,12 @@ 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 && (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))) { continue; } @@ -598,13 +590,6 @@ ngx_rtmp_relay_publish(ngx_rtmp_session_t *s, ngx_rtmp_publish_t *v) continue; } - ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_relay_module); - if (ctx == NULL || ctx->push_map == NULL) { - return NGX_ERROR; - } - - ctx->push_map[n] = 1; - ngx_log_error(NGX_LOG_ERR, s->connection->log, 0, "relay: push failed name='%V' app='%V' " "playpath='%V' url='%V'", @@ -625,7 +610,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; @@ -643,10 +628,12 @@ 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 && (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))) { continue; } @@ -1255,11 +1242,8 @@ ngx_rtmp_relay_delete_stream(ngx_rtmp_session_t *s, ngx_rtmp_delete_stream_t *v) &ctx->app, &ctx->name); /* push reconnect */ - if (ctx->relay && ctx->publish->push_map && ctx->push_id) { - ctx->publish->push_map[*ctx->push_id] = 1; - if (!ctx->publish->push_evt.timer_set) { - ngx_add_timer(&ctx->publish->push_evt, racf->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 @@ -1320,7 +1304,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; @@ -1329,11 +1313,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; diff --git a/ngx_rtmp_relay_module.h b/ngx_rtmp_relay_module.h index 3341d90..e949112 100644 --- a/ngx_rtmp_relay_module.h +++ b/ngx_rtmp_relay_module.h @@ -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; From dbc3ac2438757b23a65e1612c36b5691e7173585 Mon Sep 17 00:00:00 2001 From: Roman Arutyunyan Date: Wed, 18 Jul 2012 13:36:17 +0400 Subject: [PATCH 3/3] fixed compilation --- ngx_rtmp_relay_module.c | 3 --- 1 file changed, 3 deletions(-) diff --git a/ngx_rtmp_relay_module.c b/ngx_rtmp_relay_module.c index 0fe9998..e3d234f 100644 --- a/ngx_rtmp_relay_module.c +++ b/ngx_rtmp_relay_module.c @@ -440,9 +440,6 @@ ngx_rtmp_relay_create_local_ctx(ngx_rtmp_session_t *s, ngx_str_t *name, ngx_rtmp_relay_target_t *target) { ngx_rtmp_relay_ctx_t *ctx; - ngx_rtmp_relay_app_conf_t *racf; - - 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) {