Split sub goroutines into funcs
@@ -127,37 +127,25 @@ func (r *Router) GetPub() transport.Transport {
return r.pub
}
-// AddSub add a pub to router
-func (r *Router) AddSub(id string, t transport.Transport) transport.Transport {
- //fix panic: assignment to entry in nil map
- if r.stop {
- return nil
- }
- r.subLock.Lock()
- defer r.subLock.Unlock()
- r.subs[id] = t
- r.subChans[id] = make(chan *rtp.Packet, 1000)
- log.Infof("Router.AddSub id=%s t=%p", id, t)
- // Sub writer
- go func() {
- for pkt := range r.subChans[id] {
+func (r *Router) subWriteLoop(subID string, trans transport.Transport) {
+ for pkt := range r.subChans[subID] {
// log.Infof(" WriteRTP %v:%v to %v ", pkt.SSRC, pkt.SequenceNumber, t.ID())
- if err := t.WriteRTP(pkt); err != nil {
+
+ if err := trans.WriteRTP(pkt); err != nil {
// log.Errorf("wt.WriteRTP err=%v", err)
// del sub when err is increasing
- if t.WriteErrTotal() > maxWriteErr {
- r.DelSub(t.ID())
+ if trans.WriteErrTotal() > maxWriteErr {
+ r.DelSub(trans.ID())
}
}
- t.WriteErrReset()
+ trans.WriteErrReset()
}
log.Infof("Closing sub writer")
- }()
+}
- // Sub reader
- go func() {
+func (r *Router) subFeedbackLoop(subID string, trans transport.Transport) {
for {
- pkt := <-t.GetRTCPChan()
+ pkt := <-trans.GetRTCPChan()
if r.stop {
return
}
@@ -175,7 +163,7 @@ func (r *Router) AddSub(id string, t transport.Transport) transport.Transport {
// log.Infof("Router.AddSub got nack: %+v", pkt)
nack := pkt
for _, nackPair := range nack.Nacks {
- if !r.ReSendRTP(id, nack.MediaSSRC, nackPair.PacketID) {
+ if !r.ReSendRTP(subID, nack.MediaSSRC, nackPair.PacketID) {
n := &rtcp.TransportLayerNack{
//origin ssrc
SenderSSRC: nack.SenderSSRC,
@@ -194,7 +182,23 @@ func (r *Router) AddSub(id string, t transport.Transport) transport.Transport {
default:
}
}
- }()
+}
+
+// AddSub add a pub to router
+func (r *Router) AddSub(id string, t transport.Transport) transport.Transport {
+ //fix panic: assignment to entry in nil map
+ if r.stop {
+ return nil
+ }
+ r.subLock.Lock()
+ defer r.subLock.Unlock()
+ r.subs[id] = t
+ r.subChans[id] = make(chan *rtp.Packet, 1000)
+ log.Infof("Router.AddSub id=%s t=%p", id, t)
+
+ // Sub loops
+ go r.subWriteLoop(id, t)
+ go r.subFeedbackLoop(id, t)
return t
}
