merged vod-seek

TODO

@@ -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

config

@@ -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 \

hls/ngx_rtmp_hls_module.c

@@ -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, 

ngx_rtmp.h

@@ -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;
 

ngx_rtmp_cmd_module.c

@@ -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