Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

merged sync branch

  • Loading branch information...
commit 101b43a478bb44018733e75a3bf3b1f8adc5b2e5 2 parents 4adc5f7 + 9b4725a
@arut authored
View
5 hls/ngx_rtmp_hls_module.c
@@ -992,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,
View
2  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;
View
7 ngx_rtmp_handler.c
@@ -403,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;
}
}
@@ -701,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);
View
95 ngx_rtmp_live_module.c
@@ -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,17 +346,22 @@ 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.timestamp = timestamp;
ch.msid = NGX_RTMP_LIVE_MSID;
ch.type = h->type;
lh.msid = ch.msid;
@@ -345,14 +370,16 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
ch.csid = NGX_RTMP_LIVE_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;
+ prio = 0;
ch.csid = NGX_RTMP_LIVE_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);
View
2  ngx_rtmp_live_module.h
@@ -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;
Please sign in to comment.
Something went wrong with that request. Please try again.