diff --git a/ngx_rtmp_relay_module.c b/ngx_rtmp_relay_module.c index 51be0dc..e3d234f 100644 --- a/ngx_rtmp_relay_module.c +++ b/ngx_rtmp_relay_module.c @@ -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; 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;