Resolve issues with multiple audio streams

pkg/web/client.go

@@ -20,13 +20,14 @@ type Client struct {
 	In       chan *Message
 	Out      chan *Message
 
-	AudioTrack *webrtc.Track
+	AudioTrack     *webrtc.Track
+	SequenceNumber uint16
 
 	Terminated chan bool
 }
 
 func NewClient(conn *websocket.Conn) *Client {
-	c := Client{Conn: conn, Name: "Anonymous", In: make(chan *Message, 10), Out: make(chan *Message, 10), Terminated: make(chan bool)}
+	c := Client{Conn: conn, Name: "Anonymous", In: make(chan *Message, 10), Out: make(chan *Message, 10), SequenceNumber: 1, Terminated: make(chan bool)}
 
 	go c.handleRead()
 	go c.handleWrite()
@@ -116,3 +117,14 @@ func (c *Client) Close() {
 		c.Terminated <- true
 	}()
 }
+
+func (c *Client) ClosePeerConn() {
+	if c.PeerConn == nil {
+		return
+	}
+
+	c.PeerConn.Close()
+	c.AudioTrack = nil
+	c.PeerConn = nil
+	c.SequenceNumber = 1
+}

pkg/web/public/assets/js/harmony.js

@@ -95,7 +95,7 @@ $(document).ready(function () {
             pc.getSenders()[0].replaceTrack(null);
 
             pc.ontrack = function (event) {
-                console.log(`onTrack ${event.streams.length}`);
+                console.log(`onTrack ${event.streams.length} ${event.streams[0].getAudioTracks().length}`);
 
                 pc.addTransceiver(event.streams[0].getAudioTracks()[0], {'direction': 'sendrecv'});
 
@@ -316,11 +316,12 @@ function Connect() {
             return;
         }
 
+        ReconnectDelay += (ReconnectDelay * 2) + 1;
+
         var waitTime = ReconnectDelay;
         console.log("Reconnecting in " + ReconnectDelay + " seconds...");
         reconnectTimeout = setTimeout(Connect, waitTime * 1000);
 
-        ReconnectDelay += (ReconnectDelay * 2) + 1;
         if (ReconnectDelay > 10) {
             ReconnectDelay = 10;
         }

pkg/web/web.go

@@ -6,7 +6,6 @@ import (
 	"math/rand"
 	"net/http"
 	"sync"
-	"time"
 
 	rice "github.com/GeertJohan/go.rice"
 	"github.com/gorilla/mux"
@@ -56,7 +55,6 @@ func NewWebInterface(address string, path string) *WebInterface {
 	r.PathPrefix(path).Handler(http.StripPrefix(path, http.FileServer(rice.MustFindBox("public").HTTPBox())))
 
 	go w.handleIncomingClients()
-	go w.handleTerminatedClients()
 
 	go func() {
 		if err := http.ListenAndServe(address, r); err != nil {
@@ -84,27 +82,6 @@ func (w *WebInterface) handleIncomingClients() {
 	}
 }
 
-func (w *WebInterface) handleTerminatedClients() {
-	for {
-		time.Sleep(15 * time.Second)
-
-		w.ClientsLock.Lock()
-		for id := range w.Clients {
-			if w.Clients[id].Status != -1 {
-				continue
-			}
-
-			name := w.Clients[id].Name
-			delete(w.Clients, id)
-
-			for _, wc := range w.Clients {
-				wc.Out <- &Message{T: MessageDisconnect, M: []byte(name)}
-			}
-		}
-		w.ClientsLock.Unlock()
-	}
-}
-
 func (w *WebInterface) handleRead(c *Client) {
 	for msg := range c.In {
 		if msg == nil {
@@ -146,8 +123,9 @@ func (w *WebInterface) handleRead(c *Client) {
 				c.AudioTrack = nil
 
 				if c.PeerConn != nil {
-					c.PeerConn.Close()
-					c.PeerConn = nil
+					c.ClosePeerConn()
+
+					log.Printf("closing peerconn, peer sent %s %d", msg.T, c.ID)
 				}
 
 				if msg.T == MessageDisconnect {
@@ -195,6 +173,21 @@ func (w *WebInterface) webSocketHandler(wr http.ResponseWriter, r *http.Request)
 	incomingClients <- c
 
 	<-c.Terminated
+
+	w.ClientsLock.Lock()
+	for id := range w.Clients {
+		if w.Clients[id].Status != -1 {
+			continue
+		}
+
+		name := w.Clients[id].Name
+		delete(w.Clients, id)
+
+		for _, wc := range w.Clients {
+			wc.Out <- &Message{T: MessageDisconnect, M: []byte(name)}
+		}
+	}
+	w.ClientsLock.Unlock()
 }
 
 func (w *WebInterface) answerRTC(c *Client, sdp []byte) ([]byte, error) {
@@ -238,7 +231,9 @@ func (w *WebInterface) answerRTC(c *Client, sdp []byte) ([]byte, error) {
 		}
 	}
 	if payloadType == 0 {
-		panic("no payloadType")
+		c.ClosePeerConn()
+
+		return nil, errors.New("no payloadType")
 	}
 
 	t, err := pc.NewTrack(payloadType, rand.Uint32(), "audio", "harmony")
@@ -248,7 +243,7 @@ func (w *WebInterface) answerRTC(c *Client, sdp []byte) ([]byte, error) {
 
 	c.AudioTrack = t
 
-	_, err = pc.AddTrack(t)
+	_, err = pc.AddTrack(c.AudioTrack)
 	if err != nil {
 		panic(err)
 	}
@@ -261,8 +256,8 @@ func (w *WebInterface) answerRTC(c *Client, sdp []byte) ([]byte, error) {
 		for {
 			p, err = remoteTrack.ReadRTP()
 			if err != nil {
-				c.AudioTrack = nil
-				c.PeerConn.Close()
+				c.ClosePeerConn()
+				log.Printf("failed to read RTP from %d", c.ID)
 				return
 			}
 
@@ -274,10 +269,13 @@ func (w *WebInterface) answerRTC(c *Client, sdp []byte) ([]byte, error) {
 
 				p.SSRC = wc.AudioTrack.SSRC()
 				p.PayloadType = wc.AudioTrack.PayloadType()
+				p.SequenceNumber = wc.SequenceNumber
 
 				if err = wc.AudioTrack.WriteRTP(p); err != nil && err != io.ErrClosedPipe {
 					panic(err)
 				}
+
+				wc.SequenceNumber++
 			}
 			w.ClientsLock.Unlock()
 		}
@@ -290,10 +288,9 @@ func (w *WebInterface) answerRTC(c *Client, sdp []byte) ([]byte, error) {
 			w.ClientsLock.Lock()
 
 			if connectionState == webrtc.PeerConnectionStateDisconnected {
-				c.AudioTrack = nil
+				c.ClosePeerConn()
 
-				c.PeerConn.Close()
-				c.PeerConn = nil
+				log.Printf("closing peerconn, peer disconnected %d", c.ID)
 			}
 
 			for _, wc := range w.Clients {