diff --git a/internal/xiaomi/README.md b/internal/xiaomi/README.md index 330183e8c..87eb1dca8 100644 --- a/internal/xiaomi/README.md +++ b/internal/xiaomi/README.md @@ -62,3 +62,10 @@ You can use a second channel for dual cameras: `channel=2`. streams: xiaomi1: xiaomi://***&channel=2 ``` + +You can change the media buffer size (in packets) if your consumers are slow: `buffer=256` (default). + +```yaml +streams: + xiaomi1: xiaomi://***&buffer=1024 +``` diff --git a/pkg/xiaomi/miss/client.go b/pkg/xiaomi/miss/client.go index 3113632a1..8c5519005 100644 --- a/pkg/xiaomi/miss/client.go +++ b/pkg/xiaomi/miss/client.go @@ -8,6 +8,7 @@ import ( "fmt" "net" "net/url" + "strconv" "time" "github.com/AlexxIT/go2rtc/pkg/tutk" @@ -55,7 +56,8 @@ func NewClient(rawURL string) (*Client, error) { var conn Conn switch s := query.Get("vendor"); s { case "cs2": - conn, err = cs2.Dial(u.Host, query.Get("transport")) + popSize, _ := strconv.Atoi(query.Get("buffer")) + conn, err = cs2.Dial(u.Host, query.Get("transport"), popSize) case "tutk": conn, err = tutk.Dial(u.Host, query.Get("uid"), "Miss", "client") default: diff --git a/pkg/xiaomi/miss/cs2/conn.go b/pkg/xiaomi/miss/cs2/conn.go index 2c1b395e9..b0577d1e3 100644 --- a/pkg/xiaomi/miss/cs2/conn.go +++ b/pkg/xiaomi/miss/cs2/conn.go @@ -12,7 +12,7 @@ import ( "time" ) -func Dial(host, transport string) (*Conn, error) { +func Dial(host, transport string, popSize int) (*Conn, error) { conn, err := handshake(host, transport) if err != nil { return nil, err @@ -20,11 +20,15 @@ func Dial(host, transport string) (*Conn, error) { _, isTCP := conn.(*tcpConn) + if popSize <= 0 { + popSize = 256 + } + c := &Conn{ Conn: conn, isTCP: isTCP, channels: [4]*dataChannel{ - newDataChannel(0, 10), nil, newDataChannel(250, 100), nil, + newDataChannel(0, 10), nil, newDataChannel(250, popSize), nil, }, } go c.worker() @@ -445,7 +449,16 @@ func (c *dataChannel) Push(b []byte) error { select { case c.popBuf <- c.waitData[:c.waitSize]: default: - return fmt.Errorf("pop buffer is full") + // Slow consumer or retransmit burst. Drop the oldest packet + // instead of killing the whole connection (stream restart). + select { + case <-c.popBuf: + default: + } + select { + case c.popBuf <- c.waitData[:c.waitSize]: + default: // consumer raced us, drop the new packet + } } c.waitData = c.waitData[c.waitSize:]