Don't re-create chanCandidate every gather

Instead keep the channel around the entire time, and use
the same thread to distribute state changes and candidates.

Resolves #217
Resolves #218

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
+}

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)
 		}
 	}()