zotan/nginx-mod-rtmp
1
1
Fork 0
mirror of https://github.com/zotanmew/nginx-rtmp-module.git synced 2024-05-14 16:01:09 +02:00
Code Issues Projects Releases Wiki Activity

merged vod-seek

Browse source
This commit is contained in:
Roman Arutyunyan 2012-10-24 17:26:20 +04:00
parent a4c790e0ee 5824635938
commit 20b55b394a
23 changed files with 1371 additions and 150 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
TODO
View file

@ -1,17 +1,12 @@
- Auto-pushing pulled stream
- option to start live streaming with a keyframe
- Pull secondary address support
- interleaved mode
- Binary search in play module
- vod seamless pause
- More Wiki docs
- DNS round-robin url
Style:
======
- a/v merge from different sources
- Move out & merge stream ids from live & cmd modules
- Clean cmd module:
* make shortcuts for status messages
* move protocol message sending to live/play modules
- multiple streams perconnection

18
config
View file

@ -22,6 +22,23 @@ CORE_MODULES="$CORE_MODULES
HTTP_MODULES="$HTTP_MODULES \
ngx_rtmp_stat_module \
ngx_rtmp_control_module \
"
NGX_ADDON_DEPS="$NGX_ADDON_DEPS \
$ngx_addon_dir/ngx_rtmp_amf.h \
$ngx_addon_dir/ngx_rtmp_bandwidth.h \
$ngx_addon_dir/ngx_rtmp_cmd_module.h \
$ngx_addon_dir/ngx_rtmp_codec_module.h \
$ngx_addon_dir/ngx_rtmp_eval.h \
$ngx_addon_dir/ngx_rtmp.h \
$ngx_addon_dir/ngx_rtmp_live_module.h \
$ngx_addon_dir/ngx_rtmp_netcall_module.h \
$ngx_addon_dir/ngx_rtmp_play_module.h \
$ngx_addon_dir/ngx_rtmp_record_module.h \
$ngx_addon_dir/ngx_rtmp_relay_module.h \
$ngx_addon_dir/ngx_rtmp_streams.h \
"
@ -45,6 +62,7 @@ NGX_ADDON_SRCS="$NGX_ADDON_SRCS \
$ngx_addon_dir/ngx_rtmp_mp4_module.c \
$ngx_addon_dir/ngx_rtmp_netcall_module.c \
$ngx_addon_dir/ngx_rtmp_stat_module.c \
$ngx_addon_dir/ngx_rtmp_control_module.c \
$ngx_addon_dir/ngx_rtmp_relay_module.c \
$ngx_addon_dir/ngx_rtmp_bandwidth.c \
$ngx_addon_dir/ngx_rtmp_exec_module.c \

53
hls/ngx_rtmp_hls_module.c
View file

