diff --git a/client.go b/client.go index 7150cce..7b0919c 100644 --- a/client.go +++ b/client.go @@ -366,7 +366,7 @@ func addUpConn(c *client, id string) (*upConnection, error) { rate: estimator.New(time.Second), jitter: jitter.New(remote.Codec().ClockRate), maxBitrate: ^uint64(0), - localCh: make(chan struct{}, 2), + localCh: make(chan localTrackAction, 2), writerDone: make(chan struct{}), } u.tracks = append(u.tracks, track) @@ -380,16 +380,16 @@ func addUpConn(c *client, id string) (*upConnection, error) { } c.mu.Unlock() + go readLoop(conn, track) + + go rtcpUpListener(conn, track, receiver) + if tracks != nil { clients := c.group.getClients(c) for _, cc := range clients { pushConn(cc, u, tracks, u.label) } } - - go readLoop(conn, track) - - go rtcpUpListener(conn, track, receiver) }) return conn, nil @@ -469,12 +469,26 @@ func writeLoop(conn *upConnection, track *upTrack, ch <-chan packetIndex) { buf := make([]byte, packetcache.BufSize) var packet rtp.Packet - local := track.getLocal() + local := make([]downTrack, 0) for { select { - case <-track.localCh: - local = track.getLocal() + case action := <-track.localCh: + if action.add { + local = append(local, action.track) + } else { + found := false + for i, t := range local { + if t == action.track { + local = append(local[:i], local[i+1:]...) + found = true + break + } + } + if !found { + log.Printf("Deleting unknown track!") + } + } case pi, ok := <-ch: if !ok { return diff --git a/conn.go b/conn.go index 36567ea..5edb185 100644 --- a/conn.go +++ b/conn.go @@ -18,6 +18,11 @@ import ( "github.com/pion/webrtc/v2" ) +type localTrackAction struct { + add bool + track downTrack +} + type upTrack struct { track *webrtc.Track label string @@ -29,17 +34,16 @@ type upTrack struct { lastSenderReport uint32 lastSenderReportTime uint32 - localCh chan struct{} // signals that local has changed - writerDone chan struct{} // closed when the loop dies + localCh chan localTrackAction // signals that local has changed + writerDone chan struct{} // closed when the loop dies mu sync.Mutex local []downTrack } -func (up *upTrack) notifyLocal() { - var s struct{} +func (up *upTrack) notifyLocal(add bool, track downTrack) { select { - case up.localCh <- s: + case up.localCh <- localTrackAction{add, track}: case <-up.writerDone: } } @@ -54,7 +58,7 @@ func (up *upTrack) addLocal(local downTrack) { } up.local = append(up.local, local) up.mu.Unlock() - up.notifyLocal() + up.notifyLocal(true, local) } func (up *upTrack) delLocal(local downTrack) bool { @@ -63,7 +67,7 @@ func (up *upTrack) delLocal(local downTrack) bool { if l == local { up.local = append(up.local[:i], up.local[i+1:]...) up.mu.Unlock() - up.notifyLocal() + up.notifyLocal(false, l) return true } }