pkg/peer.go

@@ -111,6 +111,8 @@ func (p *Peer) Join(sid string, sdp webrtc.SessionDescription) (*webrtc.SessionD
 		}
 	})
 
+	p.session.Subscribe(p)
+
 	p.subscriber.OnICECandidate(func(c *webrtc.ICECandidate) {
 		log.Debugf("on ice candidate called")
 		if c == nil {
@@ -222,6 +224,8 @@ func (p *Peer) Close() error {
 			return err
 		}
 	}
-	p.session.RemovePeer(p.id)
+	if p.session != nil {
+		p.session.RemovePeer(p.id)
+	}
 	return nil
 }

pkg/publisher.go

@@ -21,7 +21,6 @@ type Publisher struct {
 	onICEConnectionStateChangeHandler func(webrtc.ICEConnectionState)
 
 	closeOnce sync.Once
-	subOnce   sync.Once
 }
 
 // NewPublisher creates a new Publisher
@@ -54,11 +53,6 @@ func NewPublisher(session *Session, id string, me MediaEngine, cfg WebRTCTranspo
 	pc.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
 		log.Debugf("ice connection state: %s", connectionState)
 		switch connectionState {
-		case webrtc.ICEConnectionStateConnected:
-			p.subOnce.Do(func() {
-				// Subscribe to existing peers
-				p.session.Subscribe(p.id)
-			})
 		case webrtc.ICEConnectionStateFailed:
 			fallthrough
 		case webrtc.ICEConnectionStateClosed:

pkg/session.go

@@ -69,19 +69,17 @@ func (s *Session) Publish(router Router, rr *receiverRouter) {
 }
 
 // Subscribe will create a Sender for every other Receiver in the session
-func (s *Session) Subscribe(id string) {
+func (s *Session) Subscribe(peer *Peer) {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
-	if peer, ok := s.peers[id]; ok {
-		for pid, p := range s.peers {
-			if pid == id {
-				continue
-			}
-			err := p.publisher.GetRouter().AddSender(peer.subscriber, nil)
-			if err != nil {
-				log.Errorf("Subscribing to router err: %v", err)
-				continue
-			}
+	for pid, p := range s.peers {
+		if pid == peer.id {
+			continue
+		}
+		err := p.publisher.GetRouter().AddSender(peer.subscriber, nil)
+		if err != nil {
+			log.Errorf("Subscribing to router err: %v", err)
+			continue
 		}
 	}
 }