Skip to content

Commit 25ba9e2

Browse files
daniel-abramovdaniel-abramov
committed
refactor: finalize signaling and peer communication
1 parent 4d6cee6 commit 25ba9e2

File tree

14 files changed

+572
-222
lines changed

14 files changed

+572
-222
lines changed

‎config.yaml.sample‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,6 @@
1-
homeserverurl: "http://localhost:8008"
2-
userid: "@sfu:shadowfax"
3-
accesstoken: "..."
4-
timeout: 30
1+
matrix:
2+
homeserverurl: "http://localhost:8008"
3+
userid: "@sfu:shadowfax"
4+
accesstoken: "..."
5+
conference:
6+
timeout: 30

‎docker-compose.yaml‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,9 @@ services:
99
environment:
1010
# Set the `CONFIG` to the configuration you want.
1111
CONFIG: |
12-
homeserverurl: "http://localhost:8008"
13-
userid: "@sfu:shadowfax"
14-
accesstoken: "..."
15-
timeout: 30
12+
matrix:
13+
homeserverurl: "http://localhost:8008"
14+
userid: "@sfu:shadowfax"
15+
accesstoken: "..."
16+
conference:
17+
timeout: 30

‎src/conference/conference.go‎

Lines changed: 47 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -18,46 +18,33 @@ package conference
1818

1919
import (
2020
"github.com/matrix-org/waterfall/src/peer"
21+
"github.com/matrix-org/waterfall/src/signaling"
2122
"github.com/pion/webrtc/v3"
2223
"github.com/sirupsen/logrus"
2324
"maunium.net/go/mautrix/event"
24-
"maunium.net/go/mautrix/id"
2525
)
2626

27-
// Configuration for the group conferences (calls).
28-
type CallConfig struct {
29-
// Keep-alive timeout for WebRTC connections. If no keep-alive has been received
30-
// from the client for this duration, the connection is considered dead.
31-
KeepAliveTimeout int
32-
}
33-
34-
type Participant struct {
35-
Peer *peer.Peer
36-
Data *ParticipantData
37-
}
38-
39-
type ParticipantData struct {
40-
RemoteSessionID id.SessionID
41-
StreamMetadata event.CallSDPStreamMetadata
42-
}
43-
4427
type Conference struct {
45-
conferenceID string
46-
config *CallConfig
47-
participants map[peer.ID]*Participant
48-
participantsChannel peer.MessageChannel
49-
logger *logrus.Entry
28+
id string
29+
config Config
30+
signaling signaling.MatrixSignaling
31+
participants map[peer.ID]*Participant
32+
peerEventsStream chan peer.Message
33+
logger *logrus.Entry
5034
}
5135

