From e7f9a8f3dcef479d38311204ddcbb61345cccf38 Mon Sep 17 00:00:00 2001 From: Juliusz Chroboczek Date: Tue, 26 May 2020 17:44:21 +0200 Subject: [PATCH] Make downTrack and downConnection into interfaces. --- client.go | 24 ++++++++++++------------ conn.go | 56 ++++++++++++++++++++++++++++--------------------------- group.go | 2 +- 3 files changed, 42 insertions(+), 40 deletions(-) diff --git a/client.go b/client.go index 859d64c..7150cce 100644 --- a/client.go +++ b/client.go @@ -618,7 +618,7 @@ func delUpConn(c *client, id string) bool { return true } -func getDownConn(c *client, id string) *downConnection { +func getDownConn(c *client, id string) *rtpDownConnection { if c.down == nil { return nil } @@ -644,7 +644,7 @@ func getConn(c *client, id string) iceConnection { return nil } -func addDownConn(c *client, id string, remote *upConnection) (*downConnection, error) { +func addDownConn(c *client, id string, remote *upConnection) (*rtpDownConnection, error) { pc, err := groups.api.NewPeerConnection(iceConfiguration()) if err != nil { return nil, err @@ -659,9 +659,9 @@ func addDownConn(c *client, id string, remote *upConnection) (*downConnection, e }) if c.down == nil { - c.down = make(map[string]*downConnection) + c.down = make(map[string]*rtpDownConnection) } - conn := &downConnection{ + conn := &rtpDownConnection{ id: id, client: c, pc: pc, @@ -705,7 +705,7 @@ func delDownConn(c *client, id string) bool { return true } -func addDownTrack(c *client, conn *downConnection, remoteTrack *upTrack, remoteConn *upConnection) (*webrtc.RTPSender, error) { +func addDownTrack(c *client, conn *rtpDownConnection, remoteTrack *upTrack, remoteConn *upConnection) (*webrtc.RTPSender, error) { local, err := conn.pc.NewTrack( remoteTrack.track.PayloadType(), remoteTrack.track.SSRC(), @@ -721,7 +721,7 @@ func addDownTrack(c *client, conn *downConnection, remoteTrack *upTrack, remoteC return nil, err } - track := &downTrack{ + track := &rtpDownTrack{ track: local, remote: remoteTrack, maxLossBitrate: new(bitrate), @@ -742,7 +742,7 @@ const ( maxLossRate = 1 << 30 ) -func (track *downTrack) updateRate(loss uint8, now uint64) { +func (track *rtpDownTrack) updateRate(loss uint8, now uint64) { rate := track.maxLossBitrate.Get(now) if rate > maxLossRate { // no recent feedback, reset @@ -771,7 +771,7 @@ func (track *downTrack) updateRate(loss uint8, now uint64) { track.maxLossBitrate.Set(rate, now) } -func rtcpDownListener(conn *downConnection, track *downTrack, s *webrtc.RTPSender) { +func rtcpDownListener(conn *rtpDownConnection, track *rtpDownTrack, s *webrtc.RTPSender) { for { ps, err := s.ReadRTCP() if err != nil { @@ -820,7 +820,7 @@ func rtcpDownListener(conn *downConnection, track *downTrack, s *webrtc.RTPSende } } -func trackKinds(down *downConnection) (audio bool, video bool) { +func trackKinds(down *rtpDownConnection) (audio bool, video bool) { if down.pc == nil { return } @@ -926,7 +926,7 @@ func sendNACK(pc *webrtc.PeerConnection, ssrc uint32, first uint16, bitmap uint1 return pc.WriteRTCP([]rtcp.Packet{packet}) } -func sendRecovery(p *rtcp.TransportLayerNack, track *downTrack) { +func sendRecovery(p *rtcp.TransportLayerNack, track *rtpDownTrack) { var packet rtp.Packet buf := make([]byte, packetcache.BufSize) for _, nack := range p.Nacks { @@ -949,7 +949,7 @@ func sendRecovery(p *rtcp.TransportLayerNack, track *downTrack) { } } -func negotiate(c *client, down *downConnection) error { +func negotiate(c *client, down *rtpDownConnection) error { offer, err := down.pc.CreateOffer(nil) if err != nil { return err @@ -1094,7 +1094,7 @@ func (c *client) isRequested(label string) bool { return c.requested[label] != 0 } -func addDownConnTracks(c *client, remote *upConnection, tracks []*upTrack) (*downConnection, error) { +func addDownConnTracks(c *client, remote *upConnection, tracks []*upTrack) (*rtpDownConnection, error) { requested := false for _, t := range tracks { if c.isRequested(t.label) { diff --git a/conn.go b/conn.go index da86cb6..36567ea 100644 --- a/conn.go +++ b/conn.go @@ -33,7 +33,7 @@ type upTrack struct { writerDone chan struct{} // closed when the loop dies mu sync.Mutex - local []*downTrack + local []downTrack } func (up *upTrack) notifyLocal() { @@ -44,7 +44,7 @@ func (up *upTrack) notifyLocal() { } } -func (up *upTrack) addLocal(local *downTrack) { +func (up *upTrack) addLocal(local downTrack) { up.mu.Lock() for _, t := range up.local { if t == local { @@ -57,7 +57,7 @@ func (up *upTrack) addLocal(local *downTrack) { up.notifyLocal() } -func (up *upTrack) delLocal(local *downTrack) bool { +func (up *upTrack) delLocal(local downTrack) bool { up.mu.Lock() for i, l := range up.local { if l == local { @@ -71,10 +71,10 @@ func (up *upTrack) delLocal(local *downTrack) bool { return false } -func (up *upTrack) getLocal() []*downTrack { +func (up *upTrack) getLocal() []downTrack { up.mu.Lock() defer up.mu.Unlock() - local := make([]*downTrack, len(up.local)) + local := make([]downTrack, len(up.local)) copy(local, up.local) return local } @@ -102,14 +102,10 @@ type upConnection struct { iceCandidates []*webrtc.ICECandidateInit mu sync.Mutex - local []*downConnection + local []downConnection } -func (up *upConnection) getPC() *webrtc.PeerConnection { - return up.pc -} - -func (up *upConnection) addLocal(local *downConnection) { +func (up *upConnection) addLocal(local downConnection) { up.mu.Lock() defer up.mu.Unlock() for _, t := range up.local { @@ -121,7 +117,7 @@ func (up *upConnection) addLocal(local *downConnection) { up.local = append(up.local, local) } -func (up *upConnection) delLocal(local *downConnection) bool { +func (up *upConnection) delLocal(local downConnection) bool { up.mu.Lock() defer up.mu.Unlock() for i, l := range up.local { @@ -133,10 +129,10 @@ func (up *upConnection) delLocal(local *downConnection) bool { return false } -func (up *upConnection) getLocal() []*downConnection { +func (up *upConnection) getLocal() []downConnection { up.mu.Lock() defer up.mu.Unlock() - local := make([]*downConnection, len(up.local)) + local := make([]downConnection, len(up.local)) copy(local, up.local) return local } @@ -239,7 +235,13 @@ func (s *receiverStats) Get(now uint64) (uint8, uint32) { return uint8(atomic.LoadUint32(&s.loss)), atomic.LoadUint32(&s.jitter) } -type downTrack struct { +type downTrack interface { + WriteRTP(packat *rtp.Packet) error + Accumulate(bytes uint32) + GetMaxBitrate(now uint64) uint64 +} + +type rtpDownTrack struct { track *webrtc.Track remote *upTrack maxLossBitrate *bitrate @@ -248,15 +250,15 @@ type downTrack struct { stats *receiverStats } -func (down *downTrack) WriteRTP(packet *rtp.Packet) error { +func (down *rtpDownTrack) WriteRTP(packet *rtp.Packet) error { return down.track.WriteRTP(packet) } -func (down *downTrack) Accumulate(bytes uint32) { +func (down *rtpDownTrack) Accumulate(bytes uint32) { down.rate.Add(bytes) } -func (down *downTrack) GetMaxBitrate(now uint64) uint64 { +func (down *rtpDownTrack) GetMaxBitrate(now uint64) uint64 { br1 := down.maxLossBitrate.Get(now) br2 := down.maxREMBBitrate.Get(now) if br1 < br2 { @@ -265,24 +267,24 @@ func (down *downTrack) GetMaxBitrate(now uint64) uint64 { return br2 } -type downConnection struct { +type downConnection interface { + Close() error +} + +type rtpDownConnection struct { id string client *client pc *webrtc.PeerConnection remote *upConnection - tracks []*downTrack + tracks []*rtpDownTrack iceCandidates []*webrtc.ICECandidateInit } -func (down *downConnection) Close() error { +func (down *rtpDownConnection) Close() error { return down.client.action(delConnAction{down.id}) } -func (down *downConnection) getPC() *webrtc.PeerConnection { - return down.pc -} - -func (down *downConnection) addICECandidate(candidate *webrtc.ICECandidateInit) error { +func (down *rtpDownConnection) addICECandidate(candidate *webrtc.ICECandidateInit) error { if down.pc.RemoteDescription() != nil { return down.pc.AddICECandidate(*candidate) } @@ -290,7 +292,7 @@ func (down *downConnection) addICECandidate(candidate *webrtc.ICECandidateInit) return nil } -func (down *downConnection) flushICECandidates() error { +func (down *rtpDownConnection) flushICECandidates() error { err := flushICECandidates(down.pc, down.iceCandidates) down.iceCandidates = nil return err diff --git a/group.go b/group.go index e33535b..0d5a2a3 100644 --- a/group.go +++ b/group.go @@ -33,7 +33,7 @@ type client struct { actionCh chan interface{} mu sync.Mutex - down map[string]*downConnection + down map[string]*rtpDownConnection up map[string]*upConnection }