storm/skald
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Label streams, not tracks.

Browse Source 
We used to label tracks individually, in a view to using the labelling
for simulcast.  Since then, the WebRTC community has converged on a
different strategy, where multiple tracks share a single mid and
are labelled with the rid extension.

We now label whole streams, which is simpler, and use the track's
kind (and, in the future, the rid) to disambiguate.  This changes the
protocol in two ways:

  * in offers, the "labels" dictionary is replaced by a single "label"
    field; and
  * the syntax of the "request" message has changed.
This commit is contained in:
Juliusz Chroboczek
2021-04-28 17:00:50 +02:00
parent b08a2e3943
commit be73380f9f
7 changed files with 147 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rtpconn/rtpconn.go
+12 -55
View File
@@ -313,6 +313,10 @@ func (up *rtpUpTrack) Label() string {
return up.label
}
func (up *rtpUpTrack) Kind() webrtc.RTPCodecType {
return up.track.Kind()
}
func (up *rtpUpTrack) Codec() webrtc.RTPCodecCapability {
return up.track.Codec().RTPCodecCapability
}
@@ -328,10 +332,10 @@ func (up *rtpUpTrack) hasRtcpFb(tpe, parameter string) bool {
type rtpUpConnection struct {
id string
label string
userId string
username string
pc *webrtc.PeerConnection
labels map[string]string
iceCandidates []*webrtc.ICECandidateInit
mu sync.Mutex
@@ -363,6 +367,10 @@ func (up *rtpUpConnection) Id() string {
return up.id
}
func (up *rtpUpConnection) Label() string {
return up.label
}
func (up *rtpUpConnection) User() (string, string) {
return up.userId, up.username
}
@@ -413,33 +421,6 @@ func (up *rtpUpConnection) flushICECandidates() error {
return err
}
func getTrackMid(pc *webrtc.PeerConnection, track *webrtc.TrackRemote) string {
for _, t := range pc.GetTransceivers() {
if t.Receiver() != nil && t.Receiver().Track() == track {
return t.Mid()
}
}
return ""
}
// called locked
func (up *rtpUpConnection) complete() bool {
for mid := range up.labels {
found := false
for _, t := range up.tracks {
m := getTrackMid(up.pc, t.track)
if m == mid {
found = true
break
}
}
if !found {
return false
}
}
return true
}
// pushConnNow pushes a connection to all of the clients in a group
func pushConnNow(up *rtpUpConnection, g *group.Group, cs []group.Client) {
up.mu.Lock()
@@ -460,17 +441,11 @@ func pushConnNow(up *rtpUpConnection, g *group.Group, cs []group.Client) {
// pushConn schedules a call to pushConnNow
func pushConn(up *rtpUpConnection, g *group.Group, cs []group.Client) {
up.mu.Lock()
if up.complete() {
up.mu.Unlock()
pushConnNow(up, g, cs)
return
}
up.pushed = false
up.mu.Unlock()
go func(g *group.Group, cs []group.Client) {
time.Sleep(4 * time.Second)
time.Sleep(200 * time.Millisecond)
up.mu.Lock()
pushed := up.pushed
up.pushed = true
@@ -481,7 +456,7 @@ func pushConn(up *rtpUpConnection, g *group.Group, cs []group.Client) {
}(g, cs)
}
func newUpConn(c group.Client, id string, labels map[string]string, offer string) (*rtpUpConnection, error) {
func newUpConn(c group.Client, id string, label string, offer string) (*rtpUpConnection, error) {
var o sdp.SessionDescription
err := o.Unmarshal([]byte(offer))
if err != nil {
@@ -506,31 +481,13 @@ func newUpConn(c group.Client, id string, labels map[string]string, offer string
}
}
up := &rtpUpConnection{id: id, pc: pc, labels: labels}
up := &rtpUpConnection{id: id, label: label, pc: pc}
pc.OnTrack(func(remote *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
up.mu.Lock()
mid := getTrackMid(pc, remote)
if mid == "" {
log.Printf("Couldn't get track's mid")
return
}
label, ok := up.labels[mid]
if !ok {
log.Printf("Couldn't get track's label")
isvideo := remote.Kind() == webrtc.RTPCodecTypeVideo
if isvideo {
label = "video"
} else {
label = "audio"
}
}
track := &rtpUpTrack{
track: remote,
label: label,
cache: packetcache.New(minPacketCache(remote)),
rate: estimator.New(time.Second),
jitter: jitter.New(remote.Codec().ClockRate),
rtpconn/webclient.go
+52 -90
View File
@@ -59,7 +59,7 @@ type webClient struct {
password string
permissions group.ClientPermissions
status map[string]interface{}
requested map[string]uint32
requested map[string][]string
done chan struct{}
writeCh chan interface{}
writerDone chan struct{}
@@ -122,53 +122,6 @@ func (c *webClient) PushClient(id, username string, permissions group.ClientPerm
})
}
type rateMap map[string]uint32
func (v *rateMap) UnmarshalJSON(b []byte) error {
var m map[string]interface{}
err := json.Unmarshal(b, &m)
if err != nil {
return err
}
n := make(map[string]uint32, len(m))
for k, w := range m {
switch w := w.(type) {
case bool:
if w {
n[k] = ^uint32(0)
} else {
n[k] = 0
}
case float64:
if w < 0 || w >= float64(^uint32(0)) {
return errors.New("overflow")
}
n[k] = uint32(w)
default:
return errors.New("unexpected type in JSON map")
}
}
*v = n
return nil
}
func (v rateMap) MarshalJSON() ([]byte, error) {
m := make(map[string]interface{}, len(v))
for k, w := range v {
switch w {
case 0:
m[k] = false
case ^uint32(0):
m[k] = true
default:
m[k] = w
}
}
return json.Marshal(m)
}
type clientMessage struct {
Type string `json:"type"`
Kind string `json:"kind,omitempty"`
@@ -187,8 +140,8 @@ type clientMessage struct {
Time int64 `json:"time,omitempty"`
SDP string `json:"sdp,omitempty"`
Candidate *webrtc.ICECandidateInit `json:"candidate,omitempty"`
Labels map[string]string `json:"labels,omitempty"`
Request rateMap `json:"request,omitempty"`
Label string `json:"label,omitempty"`
Request map[string][]string `json:"request,omitempty"`
RTCConfiguration *webrtc.Configuration `json:"rtcConfiguration,omitempty"`
}
@@ -216,7 +169,7 @@ func getUpConns(c *webClient) []*rtpUpConnection {
return up
}
func addUpConn(c *webClient, id string, labels map[string]string, offer string) (*rtpUpConnection, bool, error) {
func addUpConn(c *webClient, id, label string, offer string) (*rtpUpConnection, bool, error) {
c.mu.Lock()
defer c.mu.Unlock()
@@ -232,7 +185,7 @@ func addUpConn(c *webClient, id string, labels map[string]string, offer string)
return old, false, nil
}
conn, err := newUpConn(c, id, labels, offer)
conn, err := newUpConn(c, id, label, offer)
if err != nil {
return nil, false, err
}
@@ -540,33 +493,16 @@ func negotiate(c *webClient, down *rtpDownConnection, restartIce bool, replace s
return err
}
labels := make(map[string]string)
for _, t := range down.pc.GetTransceivers() {
var track webrtc.TrackLocal
if t.Sender() != nil {
track = t.Sender().Track()
}
if track == nil {
continue
}
for _, tr := range down.tracks {
if tr.track == track {
labels[t.Mid()] = tr.remote.Label()
}
}
}
source, username := down.remote.User()
return c.write(clientMessage{
Type: "offer",
Id: down.id,
Label: down.remote.Label(),
Replace: replace,
Source: source,
Username: username,
SDP: down.pc.LocalDescription().SDP,
Labels: labels,
})
}
@@ -582,8 +518,8 @@ func sendICE(c *webClient, id string, candidate *webrtc.ICECandidate) error {
})
}
func gotOffer(c *webClient, id string, sdp string, labels map[string]string, replace string) error {
up, _, err := addUpConn(c, id, labels, sdp)
func gotOffer(c *webClient, id, label string, sdp string, replace string) error {
up, _, err := addUpConn(c, id, label, sdp)
if err != nil {
return err
}
@@ -681,7 +617,7 @@ func gotICE(c *webClient, candidate *webrtc.ICECandidateInit, id string) error {
return conn.addICECandidate(candidate)
}
func (c *webClient) setRequested(requested map[string]uint32) error {
func (c *webClient) setRequested(requested map[string][]string) error {
if c.group == nil {
return errors.New("attempted to request with no group joined")
}
@@ -701,8 +637,46 @@ func pushConns(c group.Client, g *group.Group) {
}
}
func (c *webClient) isRequested(label string) bool {
return c.requested[label] != 0
func requestedTracks(c *webClient, up conn.Up, tracks []conn.UpTrack) []conn.UpTrack {
r, ok := c.requested[up.Label()]
if !ok {
r, ok = c.requested[""]
}
if !ok || len(r) == 0 {
return nil
}
var audio, video bool
for _, s := range r {
switch s {
case "audio":
audio = true
case "video":
video = true
default:
log.Printf("client requested unknown value %v", s)
}
}
var ts []conn.UpTrack
if audio {
for _, t := range tracks {
if t.Kind() == webrtc.RTPCodecTypeAudio {
ts = append(ts, t)
break
}
}
}
if video {
for _, t := range tracks {
if t.Kind() == webrtc.RTPCodecTypeVideo {
ts = append(ts, t)
break
}
}
}
return ts
}
func (c *webClient) PushConn(g *group.Group, id string, up conn.Up, tracks []conn.UpTrack, replace string) error {
@@ -871,16 +845,7 @@ func handleAction(c *webClient, a interface{}) error {
}
var tracks []conn.UpTrack
if a.conn != nil {
tracks = make([]conn.UpTrack,
0, len(a.tracks),
)
for _, t := range a.tracks {
if c.isRequested(t.Label()) {
tracks = append(
tracks, t,
)
}
}
tracks = requestedTracks(c, a.conn, a.tracks)
}
if len(tracks) == 0 {
@@ -923,9 +888,6 @@ func handleAction(c *webClient, a interface{}) error {
return nil
}
for _, u := range c.up {
if !u.complete() {
continue
}
tracks := u.getTracks()
replace := u.getReplace(false)
@@ -1051,7 +1013,7 @@ func leaveGroup(c *webClient) {
group.DelClient(c)
c.permissions = group.ClientPermissions{}
c.status = nil
c.requested = map[string]uint32{}
c.requested = make(map[string][]string)
c.group = nil
}
@@ -1238,7 +1200,7 @@ func handleClientMessage(c *webClient, m clientMessage) error {
})
return c.error(group.UserError("not authorised"))
}
err := gotOffer(c, m.Id, m.SDP, m.Labels, m.Replace)
err := gotOffer(c, m.Id, m.Label, m.SDP, m.Replace)
if err != nil {
log.Printf("gotOffer: %v", err)
return failUpConnection(c, m.Id, "negotiation failed")