Fix xiaomi/cs2 stream restart on pop buffer overflow

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
+```

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:

pkg/xiaomi/miss/cs2/conn.go

@@ -12,19 +12,23 @@ 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
 	}
 
 	_, 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:]