Add received audio data handling

audiostream.go

@@ -1,17 +1,53 @@
 package main
 
-import "github.com/nonoo/kappanhang/log"
+import (
+	"bytes"
+	"encoding/binary"
+	"time"
+
+	"github.com/nonoo/kappanhang/log"
+)
+
+const audioTimeoutDuration = 3 * time.Second
 
 type audioStream struct {
 	common streamCommon
+
+	timeoutTimer         *time.Timer
+	lastReceivedAudioSeq uint16
 }
 
 func (s *audioStream) sendDisconnect() {
 	s.common.sendDisconnect()
 }
 
+func (s *audioStream) handleAudioPacket(r []byte) {
+	gotSeq := binary.LittleEndian.Uint16(r[6:8])
+
+	if s.timeoutTimer != nil {
+		s.timeoutTimer.Stop()
+		s.timeoutTimer.Reset(audioTimeoutDuration)
+	}
+
+	expectedSeq := s.lastReceivedAudioSeq + 1
+	if expectedSeq != gotSeq {
+		var missingPkts int
+		if gotSeq > expectedSeq {
+			missingPkts = int(gotSeq) - int(expectedSeq)
+		} else {
+			missingPkts = int(gotSeq) + 65536 - int(expectedSeq)
+		}
+		log.Error("lost ", missingPkts, " audio packets")
+	}
+	s.lastReceivedAudioSeq = gotSeq
+
+	// log.Print("got audio packet ", len(r[24:]), " bytes")
+}
+
 func (s *audioStream) handleRead(r []byte) {
-	// TODO
+	if len(r) >= 580 && (bytes.Equal(r[:6], []byte{0x6c, 0x05, 0x00, 0x00, 0x00, 0x00}) || bytes.Equal(r[:6], []byte{0x44, 0x02, 0x00, 0x00, 0x00, 0x00})) {
+		s.handleAudioPacket(r)
+	}
 }
 
 func (s *audioStream) start() {
@@ -24,6 +60,8 @@ func (s *audioStream) start() {
 
 	log.Print("stream opened")
 
+	s.timeoutTimer = time.NewTimer(audioTimeoutDuration)
+
 	s.common.pkt7.sendSeq = 1
 	s.common.pkt7.startPeriodicSend(&s.common)
 
@@ -32,6 +70,8 @@ func (s *audioStream) start() {
 		select {
 		case r = <-s.common.readChan:
 			s.handleRead(r)
+		case <-s.timeoutTimer.C:
+			log.Fatal("timeout")
 		}
 	}
 }

go.mod

@@ -4,5 +4,6 @@ go 1.14
 
 require (
 	github.com/pborman/getopt v1.1.0
+	github.com/sqs/goreturns v0.0.0-20181028201513-538ac6014518 // indirect
 	go.uber.org/zap v1.16.0
 )

go.sum

@@ -11,6 +11,8 @@ github.com/pborman/getopt v1.1.0/go.mod h1:FxXoW1Re00sQG/+KIkuSqRL/LwQgSkv7uyac+
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+github.com/sqs/goreturns v0.0.0-20181028201513-538ac6014518 h1:iD+PFTQwKEmbwSdwfvP5ld2WEI/g7qbdhmHJ2ASfYGs=
+github.com/sqs/goreturns v0.0.0-20181028201513-538ac6014518/go.mod h1:CKI4AZ4XmGV240rTHfO0hfE83S6/a3/Q1siZJ/vXf7A=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=

pkt7.go

@@ -9,6 +9,8 @@ import (
 	"github.com/nonoo/kappanhang/log"
 )
 
+const pkt7TimeoutDuration = 3 * time.Second
+
 type pkt7Type struct {
 	sendSeq          uint16
 	randIDByte       [1]byte
@@ -49,7 +51,7 @@ func (p *pkt7Type) handle(s *streamCommon, r []byte) {
 			} else {
 				missingPkts = int(gotSeq) + 65536 - int(expectedSeq)
 			}
-			log.Error(s.name+"/lost ", missingPkts, " packets ")
+			log.Error(s.name+"/lost ", missingPkts, " packets")
 		}
 		p.lastConfirmedSeq = gotSeq
 	}
@@ -102,7 +104,7 @@ func (p *pkt7Type) startPeriodicSend(s *streamCommon) {
 			case <-p.sendTicker.C:
 				p.send(s)
 			case <-p.timeoutTimer.C:
-				log.Fatal(s.name + "ping timeout")
+				log.Fatal(s.name + "/ping timeout")
 			}
 		}
 	}()

streamcommon.go

@@ -11,8 +11,6 @@ import (
 	"github.com/nonoo/kappanhang/log"
 )
 
-const pkt7TimeoutDuration = 3 * time.Second
-
 type streamCommon struct {
 	name      string
 	conn      *net.UDPConn