@ -73,6 +73,9 @@ typedef struct {
ngx_int_t out_astream;
int8_t nal_bytes;
int64_t aframe_base;
int64_t aframe_num;
AVFormatContext *out_format;
} ngx_rtmp_hls_ctx_t;
@ -82,6 +85,7 @@ typedef struct {
ngx_flag_t hls;
ngx_msec_t fraglen;
ngx_msec_t muxdelay;
ngx_msec_t sync;
ngx_msec_t playlen;
size_t nfrags;
ngx_rtmp_hls_ctx_t **ctx;
@ -127,6 +131,13 @@ static ngx_command_t ngx_rtmp_hls_commands[] = {
offsetof(ngx_rtmp_hls_app_conf_t, muxdelay),
NULL },
{ ngx_string("hls_sync"),
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_hls_app_conf_t, sync),
NULL },
ngx_null_command
};
@ -909,6 +920,7 @@ ngx_rtmp_hls_audio(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_rtmp_hls_ctx_t *ctx;
ngx_rtmp_codec_ctx_t *codec_ctx;
AVPacket packet;
int64_t dts, ddts;
static u_char buffer[NGX_RTMP_HLS_BUFSIZE];
hacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_hls_module);
@ -935,11 +947,41 @@ ngx_rtmp_hls_audio(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
/* write to file */
av_init_packet(&packet);
packet.dts = h->timestamp * 90L;
packet.pts = packet.dts;
packet.stream_index = ctx->out_astream;
packet.data = buffer;
packet.size = ngx_rtmp_hls_chain2buffer(buffer, sizeof(buffer), in, 1);
if (hacf->sync && codec_ctx->sample_rate) {
/* TODO: We assume here AAC frame size is 1024
* Need to handle AAC frames with frame size of 960 */
dts = ctx->aframe_base + ctx->aframe_num * 90000 * 1024 /
codec_ctx->sample_rate;
ddts = dts - packet.dts;
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"hls: sync stat ddts=%L (%.5fs)",
ddts, ddts / 90000.);
if (ddts > (int64_t) hacf->sync * 90 ||
ddts < (int64_t) hacf->sync * -90)
{
ctx->aframe_base = packet.dts;
ctx->aframe_num = 0;
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"hls: sync breakup ddts=%L (%.5fs)",
ddts, ddts / 90000.);
} else {
packet.dts = dts;
}
ctx->aframe_num++;
}
packet.pts = packet.dts;
if (codec_ctx->audio_codec_id == NGX_RTMP_AUDIO_AAC) {
if (packet.size == 0) {
ngx_log_error(NGX_LOG_ERR, s->connection->log, 0,
@ -950,8 +992,9 @@ ngx_rtmp_hls_audio(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
--packet.size;
}
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"hls: audio buffer %uD", *(uint32_t*)packet.data);
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"hls: audio dts=%L timestamp=%uD", (int64_t)packet.dts,
h->timestamp);
if (av_interleaved_write_frame(ctx->out_format, &packet) < 0) {
ngx_log_error(NGX_LOG_ERR, s->connection->log, 0,
@ -1119,10 +1162,11 @@ ngx_rtmp_hls_create_app_conf(ngx_conf_t *cf)
conf->hls = NGX_CONF_UNSET;
conf->fraglen = NGX_CONF_UNSET;
conf->muxdelay = NGX_CONF_UNSET;
conf->sync = NGX_CONF_UNSET;
conf->playlen = NGX_CONF_UNSET;
conf->nbuckets = 1024;
return conf;
return conf;
}
@ -1135,6 +1179,7 @@ ngx_rtmp_hls_merge_app_conf(ngx_conf_t *cf, void *parent, void *child)
ngx_conf_merge_value(conf->hls, prev->hls, 0);
ngx_conf_merge_msec_value(conf->fraglen, prev->fraglen, 5000);
ngx_conf_merge_msec_value(conf->muxdelay, prev->muxdelay, 700);
ngx_conf_merge_msec_value(conf->sync, prev->sync, 0);
ngx_conf_merge_msec_value(conf->playlen, prev->playlen, 30000);
ngx_conf_merge_str_value(conf->path, prev->path, "");
conf->ctx = ngx_pcalloc(cf->pool,

3
ngx_rtmp.h
View file

@ -151,6 +151,7 @@ typedef struct {
typedef struct {
uint32_t csid; /* chunk stream id */
uint32_t timestamp; /* timestamp (delta) */
uint32_t timeshift; /* clock - timestamp */
uint32_t mlen; /* message length */
uint8_t type; /* message type id */
uint32_t msid; /* message stream id */
@ -201,6 +202,7 @@ typedef struct {
/* connection timestamps */
ngx_msec_t epoch;
ngx_msec_t peer_epoch;
ngx_msec_t base_time;
/* ping */
ngx_event_t ping_evt;
@ -287,6 +289,7 @@ typedef struct ngx_rtmp_core_srv_conf_s {
size_t max_message;
ngx_flag_t play_time_fix;
ngx_flag_t publish_time_fix;
ngx_flag_t busy;
size_t out_queue;
size_t out_cork;

32
ngx_rtmp_cmd_module.c
View file

@ -214,7 +214,7 @@ ngx_rtmp_cmd_connect(ngx_rtmp_session_t *s, ngx_rtmp_connect_t *v)
s->connected = 1;
ngx_memzero(&h, sizeof(h));
h.csid = NGX_RTMP_CMD_CSID_AMF_INI;
h.csid = NGX_RTMP_CSID_AMF_INI;
h.type = NGX_RTMP_MSG_AMF_CMD;
@ -321,10 +321,10 @@ ngx_rtmp_cmd_create_stream(ngx_rtmp_session_t *s, ngx_rtmp_create_stream_t *v)
};
trans = v->trans;
stream = NGX_RTMP_CMD_MSID;
stream = NGX_RTMP_MSID;
ngx_memzero(&h, sizeof(h));
h.csid = NGX_RTMP_CMD_CSID_AMF_INI;
h.csid = NGX_RTMP_CSID_AMF_INI;
h.type = NGX_RTMP_MSG_AMF_CMD;
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
@ -369,7 +369,7 @@ ngx_rtmp_cmd_close_stream_init(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
static ngx_int_t
ngx_rtmp_cmd_close_stream(ngx_rtmp_session_t *s, ngx_rtmp_close_stream_t *v)
{
ngx_rtmp_send_user_stream_eof(s, NGX_RTMP_CMD_MSID);
ngx_rtmp_send_user_stream_eof(s, NGX_RTMP_MSID);
/* Whatever happens return OK
* since we should be careful with destruction */
@ -537,8 +537,8 @@ ngx_rtmp_cmd_publish(ngx_rtmp_session_t *s, ngx_rtmp_publish_t *v)
/* send onStatus reply */
memset(&h, 0, sizeof(h));
h.type = NGX_RTMP_MSG_AMF_CMD;
h.csid = NGX_RTMP_CMD_CSID_AMF;
h.msid = NGX_RTMP_CMD_MSID;
h.csid = NGX_RTMP_CSID_AMF;
h.msid = NGX_RTMP_MSID;
if (ngx_rtmp_send_amf(s, &h, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) != NGX_OK)
@ -630,8 +630,8 @@ ngx_rtmp_cmd_fcpublish(ngx_rtmp_session_t *s, ngx_rtmp_fcpublish_t *v)
/* send onFCPublish reply */
memset(&h, 0, sizeof(h));
h.type = NGX_RTMP_MSG_AMF_CMD;
h.csid = NGX_RTMP_CMD_CSID_AMF;
h.msid = NGX_RTMP_CMD_MSID;
h.csid = NGX_RTMP_CSID_AMF;
h.msid = NGX_RTMP_MSID;
if (ngx_rtmp_send_amf(s, &h, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) != NGX_OK)
@ -818,15 +818,15 @@ ngx_rtmp_cmd_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
/* send onStatus reply */
memset(&h, 0, sizeof(h));
h.type = NGX_RTMP_MSG_AMF_CMD;
h.csid = NGX_RTMP_CMD_CSID_AMF;
h.msid = NGX_RTMP_CMD_MSID;
h.csid = NGX_RTMP_CSID_AMF;
h.msid = NGX_RTMP_MSID;
/*
if (ngx_rtmp_send_user_recorded(s, NGX_RTMP_CMD_MSID) != NGX_OK) {
if (ngx_rtmp_send_user_recorded(s, NGX_RTMP_MSID) != NGX_OK) {
return NGX_ERROR;
}*/
if (ngx_rtmp_send_user_stream_begin(s, NGX_RTMP_CMD_MSID) != NGX_OK) {
if (ngx_rtmp_send_user_stream_begin(s, NGX_RTMP_MSID) != NGX_OK) {
return NGX_ERROR;
}
@ -943,8 +943,8 @@ ngx_rtmp_cmd_fcsubscribe(ngx_rtmp_session_t *s, ngx_rtmp_fcsubscribe_t *v)
/* send onFCSubscribe reply */
memset(&h, 0, sizeof(h));
h.type = NGX_RTMP_MSG_AMF_CMD;
h.csid = NGX_RTMP_CMD_CSID_AMF;
h.msid = NGX_RTMP_CMD_MSID;
h.csid = NGX_RTMP_CSID_AMF;
h.msid = NGX_RTMP_MSID;
if (ngx_rtmp_send_amf(s, &h, out_elts,
sizeof(out_elts) / sizeof(out_elts[0])) != NGX_OK)
@ -1073,8 +1073,8 @@ ngx_rtmp_cmd_seek(ngx_rtmp_session_t *s, ngx_rtmp_seek_t *v)
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
"seek: offset=%i", (ngx_int_t) v->offset);
return (ngx_rtmp_send_user_stream_eof(s, NGX_RTMP_CMD_MSID) != NGX_OK
|| ngx_rtmp_send_user_stream_begin(s, NGX_RTMP_CMD_MSID) != NGX_OK
return (ngx_rtmp_send_user_stream_eof(s, NGX_RTMP_MSID) != NGX_OK
|| ngx_rtmp_send_user_stream_begin(s, NGX_RTMP_MSID) != NGX_OK
|| ngx_rtmp_send_status(s, "NetStream.Seek.Notify", "status",
"Seeking"))
? NGX_ERROR

74
ngx_rtmp_codec_module.c
View file

@ -158,10 +158,16 @@ ngx_rtmp_codec_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_chain_t **header, **pheader;
uint8_t fmt;
ngx_rtmp_header_t ch, lh;
ngx_uint_t *version;
ngx_uint_t *version, idx;
u_char *p;
static ngx_uint_t sample_rates[] =
{ 5512, 11025, 22050, 44100 };
static ngx_uint_t aac_sample_rates[] =
{ 96000, 88200, 64000, 48000,
44100, 32000, 24000, 22050,
16000, 12000, 11025, 8000,
7350, 0, 0, 0 };
if (h->type != NGX_RTMP_MSG_AUDIO && h->type != NGX_RTMP_MSG_VIDEO) {
return NGX_OK;
@ -183,7 +189,10 @@ ngx_rtmp_codec_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ctx->audio_codec_id = (fmt & 0xf0) >> 4;
ctx->audio_channels = (fmt & 0x01) + 1;
ctx->sample_size = (fmt & 0x02) ? 2 : 1;
ctx->sample_rate = sample_rates[(fmt & 0x0c) >> 2];
if (ctx->aac_sample_rate == 0) {
ctx->sample_rate = sample_rates[(fmt & 0x0c) >> 2];
}
} else {
ctx->video_codec_id = (fmt & 0x0f);
}
@ -207,8 +216,55 @@ ngx_rtmp_codec_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
header = &ctx->aac_header;
pheader = &ctx->aac_pheader;
version = &ctx->aac_version;
ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"codec: AAC header arrived");
if (in->buf->last - in->buf->pos > 3) {
p = in->buf->pos + 2;
/* MPEG-4 Audio Specific Config
5 bits: object type
if (object type == 31)
6 bits + 32: object type
--->4 bits: frequency index
if (frequency index == 15)
24 bits: frequency
4 bits: channel configuration
var bits: AOT Specific Config
*/
if ((p[0] >> 3) == 0x1f) {
idx = (p[1] >> 1) & 0x0f;
} else {
idx = ((p[0] << 1) & 0x0f) | (p[1] >> 7);
}
#ifdef NGX_DEBUG
{
u_char buf[256], *p, *pp;
u_char hex[] = "01234567890abcdef";
for (pp = buf, p = in->buf->pos;
p < in->buf->last && pp < buf + sizeof(buf) - 1;
++p)
{
*pp++ = hex[*p >> 4];
*pp++ = hex[*p & 0x0f];
}
*pp = 0;
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"codec: AAC header: %s", buf);
}
#endif
ctx->aac_sample_rate = aac_sample_rates[idx];
ctx->sample_rate = ctx->aac_sample_rate;
}
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"codec: AAC header arrived, sample_rate=%ui",
ctx->aac_sample_rate);
}
} else {
if (ctx->video_codec_id == NGX_RTMP_VIDEO_H264) {
@ -234,11 +290,11 @@ ngx_rtmp_codec_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
/* equal headers; timeout diff is zero */
ngx_memzero(&ch, sizeof(ch));
ch.msid = NGX_RTMP_LIVE_MSID;
ch.msid = NGX_RTMP_MSID;
ch.type = h->type;
ch.csid = (h->type == NGX_RTMP_MSG_VIDEO
? NGX_RTMP_LIVE_CSID_VIDEO
: NGX_RTMP_LIVE_CSID_AUDIO);
? NGX_RTMP_CSID_VIDEO
: NGX_RTMP_CSID_AUDIO);
lh = ch;
*header = ngx_rtmp_append_shared_bufs(cscf, NULL, in);
*pheader = ngx_rtmp_append_shared_bufs(cscf, NULL, in);
@ -372,8 +428,8 @@ ngx_rtmp_codec_update_meta(ngx_rtmp_session_t *s)
}
ngx_memzero(&h, sizeof(h));
h.csid = NGX_RTMP_LIVE_CSID_META;
h.msid = NGX_RTMP_LIVE_MSID;
h.csid = NGX_RTMP_CSID_AMF;
h.msid = NGX_RTMP_MSID;
h.type = NGX_RTMP_MSG_AMF_META;
ngx_rtmp_prepare_message(s, &h, NULL, ctx->meta);

1
ngx_rtmp_codec_module.h
View file

@ -55,6 +55,7 @@ typedef struct {
ngx_uint_t video_codec_id;
ngx_uint_t audio_data_rate;
ngx_uint_t audio_codec_id;
ngx_uint_t aac_sample_rate;
ngx_uint_t sample_rate; /* 5512, 11025, 22050, 44100 */
ngx_uint_t sample_size; /* 1=8bit, 2=16bit */
ngx_uint_t audio_channels; /* 1, 2 */

529
ngx_rtmp_control_module.c Normal file
View file

@ -0,0 +1,529 @@
/*
* Copyright (c) 2012 Roman Arutyunyan
*/
#include <nginx.h>
#include <ngx_http.h>
#include "ngx_rtmp.h"
#include "ngx_rtmp_live_module.h"
#include "ngx_rtmp_record_module.h"
static ngx_int_t ngx_rtmp_control_postconfiguration(ngx_conf_t *cf);
static void * ngx_rtmp_control_create_loc_conf(ngx_conf_t *cf);
static char * ngx_rtmp_control_merge_loc_conf(ngx_conf_t *cf,
void *parent, void *child);
typedef struct {
ngx_rtmp_core_main_conf_t *cmcf;
ngx_rtmp_core_srv_conf_t *cscf;
ngx_rtmp_core_app_conf_t *cacf;
} ngx_rtmp_control_core_t;
typedef struct {
ngx_rtmp_live_app_conf_t *lacf;
ngx_rtmp_live_stream_t *ls;
} ngx_rtmp_control_live_t;
#define NGX_RTMP_CONTROL_ALL 0xff
#define NGX_RTMP_CONTROL_RECORD 0x01
#define NGX_RTMP_CONTROL_DROP 0x02
typedef struct {
ngx_uint_t control;
} ngx_rtmp_control_loc_conf_t;
static ngx_conf_bitmask_t ngx_rtmp_control_masks[] = {
{ ngx_string("all"), NGX_RTMP_CONTROL_ALL },
{ ngx_string("record"), NGX_RTMP_CONTROL_RECORD },
{ ngx_string("drop"), NGX_RTMP_CONTROL_DROP },
{ ngx_null_string, 0 }
};
static ngx_command_t ngx_rtmp_control_commands[] = {
{ ngx_string("rtmp_control"),
NGX_HTTP_MAIN_CONF|NGX_HTTP_SRV_CONF|NGX_HTTP_LOC_CONF|NGX_CONF_1MORE,
ngx_conf_set_bitmask_slot,
NGX_HTTP_LOC_CONF_OFFSET,
offsetof(ngx_rtmp_control_loc_conf_t, control),
ngx_rtmp_control_masks },
ngx_null_command
};
static ngx_http_module_t ngx_rtmp_control_module_ctx = {
NULL, /* preconfiguration */
ngx_rtmp_control_postconfiguration, /* postconfiguration */
NULL, /* create main configuration */
NULL, /* init main configuration */
NULL, /* create server configuration */
NULL, /* merge server configuration */
ngx_rtmp_control_create_loc_conf, /* create location configuration */
ngx_rtmp_control_merge_loc_conf, /* merge location configuration */
};
ngx_module_t ngx_rtmp_control_module = {
NGX_MODULE_V1,
&ngx_rtmp_control_module_ctx, /* module context */
ngx_rtmp_control_commands, /* module directives */
NGX_HTTP_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
static ngx_int_t
ngx_rtmp_control_output_error(ngx_http_request_t *r, const char *msg)
{
size_t len;
ngx_buf_t *b;
ngx_chain_t cl;
len = ngx_strlen(msg);
r->headers_out.status = NGX_HTTP_BAD_REQUEST;
r->headers_out.content_length_n = len;
b = ngx_calloc_buf(r->pool);
if (b == NULL) {
return NGX_ERROR;
}
ngx_memzero(&cl, sizeof(cl));
cl.buf = b;
b->start = b->pos = (u_char *) msg;
b->end = b->last = (u_char *) msg + len;
b->memory = 1;
b->last_buf = 1;
ngx_http_send_header(r);
return ngx_http_output_filter(r, &cl);
}
static const char *
ngx_rtmp_control_parse_core(ngx_http_request_t *r,
ngx_rtmp_control_core_t *core)
{
ngx_str_t srv, app;
ngx_uint_t sn, n;
ngx_rtmp_core_srv_conf_t **pcscf;
ngx_rtmp_core_app_conf_t **pcacf;
core->cmcf = ngx_rtmp_core_main_conf;
if (core->cmcf == NULL) {
return "Missing main RTMP conf";
}
/* find server */
sn = 0;
if (ngx_http_arg(r, (u_char *) "srv", sizeof("srv") - 1, &srv) == NGX_OK) {
sn = ngx_atoi(srv.data, srv.len);
}
if (sn >= core->cmcf->servers.nelts) {
return "Server index out of range";
}
pcscf = core->cmcf->servers.elts;
pcscf += sn;
core->cscf = *pcscf;
/* find application */
if (ngx_http_arg(r, (u_char *) "app", sizeof("app") - 1, &app) != NGX_OK) {
ngx_log_error(NGX_LOG_ALERT, r->connection->log, 0,
"rtmp_control: app not specified");
return "Application not specified";
}
core->cacf = NULL;
pcacf = core->cscf->applications.elts;
for (n = 0; n < core->cscf->applications.nelts; ++n, ++pcacf) {
if ((*pcacf)->name.len == app.len &&
ngx_strncmp((*pcacf)->name.data, app.data, app.len) == 0)
{
core->cacf = *pcacf;
break;
}
}
if (core->cacf == NULL) {
return "Application not found";
}
return NGX_CONF_OK;
}
static const char *
ngx_rtmp_control_parse_live(ngx_http_request_t *r,
ngx_rtmp_control_core_t *core,
ngx_rtmp_control_live_t *live)
{
ngx_str_t name;
size_t len;
ngx_memzero(&name, sizeof(name));
ngx_http_arg(r, (u_char *) "name", sizeof("name") - 1, &name);
live->lacf = core->cacf->app_conf[ngx_rtmp_live_module.ctx_index];
/* find live stream by name */
for (live->ls = live->lacf->streams[ngx_hash_key(name.data, name.len) %
live->lacf->nbuckets];
live->ls; live->ls = live->ls->next)
{
len = ngx_strlen(live->ls->name);
if (name.len == len && ngx_strncmp(name.data, live->ls->name, name.len)
== 0)
{
break;
}
}
if (live->ls == NULL) {
return "Live stream not found";
}
return NGX_CONF_OK;
}
/* /record arguments:
* srv - server index (optional)
* app - application name
* name - stream name
* rec - recorder name
*/
static ngx_int_t
ngx_rtmp_control_record(ngx_http_request_t *r, ngx_str_t *method)
{
ngx_rtmp_control_core_t core;
ngx_rtmp_control_live_t live;
ngx_rtmp_record_app_conf_t *racf;
ngx_rtmp_live_ctx_t *lctx;
ngx_rtmp_session_t *s;
ngx_chain_t cl;
ngx_uint_t rn;
ngx_str_t rec, path;
ngx_buf_t *b;
ngx_int_t rc;
const char *msg;
msg = ngx_rtmp_control_parse_core(r, &core);
if (msg != NGX_CONF_OK) {
goto error;
}
msg = ngx_rtmp_control_parse_live(r, &core, &live);
if (msg != NGX_CONF_OK) {
goto error;
}
/* find publisher context */
for (lctx = live.ls->ctx; lctx; lctx = lctx->next) {
if (lctx->flags & NGX_RTMP_LIVE_PUBLISHING) {
break;
}
}
if (lctx == NULL) {
msg = "No publisher";
goto error;
}
s = lctx->session;
/* find recorder */
ngx_memzero(&rec, sizeof(rec));
ngx_http_arg(r, (u_char *) "rec", sizeof("rec") - 1, &rec);
racf = core.cacf->app_conf[ngx_rtmp_record_module.ctx_index];
rn = ngx_rtmp_record_find(racf, &rec);
if (rn == NGX_CONF_UNSET_UINT) {
msg = "Recorder not found";
goto error;
}
/* call the method */
ngx_memzero(&path, sizeof(path));
if (method->len == sizeof("start") - 1 &&
ngx_strncmp(method->data, "start", method->len) == 0)
{
rc = ngx_rtmp_record_open(s, rn, &path);
} else if (method->len == sizeof("stop") - 1 &&
ngx_strncmp(method->data, "stop", method->len) == 0)
{
rc = ngx_rtmp_record_close(s, rn, &path);
} else {
msg = "Undefined method";
goto error;
}
if (rc == NGX_ERROR) {
msg = "Recorder error";
goto error;
}
if (rc == NGX_AGAIN) {
/* already opened/closed */
ngx_str_null(&path);
r->header_only = 1;
}
r->headers_out.status = NGX_HTTP_OK;
r->headers_out.content_length_n = path.len;
b = ngx_create_temp_buf(r->pool, path.len);
if (b == NULL) {
return NGX_ERROR;
}
ngx_memzero(&cl, sizeof(cl));
cl.buf = b;
b->last = ngx_cpymem(b->pos, path.data, path.len);
b->last_buf = 1;
ngx_http_send_header(r);
return ngx_http_output_filter(r, &cl);
error:
return ngx_rtmp_control_output_error(r, msg);
}
static ngx_int_t
ngx_rtmp_control_drop(ngx_http_request_t *r, ngx_str_t *method)
{
ngx_rtmp_control_core_t core;
ngx_rtmp_control_live_t live;
ngx_rtmp_live_ctx_t *lctx;
ngx_str_t addr, *paddr;
const char *msg;
ngx_uint_t ndropped;
size_t len;
u_char *p;
ngx_buf_t *b;
ngx_chain_t cl;
msg = ngx_rtmp_control_parse_core(r, &core);
if (msg != NGX_CONF_OK) {
goto error;
}
msg = ngx_rtmp_control_parse_live(r, &core, &live);
if (msg != NGX_CONF_OK) {
goto error;
}
ndropped = 0;
if (method->len == sizeof("publisher") - 1 &&
ngx_strncmp(method->data, "publisher", method->len) == 0)
{
for (lctx = live.ls->ctx; lctx; lctx = lctx->next) {
if (lctx->flags & NGX_RTMP_LIVE_PUBLISHING) {
ngx_rtmp_finalize_session(lctx->session);
++ndropped;
break;
}
}
} else if (method->len == sizeof("client") - 1 &&
ngx_strncmp(method->data, "client", method->len) == 0)
{
ngx_memzero(&addr, sizeof(addr));
ngx_http_arg(r, (u_char *) "addr", sizeof("addr") - 1, &addr);
for (lctx = live.ls->ctx; lctx; lctx = lctx->next) {
if (addr.len && lctx->session && lctx->session->connection) {
paddr = &lctx->session->connection->addr_text;
if (paddr->len != addr.len ||
ngx_strncmp(paddr->data, addr.data, addr.len))
{
continue;
}
}
ngx_rtmp_finalize_session(lctx->session);
++ndropped;
}
} else {
msg = "Undefined method";
goto error;
}
/* output ndropped */
len = NGX_OFF_T_LEN;
p = ngx_palloc(r->connection->pool, len);
if (p == NULL) {
return NGX_ERROR;
}
len = (size_t) (ngx_snprintf(p, len, "%ui", ndropped) - p);
r->headers_out.status = NGX_HTTP_OK;
r->headers_out.content_length_n = len;
b = ngx_calloc_buf(r->pool);
if (b == NULL) {
return NGX_ERROR;
}
b->start = b->pos = p;
b->end = b->last = p + len;
b->temporary = 1;
b->last_buf = 1;
ngx_memzero(&cl, sizeof(cl));
cl.buf = b;
ngx_http_send_header(r);
return ngx_http_output_filter(r, &cl);
error:
return ngx_rtmp_control_output_error(r, msg);
}
static ngx_int_t
ngx_rtmp_control_handler(ngx_http_request_t *r)
{
ngx_rtmp_control_loc_conf_t *llcf;
ngx_str_t section, method;
u_char *p;
ngx_uint_t n;
llcf = ngx_http_get_module_loc_conf(r, ngx_rtmp_control_module);
if (llcf->control == 0) {
return NGX_DECLINED;
}
/* uri format: .../section/method?args */
ngx_memzero(&section, sizeof(section));
ngx_memzero(&method, sizeof(method));
for (n = r->uri.len; n; --n) {
p = &r->uri.data[n - 1];
if (*p != '/') {
continue;
}
if (method.data) {
section.data = p + 1;
section.len = method.data - section.data - 1;
break;
}
method.data = p + 1;
method.len = r->uri.data + r->uri.len - method.data;
}
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, r->connection->log, 0,
"rtmp_control: section='%V' method='%V'",
&section, &method);
#define NGX_RTMP_CONTROL_SECTION(flag, secname) \
if (llcf->control & NGX_RTMP_CONTROL_##flag && \
section.len == sizeof(#secname) - 1 && \
ngx_strncmp(section.data, #secname, sizeof(#secname) - 1) == 0) \
{ \
return ngx_rtmp_control_##secname(r, &method); \
}
NGX_RTMP_CONTROL_SECTION(RECORD, record);
NGX_RTMP_CONTROL_SECTION(DROP, drop);
#undef NGX_RTMP_CONTROL_SECTION
return NGX_DECLINED;
}
static void *
ngx_rtmp_control_create_loc_conf(ngx_conf_t *cf)
{
ngx_rtmp_control_loc_conf_t *conf;
conf = ngx_pcalloc(cf->pool, sizeof(ngx_rtmp_control_loc_conf_t));
if (conf == NULL) {
return NULL;
}
conf->control = 0;
return conf;
}
static char *
ngx_rtmp_control_merge_loc_conf(ngx_conf_t *cf, void *parent, void *child)
{
ngx_rtmp_control_loc_conf_t *prev = parent;
ngx_rtmp_control_loc_conf_t *conf = child;
ngx_conf_merge_bitmask_value(conf->control, prev->control, 0);
return NGX_CONF_OK;
}
static ngx_int_t
ngx_rtmp_control_postconfiguration(ngx_conf_t *cf)
{
ngx_http_handler_pt *h;
ngx_http_core_main_conf_t *cmcf;
cmcf = ngx_http_conf_get_module_main_conf(cf, ngx_http_core_module);
h = ngx_array_push(&cmcf->phases[NGX_HTTP_CONTENT_PHASE].handlers);
if (h == NULL) {
return NGX_ERROR;
}
*h = ngx_rtmp_control_handler;
return NGX_OK;
}

9
ngx_rtmp_core_module.c
View file

@ -127,6 +127,13 @@ static ngx_command_t ngx_rtmp_core_commands[] = {
offsetof(ngx_rtmp_core_srv_conf_t, out_cork),
NULL },
{ ngx_string("busy"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_CONF_TAKE1,
ngx_conf_set_flag_slot,
NGX_RTMP_SRV_CONF_OFFSET,
offsetof(ngx_rtmp_core_srv_conf_t, busy),
NULL },
/* time fixes are needed for flash clients */
{ ngx_string("play_time_fix"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_RTMP_APP_CONF|NGX_CONF_TAKE1,
@ -232,6 +239,7 @@ ngx_rtmp_core_create_srv_conf(ngx_conf_t *cf)
conf->out_cork = NGX_CONF_UNSET;
conf->play_time_fix = NGX_CONF_UNSET;
conf->publish_time_fix = NGX_CONF_UNSET;
conf->busy = NGX_CONF_UNSET;
return conf;
}
@ -258,6 +266,7 @@ ngx_rtmp_core_merge_srv_conf(ngx_conf_t *cf, void *parent, void *child)
conf->out_queue / 8);
ngx_conf_merge_value(conf->play_time_fix, prev->play_time_fix, 1);
ngx_conf_merge_value(conf->publish_time_fix, prev->publish_time_fix, 1);
ngx_conf_merge_value(conf->busy, prev->busy, 0);
if (prev->pool == NULL) {
prev->pool = ngx_create_pool(4096, &cf->cycle->new_log);

11
ngx_rtmp_enotify_module.c
View file

@ -242,8 +242,8 @@ ngx_rtmp_enotify_merge_app_conf(ngx_conf_t *cf, void *parent, void *child)
static ngx_int_t
ngx_rtmp_enotify_exec(ngx_rtmp_session_t *s, ngx_rtmp_enotify_conf_t *ec)
{
#ifndef NGX_WIN32
int pid, fd;
#if !(NGX_WIN32)
int pid, fd, maxfd;
ngx_str_t a, *arg;
char **args;
ngx_uint_t n;
@ -258,6 +258,13 @@ ngx_rtmp_enotify_exec(ngx_rtmp_session_t *s, ngx_rtmp_enotify_conf_t *ec)
case 0:
/* child */
/* close all descriptors */
maxfd = sysconf(_SC_OPEN_MAX);
for (fd = 0; fd < maxfd; ++fd) {
close(fd);
}
fd = open("/dev/null", O_RDWR);
dup2(fd, STDIN_FILENO);

86
ngx_rtmp_exec_module.c
View file

@ -38,6 +38,7 @@ typedef struct {
ngx_array_t execs; /* ngx_rtmp_exec_conf_t */
ngx_msec_t respawn_timeout;
ngx_flag_t respawn;
ngx_int_t kill_signal;
} ngx_rtmp_exec_app_conf_t;
@ -59,6 +60,8 @@ typedef struct {
} ngx_rtmp_exec_ctx_t;
static char *ngx_rtmp_exec_kill_signal(ngx_conf_t *cf, ngx_command_t *cmd,
void *conf);
static ngx_int_t ngx_rtmp_exec_kill(ngx_rtmp_session_t *s, ngx_rtmp_exec_t *e,
ngx_int_t term);
static ngx_int_t ngx_rtmp_exec_run(ngx_rtmp_session_t *s, size_t n);
@ -87,6 +90,13 @@ static ngx_command_t ngx_rtmp_exec_commands[] = {
offsetof(ngx_rtmp_exec_app_conf_t, respawn_timeout),
NULL },
{ ngx_string("exec_kill_signal"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_RTMP_APP_CONF|NGX_CONF_TAKE1,
ngx_rtmp_exec_kill_signal,
NGX_RTMP_APP_CONF_OFFSET,
0,
NULL },
ngx_null_command
};
@ -165,6 +175,7 @@ ngx_rtmp_exec_create_app_conf(ngx_conf_t *cf)
eacf->respawn = NGX_CONF_UNSET;
eacf->respawn_timeout = NGX_CONF_UNSET;
eacf->kill_signal = NGX_CONF_UNSET;
if (ngx_array_init(&eacf->execs, cf->pool, 1,
sizeof(ngx_rtmp_exec_conf_t)) != NGX_OK)
@ -187,6 +198,7 @@ ngx_rtmp_exec_merge_app_conf(ngx_conf_t *cf, void *parent, void *child)
ngx_conf_merge_value(conf->respawn, prev->respawn, 1);
ngx_conf_merge_msec_value(conf->respawn_timeout, prev->respawn_timeout,
5000);
ngx_conf_merge_value(conf->kill_signal, prev->kill_signal, SIGKILL);
if (prev->execs.nelts) {
ec = ngx_array_push_n(&conf->execs, prev->execs.nelts);
@ -254,6 +266,10 @@ ngx_rtmp_exec_child_dead(ngx_event_t *ev)
static ngx_int_t
ngx_rtmp_exec_kill(ngx_rtmp_session_t *s, ngx_rtmp_exec_t *e, ngx_int_t term)
{
ngx_rtmp_exec_app_conf_t *eacf;
eacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_exec_module);
if (e->respawn_evt.timer_set) {
ngx_del_timer(&e->respawn_evt);
}
@ -277,7 +293,7 @@ ngx_rtmp_exec_kill(ngx_rtmp_session_t *s, ngx_rtmp_exec_t *e, ngx_int_t term)
return NGX_OK;
}
if (kill(e->pid, SIGKILL) == -1) {
if (kill(e->pid, eacf->kill_signal) == -1) {
ngx_log_error(NGX_LOG_INFO, s->connection->log, ngx_errno,
"exec: kill failed pid=%i", (ngx_int_t)e->pid);
} else {
@ -292,10 +308,10 @@ ngx_rtmp_exec_kill(ngx_rtmp_session_t *s, ngx_rtmp_exec_t *e, ngx_int_t term)
static ngx_int_t
ngx_rtmp_exec_run(ngx_rtmp_session_t *s, size_t n)
{
#ifndef NGX_WIN32
#if !(NGX_WIN32)
ngx_rtmp_exec_app_conf_t *eacf;
ngx_rtmp_exec_ctx_t *ctx;
int pid, fd;
int pid, fd, maxfd;
int pipefd[2];
int ret;
ngx_rtmp_exec_conf_t *ec;
@ -309,6 +325,8 @@ ngx_rtmp_exec_run(ngx_rtmp_session_t *s, size_t n)
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_exec_module);
e = ctx->execs + n;
ngx_memzero(e, sizeof(*e));
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"exec: starting child '%V'",
&ec->cmd);
@ -344,6 +362,17 @@ ngx_rtmp_exec_run(ngx_rtmp_session_t *s, size_t n)
case 0:
/* child */
/* close all descriptors but pipe write end */
maxfd = sysconf(_SC_OPEN_MAX);
for (fd = 0; fd < maxfd; ++fd) {
if (fd == pipefd[1]) {
continue;
}
close(fd);
}
fd = open("/dev/null", O_RDWR);
dup2(fd, STDIN_FILENO);
@ -513,6 +542,57 @@ ngx_rtmp_exec_exec(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
}
static char *
ngx_rtmp_exec_kill_signal(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
{
ngx_rtmp_exec_app_conf_t *eacf;
ngx_str_t *value;
eacf = ngx_rtmp_conf_get_module_app_conf(cf, ngx_rtmp_exec_module);
value = cf->args->elts;
value++;
eacf->kill_signal = ngx_atoi(value->data, value->len);
if (eacf->kill_signal != NGX_ERROR) {
return NGX_CONF_OK;
}
#define NGX_RMTP_EXEC_SIGNAL(name) \
if (value->len == sizeof(#name) - 1 && \
ngx_strncasecmp(value->data, (u_char *) #name, value->len) == 0) \
{ \
eacf->kill_signal = SIG##name; \
return NGX_CONF_OK; \
}
/* POSIX.1-1990 signals */
NGX_RMTP_EXEC_SIGNAL(HUP);
NGX_RMTP_EXEC_SIGNAL(INT);
NGX_RMTP_EXEC_SIGNAL(QUIT);
NGX_RMTP_EXEC_SIGNAL(ILL);
NGX_RMTP_EXEC_SIGNAL(ABRT);
NGX_RMTP_EXEC_SIGNAL(FPE);
NGX_RMTP_EXEC_SIGNAL(KILL);
NGX_RMTP_EXEC_SIGNAL(SEGV);
NGX_RMTP_EXEC_SIGNAL(PIPE);
NGX_RMTP_EXEC_SIGNAL(ALRM);
NGX_RMTP_EXEC_SIGNAL(TERM);
NGX_RMTP_EXEC_SIGNAL(USR1);
NGX_RMTP_EXEC_SIGNAL(USR2);
NGX_RMTP_EXEC_SIGNAL(CHLD);
NGX_RMTP_EXEC_SIGNAL(CONT);
NGX_RMTP_EXEC_SIGNAL(STOP);
NGX_RMTP_EXEC_SIGNAL(TSTP);
NGX_RMTP_EXEC_SIGNAL(TTIN);
NGX_RMTP_EXEC_SIGNAL(TTOU);
#undef NGX_RMTP_EXEC_SIGNAL
return "unknown signal";
}
static ngx_int_t
ngx_rtmp_exec_postconfiguration(ngx_conf_t *cf)
{

6
ngx_rtmp_flv_module.c
View file

@ -345,8 +345,8 @@ ngx_rtmp_flv_read_meta(ngx_rtmp_session_t *s, ngx_file_t *f)
ngx_memzero(&h, sizeof(h));
h.type = NGX_RTMP_MSG_AMF_META;
h.msid = NGX_RTMP_LIVE_MSID;
h.csid = NGX_RTMP_LIVE_CSID_META;
h.msid = NGX_RTMP_MSID;
h.csid = NGX_RTMP_CSID_AMF;
size = 0;
ngx_rtmp_rmemcpy(&size, ngx_rtmp_flv_header + 1, 3);
@ -433,7 +433,7 @@ ngx_rtmp_flv_send(ngx_rtmp_session_t *s, ngx_file_t *f, ngx_uint_t *ts)
/* parse header fields */
ngx_memzero(&h, sizeof(h));
h.msid = NGX_RTMP_LIVE_MSID;
h.msid = NGX_RTMP_MSID;
h.type = ngx_rtmp_flv_header[0];
size = 0;

14
ngx_rtmp_handler.c
View file

@ -164,6 +164,13 @@ ngx_rtmp_ping(ngx_event_t *pev)
return;
}
if (cscf->busy) {
ngx_log_error(NGX_LOG_INFO, c->log, 0,
"ping: not busy between pings");
ngx_rtmp_finalize_session(s);
return;
}
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"ping: schedule %Mms", cscf->ping_timeout);
@ -396,6 +403,7 @@ ngx_rtmp_recv(ngx_event_t *rev)
st->dtime = timestamp;
} else {
h->timestamp = timestamp;
h->timeshift = (uint32_t) ngx_current_msec - timestamp;
st->dtime = 0;
}
}
@ -694,9 +702,13 @@ ngx_rtmp_send_message(ngx_rtmp_session_t *s, ngx_chain_t *out,
nmsg = (s->out_last - s->out_pos) % s->out_queue + 1;
if (priority > 3) {
priority = 3;
}
/* drop packet?
* Note we always leave 1 slot free */
if (nmsg + priority * s->out_queue / 16 >= s->out_queue) {
if (nmsg + priority * s->out_queue / 4 >= s->out_queue) {
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"RTMP drop message bufs=%ui, priority=%ui",
nmsg, priority);

101
ngx_rtmp_live_module.c
View file

@ -53,6 +53,20 @@ static ngx_command_t ngx_rtmp_live_commands[] = {
offsetof(ngx_rtmp_live_app_conf_t, buflen),
NULL },
{ ngx_string("sync"),
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_live_app_conf_t, sync),
NULL },
{ ngx_string("atc"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_RTMP_APP_CONF|NGX_CONF_TAKE1,
ngx_conf_set_flag_slot,
NGX_RTMP_APP_CONF_OFFSET,
offsetof(ngx_rtmp_live_app_conf_t, atc),
NULL },
ngx_null_command
};
@ -99,6 +113,8 @@ ngx_rtmp_live_create_app_conf(ngx_conf_t *cf)
lacf->meta = NGX_CONF_UNSET;
lacf->nbuckets = NGX_CONF_UNSET;
lacf->buflen = NGX_CONF_UNSET;
lacf->sync = NGX_CONF_UNSET;
lacf->atc = NGX_CONF_UNSET;
return lacf;
}
@ -114,6 +130,8 @@ ngx_rtmp_live_merge_app_conf(ngx_conf_t *cf, void *parent, void *child)
ngx_conf_merge_value(conf->meta, prev->meta, 1);
ngx_conf_merge_value(conf->nbuckets, prev->nbuckets, 1024);
ngx_conf_merge_msec_value(conf->buflen, prev->buflen, 0);
ngx_conf_merge_msec_value(conf->sync, prev->sync, 0);
ngx_conf_merge_value(conf->atc, prev->atc, 0);
conf->pool = ngx_create_pool(4096, &cf->cycle->new_log);
if (conf->pool == NULL) {
@ -292,7 +310,7 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
{
ngx_rtmp_live_ctx_t *ctx, *pctx;
ngx_rtmp_codec_ctx_t *codec_ctx;
ngx_chain_t *out, *peer_out, *header_out,
ngx_chain_t *out, *peer_out, *header_out,
*pheader_out, *meta;
ngx_rtmp_core_srv_conf_t *cscf;
ngx_rtmp_live_app_conf_t *lacf;
@ -300,8 +318,10 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ngx_rtmp_header_t ch, lh;
ngx_uint_t prio, peer_prio;
ngx_uint_t peers, dropped_peers;
size_t header_offset;
size_t header_offset, last_offset;
ngx_uint_t header_version, meta_version;
ngx_int_t diff_timestamp;
uint32_t *last, timestamp;
lacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_live_module);
if (lacf == NULL) {
@ -326,33 +346,40 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
return NGX_OK;
}
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"live: av: %s timestamp=%uD",
ngx_log_debug3(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"live: av: %s timestamp=%uD timeshift=%uD",
h->type == NGX_RTMP_MSG_VIDEO ? "video" : "audio",
h->timestamp);
h->timestamp, h->timeshift);
cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module);
timestamp = h->timestamp;
if (lacf->atc == 0) {
timestamp += h->timeshift;
}
/* prepare output header */
ngx_memzero(&ch, sizeof(ch));
ngx_memzero(&lh, sizeof(lh));
ch.timestamp = h->timestamp;
ch.msid = NGX_RTMP_LIVE_MSID;
ch.timestamp = timestamp;
ch.msid = NGX_RTMP_MSID;
ch.type = h->type;
lh.msid = ch.msid;
if (h->type == NGX_RTMP_MSG_VIDEO) {
prio = ngx_rtmp_get_video_frame_type(in);
ch.csid = NGX_RTMP_LIVE_CSID_VIDEO;
ch.csid = NGX_RTMP_CSID_VIDEO;
lh.timestamp = ctx->last_video;
ctx->last_video = ch.timestamp;
last_offset = offsetof(ngx_rtmp_live_ctx_t, last_video);
} else {
/* audio priority is the same as video key frame's */
prio = NGX_RTMP_VIDEO_KEY_FRAME;
ch.csid = NGX_RTMP_LIVE_CSID_AUDIO;
prio = 0;
ch.csid = NGX_RTMP_CSID_AUDIO;
lh.timestamp = ctx->last_audio;
ctx->last_audio = ch.timestamp;
last_offset = offsetof(ngx_rtmp_live_ctx_t, last_audio);
}
lh.csid = ch.csid;
diff_timestamp = ch.timestamp - lh.timestamp;
out = ngx_rtmp_append_shared_bufs(cscf, NULL, in);
ngx_rtmp_prepare_message(s, &ch, &lh, out);
@ -396,25 +423,43 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
}
++peers;
ss = pctx->session;
last = (uint32_t *)((u_char *)pctx + last_offset);
ch.timestamp = timestamp;
if (lacf->atc == 0) {
ch.timestamp -= (uint32_t)ss->epoch;
}
/* send absolute frame */
if ((pctx->msg_mask & (1 << h->type)) == 0) {
ch.timestamp = ngx_current_msec - ss->epoch;
/* packet from the past for the peer */
if (lacf->atc == 0 && timestamp < (uint32_t)ss->epoch) {
ngx_log_debug3(NGX_LOG_DEBUG_RTMP, ss->connection->log, 0,
"live: av: %s packet from the past %uD < %uD",
h->type == NGX_RTMP_MSG_VIDEO ? "video" : "audio",
timestamp, (uint32_t)ss->epoch);
continue;
}
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, ss->connection->log, 0,
"live: av: abs %s timestamp=%uD",
h->type == NGX_RTMP_MSG_VIDEO ? "video" : "audio",
ch.timestamp);
/* send codec header as abs frame if any */
peer_out = ngx_rtmp_append_shared_bufs(cscf, NULL,
header_out ? header_out : in);
ngx_rtmp_prepare_message(s, &ch, NULL, peer_out);
pctx->msg_mask |= (1 << h->type);
if (ngx_rtmp_send_message(ss, peer_out, prio) == NGX_OK
&& header_out)
{
*(ngx_uint_t *)((u_char *)pctx + header_offset)
= header_version;
if (ngx_rtmp_send_message(ss, peer_out, prio) == NGX_OK) {
pctx->msg_mask |= (1 << h->type);
if (header_out) {
*(ngx_uint_t *)((u_char *)pctx + header_offset)
= header_version;
*last = ch.timestamp;
}
}
ngx_rtmp_free_shared_chain(cscf, peer_out);
continue;
}
@ -445,7 +490,27 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
if (ngx_rtmp_send_message(ss, out, peer_prio) != NGX_OK) {
++pctx->dropped;
++dropped_peers;
continue;
}
*last += diff_timestamp;
if (lacf->sync == 0 || *last + lacf->sync >= ch.timestamp) {
continue;
}
/* send absolute frame to sync stream */
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, ss->connection->log, 0,
"live: av: sync %s: %i",
h->type == NGX_RTMP_MSG_VIDEO ? "video" : "audio",
(ngx_int_t) (ch.timestamp - *last));
peer_out = ngx_rtmp_alloc_shared_buf(cscf);
ngx_rtmp_prepare_message(s, &ch, NULL, peer_out);
if (ngx_rtmp_send_message(ss, peer_out, 0) == NGX_OK) {
*last = ch.timestamp;
}
ngx_rtmp_free_shared_chain(cscf, peer_out);
}
ngx_rtmp_free_shared_chain(cscf, out);

2
ngx_rtmp_live_module.h
View file

@ -54,6 +54,8 @@ typedef struct {
ngx_rtmp_live_stream_t **streams;
ngx_flag_t live;
ngx_flag_t meta;
ngx_msec_t sync;
ngx_flag_t atc;
ngx_msec_t buflen;
ngx_pool_t *pool;
ngx_rtmp_live_stream_t *free_streams;

112
ngx_rtmp_netcall_module.c
View file

@ -20,12 +20,9 @@ static void ngx_rtmp_netcall_recv(ngx_event_t *rev);
static void ngx_rtmp_netcall_send(ngx_event_t *wev);
ngx_str_t ngx_rtmp_netcall_content_type_urlencoded =
ngx_string("application/x-www-form-urlencoded");
typedef struct {
ngx_msec_t timeout;
size_t bufsize;
ngx_log_t *log;
} ngx_rtmp_netcall_app_conf_t;
@ -38,11 +35,13 @@ typedef struct ngx_rtmp_netcall_session_s {
void *arg;
ngx_rtmp_netcall_handle_pt handle;
ngx_rtmp_netcall_filter_pt filter;
ngx_rtmp_netcall_sink_pt sink;
ngx_chain_t *in;
ngx_chain_t *inlast;
ngx_chain_t *out;
ngx_msec_t timeout;
ngx_int_t detached;
unsigned detached:1;
size_t bufsize;
} ngx_rtmp_netcall_session_t;
@ -60,6 +59,13 @@ static ngx_command_t ngx_rtmp_netcall_commands[] = {
offsetof(ngx_rtmp_netcall_app_conf_t, timeout),
NULL },
{ ngx_string("netcall_buffer"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_CONF_TAKE1,
ngx_conf_set_size_slot,
NGX_RTMP_SRV_CONF_OFFSET,
offsetof(ngx_rtmp_netcall_app_conf_t, bufsize),
NULL },
ngx_null_command
};
@ -103,6 +109,8 @@ ngx_rtmp_netcall_create_app_conf(ngx_conf_t *cf)
}
nacf->timeout = NGX_CONF_UNSET_MSEC;
nacf->bufsize = NGX_CONF_UNSET_SIZE;
nacf->log = &cf->cycle->new_log;
return nacf;
@ -116,6 +124,7 @@ ngx_rtmp_netcall_merge_app_conf(ngx_conf_t *cf, void *parent, void *child)
ngx_rtmp_netcall_app_conf_t *conf = child;
ngx_conf_merge_msec_value(conf->timeout, prev->timeout, 10000);
ngx_conf_merge_size_value(conf->bufsize, prev->bufsize, 1024);
return NGX_CONF_OK;
}
@ -219,9 +228,11 @@ ngx_rtmp_netcall_create(ngx_rtmp_session_t *s, ngx_rtmp_netcall_init_t *ci)
}
cs->timeout = cacf->timeout;
cs->bufsize = cacf->bufsize;
cs->url = ci->url;
cs->session = s;
cs->filter = ci->filter;
cs->sink = ci->sink;
cs->handle = ci->handle;
if (cs->handle == NULL) {
cs->detached = 1;
@ -281,6 +292,7 @@ ngx_rtmp_netcall_close(ngx_connection_t *cc)
ngx_pool_t *pool;
ngx_rtmp_session_t *s;
ngx_rtmp_netcall_ctx_t *ctx;
ngx_buf_t *b;
cs = cc->data;
@ -294,6 +306,14 @@ ngx_rtmp_netcall_close(ngx_connection_t *cc)
s = cs->session;
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_netcall_module);
if (cs->in && cs->sink) {
cs->sink(cs->session, cs->in);
b = cs->in->buf;
b->pos = b->last = b->start;
}
for(css = &ctx->cs; *css; css = &((*css)->next)) {
if (*css == cs) {
*css = cs->next;
@ -301,9 +321,7 @@ ngx_rtmp_netcall_close(ngx_connection_t *cc)
}
}
if (cs->handle &&
cs->handle(s, cs->arg, cs->in) != NGX_OK)
{
if (cs->handle && cs->handle(s, cs->arg, cs->in) != NGX_OK) {
ngx_rtmp_finalize_session(s);
}
}
@ -352,29 +370,43 @@ ngx_rtmp_netcall_recv(ngx_event_t *rev)
for ( ;; ) {
if (cs->inlast == NULL
|| cs->inlast->buf->last == cs->inlast->buf->end)
if (cs->inlast == NULL ||
cs->inlast->buf->last == cs->inlast->buf->end)
{
cl = ngx_alloc_chain_link(cc->pool);
if (cl == NULL) {
ngx_rtmp_netcall_close(cc);
return;
}
cl->next = NULL;
if (cs->in && cs->sink) {
if (!cs->detached) {
if (cs->sink(cs->session, cs->in) != NGX_OK) {
ngx_rtmp_netcall_close(cc);
return;
}
}
cl->buf = ngx_create_temp_buf(cc->pool, 1024);
if (cl->buf == NULL) {
ngx_rtmp_netcall_close(cc);
return;
}
b = cs->in->buf;
b->pos = b->last = b->start;
if (cs->in == NULL) {
cs->in = cl;
} else {
cs->inlast->next = cl;
}
cl = ngx_alloc_chain_link(cc->pool);
if (cl == NULL) {
ngx_rtmp_netcall_close(cc);
return;
}
cs->inlast = cl;
cl->next = NULL;
cl->buf = ngx_create_temp_buf(cc->pool, cs->bufsize);
if (cl->buf == NULL) {
ngx_rtmp_netcall_close(cc);
return;
}
if (cs->in == NULL) {
cs->in = cl;
} else {
cs->inlast->next = cl;
}
cs->inlast = cl;
}
}
b = cs->inlast->buf;
@ -459,14 +491,17 @@ ngx_rtmp_netcall_send(ngx_event_t *wev)
ngx_chain_t *
ngx_rtmp_netcall_http_format_header(ngx_url_t *url, ngx_pool_t *pool,
size_t content_length, ngx_str_t *content_type)
ngx_rtmp_netcall_http_format_header(ngx_int_t method, ngx_str_t *uri,
ngx_str_t *host, ngx_pool_t *pool,
size_t content_length,
ngx_str_t *content_type)
{
ngx_chain_t *cl;
ngx_buf_t *b;
const char *method_s;
static char rq_tmpl[] =
"POST %V HTTP/1.0\r\n"
"%s %V HTTP/1.0\r\n"
"Host: %V\r\n"
"Content-Type: %V\r\n"
"Connection: Close\r\n"
@ -480,8 +515,9 @@ ngx_rtmp_netcall_http_format_header(ngx_url_t *url, ngx_pool_t *pool,
}
b = ngx_create_temp_buf(pool, sizeof(rq_tmpl)
+ url->uri.len
+ url->host.len
+ sizeof("POST") - 1 /* longest method */
+ uri->len
+ host->len
+ content_type->len
+ 5);
@ -490,9 +526,21 @@ ngx_rtmp_netcall_http_format_header(ngx_url_t *url, ngx_pool_t *pool,
}
cl->buf = b;
cl->next = NULL;
switch (method) {
case NGX_RTMP_NETCALL_HTTP_GET:
method_s = "GET";
break;
case NGX_RTMP_NETCALL_HTTP_POST:
method_s = "POST";
break;
default:
return NULL;
}
b->last = ngx_snprintf(b->last, b->end - b->last, rq_tmpl,
&url->uri, &url->host, content_type, content_length);
method_s, uri, host, content_type, content_length);
return cl;
}

13
ngx_rtmp_netcall_module.h
View file

@ -15,9 +15,14 @@
typedef ngx_chain_t * (*ngx_rtmp_netcall_create_pt)(ngx_rtmp_session_t *s,
void *arg, ngx_pool_t *pool);
typedef ngx_int_t (*ngx_rtmp_netcall_filter_pt)(ngx_chain_t *in);
typedef ngx_int_t (*ngx_rtmp_netcall_sink_pt)(ngx_rtmp_session_t *s,
ngx_chain_t *in);
typedef ngx_int_t (*ngx_rtmp_netcall_handle_pt)(ngx_rtmp_session_t *s,
void *arg, ngx_chain_t *in);
#define NGX_RTMP_NETCALL_HTTP_GET 1
#define NGX_RTMP_NETCALL_HTTP_POST 2
/* If handle is NULL then netcall is created detached
* which means it's completely independent of RTMP
@ -32,6 +37,7 @@ typedef struct {
ngx_url_t *url;
ngx_rtmp_netcall_create_pt create;
ngx_rtmp_netcall_filter_pt filter;
ngx_rtmp_netcall_sink_pt sink;
ngx_rtmp_netcall_handle_pt handle;
void *arg;
size_t argsize;
@ -41,14 +47,13 @@ typedef struct {
ngx_int_t ngx_rtmp_netcall_create(ngx_rtmp_session_t *s,
ngx_rtmp_netcall_init_t *ci);
extern ngx_str_t ngx_rtmp_netcall_content_type_urlencoded;
/* HTTP handling */
ngx_chain_t * ngx_rtmp_netcall_http_format_session(ngx_rtmp_session_t *s,
ngx_pool_t *pool);
ngx_chain_t * ngx_rtmp_netcall_http_format_header(ngx_url_t *url,
ngx_pool_t *pool, size_t content_length, ngx_str_t *content_type);
ngx_chain_t * ngx_rtmp_netcall_http_format_header(ngx_int_t method,
ngx_str_t *uri, ngx_str_t *host, ngx_pool_t *pool,
size_t content_length, ngx_str_t *content_type);
ngx_chain_t * ngx_rtmp_netcall_http_skip_header(ngx_chain_t *in);

41
ngx_rtmp_notify_module.c
View file

@ -28,6 +28,10 @@ static ngx_int_t ngx_rtmp_notify_done(ngx_rtmp_session_t *s, char *cbname,
ngx_url_t *url);
ngx_str_t ngx_rtmp_notify_urlencoded =
ngx_string("application/x-www-form-urlencoded");
#define NGX_RTMP_NOTIFY_PUBLISHING 0x01
#define NGX_RTMP_NOTIFY_PLAYING 0x02
@ -190,6 +194,7 @@ ngx_rtmp_notify_publish_create(ngx_rtmp_session_t *s, void *arg,
ngx_chain_t *hl, *cl, *pl;
ngx_buf_t *b;
ngx_str_t *addr_text;
ngx_url_t *url;
size_t name_len, type_len, args_len;
nacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_notify_module);
@ -215,7 +220,7 @@ ngx_rtmp_notify_publish_create(ngx_rtmp_session_t *s, void *arg,
b = ngx_create_temp_buf(pool,
sizeof("&call=publish") +
sizeof("&addr=") + addr_text->len +
sizeof("&addr=") + addr_text->len *3 +
sizeof("&name=") + name_len * 3 +
sizeof("&type=") + type_len * 3 +
1 + args_len);
@ -244,9 +249,11 @@ ngx_rtmp_notify_publish_create(ngx_rtmp_session_t *s, void *arg,
}
/* HTTP header */
hl = ngx_rtmp_netcall_http_format_header(nacf->url[NGX_RTMP_NOTIFY_PUBLISH],
url = nacf->url[NGX_RTMP_NOTIFY_PUBLISH];
hl = ngx_rtmp_netcall_http_format_header(NGX_RTMP_NETCALL_HTTP_POST,
&url->uri, &url->host,
pool, cl->buf->last - cl->buf->pos + (pl->buf->last - pl->buf->pos),
&ngx_rtmp_netcall_content_type_urlencoded);
&ngx_rtmp_notify_urlencoded);
if (hl == NULL) {
return NULL;
@ -270,6 +277,7 @@ ngx_rtmp_notify_play_create(ngx_rtmp_session_t *s, void *arg,
ngx_chain_t *hl, *cl, *pl;
ngx_buf_t *b;
ngx_str_t *addr_text;
ngx_url_t *url;
size_t name_len, args_len;
nacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_notify_module);
@ -294,7 +302,7 @@ ngx_rtmp_notify_play_create(ngx_rtmp_session_t *s, void *arg,
b = ngx_create_temp_buf(pool,
sizeof("&call=play") +
sizeof("&addr=") + addr_text->len +
sizeof("&addr=") + addr_text->len * 3 +
sizeof("&name=") + name_len * 3 +
sizeof("&start=&duration=&reset=") + NGX_OFF_T_LEN * 3
+ 1 + args_len);
@ -324,9 +332,11 @@ ngx_rtmp_notify_play_create(ngx_rtmp_session_t *s, void *arg,
}
/* HTTP header */
hl = ngx_rtmp_netcall_http_format_header(nacf->url[NGX_RTMP_NOTIFY_PLAY],
url = nacf->url[NGX_RTMP_NOTIFY_PLAY];
hl = ngx_rtmp_netcall_http_format_header(NGX_RTMP_NETCALL_HTTP_POST,
&url->uri, &url->host,
pool, cl->buf->last - cl->buf->pos + (pl->buf->last - pl->buf->pos),
&ngx_rtmp_netcall_content_type_urlencoded);
&ngx_rtmp_notify_urlencoded);
if (hl == NULL) {
return NULL;
@ -374,7 +384,7 @@ ngx_rtmp_notify_done_create(ngx_rtmp_session_t *s, void *arg,
b = ngx_create_temp_buf(pool,
sizeof("&call=") + cbname_len +
sizeof("&addr=") + addr_text->len +
sizeof("&addr=") + addr_text->len * 3 +
sizeof("&name=") + name_len * 3
+ 1 + args_len);
if (b == NULL) {
@ -401,9 +411,10 @@ ngx_rtmp_notify_done_create(ngx_rtmp_session_t *s, void *arg,
}
/* HTTP header */
hl = ngx_rtmp_netcall_http_format_header(ds->url, pool,
cl->buf->last - cl->buf->pos + (pl->buf->last - pl->buf->pos),
&ngx_rtmp_netcall_content_type_urlencoded);
hl = ngx_rtmp_netcall_http_format_header(NGX_RTMP_NETCALL_HTTP_POST,
&ds->url->uri, &ds->url->host,
pool, cl->buf->last - cl->buf->pos + (pl->buf->last - pl->buf->pos),
&ngx_rtmp_notify_urlencoded);
if (hl == NULL) {
return NULL;
@ -428,6 +439,7 @@ ngx_rtmp_notify_record_done_create(ngx_rtmp_session_t *s, void *arg,
ngx_chain_t *hl, *cl, *pl;
ngx_buf_t *b;
ngx_str_t *addr_text;
ngx_url_t *url;
size_t name_len, args_len;
nacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_notify_module);
@ -455,7 +467,7 @@ ngx_rtmp_notify_record_done_create(ngx_rtmp_session_t *s, void *arg,
b = ngx_create_temp_buf(pool,
sizeof("&call=record_done") +
sizeof("&recorder=") + v->recorder.len +
sizeof("&addr=") + addr_text->len +
sizeof("&addr=") + addr_text->len *3 +
sizeof("&name=") + name_len * 3 +
sizeof("&path=") + v->path.len * 3 +
+ 1 + args_len);
@ -489,10 +501,11 @@ ngx_rtmp_notify_record_done_create(ngx_rtmp_session_t *s, void *arg,
}
/* HTTP header */
hl = ngx_rtmp_netcall_http_format_header(
nacf->url[NGX_RTMP_NOTIFY_RECORD_DONE],
url = nacf->url[NGX_RTMP_NOTIFY_RECORD_DONE];
hl = ngx_rtmp_netcall_http_format_header(NGX_RTMP_NETCALL_HTTP_POST,
&url->uri, &url->host,
pool, cl->buf->last - cl->buf->pos + (pl->buf->last - pl->buf->pos),
&ngx_rtmp_netcall_content_type_urlencoded);
&ngx_rtmp_notify_urlencoded);
if (hl == NULL) {
return NULL;

299
ngx_rtmp_play_module.c
View file

@ -5,6 +5,7 @@
#include "ngx_rtmp_play_module.h"
#include "ngx_rtmp_cmd_module.h"
#include "ngx_rtmp_netcall_module.h"
#include "ngx_rtmp_streams.h"
@ -14,6 +15,8 @@ static ngx_rtmp_seek_pt next_seek;
static ngx_rtmp_pause_pt next_pause;
static char *ngx_rtmp_play_url(ngx_conf_t *cf, ngx_command_t *cmd,
void *conf);
static void *ngx_rtmp_play_create_main_conf(ngx_conf_t *cf);
static ngx_int_t ngx_rtmp_play_postconfiguration(ngx_conf_t *cf);
static void * ngx_rtmp_play_create_app_conf(ngx_conf_t *cf);
@ -32,15 +35,29 @@ static ngx_int_t ngx_rtmp_play_seek(ngx_rtmp_session_t *s, ngx_rtmp_seek_t *v);
static ngx_int_t ngx_rtmp_play_pause(ngx_rtmp_session_t *s,
ngx_rtmp_pause_t *v);
static void ngx_rtmp_play_send(ngx_event_t *e);
static ngx_int_t ngx_rtmp_play_open(ngx_rtmp_session_t *s, double start);
static ngx_int_t ngx_rtmp_play_remote_handle(ngx_rtmp_session_t *s,
void *arg, ngx_chain_t *in);
static ngx_chain_t * ngx_rtmp_play_remote_create(ngx_rtmp_session_t *s,
void *arg, ngx_pool_t *pool);
static ngx_int_t ngx_rtmp_play_open_remote(ngx_rtmp_session_t *s,
ngx_rtmp_play_t *v);
static ngx_command_t ngx_rtmp_play_commands[] = {
{ ngx_string("play"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_RTMP_APP_CONF|NGX_CONF_TAKE1,
ngx_rtmp_play_url,
NGX_RTMP_APP_CONF_OFFSET,
0,
NULL },
{ ngx_string("play_temp_path"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_RTMP_APP_CONF|NGX_CONF_TAKE1,
ngx_conf_set_str_slot,
NGX_RTMP_APP_CONF_OFFSET,
offsetof(ngx_rtmp_play_app_conf_t, root),
offsetof(ngx_rtmp_play_app_conf_t, temp_path),
NULL },
ngx_null_command
@ -119,6 +136,7 @@ ngx_rtmp_play_merge_app_conf(ngx_conf_t *cf, void *parent, void *child)
ngx_rtmp_play_app_conf_t *conf = child;
ngx_conf_merge_str_value(conf->root, prev->root, "");
ngx_conf_merge_str_value(conf->temp_path, prev->temp_path, "/tmp");
return NGX_CONF_OK;
}
@ -388,7 +406,6 @@ ngx_rtmp_play_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
ngx_rtmp_play_app_conf_t *pacf;
ngx_rtmp_play_ctx_t *ctx;
u_char *p;
ngx_event_t *e;
ngx_rtmp_play_fmt_t *fmt, **pfmt;
ngx_str_t *pfx, *sfx;
ngx_str_t name;
@ -400,7 +417,7 @@ ngx_rtmp_play_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
pacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_play_module);
if (pacf == NULL || pacf->root.len == 0) {
if (pacf == NULL || (pacf->root.len == 0 && pacf->url == NULL)) {
goto next;
}
@ -477,8 +494,11 @@ ngx_rtmp_play_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
goto next;
}
ctx->file.fd = NGX_INVALID_FILE;
ngx_log_error(NGX_LOG_INFO, s->connection->log, 0,
"play: %V", &ctx->fmt->name);
"play %s: %V", pacf->url ? "remote" : "local",
&ctx->fmt->name);
sfx = &ctx->fmt->sfx;
@ -490,6 +510,23 @@ ngx_rtmp_play_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
sfx = &nosfx;
}
/* remote? */
if (pacf->url) {
ctx->name.data = ngx_palloc(s->connection->pool, name.len + sfx->len);
if (ctx->name.data == NULL) {
return NGX_ERROR;
}
p = ngx_sprintf(ctx->name.data, "%V%V", &name, sfx);
*p = 0;
ctx->name.len = p - ctx->name.data;
return ngx_rtmp_play_open_remote(s, v);
}
/* open local */
p = ngx_snprintf(path, sizeof(path), "%V/%V%V", &pacf->root, &name, sfx);
*p = 0;
@ -497,13 +534,34 @@ ngx_rtmp_play_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
NGX_FILE_DEFAULT_ACCESS);
if (ctx->file.fd == NGX_INVALID_FILE) {
ngx_log_error(NGX_LOG_ERR, s->connection->log, 0,
ngx_log_error(NGX_LOG_ERR, s->connection->log, ngx_errno,
"play: error opening file '%s'", path);
return NGX_ERROR;
}
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"play: opened file '%s'", path);
"play: opened local file '%s'", path);
if (ngx_rtmp_play_open(s, v->start) != NGX_OK) {
return NGX_ERROR;
}
next:
return next_play(s, v);
}
static ngx_int_t
ngx_rtmp_play_open(ngx_rtmp_session_t *s, double start)
{
ngx_rtmp_play_ctx_t *ctx;
ngx_event_t *e;
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
if (ctx->file.fd == NGX_INVALID_FILE) {
return NGX_ERROR;
}
e = &ctx->send_evt;
e->data = s;
@ -516,7 +574,7 @@ ngx_rtmp_play_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
return NGX_ERROR;
}
if (ngx_rtmp_play_do_seek(s, v->start < 0 ? 0 : v->start) != NGX_OK) {
if (ngx_rtmp_play_do_seek(s, start < 0 ? 0 : start) != NGX_OK) {
return NGX_ERROR;
}
@ -524,8 +582,231 @@ ngx_rtmp_play_play(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
return NGX_ERROR;
}
next:
return next_play(s, v);
return NGX_OK;
}
static ngx_chain_t *
ngx_rtmp_play_remote_create(ngx_rtmp_session_t *s, void *arg, ngx_pool_t *pool)
{
ngx_rtmp_play_t *v = arg;
ngx_rtmp_play_app_conf_t *pacf;
ngx_rtmp_play_ctx_t *ctx;
ngx_chain_t *hl;
ngx_str_t *addr_text, uri;
u_char *p;
size_t args_len, len;
static ngx_str_t text_plain = ngx_string("text/plain");
pacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_play_module);
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
args_len = ngx_strlen(v->args);
addr_text = &s->connection->addr_text;
len = pacf->url->uri.len + 1 + ctx->name.len +
sizeof("?addr=") + addr_text->len * 3 +
1 + args_len;
uri.data = ngx_palloc(pool, len);
if (uri.data == NULL) {
return NULL;
}
p = uri.data;
p = ngx_cpymem(p, pacf->url->uri.data, pacf->url->uri.len);
if (p == uri.data || p[-1] != '/') {
*p++ = '/';
}
p = ngx_cpymem(p, ctx->name.data, ctx->name.len);
p = ngx_cpymem(p, (u_char*)"?addr=", sizeof("&addr=") -1);
p = (u_char*)ngx_escape_uri(p, addr_text->data, addr_text->len, 0);
if (args_len) {
*p++ = '&';
p = (u_char *) ngx_cpymem(p, v->args, args_len);
}
uri.len = p - uri.data;
/* HTTP header */
hl = ngx_rtmp_netcall_http_format_header(NGX_RTMP_NETCALL_HTTP_GET,
&uri, &pacf->url->host, pool, 0, &text_plain);
if (hl == NULL) {
return NULL;
}
return hl;
}
static ngx_int_t
ngx_rtmp_play_remote_handle(ngx_rtmp_session_t *s, void *arg, ngx_chain_t *in)
{
ngx_rtmp_play_t *v = arg;
if (ngx_rtmp_play_open(s, v->start) != NGX_OK) {
return NGX_ERROR;
}
return next_play(s, (ngx_rtmp_play_t *)arg);
}
static ngx_int_t
ngx_rtmp_play_remote_sink(ngx_rtmp_session_t *s, ngx_chain_t *in)
{
ngx_rtmp_play_ctx_t *ctx;
ngx_buf_t *b;
ngx_int_t rc;
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
/* skip HTTP header */
while (in && ctx->ncrs != 2) {
b = in->buf;
for (; b->pos != b->last && ctx->ncrs != 2; ++b->pos) {
switch (*b->pos) {
case '\n':
++ctx->ncrs;
case '\r':
break;
default:
ctx->ncrs = 0;
}
}
if (b->pos == b->last) {
in = in->next;
}
}
/* write to temp file */
for (; in; in = in->next) {
b = in->buf;
if (b->pos == b->last) {
continue;
}
rc = ngx_write_fd(ctx->file.fd, b->pos, b->last - b->pos);
if (rc == NGX_ERROR) {
ngx_log_error(NGX_LOG_INFO, s->connection->log, ngx_errno,
"play: error writing to temp file");
return NGX_ERROR;
}
}
return NGX_OK;
}
static ngx_int_t
ngx_rtmp_play_open_remote(ngx_rtmp_session_t *s, ngx_rtmp_play_t *v)
{
ngx_rtmp_play_app_conf_t *pacf;
ngx_rtmp_play_ctx_t *ctx;
ngx_rtmp_netcall_init_t ci;
u_char *p;
ngx_err_t err;
static u_char path[NGX_MAX_PATH];
static ngx_uint_t counter;
pacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_play_module);
ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_play_module);
for ( ;; ) {
p = ngx_snprintf(path, sizeof(path), "%V/nginx-rtmp-play%ui",
&pacf->temp_path, counter++);
*p = 0;
ctx->file.fd = ngx_open_tempfile(path, 0, 0);
if (ctx->file.fd != NGX_INVALID_FILE) {
break;
}
err = ngx_errno;
if (err != NGX_EEXIST) {
ngx_log_error(NGX_LOG_INFO, s->connection->log, err,
"play: failed to create temp file");
return NGX_ERROR;
}
}
ngx_memzero(&ci, sizeof(ci));
ci.url = pacf->url;;
ci.create = ngx_rtmp_play_remote_create;
ci.sink = ngx_rtmp_play_remote_sink;
ci.handle = ngx_rtmp_play_remote_handle;
ci.arg = v;
ci.argsize = sizeof(*v);
return ngx_rtmp_netcall_create(s, &ci);
}
static char *
ngx_rtmp_play_url(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
{
ngx_rtmp_play_app_conf_t *pacf = conf;
ngx_str_t url;
ngx_url_t *u;
size_t add;
ngx_str_t *value;
value = cf->args->elts;
if (ngx_strncasecmp(value[1].data, (u_char *) "http://", 7)) {
/* local file */
pacf->root = value[1];
return NGX_CONF_OK;
}
/* http case */
url = value[1];
add = sizeof("http://") - 1;
url.data += add;
url.len -= add;
u = ngx_pcalloc(cf->pool, sizeof(ngx_url_t));
if (u == NULL) {
return NGX_CONF_ERROR;
}
u->url.len = url.len;
u->url.data = url.data;
u->default_port = 80;
u->uri_part = 1;
if (ngx_parse_url(cf->pool, u) != NGX_OK) {
if (u->err) {
ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
"%s in url \"%V\"", u->err, &u->url);
}
return NGX_CONF_ERROR;
}
pacf->url = u;
return NGX_CONF_OK;
}

4
ngx_rtmp_play_module.h
View file

@ -43,11 +43,15 @@ typedef struct {
ngx_rtmp_play_fmt_t *fmt;
ngx_event_t send_evt;
unsigned playing:1;
ngx_uint_t ncrs;
ngx_str_t name;
} ngx_rtmp_play_ctx_t;
typedef struct {
ngx_str_t root;
ngx_str_t temp_path;
ngx_url_t *url;
} ngx_rtmp_play_app_conf_t;

81
ngx_rtmp_record_module.c
View file

@ -27,7 +27,7 @@ static char * ngx_rtmp_record_merge_app_conf(ngx_conf_t *cf,
void *parent, void *child);
static ngx_int_t ngx_rtmp_record_write_frame(ngx_rtmp_session_t *s,
ngx_rtmp_record_rec_ctx_t *rctx,
ngx_rtmp_header_t *h, ngx_chain_t *in);
ngx_rtmp_header_t *h, ngx_chain_t *in, ngx_int_t inc_nframes);
static ngx_int_t ngx_rtmp_record_av(ngx_rtmp_session_t *s,
ngx_rtmp_header_t *h, ngx_chain_t *in);
static ngx_int_t ngx_rtmp_record_node_av(ngx_rtmp_session_t *s,
@ -111,6 +111,14 @@ static ngx_command_t ngx_rtmp_record_commands[] = {
offsetof(ngx_rtmp_record_app_conf_t, interval),
NULL },
{ ngx_string("record_notify"),
NGX_RTMP_MAIN_CONF|NGX_RTMP_SRV_CONF|NGX_RTMP_APP_CONF|
NGX_RTMP_REC_CONF|NGX_CONF_TAKE1,
ngx_conf_set_flag_slot,
NGX_RTMP_APP_CONF_OFFSET,
offsetof(ngx_rtmp_record_app_conf_t, notify),
NULL },
{ ngx_string("recorder"),
NGX_RTMP_APP_CONF|NGX_CONF_BLOCK|NGX_CONF_TAKE1,
ngx_rtmp_record_recorder,
@ -166,10 +174,9 @@ ngx_rtmp_record_create_app_conf(ngx_conf_t *cf)
racf->max_frames = NGX_CONF_UNSET;
racf->interval = NGX_CONF_UNSET;
racf->unique = NGX_CONF_UNSET;
racf->notify = NGX_CONF_UNSET;
racf->url = NGX_CONF_UNSET_PTR;
ngx_str_set(&racf->id, "default");
if (ngx_array_init(&racf->rec, cf->pool, 1, sizeof(void *)) != NGX_OK) {
return NULL;
}
@ -190,6 +197,7 @@ ngx_rtmp_record_merge_app_conf(ngx_conf_t *cf, void *parent, void *child)
ngx_conf_merge_size_value(conf->max_size, prev->max_size, 0);
ngx_conf_merge_size_value(conf->max_frames, prev->max_frames, 0);
ngx_conf_merge_value(conf->unique, prev->unique, 0);
ngx_conf_merge_value(conf->notify, prev->notify, 0);
ngx_conf_merge_msec_value(conf->interval, prev->interval,
(ngx_msec_t) NGX_CONF_UNSET);
ngx_conf_merge_bitmask_value(conf->flags, prev->flags, 0);
@ -259,6 +267,7 @@ ngx_int_t
ngx_rtmp_record_open(ngx_rtmp_session_t *s, ngx_uint_t n, ngx_str_t *path)
{
ngx_rtmp_record_rec_ctx_t *rctx;
ngx_int_t rc;
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"record: #%ui manual open", n);
@ -269,8 +278,9 @@ ngx_rtmp_record_open(ngx_rtmp_session_t *s, ngx_uint_t n, ngx_str_t *path)
return NGX_ERROR;
}
if (ngx_rtmp_record_node_open(s, rctx) != NGX_OK) {
return NGX_ERROR;
rc = ngx_rtmp_record_node_open(s, rctx);
if (rc != NGX_OK) {
return rc;
}
if (path) {
@ -285,6 +295,7 @@ ngx_int_t
ngx_rtmp_record_close(ngx_rtmp_session_t *s, ngx_uint_t n, ngx_str_t *path)
{
ngx_rtmp_record_rec_ctx_t *rctx;
ngx_int_t rc;
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"record: #%ui manual close", n);
@ -295,8 +306,9 @@ ngx_rtmp_record_close(ngx_rtmp_session_t *s, ngx_uint_t n, ngx_str_t *path)
return NGX_ERROR;
}
if (ngx_rtmp_record_node_close(s, rctx) != NGX_OK) {
return NGX_ERROR;
rc = ngx_rtmp_record_node_close(s, rctx);
if (rc != NGX_OK) {
return rc;
}
if (path) {
@ -307,6 +319,28 @@ ngx_rtmp_record_close(ngx_rtmp_session_t *s, ngx_uint_t n, ngx_str_t *path)
}
ngx_uint_t
ngx_rtmp_record_find(ngx_rtmp_record_app_conf_t *racf, ngx_str_t *id)
{
ngx_rtmp_record_app_conf_t **pracf, *rracf;
ngx_uint_t n;
pracf = racf->rec.elts;
for (n = 0; n < racf->rec.nelts; ++n, ++pracf) {
rracf = *pracf;
if (rracf->id.len == id->len &&
ngx_strncmp(rracf->id.data, id->data, id->len) == 0)
{
return n;
}
}
return NGX_CONF_UNSET_UINT;
}
/* This funcion returns pointer to a static buffer */
static void
ngx_rtmp_record_make_path(ngx_rtmp_session_t *s,
@ -362,7 +396,7 @@ ngx_rtmp_record_node_open(ngx_rtmp_session_t *s,
rracf = rctx->conf;
if (rctx->file.fd != NGX_INVALID_FILE) {
return NGX_OK;
return NGX_AGAIN;
}
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
@ -397,6 +431,11 @@ ngx_rtmp_record_node_open(ngx_rtmp_session_t *s,
ngx_log_debug2(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"record: %V opened '%V'", &rracf->id, &path);
if (rracf->notify) {
ngx_rtmp_send_status(s, "NetStream.Record.Start", "status",
rracf->id.data ? (char *) rracf->id.data : "");
}
return NGX_OK;
}
@ -526,7 +565,7 @@ ngx_rtmp_record_node_close(ngx_rtmp_session_t *s,
rracf = rctx->conf;
if (rctx->file.fd == NGX_INVALID_FILE) {
return NGX_OK;
return NGX_AGAIN;
}
if (ngx_close_file(rctx->file.fd) == NGX_FILE_ERROR) {
@ -540,6 +579,11 @@ ngx_rtmp_record_node_close(ngx_rtmp_session_t *s,
ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
"record: %V closed", &rracf->id);
if (rracf->notify) {
ngx_rtmp_send_status(s, "NetStream.Record.Stop", "status",
rracf->id.data ? (char *) rracf->id.data : "");
}
app_conf = s->app_conf;
if (rracf->rec_conf) {
@ -589,7 +633,8 @@ next:
static ngx_int_t
ngx_rtmp_record_write_frame(ngx_rtmp_session_t *s,
ngx_rtmp_record_rec_ctx_t *rctx,
ngx_rtmp_header_t *h, ngx_chain_t *in)
ngx_rtmp_header_t *h, ngx_chain_t *in,
ngx_int_t inc_nframes)
{
u_char hdr[11], *p, *ph;
uint32_t timestamp, tag_size;
@ -666,7 +711,7 @@ ngx_rtmp_record_write_frame(ngx_rtmp_session_t *s,
return NGX_ERROR;
}
++rctx->nframes;
rctx->nframes += inc_nframes;
/* watch max size */
if ((rracf->max_size && rctx->file.offset >= (ngx_int_t) rracf->max_size) ||
@ -763,6 +808,12 @@ ngx_rtmp_record_node_av(ngx_rtmp_session_t *s, ngx_rtmp_record_rec_ctx_t *rctx,
}
}
if ((rracf->flags & NGX_RTMP_RECORD_MANUAL) &&
!brkframe && rctx->nframes == 0)
{
return NGX_OK;
}
if (rctx->file.fd == NGX_INVALID_FILE) {
return NGX_OK;
}
@ -815,7 +866,8 @@ ngx_rtmp_record_node_av(ngx_rtmp_session_t *s, ngx_rtmp_record_rec_ctx_t *rctx,
ch.type = NGX_RTMP_MSG_AUDIO;
ch.mlen = ngx_rtmp_record_get_chain_mlen(codec_ctx->aac_header);
if (ngx_rtmp_record_write_frame(s, rctx, &ch, codec_ctx->aac_header)
if (ngx_rtmp_record_write_frame(s, rctx, &ch,
codec_ctx->aac_header, 0)
!= NGX_OK)
{
return NGX_OK;
@ -833,7 +885,8 @@ ngx_rtmp_record_node_av(ngx_rtmp_session_t *s, ngx_rtmp_record_rec_ctx_t *rctx,
ch.type = NGX_RTMP_MSG_VIDEO;
ch.mlen = ngx_rtmp_record_get_chain_mlen(codec_ctx->avc_header);
if (ngx_rtmp_record_write_frame(s, rctx, &ch, codec_ctx->avc_header)
if (ngx_rtmp_record_write_frame(s, rctx, &ch,
codec_ctx->avc_header, 0)
!= NGX_OK)
{
return NGX_OK;
@ -842,7 +895,7 @@ ngx_rtmp_record_node_av(ngx_rtmp_session_t *s, ngx_rtmp_record_rec_ctx_t *rctx,
}
}
return ngx_rtmp_record_write_frame(s, rctx, h, in);
return ngx_rtmp_record_write_frame(s, rctx, h, in, 1);
}

5
ngx_rtmp_record_module.h
View file

@ -26,6 +26,7 @@ typedef struct {
ngx_msec_t interval;
ngx_str_t suffix;
ngx_flag_t unique;
ngx_flag_t notify;
ngx_url_t *url;
void **rec_conf;
@ -50,6 +51,10 @@ typedef struct {
} ngx_rtmp_record_ctx_t;
ngx_uint_t ngx_rtmp_record_find(ngx_rtmp_record_app_conf_t *racf,
ngx_str_t *id);
/* Manual recording control,
* 'n' is record node index in config array.
* Note: these functions allocate path in static buffer */

10
ngx_rtmp_streams.h
View file

@ -15,14 +15,4 @@
#define NGX_RTMP_CSID_VIDEO 7
/*legacy*/
#define NGX_RTMP_CMD_CSID_AMF_INI NGX_RTMP_CSID_AMF_INI
#define NGX_RTMP_CMD_CSID_AMF NGX_RTMP_CSID_AMF
#define NGX_RTMP_CMD_MSID NGX_RTMP_MSID
#define NGX_RTMP_LIVE_CSID_META NGX_RTMP_CSID_AMF
#define NGX_RTMP_LIVE_CSID_AUDIO NGX_RTMP_CSID_AUDIO
#define NGX_RTMP_LIVE_CSID_VIDEO NGX_RTMP_CSID_VIDEO
#define NGX_RTMP_LIVE_MSID NGX_RTMP_MSID
#endif /* _NGX_RTMP_STREAMS_H_INCLUDED_ */