Merge pull request #30 from matrix-org/SimonBrandner/fix/double-unsubs

Author: SimonBrandner

src/call.go

@@ -127,26 +127,11 @@ func (c *Call) onDCPublish(sdp string) {
 
 func (c *Call) onDCUnpublish(stop []event.SFUTrackDescription, sdp string) {
 	for _, trackDesc := range stop {
-		trackLogger := c.logger.WithFields(logrus.Fields{
-			"track_id":  trackDesc.TrackID,
-			"stream_id": trackDesc.StreamID,
-		})
-
-		trackLogger.Info("unpublishing track")
-
-		newPublishers := []*Publisher{}
-
-		c.mutex.Lock()
 		for _, publisher := range c.Publishers {
 			if publisher.Matches(trackDesc) {
 				publisher.Stop()
-			} else {
-				newPublishers = append(newPublishers, publisher)
 			}
 		}
-
-		c.Publishers = newPublishers
-		c.mutex.Unlock()
 	}
 
 	err := c.PeerConnection.SetRemoteDescription(webrtc.SessionDescription{
@@ -300,11 +285,7 @@ func (c *Call) iceCandidateHandler(candidate *webrtc.ICECandidate) {
 }
 
 func (c *Call) trackHandler(trackRemote *webrtc.TrackRemote) {
-	publisher := NewPublisher(trackRemote, c)
-
-	c.mutex.Lock()
-	c.Publishers = append(c.Publishers, publisher)
-	c.mutex.Unlock()
+	NewPublisher(trackRemote, c)
 
 	go c.conf.SendUpdatedMetadataFromCall(c.CallID)
 }
@@ -545,16 +526,40 @@ func (c *Call) CheckKeepAliveTimestamp() {
 	}
 }
 
-func (c *Call) RemoveSubscriber(toDelete *Subscriber) {
+func (c *Call) RemoveSubscriber(toDelete *Subscriber) bool {
+	removed := false
 	newSubscribers := []*Subscriber{}
 
 	c.mutex.Lock()
 	for _, subscriber := range c.Subscribers {
 		if subscriber != toDelete {
+			removed = true
+		} else {
 			newSubscribers = append(newSubscribers, subscriber)
 		}
 	}
 
 	c.Subscribers = newSubscribers
 	c.mutex.Unlock()
+
+	return removed
+}
+
+func (c *Call) RemovePublisher(toDelete *Publisher) bool {
+	removed := false
+	newPublishers := []*Publisher{}
+
+	c.mutex.Lock()
+	for _, publisher := range c.Publishers {
+		if publisher == toDelete {
+			removed = true
+		} else {
+			newPublishers = append(newPublishers, publisher)
+		}
+	}
+
+	c.Publishers = newPublishers
+	c.mutex.Unlock()
+
+	return removed
 }

src/publisher.go

@@ -53,6 +53,10 @@ func NewPublisher(
 		"stream_id":  track.StreamID(),
 	})
 
+	call.mutex.Lock()
+	call.Publishers = append(call.Publishers, publisher)
+	call.mutex.Unlock()
+
 	go WriteRTCP(track, call.PeerConnection, publisher.logger)
 	go publisher.WriteToSubscribers()
 
@@ -68,10 +72,18 @@ func (p *Publisher) Subscribe(call *Call) {
 }
 
 func (p *Publisher) Stop() {
+	removed := p.Call.RemovePublisher(p)
+
+	if len(p.subscribers) == 0 && !removed {
+		return
+	}
+
 	for _, subscriber := range p.subscribers {
 		subscriber.Unsubscribe()
 		p.RemoveSubscriber(subscriber)
 	}
+
+	p.logger.Info("unpublished track")
 }
 
 func (p *Publisher) AddSubscriber(subscriber *Subscriber) {
@@ -83,15 +95,13 @@ func (p *Publisher) AddSubscriber(subscriber *Subscriber) {
 func (p *Publisher) RemoveSubscriber(toDelete *Subscriber) {
 	newSubscribers := []*Subscriber{}
 
-	p.mutex.RLock()
+	p.mutex.Lock()
 	for _, subscriber := range p.subscribers {
 		if subscriber != toDelete {
 			newSubscribers = append(newSubscribers, subscriber)
 		}
 	}
-	p.mutex.RUnlock()
 
-	p.mutex.Lock()
 	p.subscribers = newSubscribers
 	p.mutex.Unlock()
 }

src/subscriber.go

@@ -39,10 +39,16 @@ func NewSubscriber(call *Call) *Subscriber {
 	subscriber.call = call
 	subscriber.logger = call.logger
 
+	call.mutex.Lock()
+	call.Subscribers = append(call.Subscribers, subscriber)
+	call.mutex.Unlock()
+
 	return subscriber
 }
 
 func (s *Subscriber) initLoggingWithTrack(track *webrtc.TrackRemote) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
 	s.logger = s.call.logger.WithFields(logrus.Fields{
 		"track_id":   (*track).ID(),
 		"track_kind": (*track).Kind(),
@@ -79,6 +85,10 @@ func (s *Subscriber) Subscribe(publisher *Publisher) {
 }
 
 func (s *Subscriber) Unsubscribe() {
+	if s.publisher == nil {
+		return
+	}
+
 	if s.call.PeerConnection.ConnectionState() != webrtc.PeerConnectionStateClosed {
 		err := s.call.PeerConnection.RemoveTrack(s.sender)
 		if err != nil {
@@ -88,5 +98,9 @@ func (s *Subscriber) Unsubscribe() {
 
 	s.call.RemoveSubscriber(s)
 
+	s.mutex.Lock()
+	s.publisher = nil
+	s.mutex.Unlock()
+
 	s.logger.Info("unsubscribed")
 }