Commit b26a8cad authored by Juliusz Chroboczek's avatar Juliusz Chroboczek

Label tracks explicitly.

For now, this is only used to request screen sharing as opposed to normal
videos.  In the future, it will be used for simulcasting.
parent 7281a09f
...@@ -26,7 +26,6 @@ import ( ...@@ -26,7 +26,6 @@ import (
"github.com/gorilla/websocket" "github.com/gorilla/websocket"
"github.com/pion/rtcp" "github.com/pion/rtcp"
"github.com/pion/rtp" "github.com/pion/rtp"
"github.com/pion/sdp"
"github.com/pion/webrtc/v2" "github.com/pion/webrtc/v2"
) )
...@@ -104,6 +103,7 @@ type clientMessage struct { ...@@ -104,6 +103,7 @@ type clientMessage struct {
Offer *webrtc.SessionDescription `json:"offer,omitempty"` Offer *webrtc.SessionDescription `json:"offer,omitempty"`
Answer *webrtc.SessionDescription `json:"answer,omitempty"` Answer *webrtc.SessionDescription `json:"answer,omitempty"`
Candidate *webrtc.ICECandidateInit `json:"candidate,omitempty"` Candidate *webrtc.ICECandidateInit `json:"candidate,omitempty"`
Labels map[string]string `json:"labels,omitempty"`
Del bool `json:"del,omitempty"` Del bool `json:"del,omitempty"`
Request []string `json:"request,omitempty"` Request []string `json:"request,omitempty"`
} }
...@@ -301,13 +301,14 @@ func addUpConn(c *client, id string) (*upConnection, error) { ...@@ -301,13 +301,14 @@ func addUpConn(c *client, id string) (*upConnection, error) {
} }
track := &upTrack{ track := &upTrack{
track: remote, track: remote,
label: u.labels[remote.ID()],
cache: packetcache.New(96), cache: packetcache.New(96),
rate: estimator.New(time.Second), rate: estimator.New(time.Second),
jitter: jitter.New(remote.Codec().ClockRate), jitter: jitter.New(remote.Codec().ClockRate),
maxBitrate: ^uint64(0), maxBitrate: ^uint64(0),
} }
u.tracks = append(u.tracks, track) u.tracks = append(u.tracks, track)
done := len(u.tracks) >= u.trackCount done := u.complete()
if remote.Kind() == webrtc.RTPCodecTypeVideo { if remote.Kind() == webrtc.RTPCodecTypeVideo {
atomic.AddUint32(&c.group.videoCount, 1) atomic.AddUint32(&c.group.videoCount, 1)
} }
...@@ -830,28 +831,27 @@ func sendRecovery(p *rtcp.TransportLayerNack, track *downTrack) { ...@@ -830,28 +831,27 @@ func sendRecovery(p *rtcp.TransportLayerNack, track *downTrack) {
} }
} }
func countMediaStreams(data string) (int, error) { func negotiate(c *client, id string, down *downConnection) error {
desc := sdp.NewJSEPSessionDescription(false) offer, err := down.pc.CreateOffer(nil)
err := desc.Unmarshal(data)
if err != nil { if err != nil {
return 0, err return err
} }
return len(desc.MediaDescriptions), nil
}
func negotiate(c *client, id string, pc *webrtc.PeerConnection) error { err = down.pc.SetLocalDescription(offer)
offer, err := pc.CreateOffer(nil)
if err != nil { if err != nil {
return err return err
} }
err = pc.SetLocalDescription(offer)
if err != nil { labels := make(map[string]string)
return err for _, t := range down.tracks {
labels[t.track.ID()] = t.remote.label
} }
return c.write(clientMessage{ return c.write(clientMessage{
Type: "offer", Type: "offer",
Id: id, Id: id,
Offer: &offer, Offer: &offer,
Labels: labels,
}) })
} }
...@@ -867,7 +867,7 @@ func sendICE(c *client, id string, candidate *webrtc.ICECandidate) error { ...@@ -867,7 +867,7 @@ func sendICE(c *client, id string, candidate *webrtc.ICECandidate) error {
}) })
} }
func gotOffer(c *client, offer webrtc.SessionDescription, id string) error { func gotOffer(c *client, id string, offer webrtc.SessionDescription, labels map[string]string) error {
var err error var err error
up, ok := c.up[id] up, ok := c.up[id]
if !ok { if !ok {
...@@ -879,12 +879,6 @@ func gotOffer(c *client, offer webrtc.SessionDescription, id string) error { ...@@ -879,12 +879,6 @@ func gotOffer(c *client, offer webrtc.SessionDescription, id string) error {
if c.username != "" { if c.username != "" {
up.label = c.username up.label = c.username
} }
n, err := countMediaStreams(offer.SDP)
if err != nil {
log.Printf("Couldn't parse SDP: %v", err)
n = 2
}
up.trackCount = n
err = up.pc.SetRemoteDescription(offer) err = up.pc.SetRemoteDescription(offer)
if err != nil { if err != nil {
return err return err
...@@ -900,6 +894,8 @@ func gotOffer(c *client, offer webrtc.SessionDescription, id string) error { ...@@ -900,6 +894,8 @@ func gotOffer(c *client, offer webrtc.SessionDescription, id string) error {
return err return err
} }
up.labels = labels
return c.write(clientMessage{ return c.write(clientMessage{
Type: "answer", Type: "answer",
Id: id, Id: id,
...@@ -907,7 +903,7 @@ func gotOffer(c *client, offer webrtc.SessionDescription, id string) error { ...@@ -907,7 +903,7 @@ func gotOffer(c *client, offer webrtc.SessionDescription, id string) error {
}) })
} }
func gotAnswer(c *client, answer webrtc.SessionDescription, id string) error { func gotAnswer(c *client, id string, answer webrtc.SessionDescription) error {
conn := getDownConn(c, id) conn := getDownConn(c, id)
if conn == nil { if conn == nil {
return protocolError("unknown id in answer") return protocolError("unknown id in answer")
...@@ -934,11 +930,7 @@ func gotICE(c *client, candidate *webrtc.ICECandidateInit, id string) error { ...@@ -934,11 +930,7 @@ func gotICE(c *client, candidate *webrtc.ICECandidateInit, id string) error {
return pc.AddICECandidate(*candidate) return pc.AddICECandidate(*candidate)
} }
func (c *client) setRequested(audio, video bool) error { func (c *client) setRequested(requested []string) error {
if audio == c.requestedAudio && video == c.requestedVideo {
return nil
}
if c.down != nil { if c.down != nil {
for id := range c.down { for id := range c.down {
c.write(clientMessage{ c.write(clientMessage{
...@@ -949,8 +941,7 @@ func (c *client) setRequested(audio, video bool) error { ...@@ -949,8 +941,7 @@ func (c *client) setRequested(audio, video bool) error {
} }
} }
c.requestedAudio = audio c.requested = requested
c.requestedVideo = video
go func() { go func() {
clients := c.group.getClients(c) clients := c.group.getClients(c)
...@@ -962,15 +953,13 @@ func (c *client) setRequested(audio, video bool) error { ...@@ -962,15 +953,13 @@ func (c *client) setRequested(audio, video bool) error {
return nil return nil
} }
func (c *client) requested(kind webrtc.RTPCodecType) bool { func (c *client) isRequested(label string) bool {
switch kind { for _, r := range c.requested {
case webrtc.RTPCodecTypeAudio: if label == r {
return c.requestedAudio return true
case webrtc.RTPCodecTypeVideo: }
return c.requestedVideo
default:
return false
} }
return false
} }
func pushTracks(c *client, conn *upConnection, tracks []*upTrack, done bool, label string) { func pushTracks(c *client, conn *upConnection, tracks []*upTrack, done bool, label string) {
...@@ -989,7 +978,7 @@ func clientLoop(c *client, conn *websocket.Conn) error { ...@@ -989,7 +978,7 @@ func clientLoop(c *client, conn *websocket.Conn) error {
go clientReader(conn, read, c.done) go clientReader(conn, read, c.done)
defer func() { defer func() {
c.setRequested(false, false) c.setRequested([]string{})
if c.up != nil { if c.up != nil {
for id := range c.up { for id := range c.up {
delUpConn(c, id) delUpConn(c, id)
...@@ -1044,7 +1033,7 @@ func clientLoop(c *client, conn *websocket.Conn) error { ...@@ -1044,7 +1033,7 @@ func clientLoop(c *client, conn *websocket.Conn) error {
case addTrackAction: case addTrackAction:
var down *downConnection var down *downConnection
var err error var err error
if c.requested(a.track.track.Kind()) { if c.isRequested(a.track.label) {
down, _, err = addDownTrack( down, _, err = addDownTrack(
c, a.remote.id, a.track, c, a.remote.id, a.track,
a.remote) a.remote)
...@@ -1055,7 +1044,7 @@ func clientLoop(c *client, conn *websocket.Conn) error { ...@@ -1055,7 +1044,7 @@ func clientLoop(c *client, conn *websocket.Conn) error {
down = getDownConn(c, a.remote.id) down = getDownConn(c, a.remote.id)
} }
if a.done && down != nil { if a.done && down != nil {
err = negotiate(c, a.remote.id, down.pc) err = negotiate(c, a.remote.id, down)
if err != nil { if err != nil {
return err return err
} }
...@@ -1080,7 +1069,7 @@ func clientLoop(c *client, conn *websocket.Conn) error { ...@@ -1080,7 +1069,7 @@ func clientLoop(c *client, conn *websocket.Conn) error {
copy(tracks, u.tracks) copy(tracks, u.tracks)
go pushTracks( go pushTracks(
a.c, u, tracks, a.c, u, tracks,
len(tracks) >= u.trackCount-1, u.complete(),
u.label, u.label,
) )
} }
...@@ -1141,15 +1130,7 @@ func clientLoop(c *client, conn *websocket.Conn) error { ...@@ -1141,15 +1130,7 @@ func clientLoop(c *client, conn *websocket.Conn) error {
func handleClientMessage(c *client, m clientMessage) error { func handleClientMessage(c *client, m clientMessage) error {
switch m.Type { switch m.Type {
case "request": case "request":
var audio, video bool err := c.setRequested(m.Request)
for _, s := range m.Request {
switch(s) {
case "audio": audio = true
case "video": video = true
default: log.Printf("Unknown request %v", s)
}
}
err := c.setRequested(audio, video)
if err != nil { if err != nil {
return err return err
} }
...@@ -1164,7 +1145,7 @@ func handleClientMessage(c *client, m clientMessage) error { ...@@ -1164,7 +1145,7 @@ func handleClientMessage(c *client, m clientMessage) error {
if m.Offer == nil { if m.Offer == nil {
return protocolError("null offer") return protocolError("null offer")
} }
err := gotOffer(c, *m.Offer, m.Id) err := gotOffer(c, m.Id, *m.Offer, m.Labels)
if err != nil { if err != nil {
return err return err
} }
...@@ -1172,7 +1153,7 @@ func handleClientMessage(c *client, m clientMessage) error { ...@@ -1172,7 +1153,7 @@ func handleClientMessage(c *client, m clientMessage) error {
if m.Answer == nil { if m.Answer == nil {
return protocolError("null answer") return protocolError("null answer")
} }
err := gotAnswer(c, *m.Answer, m.Id) err := gotAnswer(c, m.Id, *m.Answer)
if err != nil { if err != nil {
return err return err
} }
......
...@@ -26,6 +26,7 @@ import ( ...@@ -26,6 +26,7 @@ import (
type upTrack struct { type upTrack struct {
track *webrtc.Track track *webrtc.Track
label string
rate *estimator.Estimator rate *estimator.Estimator
cache *packetcache.Cache cache *packetcache.Cache
jitter *jitter.Estimator jitter *jitter.Estimator
...@@ -74,11 +75,27 @@ func (up *upTrack) hasRtcpFb(tpe, parameter string) bool { ...@@ -74,11 +75,27 @@ func (up *upTrack) hasRtcpFb(tpe, parameter string) bool {
} }
type upConnection struct { type upConnection struct {
id string id string
label string label string
pc *webrtc.PeerConnection pc *webrtc.PeerConnection
trackCount int tracks []*upTrack
tracks []*upTrack labels map[string]string
}
func (up *upConnection) complete() bool {
for id, _ := range up.labels {
found := false
for _, t := range up.tracks {
if t.track.ID() == id {
found = true
break
}
}
if !found {
return false
}
}
return true
} }
type bitrate struct { type bitrate struct {
...@@ -150,16 +167,15 @@ type downConnection struct { ...@@ -150,16 +167,15 @@ type downConnection struct {
} }
type client struct { type client struct {
group *group group *group
id string id string
username string username string
permissions userPermission permissions userPermission
requestedAudio bool requested []string
requestedVideo bool done chan struct{}
done chan struct{} writeCh chan interface{}
writeCh chan interface{} writerDone chan struct{}
writerDone chan struct{} actionCh chan interface{}
actionCh chan interface{}
mu sync.Mutex mu sync.Mutex
down map[string]*downConnection down map[string]*downConnection
......
...@@ -65,6 +65,11 @@ h1 { ...@@ -65,6 +65,11 @@ h1 {
margin-right: 0.4em; margin-right: 0.4em;
} }
#requestselect {
width: 8em;
text-align-last: center;
}
#main { #main {
display: flex; display: flex;
} }
......
...@@ -47,8 +47,12 @@ ...@@ -47,8 +47,12 @@
<label for="sharebox">Share screen:</label> <label for="sharebox">Share screen:</label>
<input id="sharebox" type="checkbox" disabled/> <input id="sharebox" type="checkbox" disabled/>
<label for="requestbox">Receive video:</label> <label for="requestselect">Receive:</label>
<input id="requestbox" type="checkbox" checked> <select id="requestselect">
<option value="audio">audio only</option>
<option value="screenshare">screen share</option>
<option value="everything" selected>everything</option>
</select>
</div> </div>
</div> </div>
......
...@@ -39,6 +39,7 @@ function Connection(id, pc) { ...@@ -39,6 +39,7 @@ function Connection(id, pc) {
this.label = null; this.label = null;
this.pc = pc; this.pc = pc;
this.stream = null; this.stream = null;
this.labels = {};
this.iceCandidates = []; this.iceCandidates = [];
this.timers = []; this.timers = [];
this.audioStats = {}; this.audioStats = {};
...@@ -140,9 +141,9 @@ document.getElementById('sharebox').onchange = function(e) { ...@@ -140,9 +141,9 @@ document.getElementById('sharebox').onchange = function(e) {
setShareMedia(this.checked); setShareMedia(this.checked);
}; };
document.getElementById('requestbox').onchange = function(e) { document.getElementById('requestselect').onchange = function(e) {
e.preventDefault(); e.preventDefault();
sendRequest(this.checked); sendRequest(this.value);
}; };
async function updateStats(conn, sender) { async function updateStats(conn, sender) {
...@@ -312,6 +313,7 @@ async function setLocalMedia(setup) { ...@@ -312,6 +313,7 @@ async function setLocalMedia(setup) {
let c = up[localMediaId]; let c = up[localMediaId];
c.stream = stream; c.stream = stream;
stream.getTracks().forEach(t => { stream.getTracks().forEach(t => {
c.labels[t.id] = t.kind
let sender = c.pc.addTrack(t, stream); let sender = c.pc.addTrack(t, stream);
c.setInterval(() => { c.setInterval(() => {
updateStats(c, sender); updateStats(c, sender);
...@@ -358,6 +360,7 @@ async function setShareMedia(setup) { ...@@ -358,6 +360,7 @@ async function setShareMedia(setup) {
document.getElementById('sharebox').checked = false; document.getElementById('sharebox').checked = false;
setShareMedia(false); setShareMedia(false);
}; };
c.labels[t.id] = 'screenshare';
c.setInterval(() => { c.setInterval(() => {
updateStats(c, sender); updateStats(c, sender);
}, 2000); }, 2000);
...@@ -485,7 +488,7 @@ function serverConnect() { ...@@ -485,7 +488,7 @@ function serverConnect() {
username: up.username, username: up.username,
password: up.password, password: up.password,
}); });
sendRequest(document.getElementById('requestbox').checked); sendRequest(document.getElementById('requestselect').value);
resolve(); resolve();
}; };
socket.onclose = function(e) { socket.onclose = function(e) {
...@@ -508,7 +511,7 @@ function serverConnect() { ...@@ -508,7 +511,7 @@ function serverConnect() {
let m = JSON.parse(e.data); let m = JSON.parse(e.data);
switch(m.type) { switch(m.type) {
case 'offer': case 'offer':
gotOffer(m.id, m.offer); gotOffer(m.id, m.labels, m.offer);
break; break;
case 'answer': case 'answer':
gotAnswer(m.id, m.answer); gotAnswer(m.id, m.answer);
...@@ -556,14 +559,30 @@ function serverConnect() { ...@@ -556,14 +559,30 @@ function serverConnect() {
}); });
} }
function sendRequest(video) { function sendRequest(value) {
let request = [];
switch(value) {
case 'audio':
request = ['audio'];
break;
case 'screenshare':
request = ['audio', 'screenshare'];
break;
case 'everything':
request = ['audio', 'screenshare', 'video'];
break;
default:
console.error(`Uknown value ${value} in sendRequest`);
break;
}
send({ send({
type: 'request', type: 'request',
request: video ? ['audio', 'video'] : ['audio'], request: request,
}); });
} }
async function gotOffer(id, offer) { async function gotOffer(id, labels, offer) {
let c = down[id]; let c = down[id];
if(!c) { if(!c) {
let pc = new RTCPeerConnection({ let pc = new RTCPeerConnection({
...@@ -587,6 +606,8 @@ async function gotOffer(id, offer) { ...@@ -587,6 +606,8 @@ async function gotOffer(id, offer) {
}; };
} }
c.labels = labels;
await c.pc.setRemoteDescription(offer); await c.pc.setRemoteDescription(offer);
await addIceCandidates(c); await addIceCandidates(c);
let answer = await c.pc.createAnswer(); let answer = await c.pc.createAnswer();
...@@ -1007,6 +1028,7 @@ async function negotiate(id) { ...@@ -1007,6 +1028,7 @@ async function negotiate(id) {
send({ send({
type: 'offer', type: 'offer',
id: id, id: id,
labels: c.labels,
offer: offer, offer: offer,
}); });
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment