Compare commits
1 Commits
723ae1f1ac
...
v2
| Author | SHA1 | Date | |
|---|---|---|---|
| 03296e7572 |
@@ -32,6 +32,9 @@ func (a *App) HandleTS(w http.ResponseWriter, r *http.Request) {
|
|||||||
return
|
return
|
||||||
case <-sub.Done():
|
case <-sub.Done():
|
||||||
return
|
return
|
||||||
|
case <-sub.Overflow():
|
||||||
|
// Client fell too far behind; close so it reconnects and resyncs.
|
||||||
|
return
|
||||||
case chunk, ok := <-sub.Chan():
|
case chunk, ok := <-sub.Chan():
|
||||||
if !ok {
|
if !ok {
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -16,13 +16,15 @@ import (
|
|||||||
const chunkSize = 64 * 1024
|
const chunkSize = 64 * 1024
|
||||||
|
|
||||||
// subscriberBuffer is how many chunks may queue per client before it is
|
// subscriberBuffer is how many chunks may queue per client before it is
|
||||||
// considered too slow and dropped (prevents one slow client stalling others).
|
// considered too slow. At ~64KB/chunk this is a large cushion so transient
|
||||||
const subscriberBuffer = 256
|
// MSE backpressure never overflows for a normal viewer.
|
||||||
|
const subscriberBuffer = 2048
|
||||||
|
|
||||||
// Subscriber receives copies of upstream TS chunks.
|
// Subscriber receives copies of upstream TS chunks.
|
||||||
type subscriber struct {
|
type subscriber struct {
|
||||||
ch chan []byte
|
ch chan []byte
|
||||||
closed chan struct{}
|
closed chan struct{}
|
||||||
|
overflow chan struct{} // closed by the hub when the client falls too far behind
|
||||||
}
|
}
|
||||||
|
|
||||||
// Hub manages a single shared upstream connection and fans its bytes out to
|
// Hub manages a single shared upstream connection and fans its bytes out to
|
||||||
@@ -120,6 +122,7 @@ func (h *Hub) Subscribe() (*subscriber, func()) {
|
|||||||
sub := &subscriber{
|
sub := &subscriber{
|
||||||
ch: make(chan []byte, subscriberBuffer),
|
ch: make(chan []byte, subscriberBuffer),
|
||||||
closed: make(chan struct{}),
|
closed: make(chan struct{}),
|
||||||
|
overflow: make(chan struct{}),
|
||||||
}
|
}
|
||||||
|
|
||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
@@ -149,7 +152,13 @@ func (s *subscriber) Chan() <-chan []byte { return s.ch }
|
|||||||
// Closed signals the subscriber has been removed by the hub.
|
// Closed signals the subscriber has been removed by the hub.
|
||||||
func (s *subscriber) Done() <-chan struct{} { return s.closed }
|
func (s *subscriber) Done() <-chan struct{} { return s.closed }
|
||||||
|
|
||||||
// broadcast sends a chunk to all subscribers, dropping any that are too slow.
|
// Overflow signals the client fell too far behind and must reconnect. We never
|
||||||
|
// drop bytes mid-stream (that corrupts TS packet alignment); instead we close
|
||||||
|
// the connection so the client reconnects and resyncs from a packet boundary.
|
||||||
|
func (s *subscriber) Overflow() <-chan struct{} { return s.overflow }
|
||||||
|
|
||||||
|
// broadcast sends a chunk to all subscribers. A subscriber whose buffer is full
|
||||||
|
// is flagged for disconnect rather than having bytes silently dropped.
|
||||||
func (h *Hub) broadcast(chunk []byte) {
|
func (h *Hub) broadcast(chunk []byte) {
|
||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
if !h.gotData {
|
if !h.gotData {
|
||||||
@@ -165,7 +174,14 @@ func (h *Hub) broadcast(chunk []byte) {
|
|||||||
select {
|
select {
|
||||||
case s.ch <- chunk:
|
case s.ch <- chunk:
|
||||||
default:
|
default:
|
||||||
// Slow client: drop this chunk for them rather than blocking everyone.
|
// Buffer full: signal overflow once so the handler closes the
|
||||||
|
// connection cleanly. Never drop bytes — that breaks TS alignment.
|
||||||
|
select {
|
||||||
|
case <-s.overflow:
|
||||||
|
// already flagged
|
||||||
|
default:
|
||||||
|
close(s.overflow)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -83,8 +83,13 @@ export default function Player({ live, streamKey }: Props) {
|
|||||||
}, 5000);
|
}, 5000);
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
function start() {
|
function start() {
|
||||||
destroyPlayer();
|
destroyPlayer();
|
||||||
|
// Config mirrors webplayer.online (proven on this exact bursty upstream):
|
||||||
|
// a deep 30-90s live buffer absorbs the upstream's multi-second delivery
|
||||||
|
// gaps, while gentle 1.1x catch-up keeps latency bounded.
|
||||||
const player = mpegts.createPlayer(
|
const player = mpegts.createPlayer(
|
||||||
{ type: "mpegts", isLive: true, url: src },
|
{ type: "mpegts", isLive: true, url: src },
|
||||||
{
|
{
|
||||||
@@ -96,7 +101,11 @@ export default function Player({ live, streamKey }: Props) {
|
|||||||
liveSyncTargetLatency: 45,
|
liveSyncTargetLatency: 45,
|
||||||
liveSyncPlaybackRate: 1.1,
|
liveSyncPlaybackRate: 1.1,
|
||||||
liveBufferLatencyChasing: true,
|
liveBufferLatencyChasing: true,
|
||||||
|
liveBufferLatencyMaxLatency: 90,
|
||||||
|
liveBufferLatencyMinRemain: 30,
|
||||||
autoCleanupSourceBuffer: true,
|
autoCleanupSourceBuffer: true,
|
||||||
|
autoCleanupMaxBackwardDuration: 120,
|
||||||
|
autoCleanupMinBackwardDuration: 90,
|
||||||
fixAudioTimestampGap: true,
|
fixAudioTimestampGap: true,
|
||||||
lazyLoad: false,
|
lazyLoad: false,
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user