Keep Xiaomi CS2 control notifications from stopping media

pkg/xiaomi/miss/cs2/conn.go

@@ -20,11 +20,18 @@ func Dial(host, transport string) (*Conn, error) {
 
 	_, isTCP := conn.(*tcpConn)
 
+	commandChannel := newDataChannel(0, 10)
+	// Some cameras keep sending unsolicited control notifications after login.
+	// There is no long-running command reader, so retaining every notification
+	// eventually fills the queue and used to tear down the media connection.
+	// Keep the newest notifications while leaving the media channel strict.
+	commandChannel.dropOldest = true
+
 	c := &Conn{
 		Conn:  conn,
 		isTCP: isTCP,
 		channels: [4]*dataChannel{
-			newDataChannel(0, 10), nil, newDataChannel(250, 100), nil,
+			commandChannel, nil, newDataChannel(250, 100), nil,
 		},
 	}
 	go c.worker()
@@ -427,6 +434,9 @@ type dataChannel struct {
 	waitData []byte
 	waitSize int
 	popBuf   chan []byte
+	// dropOldest is only enabled for the control channel. Media channels must
+	// remain strict because silently dropping media corrupts the stream.
+	dropOldest bool
 }
 
 func (c *dataChannel) Push(b []byte) error {
@@ -442,10 +452,24 @@ func (c *dataChannel) Push(b []byte) error {
 			break
 		}
 
+		data := c.waitData[:c.waitSize]
 		select {
-		case c.popBuf <- c.waitData[:c.waitSize]:
+		case c.popBuf <- data:
 		default:
-			return fmt.Errorf("pop buffer is full")
+			if !c.dropOldest {
+				return fmt.Errorf("pop buffer is full")
+			}
+
+			// Preserve recent control responses/notifications without allowing an
+			// unread notification queue to terminate the video connection.
+			select {
+			case <-c.popBuf:
+			default:
+			}
+			select {
+			case c.popBuf <- data:
+			default:
+			}
 		}
 
 		c.waitData = c.waitData[c.waitSize:]

pkg/xiaomi/miss/cs2/data_channel_test.go

@@ -0,0 +1,43 @@
+package cs2
+
+import (
+	"encoding/binary"
+	"testing"
+)
+
+func framedData(payload ...byte) []byte {
+	data := make([]byte, 4+len(payload))
+	binary.BigEndian.PutUint32(data, uint32(len(payload)))
+	copy(data[4:], payload)
+	return data
+}
+
+func TestDataChannelMediaQueueRemainsStrict(t *testing.T) {
+	channel := newDataChannel(0, 1)
+	if err := channel.Push(framedData(1)); err != nil {
+		t.Fatal(err)
+	}
+	if err := channel.Push(framedData(2)); err == nil || err.Error() != "pop buffer is full" {
+		t.Fatalf("expected strict full-buffer error, got %v", err)
+	}
+}
+
+func TestDataChannelControlQueueKeepsNewestMessage(t *testing.T) {
+	channel := newDataChannel(0, 1)
+	channel.dropOldest = true
+
+	if err := channel.Push(framedData(1)); err != nil {
+		t.Fatal(err)
+	}
+	if err := channel.Push(framedData(2)); err != nil {
+		t.Fatal(err)
+	}
+
+	data, ok := channel.Pop()
+	if !ok {
+		t.Fatal("control channel closed unexpectedly")
+	}
+	if len(data) != 1 || data[0] != 2 {
+		t.Fatalf("expected newest control message, got %x", data)
+	}
+}