From a813cc9ce44743813b7ba42d68c120faafe99b74 Mon Sep 17 00:00:00 2001 From: Juliusz Chroboczek Date: Wed, 29 Apr 2020 01:57:37 +0200 Subject: [PATCH] Split out sending loop into a separate function. --- client.go | 107 +++++++++++++++++++++++++++--------------------------- 1 file changed, 54 insertions(+), 53 deletions(-) diff --git a/client.go b/client.go index a72256f..dc4841e 100644 --- a/client.go +++ b/client.go @@ -286,10 +286,9 @@ func addUpConn(c *client, id string) (*upConnection, error) { c.mu.Unlock() return } - list := packetlist.New(32) track := &upTrack{ track: remote, - list: list, + list: packetlist.New(32), maxBitrate: ^uint64(0), } u.tracks = append(u.tracks, track) @@ -304,62 +303,64 @@ func addUpConn(c *client, id string) (*upConnection, error) { } } - go func() { - buf := make([]byte, packetlist.BufSize) - var packet rtp.Packet - var local []*downTrack - var localTime time.Time - window := packetwindow.New() - for { - now := time.Now() - if now.Sub(localTime) > time.Second/2 { - local = track.getLocal() - localTime = now - } - - i, err := remote.Read(buf) - if err != nil { - if err != io.EOF { - log.Printf("%v", err) - } - break - } - - err = packet.Unmarshal(buf[:i]) - if err != nil { - log.Printf("%v", err) - continue - } - - window.Set(packet.SequenceNumber) - if packet.SequenceNumber-window.First() > 24 { - first, bitmap := window.Get17() - if bitmap != ^uint16(0) { - err := conn.sendNACK(track, first, ^bitmap) - if err != nil { - log.Printf("%v", err) - } - } - } - - list.Store(packet.SequenceNumber, buf[:i]) - - for _, l := range local { - if l.muted() { - continue - } - err := l.track.WriteRTP(&packet) - if err != nil && err != io.ErrClosedPipe { - log.Printf("%v", err) - } - } - } - }() + go upLoop(conn, track) }) return conn, nil } +func upLoop(conn *upConnection, track *upTrack) { + buf := make([]byte, packetlist.BufSize) + var packet rtp.Packet + var local []*downTrack + var localTime time.Time + window := packetwindow.New() + for { + now := time.Now() + if now.Sub(localTime) > time.Second/2 { + local = track.getLocal() + localTime = now + } + + i, err := track.track.Read(buf) + if err != nil { + if err != io.EOF { + log.Printf("%v", err) + } + break + } + + err = packet.Unmarshal(buf[:i]) + if err != nil { + log.Printf("%v", err) + continue + } + + window.Set(packet.SequenceNumber) + if packet.SequenceNumber-window.First() > 24 { + first, bitmap := window.Get17() + if bitmap != ^uint16(0) { + err := conn.sendNACK(track, first, ^bitmap) + if err != nil { + log.Printf("%v", err) + } + } + } + + track.list.Store(packet.SequenceNumber, buf[:i]) + + for _, l := range local { + if l.muted() { + continue + } + err := l.track.WriteRTP(&packet) + if err != nil && err != io.ErrClosedPipe { + log.Printf("%v", err) + } + } + } +} + func delUpConn(c *client, id string) { c.mu.Lock() defer c.mu.Unlock()