Add ICE Trickle support

Resolves pion/ice#51

Co-authored-by: Konstantin Itskov <konstantin.itskov@kovits.com>

icegatherer.go

@@ -21,15 +21,20 @@ type ICEGatherer struct {
 
 	validatedServers []*ice.URL
 
-	agent *ice.Agent
+	agentIsTrickle bool
+	agent          *ice.Agent
 
 	portMin           uint16
 	portMax           uint16
 	candidateTypes    []ice.CandidateType
 	connectionTimeout *time.Duration
 	keepaliveInterval *time.Duration
 	loggerFactory     logging.LoggerFactory
+	log               logging.LeveledLogger
 	networkTypes      []NetworkType
+
+	onLocalCandidateHdlr func(candidate *ICECandidate)
+	onStateChangeHdlr    func(state ICEGathererState)
 }
 
 // NewICEGatherer creates a new NewICEGatherer.
@@ -66,24 +71,27 @@ func NewICEGatherer(
 		connectionTimeout: connectionTimeout,
 		keepaliveInterval: keepaliveInterval,
 		loggerFactory:     loggerFactory,
+		log:               loggerFactory.NewLogger("ice"),
 		networkTypes:      networkTypes,
 		candidateTypes:    candidateTypes,
 	}, nil
 }
 
-// State indicates the current state of the ICE gatherer.
-func (g *ICEGatherer) State() ICEGathererState {
-	g.lock.RLock()
-	defer g.lock.RUnlock()
-	return g.state
-}
-
-// Gather ICE candidates.
-func (g *ICEGatherer) Gather() error {
+func (g *ICEGatherer) createAgent() error {
 	g.lock.Lock()
 	defer g.lock.Unlock()
 
+	if g.agent != nil {
+		if !g.agentIsTrickle && g.onLocalCandidateHdlr != nil {
+			return errors.New("ICEAgent created without OnCandidate handler, but now has one set")
+		}
+
+		return nil
+	}
+
+	agentIsTrickle := g.onLocalCandidateHdlr != nil
 	config := &ice.AgentConfig{
+		Trickle:           agentIsTrickle,
 		Urls:              g.validatedServers,
 		PortMin:           g.portMin,
 		PortMax:           g.portMax,
@@ -108,8 +116,67 @@ func (g *ICEGatherer) Gather() error {
 	}
 
 	g.agent = agent
-	g.state = ICEGathererStateComplete
+	g.agentIsTrickle = agentIsTrickle
+	return nil
+}
+
+// Gather ICE candidates.
+func (g *ICEGatherer) Gather() error {
+	if err := g.createAgent(); err != nil {
+		return err
+	}
 
+	g.lock.Lock()
+	onLocalCandidateHdlr := g.onLocalCandidateHdlr
+	isTrickle := g.agentIsTrickle
+	agent := g.agent
+	g.lock.Unlock()
+
+	g.setState(ICEGathererStateGathering)
+	if isTrickle {
+		if err := agent.OnCandidate(func(candidate ice.Candidate) {
+			if candidate != nil {
+				c, err := newICECandidateFromICE(candidate)
+				if err != nil {
+					g.log.Warnf("Failed to convert ice.Candidate: %s", err)
+					return
+				}
+				onLocalCandidateHdlr(&c)
+			} else {
+				g.setState(ICEGathererStateComplete)
+				onLocalCandidateHdlr(nil)
+			}
+		}); err != nil {
+			return err
+		}
+		if err := agent.GatherCandidates(); err != nil {
+			return err
+		}
+	} else if err := g.simulateTrickle(); err != nil {
+		return err
+	}
+	return nil
+}
+
+// We support both Trickle (and non-Trickle) gathering
+// If the user opts for non-Trickle we still emit all candidates
+// after gathering is done blocking
+func (g *ICEGatherer) simulateTrickle() error {
+	g.lock.Lock()
+	if g.onLocalCandidateHdlr != nil {
+		candidates, err := g.GetLocalCandidates()
+		if err != nil {
+			g.lock.Unlock()
+			return err
+		}
+		for i := range candidates {
+			go g.onLocalCandidateHdlr(&candidates[i])
+		}
+		go g.onLocalCandidateHdlr(nil)
+	}
+
+	g.lock.Unlock()
+	g.setState(ICEGathererStateComplete)
 	return nil
 }
 
@@ -133,14 +200,11 @@ func (g *ICEGatherer) Close() error {
 
 // GetLocalParameters returns the ICE parameters of the ICEGatherer.
 func (g *ICEGatherer) GetLocalParameters() (ICEParameters, error) {
-	g.lock.RLock()
-	defer g.lock.RUnlock()
-	if g.agent == nil {
-		return ICEParameters{}, errors.New("gatherer not started")
+	if err := g.createAgent(); err != nil {
+		return ICEParameters{}, err
 	}
 
 	frag, pwd := g.agent.GetLocalUserCredentials()
-
 	return ICEParameters{
 		UsernameFragment: frag,
 		Password:         pwd,
@@ -150,17 +214,51 @@ func (g *ICEGatherer) GetLocalParameters() (ICEParameters, error) {
 
 // GetLocalCandidates returns the sequence of valid local candidates associated with the ICEGatherer.
 func (g *ICEGatherer) GetLocalCandidates() ([]ICECandidate, error) {
-	g.lock.RLock()
-	defer g.lock.RUnlock()
-
-	if g.agent == nil {
-		return nil, errors.New("gatherer not started")
+	if err := g.createAgent(); err != nil {
+		return nil, err
 	}
-
 	iceCandidates, err := g.agent.GetLocalCandidates()
 	if err != nil {
 		return nil, err
 	}
 
 	return newICECandidatesFromICE(iceCandidates)
 }
+
+// OnLocalCandidate sets an event handler which fires when a new local ICE candidate is available
+func (g *ICEGatherer) OnLocalCandidate(f func(*ICECandidate)) {
+	g.lock.Lock()
+	defer g.lock.Unlock()
+	g.onLocalCandidateHdlr = f
+}
+
+// OnStateChange fires any time the ICEGatherer changes
+func (g *ICEGatherer) OnStateChange(f func(ICEGathererState)) {
+	g.lock.Lock()
+	defer g.lock.Unlock()
+	g.onStateChangeHdlr = f
+}
+
+// State indicates the current state of the ICE gatherer.
+func (g *ICEGatherer) State() ICEGathererState {
+	g.lock.RLock()
+	defer g.lock.RUnlock()
+	return g.state
+}
+
+func (g *ICEGatherer) setState(s ICEGathererState) {
+	g.lock.Lock()
+	g.state = s
+	hdlr := g.onStateChangeHdlr
+	g.lock.Unlock()
+
+	if hdlr != nil {
+		go hdlr(s)
+	}
+}
+
+func (g *ICEGatherer) getAgent() *ice.Agent {
+	g.lock.RLock()
+	defer g.lock.RUnlock()
+	return g.agent
+}

icetransport.go

@@ -260,8 +260,7 @@ func (t *ICETransport) NewEndpoint(f mux.MatchFunc) *mux.Endpoint {
 }
 
 func (t *ICETransport) ensureGatherer() error {
-	if t.gatherer == nil ||
-		t.gatherer.agent == nil {
+	if t.gatherer == nil || t.gatherer.getAgent() == nil {
 		return errors.New("gatherer not started")
 	}