Commit a813cc9c authored by Juliusz Chroboczek's avatar Juliusz Chroboczek

Split out sending loop into a separate function.

parent 9bd093e7
...@@ -286,10 +286,9 @@ func addUpConn(c *client, id string) (*upConnection, error) { ...@@ -286,10 +286,9 @@ func addUpConn(c *client, id string) (*upConnection, error) {
c.mu.Unlock() c.mu.Unlock()
return return
} }
list := packetlist.New(32)
track := &upTrack{ track := &upTrack{
track: remote, track: remote,
list: list, list: packetlist.New(32),
maxBitrate: ^uint64(0), maxBitrate: ^uint64(0),
} }
u.tracks = append(u.tracks, track) u.tracks = append(u.tracks, track)
...@@ -304,60 +303,62 @@ func addUpConn(c *client, id string) (*upConnection, error) { ...@@ -304,60 +303,62 @@ func addUpConn(c *client, id string) (*upConnection, error) {
} }
} }
go func() { go upLoop(conn, track)
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) return conn, nil
if err != nil { }
if err != io.EOF {
log.Printf("%v", err)
}
break
}
err = packet.Unmarshal(buf[:i]) func upLoop(conn *upConnection, track *upTrack) {
if err != nil { buf := make([]byte, packetlist.BufSize)
log.Printf("%v", err) var packet rtp.Packet
continue 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
}
window.Set(packet.SequenceNumber) i, err := track.track.Read(buf)
if packet.SequenceNumber-window.First() > 24 { if err != nil {
first, bitmap := window.Get17() if err != io.EOF {
if bitmap != ^uint16(0) { log.Printf("%v", err)
err := conn.sendNACK(track, first, ^bitmap) }
if err != nil { break
log.Printf("%v", err) }
}
}
}
list.Store(packet.SequenceNumber, buf[:i]) err = packet.Unmarshal(buf[:i])
if err != nil {
log.Printf("%v", err)
continue
}
for _, l := range local { window.Set(packet.SequenceNumber)
if l.muted() { if packet.SequenceNumber-window.First() > 24 {
continue first, bitmap := window.Get17()
} if bitmap != ^uint16(0) {
err := l.track.WriteRTP(&packet) err := conn.sendNACK(track, first, ^bitmap)
if err != nil && err != io.ErrClosedPipe { if err != nil {
log.Printf("%v", err) log.Printf("%v", err)
}
} }
} }
}() }
})
return conn, nil 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) { func delUpConn(c *client, id string) {
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment