Merge pull request #29 from matrix-org/SimonBrandner/feat/abstract

Author: SimonBrandner

go.mod

@@ -6,6 +6,7 @@ require github.com/pion/webrtc/v3 v3.1.31
 
 require (
 	github.com/pion/rtcp v1.2.9
+	github.com/sirupsen/logrus v1.9.0
 	gopkg.in/yaml.v3 v3.0.1
 	maunium.net/go/mautrix v0.11.0
 )
@@ -27,7 +28,6 @@ require (
 	github.com/pion/transport v0.13.0 // indirect
 	github.com/pion/turn/v2 v2.0.8 // indirect
 	github.com/pion/udp v0.1.1 // indirect
-	github.com/sirupsen/logrus v1.9.0 // indirect
 	github.com/tidwall/gjson v1.14.1 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.0 // indirect

go.sum

@@ -134,8 +134,6 @@ golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a h1:dGzPydgVsqGcTRVwiLJ1jVbufYwmzD3LfVPLKsKg+0k=
-golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

src/call.go

@@ -18,6 +18,7 @@ package main
 
 import (
 	"encoding/json"
+	"sync"
 	"time"
 
 	"github.com/pion/webrtc/v3"
@@ -28,18 +29,49 @@ import (
 )
 
 type Call struct {
-	CallID                 string
-	UserID                 id.UserID
-	DeviceID               id.DeviceID
-	LocalSessionID         id.SessionID
-	RemoteSessionID        id.SessionID
-	Client                 *mautrix.Client
-	PeerConnection         *webrtc.PeerConnection
-	Conf                   *Conference
+	PeerConnection *webrtc.PeerConnection
+
+	CallID          string
+	UserID          id.UserID
+	DeviceID        id.DeviceID
+	LocalSessionID  id.SessionID
+	RemoteSessionID id.SessionID
+
+	Publishers  []*Publisher
+	Subscribers []*Subscriber
+
+	mutex  sync.RWMutex
+	logger *logrus.Entry
+	client *mautrix.Client
+	conf   *Conference
+
 	dataChannel            *webrtc.DataChannel
 	lastKeepAliveTimestamp time.Time
 	sentEndOfCandidates    bool
-	logger                 *logrus.Entry
+}
+
+func NewCall(callID string, conf *Conference) *Call {
+	call := new(Call)
+
+	call.CallID = callID
+	call.conf = conf
+
+	return call
+}
+
+func (c *Call) InitWithInvite(evt *event.Event, client *mautrix.Client) {
+	invite := evt.Content.AsCallInvite()
+
+	c.UserID = evt.Sender
+	c.DeviceID = invite.DeviceID
+	// XXX: What if an SFU gets restarted?
+	c.LocalSessionID = localSessionID
+	c.RemoteSessionID = invite.SenderSessionID
+	c.client = client
+	c.logger = logrus.WithFields(logrus.Fields{
+		"user_id": evt.Sender,
+		"conf_id": invite.ConfID,
+	})
 }
 
 func (c *Call) onDCSelect(start []event.SFUTrackDescription) {
@@ -55,21 +87,9 @@ func (c *Call) onDCSelect(start []event.SFUTrackDescription) {
 
 		trackLogger.Info("selecting track")
 
-		foundTracks := c.Conf.GetLocalTrackByInfo(LocalTrackInfo{
-			StreamID: trackDesc.StreamID,
-			TrackID:  trackDesc.TrackID,
-		})
-
-		if len(foundTracks) == 0 {
-			trackLogger.Info("no track found")
-			continue
-		}
-
-		for _, track := range foundTracks {
-			if _, err := c.PeerConnection.AddTrack(track); err == nil {
-				trackLogger.Info("added track")
-			} else {
-				trackLogger.WithError(err).Error("failed to add track")
+		for _, publisher := range c.conf.GetPublishers() {
+			if publisher.Matches(trackDesc) {
+				publisher.Subscribe(c)
 			}
 		}
 	}
@@ -114,12 +134,19 @@ func (c *Call) onDCUnpublish(stop []event.SFUTrackDescription, sdp string) {
 
 		trackLogger.Info("unpublishing track")
 
-		if removedTracksCount := c.Conf.RemoveTracksFromPeerConnectionsByInfo(LocalTrackInfo{
-			StreamID: trackDesc.StreamID,
-			TrackID:  trackDesc.TrackID,
-		}); removedTracksCount == 0 {
-			trackLogger.Info("no tracks to remove")
+		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{
@@ -169,7 +196,7 @@ func (c *Call) onDCAlive() {
 func (c *Call) onDCMetadata() {
 	c.logger.Info("received DC metadata")
 
-	c.Conf.SendUpdatedMetadataFromCall(c.CallID)
+	c.conf.SendUpdatedMetadataFromCall(c.CallID)
 }
 
 func (c *Call) dataChannelHandler(channel *webrtc.DataChannel) {
@@ -196,7 +223,7 @@ func (c *Call) dataChannelHandler(channel *webrtc.DataChannel) {
 		}
 
 		if msg.Metadata != nil {
-			c.Conf.UpdateSDPStreamMetadata(c.DeviceID, msg.Metadata)
+			c.conf.Metadata.Update(c.DeviceID, msg.Metadata)
 		}
 
 		switch msg.Op {
@@ -255,11 +282,11 @@ func (c *Call) iceCandidateHandler(candidate *webrtc.ICECandidate) {
 		Parsed: event.CallCandidatesEventContent{
 			BaseCallEventContent: event.BaseCallEventContent{
 				CallID:          c.CallID,
-				ConfID:          c.Conf.ConfID,
-				DeviceID:        c.Client.DeviceID,
+				ConfID:          c.conf.ConfID,
+				DeviceID:        c.client.DeviceID,
 				SenderSessionID: c.LocalSessionID,
 				DestSessionID:   c.RemoteSessionID,
-				PartyID:         string(c.Client.DeviceID),
+				PartyID:         string(c.client.DeviceID),
 				Version:         event.CallVersion("1"),
 			},
 			Candidates: []event.CallCandidate{{
@@ -273,43 +300,13 @@ func (c *Call) iceCandidateHandler(candidate *webrtc.ICECandidate) {
 }
 
 func (c *Call) trackHandler(trackRemote *webrtc.TrackRemote) {
-	trackLogger := c.logger.WithFields(logrus.Fields{
-		"track_id":   trackRemote.ID(),
-		"track_kind": trackRemote.Kind(),
-		"stream_id":  trackRemote.StreamID(),
-	})
+	publisher := NewPublisher(trackRemote, c)
 
-	go WriteRTCP(trackRemote, c.PeerConnection, trackLogger)
+	c.mutex.Lock()
+	c.Publishers = append(c.Publishers, publisher)
+	c.mutex.Unlock()
 
-	trackLocal, err := webrtc.NewTrackLocalStaticRTP(
-		trackRemote.Codec().RTPCodecCapability,
-		trackRemote.ID(),
-		trackRemote.StreamID(),
-	)
-	if err != nil {
-		trackLogger.
-			WithField("capability", trackRemote.Codec().RTPCodecCapability).
-			WithError(err).
-			Error("failed to create new track local static RTP - ignoring")
-
-		return
-	}
-
-	c.Conf.Tracks.Mutex.Lock()
-	c.Conf.Tracks.Tracks = append(c.Conf.Tracks.Tracks, LocalTrackWithInfo{
-		Track: trackLocal,
-		Info: LocalTrackInfo{
-			TrackID:  trackLocal.ID(),
-			StreamID: trackLocal.StreamID(),
-			Call:     c,
-		},
-	})
-	c.Conf.Tracks.Mutex.Unlock()
-
-	trackLogger.Info("published track")
-
-	go c.Conf.SendUpdatedMetadataFromCall(c.CallID)
-	go CopyRemoteToLocal(trackRemote, trackLocal, trackLogger)
+	go c.conf.SendUpdatedMetadataFromCall(c.CallID)
 }
 
 func (c *Call) iceConnectionStateHandler(state webrtc.ICEConnectionState) {
@@ -322,11 +319,11 @@ func (c *Call) iceConnectionStateHandler(state webrtc.ICEConnectionState) {
 				Parsed: event.CallCandidatesEventContent{
 					BaseCallEventContent: event.BaseCallEventContent{
 						CallID:          c.CallID,
-						ConfID:          c.Conf.ConfID,
-						DeviceID:        c.Client.DeviceID,
+						ConfID:          c.conf.ConfID,
+						DeviceID:        c.client.DeviceID,
 						SenderSessionID: c.LocalSessionID,
 						DestSessionID:   c.RemoteSessionID,
-						PartyID:         string(c.Client.DeviceID),
+						PartyID:         string(c.client.DeviceID),
 						Version:         event.CallVersion("1"),
 					},
 					Candidates: []event.CallCandidate{{Candidate: ""}},
@@ -339,7 +336,7 @@ func (c *Call) iceConnectionStateHandler(state webrtc.ICEConnectionState) {
 }
 
 func (c *Call) OnInvite(content *event.CallInviteEventContent) {
-	c.Conf.UpdateSDPStreamMetadata(c.DeviceID, content.SDPStreamMetadata)
+	c.conf.Metadata.Update(c.DeviceID, content.SDPStreamMetadata)
 	offer := content.Offer
 
 	peerConnection, err := webrtc.NewPeerConnection(webrtc.Configuration{})
@@ -394,26 +391,26 @@ func (c *Call) OnInvite(content *event.CallInviteEventContent) {
 		Parsed: event.CallAnswerEventContent{
 			BaseCallEventContent: event.BaseCallEventContent{
 				CallID:          c.CallID,
-				ConfID:          c.Conf.ConfID,
-				DeviceID:        c.Client.DeviceID,
+				ConfID:          c.conf.ConfID,
+				DeviceID:        c.client.DeviceID,
 				SenderSessionID: c.LocalSessionID,
 				DestSessionID:   c.RemoteSessionID,
-				PartyID:         string(c.Client.DeviceID),
+				PartyID:         string(c.client.DeviceID),
 				Version:         event.CallVersion("1"),
 			},
 			Answer: event.CallData{
 				Type: "answer",
 				SDP:  peerConnection.LocalDescription().SDP,
 			},
-			SDPStreamMetadata: c.Conf.GetRemoteMetadataForDevice(c.DeviceID),
+			SDPStreamMetadata: c.conf.Metadata.GetForDevice(c.DeviceID),
 		},
 	}
 	c.sendToDevice(event.CallAnswer, answerEvtContent)
 }
 
 func (c *Call) OnSelectAnswer(content *event.CallSelectAnswerEventContent) {
 	selectedPartyID := content.SelectedPartyID
-	if selectedPartyID != string(c.Client.DeviceID) {
+	if selectedPartyID != string(c.client.DeviceID) {
 		c.logger.WithField("selected_party_id", selectedPartyID).Warn("call was answered on a different device")
 		c.Terminate()
 	}
@@ -446,27 +443,31 @@ func (c *Call) Terminate() {
 		c.logger.WithError(err).Error("error closing peer connection")
 	}
 
-	c.Conf.Calls.CallsMu.Lock()
-	delete(c.Conf.Calls.Calls, c.CallID)
-	c.Conf.Calls.CallsMu.Unlock()
+	c.conf.mutex.Lock()
+	delete(c.conf.Calls, c.CallID)
+	c.conf.mutex.Unlock()
 
-	info := LocalTrackInfo{Call: c}
-	c.Conf.RemoveTracksFromPeerConnectionsByInfo(info)
-	c.Conf.RemoveTracksFromConfByInfo(info)
-	c.Conf.RemoveMetadataByDeviceID(c.DeviceID)
-	c.Conf.SendUpdatedMetadataFromCall(c.CallID)
+	for _, publisher := range c.Publishers {
+		publisher.Stop()
+	}
+
+	for _, subscriber := range c.Subscribers {
+		subscriber.Unsubscribe()
+	}
+
+	c.conf.SendUpdatedMetadataFromCall(c.CallID)
 }
 
 func (c *Call) Hangup(reason event.CallHangupReason) {
 	hangupEvtContent := &event.Content{
 		Parsed: event.CallHangupEventContent{
 			BaseCallEventContent: event.BaseCallEventContent{
 				CallID:          c.CallID,
-				ConfID:          c.Conf.ConfID,
-				DeviceID:        c.Client.DeviceID,
+				ConfID:          c.conf.ConfID,
+				DeviceID:        c.client.DeviceID,
 				SenderSessionID: c.LocalSessionID,
 				DestSessionID:   c.RemoteSessionID,
-				PartyID:         string(c.Client.DeviceID),
+				PartyID:         string(c.client.DeviceID),
 				Version:         event.CallVersion("1"),
 			},
 			Reason: reason,
@@ -495,7 +496,7 @@ func (c *Call) sendToDevice(callType event.Type, content *event.Content) {
 
 	// TODO: E2EE
 	// TODO: to-device reliability
-	if _, err := c.Client.SendToDevice(callType, toDevice); err != nil {
+	if _, err := c.client.SendToDevice(callType, toDevice); err != nil {
 		evtLogger.WithField("content", content).WithError(err).Error("error sending to-device")
 	}
 }
@@ -509,9 +510,11 @@ func (c *Call) SendDataChannelMessage(msg event.SFUMessage) {
 		"op": msg.Op,
 	})
 
-	msg.Metadata = c.Conf.GetRemoteMetadataForDevice(c.DeviceID)
-	if msg.Op == "metadata" && len(msg.Metadata) == 0 {
-		return
+	if msg.Metadata == nil {
+		msg.Metadata = c.conf.Metadata.GetForDevice(c.DeviceID)
+		if msg.Op == event.SFUOperationMetadata && len(msg.Metadata) == 0 {
+			return
+		}
 	}
 
 	marshaled, err := json.Marshal(msg)
@@ -541,3 +544,17 @@ func (c *Call) CheckKeepAliveTimestamp() {
 		}
 	}
 }
+
+func (c *Call) RemoveSubscriber(toDelete *Subscriber) {
+	newSubscribers := []*Subscriber{}
+
+	c.mutex.Lock()
+	for _, subscriber := range c.Subscribers {
+		if subscriber != toDelete {
+			newSubscribers = append(newSubscribers, subscriber)
+		}
+	}
+
+	c.Subscribers = newSubscribers
+	c.mutex.Unlock()
+}