From 49382c826b041589badb32b5f5f512592405a736 Mon Sep 17 00:00:00 2001 From: Roman Arutyunyan Date: Sat, 26 May 2012 08:33:41 +0400 Subject: [PATCH] implemented RTMP ping --- ngx_rtmp.h | 7 +++++ ngx_rtmp_core_module.c | 18 +++++++++++ ngx_rtmp_handler.c | 69 +++++++++++++++++++++++++++++++++++++++--- ngx_rtmp_init.c | 4 +++ ngx_rtmp_receive.c | 1 + 5 files changed, 95 insertions(+), 4 deletions(-) diff --git a/ngx_rtmp.h b/ngx_rtmp.h index 2855ee0..648f629 100644 --- a/ngx_rtmp.h +++ b/ngx_rtmp.h @@ -203,6 +203,10 @@ typedef struct { ngx_msec_t epoch; ngx_msec_t peer_epoch; + /* ping */ + ngx_event_t ping_evt; + ngx_int_t ping_pending; + /* input stream 0 (reserved by RTMP spec) * is used as free chain link */ @@ -263,6 +267,8 @@ typedef struct ngx_rtmp_core_srv_conf_s { ngx_array_t applications; /* ngx_rtmp_core_app_conf_t */ ngx_msec_t timeout; + ngx_msec_t ping; + ngx_msec_t ping_timeout; ngx_flag_t so_keepalive; ngx_int_t max_streams; @@ -353,6 +359,7 @@ void ngx_rtmp_handshake(ngx_rtmp_session_t *s); void ngx_rtmp_client_handshake(ngx_rtmp_session_t *s, unsigned async); void ngx_rtmp_free_handshake_buffers(ngx_rtmp_session_t *s); void ngx_rtmp_cycle(ngx_rtmp_session_t *s); +void ngx_rtmp_reset_ping(ngx_rtmp_session_t *s); ngx_int_t ngx_rtmp_fire_event(ngx_rtmp_session_t *s, ngx_uint_t evt, ngx_rtmp_header_t *h, ngx_chain_t *in); diff --git a/ngx_rtmp_core_module.c b/ngx_rtmp_core_module.c index ffb296c..a82f62a 100644 --- a/ngx_rtmp_core_module.c +++ b/ngx_rtmp_core_module.c @@ -71,6 +71,20 @@ static ngx_command_t ngx_rtmp_core_commands[] = { offsetof(ngx_rtmp_core_srv_conf_t, timeout), NULL }, + { ngx_string("ping"), + NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_CONF_TAKE1, + ngx_conf_set_msec_slot, + NGX_RTMP_SRV_CONF_OFFSET, + offsetof(ngx_rtmp_core_srv_conf_t, ping), + NULL }, + + { ngx_string("ping_timeout"), + NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_CONF_TAKE1, + ngx_conf_set_msec_slot, + NGX_RTMP_SRV_CONF_OFFSET, + offsetof(ngx_rtmp_core_srv_conf_t, ping_timeout), + NULL }, + { ngx_string("max_streams"), NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_CONF_TAKE1, ngx_conf_set_num_slot, @@ -200,6 +214,8 @@ ngx_rtmp_core_create_srv_conf(ngx_conf_t *cf) } conf->timeout = NGX_CONF_UNSET_MSEC; + conf->ping = NGX_CONF_UNSET_MSEC; + conf->ping_timeout = NGX_CONF_UNSET_MSEC; conf->so_keepalive = NGX_CONF_UNSET; conf->max_streams = NGX_CONF_UNSET; conf->chunk_size = NGX_CONF_UNSET; @@ -220,6 +236,8 @@ ngx_rtmp_core_merge_srv_conf(ngx_conf_t *cf, void *parent, void *child) ngx_rtmp_core_srv_conf_t *conf = child; ngx_conf_merge_msec_value(conf->timeout, prev->timeout, 60000); + ngx_conf_merge_msec_value(conf->ping, prev->ping, 0); + ngx_conf_merge_msec_value(conf->ping_timeout, prev->ping_timeout, 30000); ngx_conf_merge_value(conf->so_keepalive, prev->so_keepalive, 0); ngx_conf_merge_value(conf->max_streams, prev->max_streams, 32); diff --git a/ngx_rtmp_handler.c b/ngx_rtmp_handler.c index a398583..367060a 100644 --- a/ngx_rtmp_handler.c +++ b/ngx_rtmp_handler.c @@ -9,6 +9,7 @@ static void ngx_rtmp_recv(ngx_event_t *rev); static void ngx_rtmp_send(ngx_event_t *rev); +static void ngx_rtmp_ping(ngx_event_t *rev); static ngx_int_t ngx_rtmp_receive_message(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h, ngx_chain_t *in); static ngx_int_t ngx_rtmp_finalize_set_chunk_size(ngx_rtmp_session_t *s); @@ -78,11 +79,19 @@ void ngx_rtmp_cycle(ngx_rtmp_session_t *s) { ngx_connection_t *c; + ngx_rtmp_core_srv_conf_t *cscf; + + cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module); c = s->connection; c->read->handler = ngx_rtmp_recv; c->write->handler = ngx_rtmp_send; + s->ping_evt.data = c; + s->ping_evt.log = c->log; + s->ping_evt.handler = ngx_rtmp_ping; + ngx_rtmp_reset_ping(s); + ngx_rtmp_recv(c->read); } @@ -114,6 +123,58 @@ ngx_rtmp_alloc_in_buf(ngx_rtmp_session_t *s) } +void +ngx_rtmp_reset_ping(ngx_rtmp_session_t *s) +{ + ngx_rtmp_core_srv_conf_t *cscf; + + cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module); + if (cscf->ping == 0) { + return; + } + + s->ping_pending = 0; + ngx_add_timer(&s->ping_evt, cscf->ping); + + ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, + "ping: wait %Mms", cscf->ping); +} + + +static void +ngx_rtmp_ping(ngx_event_t *pev) +{ + ngx_connection_t *c; + ngx_rtmp_session_t *s; + ngx_rtmp_core_srv_conf_t *cscf; + + c = pev->data; + s = c->data; + + cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module); + + if (s->ping_pending) { + ngx_log_error(NGX_LOG_INFO, c->log, 0, + "ping: unresponded"); + ngx_rtmp_finalize_session(s); + return; + } + + ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, + "ping: schedule %Mms", cscf->ping_timeout); + + if (ngx_rtmp_send_user_ping_request(s, (uint32_t)ngx_current_msec) + != NGX_OK) + { + ngx_rtmp_finalize_session(s); + return; + } + + s->ping_pending = 1; + ngx_add_timer(pev, cscf->ping_timeout); +} + + static void ngx_rtmp_recv(ngx_event_t *rev) { @@ -149,7 +210,7 @@ ngx_rtmp_recv(ngx_event_t *rev) if (st->in == NULL) { st->in = ngx_rtmp_alloc_in_buf(s); if (st->in == NULL) { - ngx_log_error(NGX_LOG_INFO, c->log, NGX_ERROR, + ngx_log_error(NGX_LOG_INFO, c->log, 0, "in buf alloc failed"); ngx_rtmp_finalize_session(s); return; @@ -240,7 +301,7 @@ ngx_rtmp_recv(ngx_event_t *rev) (int)fmt, csid); if (csid >= (uint32_t)cscf->max_streams) { - ngx_log_error(NGX_LOG_INFO, c->log, NGX_ERROR, + ngx_log_error(NGX_LOG_INFO, c->log, 0, "RTMP in chunk stream too big: %D >= %D", csid, cscf->max_streams); ngx_rtmp_finalize_session(s); @@ -344,7 +405,7 @@ ngx_rtmp_recv(ngx_event_t *rev) b->pos = p; if (h->mlen > cscf->max_message) { - ngx_log_error(NGX_LOG_INFO, c->log, NGX_ERROR, + ngx_log_error(NGX_LOG_INFO, c->log, 0, "too big message: %uz", cscf->max_message); ngx_rtmp_finalize_session(s); return; @@ -491,7 +552,7 @@ ngx_rtmp_prepare_message(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h, cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module); if (h->csid >= (uint32_t)cscf->max_streams) { - ngx_log_error(NGX_LOG_INFO, c->log, NGX_ERROR, + ngx_log_error(NGX_LOG_INFO, c->log, 0, "RTMP out chunk stream too big: %D >= %D", h->csid, cscf->max_streams); ngx_rtmp_finalize_session(s); diff --git a/ngx_rtmp_init.c b/ngx_rtmp_init.c index e3e40b9..988b1ec 100644 --- a/ngx_rtmp_init.c +++ b/ngx_rtmp_init.c @@ -248,6 +248,10 @@ ngx_rtmp_close_session_handler(ngx_event_t *e) if (s) { ngx_rtmp_fire_event(s, NGX_RTMP_DISCONNECT, NULL, NULL); + if (s->ping_evt.timer_set) { + ngx_del_timer(&s->ping_evt); + } + if (s->in_old_pool) { ngx_destroy_pool(s->in_old_pool); } diff --git a/ngx_rtmp_receive.c b/ngx_rtmp_receive.c index 1994f06..845822f 100644 --- a/ngx_rtmp_receive.c +++ b/ngx_rtmp_receive.c @@ -153,6 +153,7 @@ ngx_rtmp_user_message_handler(ngx_rtmp_session_t *s, case NGX_RTMP_USER_PING_RESPONSE: /* use =val as incoming timestamp */ + ngx_rtmp_reset_ping(s); break; default: