From bd51d4b18d8261b5e9a3a41360434ca4359f842f Mon Sep 17 00:00:00 2001 From: Roman Arutyunyan Date: Wed, 16 May 2012 18:04:35 +0400 Subject: [PATCH] push is working --- ngx_rtmp_access_module.c | 5 ++++ ngx_rtmp_cmd_module.c | 2 +- ngx_rtmp_handshake.c | 55 +++++++++++++++++++++++++++++----------- ngx_rtmp_relay_module.c | 52 +++++++++++++++++++++++++------------ stat.xsl | 4 +-- test/nginx.conf | 5 ++++ 6 files changed, 89 insertions(+), 34 deletions(-) diff --git a/ngx_rtmp_access_module.c b/ngx_rtmp_access_module.c index 3f656d8..4551ae8 100644 --- a/ngx_rtmp_access_module.c +++ b/ngx_rtmp_access_module.c @@ -230,6 +230,11 @@ ngx_rtmp_access(ngx_rtmp_session_t *s, ngx_uint_t flag) return NGX_ERROR; } + /* relay etc */ + if (s->connection->sockaddr == NULL) { + return NGX_OK; + } + switch (s->connection->sockaddr->sa_family) { case AF_INET: diff --git a/ngx_rtmp_cmd_module.c b/ngx_rtmp_cmd_module.c index d855751..6ff74ac 100644 --- a/ngx_rtmp_cmd_module.c +++ b/ngx_rtmp_cmd_module.c @@ -205,7 +205,7 @@ ngx_rtmp_cmd_connect(ngx_rtmp_session_t *s, ngx_rtmp_connect_t *v) #define NGX_RTMP_SET_STRPAR(name) \ - s->name.len = ngx_strlen(v->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) diff --git a/ngx_rtmp_handshake.c b/ngx_rtmp_handshake.c index 960716d..26ee4cc 100644 --- a/ngx_rtmp_handshake.c +++ b/ngx_rtmp_handshake.c @@ -172,6 +172,10 @@ ngx_rtmp_alloc_handshake_buffer(ngx_rtmp_session_t *s, int short_buf) ngx_chain_t *cl; ngx_buf_t *b; + ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, + "handshake: allocating %sbuffer", + short_buf ? "short " : ""); + cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module); if (cscf->free_hs) { @@ -243,7 +247,7 @@ ngx_rtmp_old_handshake_response(ngx_rtmp_session_t *s) size_t len; ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, - "RTMP old-style handshake"); + "handshake: old-style handshake"); src = s->hs_bufs[0]->pos + 8; len = s->hs_bufs[0]->last - src; @@ -276,7 +280,7 @@ ngx_rtmp_handshake_response(ngx_rtmp_session_t *s) b = s->hs_bufs[0]; if (*b->pos != '\x03') { ngx_log_error(NGX_LOG_INFO, s->connection->log, 0, - "Unexpected RTMP version: %i", (ngx_int_t)*b->pos); + "handshake: unexpected RTMP version: %i", (ngx_int_t)*b->pos); return NGX_ERROR; } ++b->pos; @@ -284,7 +288,7 @@ ngx_rtmp_handshake_response(ngx_rtmp_session_t *s) p = b->pos + 4; ngx_log_debug5(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, - "RTMP client version=%i.%i.%i.%i epoch=%uD", + "handshake: client version=%i.%i.%i.%i epoch=%uD", (ngx_int_t)p[3], (ngx_int_t)p[2], (ngx_int_t)p[1], (ngx_int_t)p[0], s->peer_epoch); @@ -300,7 +304,7 @@ ngx_rtmp_handshake_response(ngx_rtmp_session_t *s) } if (offs == NGX_ERROR) { ngx_log_error(NGX_LOG_INFO, s->connection->log, 0, - "RTMP digest not found"); + "handshake: digest not found"); return ngx_rtmp_old_handshake_response(s); } b->pos += offs; @@ -311,7 +315,7 @@ ngx_rtmp_handshake_response(ngx_rtmp_session_t *s) return NGX_ERROR; } ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, - "RTMP digest found at pos=%i", offs); + "handshake: digest found at pos=%i", offs); /* create first output buffer */ b = s->hs_bufs[1]; @@ -364,10 +368,12 @@ ngx_rtmp_handshake_make_client_request(ngx_rtmp_session_t *s) b->last = ngx_rtmp_rcpymem(b->last, &s->epoch, 4); b->last = ngx_rtmp_rcpymem(b->last, ngx_rtmp_client_version, 4); ngx_rtmp_fill_random_buffer(b); + ++b->pos; if (ngx_rtmp_write_digest(b, &ngx_rtmp_client_partial_key, 0, s->connection->log) != NGX_OK) { return NGX_ERROR; } + --b->pos; return NGX_OK; } @@ -379,7 +385,7 @@ ngx_rtmp_handshake_done(ngx_rtmp_session_t *s) ngx_rtmp_free_handshake_buffers(s); ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, - "RTMP handshake done"); + "handshake: done"); if (ngx_rtmp_fire_event(s, NGX_RTMP_HANDSHAKE_DONE, NULL, NULL) != NGX_OK) @@ -408,7 +414,8 @@ ngx_rtmp_handshake_recv(ngx_event_t *rev) } if (rev->timedout) { - ngx_log_error(NGX_LOG_INFO, c->log, NGX_ETIMEDOUT, "client timed out"); + ngx_log_error(NGX_LOG_INFO, c->log, NGX_ETIMEDOUT, + "handshake: client timed out"); c->timedout = 1; ngx_rtmp_finalize_session(s); return; @@ -440,20 +447,24 @@ ngx_rtmp_handshake_recv(ngx_event_t *rev) } if (rev->active) { - ngx_del_event(c->read, NGX_READ_EVENT, 0); + ngx_del_event(rev, NGX_READ_EVENT, 0); } - switch (++s->hs_stage) { + ++s->hs_stage; + ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, + "handshake: stage %ui", s->hs_stage); + + switch (s->hs_stage) { case NGX_RTMP_HANDSHAKE_SERVER_SEND_CHALLENGE: s->hs_bufs[1] = ngx_rtmp_alloc_handshake_buffer(s, 0); s->hs_bufs[2] = ngx_rtmp_alloc_handshake_buffer(s, 1); if (ngx_rtmp_handshake_response(s) != NGX_OK) { ngx_log_error(NGX_LOG_INFO, c->log, 0, - "RTMP handshake error"); + "handshake: response error"); ngx_rtmp_finalize_session(s); return; } - s->hs_buf = s->hs_bufs[0]; + s->hs_buf = s->hs_bufs[1]; ngx_rtmp_handshake_send(c->write); break; @@ -470,7 +481,7 @@ ngx_rtmp_handshake_recv(ngx_event_t *rev) case NGX_RTMP_HANDSHAKE_CLIENT_SEND_RESPONSE: if (ngx_rtmp_handshake_client_response(s) != NGX_OK) { ngx_log_error(NGX_LOG_INFO, c->log, 0, - "RTMP client handshake error"); + "handshake: client response error"); ngx_rtmp_finalize_session(s); return; } @@ -498,7 +509,7 @@ ngx_rtmp_handshake_send(ngx_event_t *wev) if (wev->timedout) { ngx_log_error(NGX_LOG_INFO, c->log, NGX_ETIMEDOUT, - "client timed out"); + "handshake: client timed out"); c->timedout = 1; ngx_rtmp_finalize_session(s); return; @@ -529,7 +540,15 @@ ngx_rtmp_handshake_send(ngx_event_t *wev) b->pos += n; } - switch (++s->hs_stage) { + if (wev->active) { + ngx_del_event(wev, NGX_WRITE_EVENT, 0); + } + + ++s->hs_stage; + ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, + "handshake: stage %ui", s->hs_stage); + + switch (s->hs_stage) { case NGX_RTMP_HANDSHAKE_SERVER_SEND_RESPONSE: s->hs_buf = s->hs_bufs[2]; ngx_rtmp_handshake_send(wev); @@ -563,6 +582,9 @@ ngx_rtmp_handshake(ngx_rtmp_session_t *s) c->read->handler = ngx_rtmp_handshake_recv; c->write->handler = ngx_rtmp_handshake_send; + ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, + "handshake: start server handshake"); + s->hs_bufs[0] = ngx_rtmp_alloc_handshake_buffer(s, 0); s->hs_buf = s->hs_bufs[0]; s->hs_stage = NGX_RTMP_HANDSHAKE_SERVER_RECV_CHALLENGE; @@ -580,6 +602,9 @@ ngx_rtmp_client_handshake(ngx_rtmp_session_t *s) c->read->handler = ngx_rtmp_handshake_recv; c->write->handler = ngx_rtmp_handshake_send; + ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, + "handshake: start client handshake"); + s->hs_bufs[0] = ngx_rtmp_alloc_handshake_buffer(s, 0); s->hs_buf = s->hs_bufs[0]; s->hs_stage = NGX_RTMP_HANDSHAKE_CLIENT_SEND_CHALLENGE; @@ -589,6 +614,6 @@ ngx_rtmp_client_handshake(ngx_rtmp_session_t *s) return; } - ngx_rtmp_handshake_send(c->read); + ngx_rtmp_handshake_send(c->write); } diff --git a/ngx_rtmp_relay_module.c b/ngx_rtmp_relay_module.c index cc6258e..c8f85a1 100644 --- a/ngx_rtmp_relay_module.c +++ b/ngx_rtmp_relay_module.c @@ -138,7 +138,6 @@ ngx_rtmp_relay_create_app_conf(ngx_conf_t *cf) } ngx_array_init(&racf->targets, cf->pool, 1, sizeof(ngx_rtmp_relay_target_t)); - racf->nbuckets = 1024; racf->log = &cf->cycle->new_log; @@ -149,6 +148,11 @@ 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 *conf = child; + + conf->ctx = ngx_pcalloc(cf->pool, sizeof(ngx_rtmp_relay_ctx_t *) + * conf->nbuckets); + return NGX_CONF_OK; } @@ -168,7 +172,7 @@ ngx_rtmp_relay_free_peer(ngx_peer_connection_t *pc, void *data, static void -ngx_rtmp_relay_create(ngx_rtmp_relay_ctx_t *ctx, +ngx_rtmp_relay_create(ngx_rtmp_session_t *s, ngx_rtmp_relay_target_t *target) { ngx_rtmp_relay_app_conf_t *racf; @@ -177,12 +181,17 @@ ngx_rtmp_relay_create(ngx_rtmp_relay_ctx_t *ctx, ngx_connection_t *c; ngx_pool_t *pool; ngx_int_t rc; - ngx_rtmp_session_t *s; ngx_rtmp_relay_ctx_t *rctx, **cctx; ngx_uint_t hash; ngx_rtmp_addr_conf_t addr_conf; + ngx_rtmp_conf_ctx_t addr_ctx; + ngx_rtmp_relay_ctx_t *ctx; - racf = ngx_rtmp_get_module_app_conf(ctx->session, ngx_rtmp_relay_module); + 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; + } pool = NULL; pool = ngx_create_pool(4096, racf->log); @@ -241,8 +250,9 @@ ngx_rtmp_relay_create(ngx_rtmp_relay_ctx_t *ctx, c->pool = pool; ngx_memzero(&addr_conf, sizeof(addr_conf)); - addr_conf.ctx->main_conf = s->main_conf; - addr_conf.ctx->srv_conf = s->srv_conf; + addr_conf.ctx = &addr_ctx; + addr_ctx.main_conf = s->main_conf; + addr_ctx.srv_conf = s->srv_conf; ngx_str_set(&addr_conf.addr_text, "ngx-relay"); rs = ngx_rtmp_init_session(c, &addr_conf); @@ -251,11 +261,13 @@ ngx_rtmp_relay_create(ngx_rtmp_relay_ctx_t *ctx, return; } - rs->app_conf = ctx->session->app_conf; + rs->app_conf = s->app_conf; rctx->session = rs; ngx_rtmp_set_ctx(rs, rctx, ngx_rtmp_relay_module); - ngx_rtmp_client_handshake(s); + ngx_str_set(&rs->flashver, "ngx-local-relay"); + + ngx_rtmp_client_handshake(rs); return; @@ -296,8 +308,8 @@ ngx_rtmp_relay_init(ngx_rtmp_session_t *s, u_char *name) ngx_memcpy(ctx->name.data, name, len); /* find relay stream */ - hash = ngx_hash_key(name, len) % racf->nbuckets; - cctx = &racf->ctx[hash]; + hash = ngx_hash_key(name, len); + cctx = &racf->ctx[hash % racf->nbuckets]; for (sctx = *cctx; sctx; sctx = sctx->next) { if (sctx->name.len == len && !ngx_memcmp(name, sctx->name.data, len)) @@ -327,7 +339,7 @@ ngx_rtmp_relay_init(ngx_rtmp_session_t *s, u_char *name) "relay: create: name='%s' url='%V'", name, &target->url); - ngx_rtmp_relay_create(ctx, target); + ngx_rtmp_relay_create(s, target); } } @@ -420,7 +432,7 @@ ngx_rtmp_relay_send_connect(ngx_rtmp_session_t *s) { NGX_RTMP_AMF_STRING, ngx_string("flashVer"), - "ngx-rtmp-relay", 0 } + "ngx-remote-relay", 0 } }; static ngx_rtmp_amf_elt_t out_elts[] = { @@ -502,6 +514,10 @@ ngx_rtmp_relay_send_publish(ngx_rtmp_session_t *s) ngx_null_string, &trans, 0 }, + { NGX_RTMP_AMF_NULL, + ngx_null_string, + NULL, 0 }, + { NGX_RTMP_AMF_STRING, ngx_null_string, NULL, 0 }, /* <- to fill */ @@ -520,8 +536,8 @@ ngx_rtmp_relay_send_publish(ngx_rtmp_session_t *s) return NGX_ERROR; } - out_elts[2].data = ctx->name.data; - out_elts[2].len = ctx->name.len; + out_elts[3].data = ctx->name.data; + out_elts[3].len = ctx->name.len; ngx_memzero(&h, sizeof(h)); h.csid = NGX_RTMP_RELAY_CSID_AMF; @@ -547,6 +563,10 @@ ngx_rtmp_relay_send_play(ngx_rtmp_session_t *s) ngx_null_string, &trans, 0 }, + { NGX_RTMP_AMF_NULL, + ngx_null_string, + NULL, 0 }, + { NGX_RTMP_AMF_STRING, ngx_null_string, NULL, 0 }, /* <- to fill */ @@ -561,8 +581,8 @@ ngx_rtmp_relay_send_play(ngx_rtmp_session_t *s) return NGX_ERROR; } - out_elts[2].data = ctx->name.data; - out_elts[2].len = ctx->name.len; + out_elts[3].data = ctx->name.data; + out_elts[3].len = ctx->name.len; ngx_memzero(&h, sizeof(h)); h.csid = NGX_RTMP_RELAY_CSID_AMF; diff --git a/stat.xsl b/stat.xsl index b171ec1..6b4f084 100644 --- a/stat.xsl +++ b/stat.xsl @@ -71,7 +71,7 @@ - var d=document.getElementById(''); + var d=document.getElementById('-'); d.style.display=d.style.display=='none'?'':'none'; return false @@ -91,7 +91,7 @@ - + - diff --git a/test/nginx.conf b/test/nginx.conf index 542b3eb..3f57d17 100644 --- a/test/nginx.conf +++ b/test/nginx.conf @@ -44,6 +44,11 @@ rtmp { on_record_done http://localhost:8080/record_done; } + application mypull { + live on; + pull myapp mystream localhost; + } + } }