52-
func NewConference(confID string, config *CallConfig) *Conference {
53-
conference := new(Conference)
54-
conference.config = config
55-
conference.conferenceID = confID
56-
conference.participants = make(map[peer.ID]*Participant)
57-
conference.participantsChannel = make(peer.MessageChannel)
58-
conference.logger = logrus.WithFields(logrus.Fields{
59-
"conf_id": confID,
60-
})
36+
func NewConference(confID string, config Config, signaling signaling.MatrixSignaling) *Conference {
37+
conference := &Conference{
38+
id: confID,
39+
config: config,
40+
signaling: signaling,
41+
participants: make(map[peer.ID]*Participant),
42+
peerEventsStream: make(chan peer.Message),
43+
logger: logrus.WithFields(logrus.Fields{"conf_id": confID}),
44+
}
45+
46+
// Start conference "main loop".
47+
go conference.processMessages()
6148
return conference
6249
}
6350

@@ -66,41 +53,43 @@ func (c *Conference) OnNewParticipant(participantID peer.ID, inviteEvent *event.
6653
// As per MSC3401, when the `session_id` field changes from an incoming `m.call.member` event,
6754
// any existing calls from this device in this call should be terminated.
6855
// TODO: Implement this.
69-
/*
70-
for _, participant := range c.participants {
71-
if participant.data.DeviceID == inviteEvent.DeviceID {
72-
if participant.data.RemoteSessionID == inviteEvent.SenderSessionID {
73-
c.logger.WithFields(logrus.Fields{
74-
"device_id": inviteEvent.DeviceID,
75-
"session_id": inviteEvent.SenderSessionID,
76-
}).Errorf("Found existing participant with equal DeviceID and SessionID")
77-
return
78-
} else {
79-
participant.Terminate()
80-
delete(c.participants, participant.data.UserID)
81-
}
56+
for id, participant := range c.participants {
57+
if id.DeviceID == inviteEvent.DeviceID {
58+
if participant.remoteSessionID == inviteEvent.SenderSessionID {
59+
c.logger.WithFields(logrus.Fields{
60+
"device_id": inviteEvent.DeviceID,
61+
"session_id": inviteEvent.SenderSessionID,
62+
}).Errorf("Found existing participant with equal DeviceID and SessionID")
63+
return
64+
} else {
65+
participant.peer.Terminate()
8266
}
8367
}
84-
*/
68+
}
8569

86-
peer, _, err := peer.NewPeer(participantID, c.conferenceID, inviteEvent.Offer.SDP, c.participantsChannel)
70+
peer, sdpOffer, err := peer.NewPeer(participantID, c.id, inviteEvent.Offer.SDP, c.peerEventsStream)
8771
if err != nil {
8872
c.logger.WithError(err).Errorf("Failed to create new peer")
8973
return
9074
}
9175

92-
participantData := &ParticipantData{
93-
RemoteSessionID: inviteEvent.SenderSessionID,
94-
StreamMetadata: inviteEvent.SDPStreamMetadata,
76+
participant := &Participant{
77+
id: participantID,
78+
peer: peer,
79+
remoteSessionID: inviteEvent.SenderSessionID,
80+
streamMetadata: inviteEvent.SDPStreamMetadata,
81+
publishedTracks: make(map[event.SFUTrackDescription]*webrtc.TrackLocalStaticRTP),
9582
}
9683

97-
c.participants[participantID] = &Participant{Peer: peer, Data: participantData}
84+
c.participants[participantID] = participant
9885

99-
// TODO: Send the SDP answer back to the participant's device.
86+
recipient := participant.asMatrixRecipient()
87+
streamMetadata := c.getStreamsMetadata(participantID)
88+
c.signaling.SendSDPAnswer(recipient, streamMetadata, sdpOffer.SDP)
10089
}
10190

10291
func (c *Conference) OnCandidates(peerID peer.ID, candidatesEvent *event.CallCandidatesEventContent) {
103-
if participant := c.getParticipant(peerID); participant != nil {
92+
if participant := c.getParticipant(peerID, nil); participant != nil {
10493
// Convert the candidates to the WebRTC format.
10594
candidates := make([]webrtc.ICECandidateInit, len(candidatesEvent.Candidates))
10695
for i, candidate := range candidatesEvent.Candidates {
@@ -112,36 +101,23 @@ func (c *Conference) OnCandidates(peerID peer.ID, candidatesEvent *event.CallCan
112101
}
113102
}
114103

115-
participant.Peer.AddICECandidates(candidates)
104+
participant.peer.AddICECandidates(candidates)
116105
}
117106
}
118107

119108
func (c *Conference) OnSelectAnswer(peerID peer.ID, selectAnswerEvent *event.CallSelectAnswerEventContent) {
120-
if participant := c.getParticipant(peerID); participant != nil {
109+
if participant := c.getParticipant(peerID, nil); participant != nil {
121110
if selectAnswerEvent.SelectedPartyID != peerID.DeviceID.String() {
122111
c.logger.WithFields(logrus.Fields{
123112
"device_id": selectAnswerEvent.SelectedPartyID,
124113
}).Errorf("Call was answered on a different device, kicking this peer")
125-
participant.Peer.Terminate()
114+
participant.peer.Terminate()
126115
}
127116
}
128117
}
129118

130119
func (c *Conference) OnHangup(peerID peer.ID, hangupEvent *event.CallHangupEventContent) {
131-
if participant := c.getParticipant(peerID); participant != nil {
132-
participant.Peer.Terminate()
133-
}
134-
}
135-
136-
func (c *Conference) getParticipant(peerID peer.ID) *Participant {
137-
participant, ok := c.participants[peerID]
138-
if !ok {
139-
c.logger.WithFields(logrus.Fields{
140-
"user_id": peerID.UserID,
141-
"device_id": peerID.DeviceID,
142-
}).Errorf("Failed to find participant")
143-
return nil
120+
if participant := c.getParticipant(peerID, nil); participant != nil {
121+
participant.peer.Terminate()
144122
}
145-
146-
return participant
147123
}

‎src/conference/config.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
package conference
2+
3+
// Configuration for the group conferences (calls).
4+
type Config struct {
5+
// Keep-alive timeout for WebRTC connections. If no keep-alive has been received
6+
// from the client for this duration, the connection is considered dead.
7+
KeepAliveTimeout int `yaml:"timeout"`
8+
}

‎src/conference/messages.go‎

Lines changed: 179 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,179 @@
1+
package conference
2+
3+
import (
4+
"encoding/json"
5+
"errors"
6+
7+
"github.com/matrix-org/waterfall/src/peer"
8+
"maunium.net/go/mautrix/event"
9+
)
10+
11+
func (c *Conference) processMessages() {
12+
for {
13+
// Read a message from the stream (of type peer.Message) and process it.
14+
message := <-c.peerEventsStream
15+
c.processPeerMessage(message)
16+
}
17+
}
18+
19+
//nolint:funlen
20+
func (c *Conference) processPeerMessage(message peer.Message) {
21+
// Since Go does not support ADTs, we have to use a switch statement to
22+
// determine the actual type of the message.
23+
switch msg := message.(type) {
24+
case peer.JoinedTheCall:
25+
case peer.LeftTheCall:
26+
delete(c.participants, msg.Sender)
27+
// TODO: Send new metadata about available streams to all participants.
28+
// TODO: Send the hangup event over the Matrix back to the user.
29+
30+
case peer.NewTrackPublished:
31+
participant := c.getParticipant(msg.Sender, errors.New("New track published from unknown participant"))
32+
if participant == nil {
33+
return
34+
}
35+
36+
key := event.SFUTrackDescription{
37+
StreamID: msg.Track.StreamID(),
38+
TrackID: msg.Track.ID(),
39+
}
40+
41+
if _, ok := participant.publishedTracks[key]; ok {
42+
c.logger.Errorf("Track already published: %v", key)
43+
return
44+
}
45+
46+
participant.publishedTracks[key] = msg.Track
47+
48+
case peer.PublishedTrackFailed:
49+
participant := c.getParticipant(msg.Sender, errors.New("Published track failed from unknown participant"))
50+
if participant == nil {
51+
return
52+
}
53+
54+
delete(participant.publishedTracks, event.SFUTrackDescription{
55+
StreamID: msg.Track.StreamID(),
56+
TrackID: msg.Track.ID(),
57+
})
58+
59+
// TODO: Should we remove the local tracks from every subscriber as well? Or will it happen automatically?
60+
61+
case peer.NewICECandidate:
62+
participant := c.getParticipant(msg.Sender, errors.New("ICE candidate from unknown participant"))
63+
if participant == nil {
64+
return
65+
}
66+
67+
// Convert WebRTC ICE candidate to Matrix ICE candidate.
68+
jsonCandidate := msg.Candidate.ToJSON()
69+
candidates := []event.CallCandidate{{
70+
Candidate: jsonCandidate.Candidate,
71+
SDPMLineIndex: int(*jsonCandidate.SDPMLineIndex),
72+
SDPMID: *jsonCandidate.SDPMid,
73+
}}
74+
c.signaling.SendICECandidates(participant.asMatrixRecipient(), candidates)
75+
76+
case peer.ICEGatheringComplete:
77+
participant := c.getParticipant(msg.Sender, errors.New("Received ICE complete from unknown participant"))
78+
if participant == nil {
79+
return
80+
}
81+
82+
// Send an empty array of candidates to indicate that ICE gathering is complete.
83+
c.signaling.SendCandidatesGatheringFinished(participant.asMatrixRecipient())
84+
85+
case peer.RenegotiationRequired:
86+
participant := c.getParticipant(msg.Sender, errors.New("Renegotiation from unknown participant"))
87+
if participant == nil {
88+
return
89+
}
90+
91+
toSend := event.SFUMessage{
92+
Op: event.SFUOperationOffer,
93+
SDP: msg.Offer.SDP,
94+
Metadata: c.getStreamsMetadata(participant.id),
95+
}
96+
97+
participant.sendDataChannelMessage(toSend)
98+
99+
case peer.DataChannelMessage:
100+
participant := c.getParticipant(msg.Sender, errors.New("Data channel message from unknown participant"))
101+
if participant == nil {
102+
return
103+
}
104+
105+
var sfuMessage event.SFUMessage
106+
if err := json.Unmarshal([]byte(msg.Message), &sfuMessage); err != nil {
107+
c.logger.Errorf("Failed to unmarshal SFU message: %v", err)
108+
return
109+
}
110+
111+
c.handleDataChannelMessage(participant, sfuMessage)
112+
113+
case peer.DataChannelAvailable:
114+
participant := c.getParticipant(msg.Sender, errors.New("Data channel available from unknown participant"))
115+
if participant == nil {
116+
return
117+
}
118+
119+
toSend := event.SFUMessage{
120+
Op: event.SFUOperationMetadata,
121+
Metadata: c.getStreamsMetadata(participant.id),
122+
}
123+
124+
if err := participant.sendDataChannelMessage(toSend); err != nil {
125+
c.logger.Errorf("Failed to send SFU message to open data channel: %v", err)
126+
return
127+
}
128+
129+
default:
130+
c.logger.Errorf("Unknown message type: %T", msg)
131+
}
132+
}
133+
134+
// Handle the `SFUMessage` event from the DataChannel message.
135+
func (c *Conference) handleDataChannelMessage(participant *Participant, sfuMessage event.SFUMessage) {
136+
switch sfuMessage.Op {
137+
case event.SFUOperationSelect:
138+
// Get the tracks that correspond to the tracks that the participant wants to receive.
139+
for _, track := range c.getTracks(sfuMessage.Start) {
140+
if err := participant.peer.SubscribeToTrack(track); err != nil {
141+
c.logger.Errorf("Failed to subscribe to track: %v", err)
142+
return
143+
}
144+
}
145+
146+
case event.SFUOperationAnswer:
147+
if err := participant.peer.NewSDPAnswerReceived(sfuMessage.SDP); err != nil {
148+
c.logger.Errorf("Failed to set SDP answer: %v", err)
149+
return
150+
}
151+
152+
// TODO: Clarify the semantics of publish (just a new sdp offer?).
153+
case event.SFUOperationPublish:
154+
// TODO: Clarify the semantics of publish (how is it different from unpublish?).
155+
case event.SFUOperationUnpublish:
156+
// TODO: Handle the heartbeat message here (updating the last timestamp etc).
157+
case event.SFUOperationAlive:
158+
case event.SFUOperationMetadata:
159+
participant.streamMetadata = sfuMessage.Metadata
160+
161+
// Inform all participants about new metadata available.
162+
for id, participant := range c.participants {
163+
// Skip ourselves.
164+
if id == participant.id {
165+
continue
166+
}
167+
168+
toSend := event.SFUMessage{
169+
Op: event.SFUOperationMetadata,
170+
Metadata: c.getStreamsMetadata(id),
171+
}
172+
173+
if err := participant.sendDataChannelMessage(toSend); err != nil {
174+
c.logger.Errorf("Failed to send SFU message: %v", err)
175+
return
176+
}
177+
}
178+
}
179+
}

0 commit comments

Comments
 (0)