diff --git a/agent.go b/agent.go index e8c2272..1be5ad7 100644 --- a/agent.go +++ b/agent.go @@ -230,7 +230,8 @@ func NewAgent(config *AgentConfig) (*Agent, error) { done: make(chan struct{}), startedCh: startedCtx.Done(), startedFn: startedFn, - chanState: make(chan ConnectionState, 1), + chanState: make(chan ConnectionState), + chanCandidate: make(chan Candidate), portmin: config.PortMin, portmax: config.PortMax, loggerFactory: loggerFactory, @@ -286,6 +287,7 @@ func NewAgent(config *AgentConfig) (*Agent, error) { return nil, err } + a.startOnConnectionStateChangeRoutine() return a, nil } @@ -319,9 +321,25 @@ func (a *Agent) onSelectedCandidatePairChange(p *candidatePair) { func (a *Agent) startOnConnectionStateChangeRoutine() { go func() { - for s := range a.chanState { - if hdlr, ok := a.onConnectionStateChangeHdlr.Load().(func(ConnectionState)); ok { - hdlr(s) + for { + select { + case s, isOpen := <-a.chanState: + if !isOpen { + return + } + + if hdlr, ok := a.onConnectionStateChangeHdlr.Load().(func(ConnectionState)); ok { + hdlr(s) + } + + case c, isOpen := <-a.chanCandidate: + if !isOpen { + return + } + + if onCandidateHdlr, ok := a.onCandidateHdlr.Load().(func(Candidate)); ok { + onCandidateHdlr(c) + } } } }() @@ -339,7 +357,6 @@ func (a *Agent) startConnectivityChecks(isControlling bool, remoteUfrag, remoteP return err } - a.startOnConnectionStateChangeRoutine() a.log.Debugf("Started agent: isControlling? %t, remoteUfrag: %q, remotePwd: %q", isControlling, remoteUfrag, remotePwd) return a.run(func(agent *Agent) { @@ -738,6 +755,7 @@ func (a *Agent) Close() error { defer func() { close(done) close(agent.chanState) + close(agent.chanCandidate) }() agent.err.Store(ErrClosed) close(agent.done) @@ -1069,3 +1087,20 @@ func (a *Agent) Restart(ufrag, pwd string) error { } return <-err } + +func (a *Agent) setGatheringState(newState GatheringState) error { + done := make(chan struct{}) + if err := a.run(func(agent *Agent) { + if a.gatheringState != newState && newState == GatheringStateComplete { + a.chanCandidate <- nil + } + + a.gatheringState = newState + close(done) + }, nil); err != nil { + return err + } + + <-done + return nil +} diff --git a/gather.go b/gather.go index 130f5c4..9465217 100644 --- a/gather.go +++ b/gather.go @@ -76,39 +76,17 @@ func (a *Agent) GatherCandidates() error { } func (a *Agent) gatherCandidates() <-chan struct{} { - gatherStateUpdated := make(chan bool) - - a.chanCandidate = make(chan Candidate, 1) - var closeChanCandidateOnce sync.Once - go func() { - for c := range a.chanCandidate { - if onCandidateHdlr, ok := a.onCandidateHdlr.Load().(func(Candidate)); ok { - onCandidateHdlr(c) - } - } - if onCandidateHdlr, ok := a.onCandidateHdlr.Load().(func(Candidate)); ok { - onCandidateHdlr(nil) - } - }() - done := make(chan struct{}) go func() { defer func() { - closeChanCandidateOnce.Do(func() { - close(a.chanCandidate) - }) close(done) }() - if err := a.run(func(agent *Agent) { - a.gatheringState = GatheringStateGathering - close(gatherStateUpdated) - }, nil); err != nil { - a.log.Warnf("failed to set gatheringState to GatheringStateGathering for gatherCandidates: %v", err) + if err := a.setGatheringState(GatheringStateGathering); err != nil { + a.log.Warnf("failed to set gatheringState to GatheringStateGathering: %v", err) return } - <-gatherStateUpdated var wg sync.WaitGroup for _, t := range a.candidateTypes { @@ -128,14 +106,9 @@ func (a *Agent) gatherCandidates() <-chan struct{} { } // Block until all STUN and TURN URLs have been gathered (or timed out) wg.Wait() - if err := a.run(func(agent *Agent) { - closeChanCandidateOnce.Do(func() { - close(agent.chanCandidate) - }) - a.gatheringState = GatheringStateComplete - }, nil); err != nil { - a.log.Warnf("Failed to stop OnCandidate handler routine and update gatheringState: %v\n", err) - return + + if err := a.setGatheringState(GatheringStateComplete); err != nil { + a.log.Warnf("failed to set gatheringState to GatheringStateComplete: %v", err) } }()