diff --git a/pkg/xiaomi/miss/cs2/conn.go b/pkg/xiaomi/miss/cs2/conn.go index 2c1b395e9..619085fc4 100644 --- a/pkg/xiaomi/miss/cs2/conn.go +++ b/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:] diff --git a/pkg/xiaomi/miss/cs2/data_channel_test.go b/pkg/xiaomi/miss/cs2/data_channel_test.go new file mode 100644 index 000000000..d90a27af7 --- /dev/null +++ b/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) + } +}