-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #52 from matrix-org/refactoring
SFU Refactoring
- Loading branch information
Showing
39 changed files
with
1,928 additions
and
1,583 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,6 @@ | ||
| homeserverurl: "http://localhost:8008" | ||
| userid: "@sfu:shadowfax" | ||
| accesstoken: "..." | ||
| timeout: 30 | ||
| matrix: | ||
| homeserverurl: "http://localhost:8008" | ||
| userid: "@sfu:shadowfax" | ||
| accesstoken: "..." | ||
| conference: | ||
| timeout: 30 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| package common | ||
|
|
||
| import "sync/atomic" | ||
|
|
||
| // In Go, unbounded channel means something different than what it means in Rust. | ||
| // I.e. unlike Rust, "unbounded" in Go means that the channel has **no buffer**, | ||
| // meaning that each attempt to send will block the channel until the receiver | ||
| // reads it. Majority of primitives here in `waterfall` are designed under assumption | ||
| // that sending is not blocking. | ||
| const UnboundedChannelSize = 128 | ||
|
|
||
| // Creates a new channel, returns two counterparts of it where one can only send and another can only receive. | ||
| // Unlike traditional Go channels, these allow the receiver to mark the channel as closed which would then fail | ||
| // to send any messages to the channel over `Send“. | ||
| func NewChannel[M any]() (Sender[M], Receiver[M]) { | ||
| channel := make(chan M, UnboundedChannelSize) | ||
| closed := &atomic.Bool{} | ||
| sender := Sender[M]{channel, closed} | ||
| receiver := Receiver[M]{channel, closed} | ||
| return sender, receiver | ||
| } | ||
|
|
||
| // Sender counterpart of the channel. | ||
| type Sender[M any] struct { | ||
| // The channel itself. | ||
| channel chan<- M | ||
| // Atomic variable that indicates whether the channel is closed. | ||
| receiverClosed *atomic.Bool | ||
| } | ||
|
|
||
| // Tries to send a message if the channel is not closed. | ||
| // Returns the message back if the channel is closed. | ||
| func (s *Sender[M]) Send(message M) *M { | ||
| if !s.receiverClosed.Load() { | ||
| s.channel <- message | ||
| return nil | ||
| } else { | ||
| return &message | ||
| } | ||
| } | ||
|
|
||
| // The receiver counterpart of the channel. | ||
| type Receiver[M any] struct { | ||
| // The channel itself. It's public, so that we can combine it in `select` statements. | ||
| Channel <-chan M | ||
| // Atomic variable that indicates whether the channel is closed. | ||
| receiverClosed *atomic.Bool | ||
| } | ||
|
|
||
| // Marks the channel as closed, which means that no messages could be sent via this channel. | ||
| // Any attempt to send a message would result in an error. This is similar to closing the | ||
| // channel except that we don't close the underlying channel (since in Go receivers can't | ||
| // close the channel). | ||
| // | ||
| // This function reads (in a non-blocking way) all pending messages until blocking. Otherwise, | ||
| // they will stay forver in a channel and get lost. | ||
| func (r *Receiver[M]) Close() []M { | ||
| r.receiverClosed.Store(true) | ||
|
|
||
| messages := make([]M, 0) | ||
| for { | ||
| msg, ok := <-r.Channel | ||
| if !ok { | ||
| break | ||
| } | ||
| messages = append(messages, msg) | ||
| } | ||
|
|
||
| return messages | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| package common | ||
|
|
||
| import ( | ||
| "errors" | ||
| "sync/atomic" | ||
| ) | ||
|
|
||
| // MessageSink is a helper struct that allows to send messages to a message sink. | ||
| // The MessageSink abstracts the message sink which has a certain sender, so that | ||
| // the sender does not have to be specified every time a message is sent. | ||
| // At the same it guarantees that the caller can't alter the `sender`, which means that | ||
| // the sender can't impersonate another sender (and we guarantee this on a compile-time). | ||
| type MessageSink[SenderType comparable, MessageType any] struct { | ||
| // The sender of the messages. This is useful for multiple-producer-single-consumer scenarios. | ||
| sender SenderType | ||
| // The message sink to which the messages are sent. | ||
| messageSink chan<- Message[SenderType, MessageType] | ||
| // Atomic variable that indicates whether the message sink is sealed. | ||
| // This is used to prevent sending messages to a sealed message sink. | ||
| // The variable is atomic because it may be accessed from multiple goroutines. | ||
| sealed atomic.Bool | ||
| } | ||
|
|
||
| // Creates a new MessageSink. The function is generic allowing us to use it for multiple use cases. | ||
| func NewMessageSink[S comparable, M any](sender S, messageSink chan<- Message[S, M]) *MessageSink[S, M] { | ||
| return &MessageSink[S, M]{ | ||
| sender: sender, | ||
| messageSink: messageSink, | ||
| } | ||
| } | ||
|
|
||
| // Sends a message to the message sink. | ||
| func (s *MessageSink[S, M]) Send(message M) error { | ||
| if s.sealed.Load() { | ||
| return errors.New("The channel is sealed, you can't send any messages over it") | ||
| } | ||
|
|
||
| s.messageSink <- Message[S, M]{ | ||
| Sender: s.sender, | ||
| Content: message, | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // Seals the channel, which means that no messages could be sent via this channel. | ||
| // Any attempt to send a message would result in an error. This is similar to closing the | ||
| // channel except that we don't close the underlying channel (since there might be other | ||
| // senders that may want to use it). | ||
| func (s *MessageSink[S, M]) Seal() { | ||
| s.sealed.Store(true) | ||
| } | ||
|
|
||
| // Messages that are sent from the peer to the conference in order to communicate with other peers. | ||
| // Since each peer is isolated from others, it can't influence the state of other peers directly. | ||
| type Message[SenderType comparable, MessageType any] struct { | ||
| // The sender of the message. | ||
| Sender SenderType | ||
| // The content of the message. | ||
| Content MessageType | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,8 @@ | ||
| package conference | ||
|
|
||
| // Configuration for the group conferences (calls). | ||
| type Config struct { | ||
| // Keep-alive timeout for WebRTC connections. If no keep-alive has been received | ||
| // from the client for this duration, the connection is considered dead (in seconds). | ||
| KeepAliveTimeout int `yaml:"timeout"` | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,78 @@ | ||
| package conference | ||
|
|
||
| import ( | ||
| "github.com/pion/webrtc/v3" | ||
| "golang.org/x/exp/slices" | ||
| "maunium.net/go/mautrix/event" | ||
| ) | ||
|
|
||
| // Handle the `SFUMessage` event from the DataChannel message. | ||
| func (c *Conference) processSelectDCMessage(participant *Participant, msg event.SFUMessage) { | ||
| participant.logger.Info("Received select request over DC") | ||
|
|
||
| // Find tracks based on what we were asked for. | ||
| tracks := c.getTracks(msg.Start) | ||
|
|
||
| // Let's check if we have all the tracks that we were asked for are there. | ||
| // If not, we will list which are not available (later on we must inform participant | ||
| // about it unless the participant retries it). | ||
| if len(tracks) != len(msg.Start) { | ||
| for _, expected := range msg.Start { | ||
| found := slices.IndexFunc(tracks, func(track *webrtc.TrackLocalStaticRTP) bool { | ||
| return track.StreamID() == expected.StreamID && track.ID() == expected.TrackID | ||
| }) | ||
|
|
||
| if found == -1 { | ||
| c.logger.Warnf("Track not found: %s", expected.TrackID) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Subscribe to the found tracks. | ||
| for _, track := range tracks { | ||
| if err := participant.peer.SubscribeTo(track); err != nil { | ||
| participant.logger.Errorf("Failed to subscribe to track: %v", err) | ||
| return | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (c *Conference) processAnswerDCMessage(participant *Participant, msg event.SFUMessage) { | ||
| participant.logger.Info("Received SDP answer over DC") | ||
|
|
||
| if err := participant.peer.ProcessSDPAnswer(msg.SDP); err != nil { | ||
| participant.logger.Errorf("Failed to set SDP answer: %v", err) | ||
| return | ||
| } | ||
| } | ||
|
|
||
| func (c *Conference) processPublishDCMessage(participant *Participant, msg event.SFUMessage) { | ||
| participant.logger.Info("Received SDP offer over DC") | ||
|
|
||
| answer, err := participant.peer.ProcessSDPOffer(msg.SDP) | ||
| if err != nil { | ||
| participant.logger.Errorf("Failed to set SDP offer: %v", err) | ||
| return | ||
| } | ||
|
|
||
| participant.streamMetadata = msg.Metadata | ||
|
|
||
| participant.sendDataChannelMessage(event.SFUMessage{ | ||
| Op: event.SFUOperationAnswer, | ||
| SDP: answer.SDP, | ||
| Metadata: c.getAvailableStreamsFor(participant.id), | ||
| }) | ||
| } | ||
|
|
||
| func (c *Conference) processUnpublishDCMessage(participant *Participant) { | ||
| participant.logger.Info("Received unpublish over DC") | ||
| } | ||
|
|
||
| func (c *Conference) processAliveDCMessage(participant *Participant) { | ||
| participant.peer.ProcessHeartbeat() | ||
| } | ||
|
|
||
| func (c *Conference) processMetadataDCMessage(participant *Participant, msg event.SFUMessage) { | ||
| participant.streamMetadata = msg.Metadata | ||
| c.resendMetadataToAllExcept(participant.id) | ||
| } |
Oops, something went wrong.