diff --git a/README.FRONTEND b/README.FRONTEND index ef6427d..8987caf 100644 --- a/README.FRONTEND +++ b/README.FRONTEND @@ -114,11 +114,7 @@ this point, you may set up an `audio` or `video` component straight away, or you may choose to wait until the `ondowntrack` callback is called. After a new stream is created, `ondowntrack` will be called whenever -a track is added. If the `MediaStream` passed to `ondowntrack` differs -from the one previously received, then the stream has been torn down and -recreated, and you must drop all previously received tracks; in practice, -it is enough to set the `srcObject` property of the video component to the -new stream. +a track is added. The `onstatus` callback is invoked whenever the client library detects a change in the status of the stream; states `connected` and `complete` @@ -126,7 +122,10 @@ indicate a functioning stream; other states indicate that the stream is not working right now but might recover in the future. The `onclose` callback is called when the stream is destroyed, either by -the server or in response to a call to the `close` method. +the server or in response to a call to the `close` method. The optional +parameter is true when the stream is being replaced by a new stream; in +that case, the call to `onclose` will be followed with a call to +`onstream` with the same `localId` value. ## Pushing outgoing video streams @@ -145,6 +144,11 @@ localStream.getTracks().forEach(t => { }); ``` +The `newUpStream` method takes an optional parameter. If this is set to +the `localId` property of an existing stream, then the existing stream +will be closed and the server will be informed that the new stream +replaces the existing stream. + See above for information about setting up the `labels` dictionary. ## Stream statistics diff --git a/README.PROTOCOL b/README.PROTOCOL index 069be5f..96f29db 100644 --- a/README.PROTOCOL +++ b/README.PROTOCOL @@ -142,8 +142,8 @@ A stream is created by the sender with the `offer` message: ```javascript { type: 'offer', - kind: '' or 'renegotiate' id: id, + replace: id, source: source-id, username: username, sdp: sdp, @@ -151,12 +151,10 @@ A stream is created by the sender with the `offer` message: } ``` -If kind is the empty string, then this is a new offer that might or might -not replace an existing stream; if a stream with the same id exists, it -must be torn down before the new stream is created. If kind is -`renegotiate`, then a stream with the given id already exists, and the -receiving peer may either tear down the existing stream or merely perform -a renegotiation. +If a stream with the same id exists, then this is a renegotation; +otherwise this message creates a new stream. If the field `replace` is +not empty, then this request additionally requests that an existing stream +with the given id should be closed, and the new stream should replace it. The field `sdp` contains the raw SDP string (i.e. the `sdp` field of a JSEP session description). Galène will interpret the `nack`, diff --git a/diskwriter/diskwriter.go b/diskwriter/diskwriter.go index 52db542..cda9419 100644 --- a/diskwriter/diskwriter.go +++ b/diskwriter/diskwriter.go @@ -93,7 +93,7 @@ func (client *Client) Kick(id, user, message string) error { return err } -func (client *Client) PushConn(g *group.Group, id string, up conn.Up, tracks []conn.UpTrack) error { +func (client *Client) PushConn(g *group.Group, id string, up conn.Up, tracks []conn.UpTrack, replace string) error { if client.group != g { return nil } @@ -105,6 +105,14 @@ func (client *Client) PushConn(g *group.Group, id string, up conn.Up, tracks []c return errors.New("disk client is closed") } + rp := client.down[replace] + if rp != nil { + rp.Close() + delete(client.down, replace) + } else { + log.Printf("Replacing unknown connection") + } + old := client.down[id] if old != nil { old.Close() diff --git a/group/client.go b/group/client.go index db51b87..c0ca285 100644 --- a/group/client.go +++ b/group/client.go @@ -99,7 +99,7 @@ type Client interface { Permissions() ClientPermissions SetPermissions(ClientPermissions) OverridePermissions(*Group) bool - PushConn(g *Group, id string, conn conn.Up, tracks []conn.UpTrack) error + PushConn(g *Group, id string, conn conn.Up, tracks []conn.UpTrack, replace string) error PushClient(id, username string, add bool) error } diff --git a/rtpconn/rtpconn.go b/rtpconn/rtpconn.go index 984e377..30dc84c 100644 --- a/rtpconn/rtpconn.go +++ b/rtpconn/rtpconn.go @@ -143,8 +143,8 @@ type rtpDownConnection struct { iceCandidates []*webrtc.ICECandidateInit negotiationNeeded int - mu sync.Mutex - tracks []*rtpDownTrack + mu sync.Mutex + tracks []*rtpDownTrack } func (down *rtpDownConnection) getTracks() []*rtpDownTrack { @@ -333,10 +333,11 @@ type rtpUpConnection struct { labels map[string]string iceCandidates []*webrtc.ICECandidateInit - mu sync.Mutex - pushed bool - tracks []*rtpUpTrack - local []conn.Down + mu sync.Mutex + pushed bool + replace string + tracks []*rtpUpTrack + local []conn.Down } func (up *rtpUpConnection) getTracks() []*rtpUpTrack { @@ -347,6 +348,16 @@ func (up *rtpUpConnection) getTracks() []*rtpUpTrack { return tracks } +func (up *rtpUpConnection) getReplace(reset bool) string { + up.mu.Lock() + defer up.mu.Unlock() + replace := up.replace + if reset { + up.replace = "" + } + return replace +} + func (up *rtpUpConnection) Id() string { return up.id } @@ -443,6 +454,8 @@ func (up *rtpUpConnection) complete() bool { func pushConnNow(up *rtpUpConnection, g *group.Group, cs []group.Client) { up.mu.Lock() up.pushed = true + replace := up.replace + up.replace = "" tracks := make([]conn.UpTrack, len(up.tracks)) for i, t := range up.tracks { tracks[i] = t @@ -450,7 +463,7 @@ func pushConnNow(up *rtpUpConnection, g *group.Group, cs []group.Client) { up.mu.Unlock() for _, c := range cs { - c.PushConn(g, up.id, up, tracks) + c.PushConn(g, up.id, up, tracks, replace) } } diff --git a/rtpconn/webclient.go b/rtpconn/webclient.go index 76aa8be..62c0a79 100644 --- a/rtpconn/webclient.go +++ b/rtpconn/webclient.go @@ -169,6 +169,7 @@ type clientMessage struct { Type string `json:"type"` Kind string `json:"kind,omitempty"` Id string `json:"id,omitempty"` + Replace string `json:"replace,omitempty"` Source string `json:"source,omitempty"` Dest string `json:"dest,omitempty"` Username string `json:"username,omitempty"` @@ -246,17 +247,26 @@ func addUpConn(c *webClient, id string, labels map[string]string, offer string) return conn, true, nil } -func delUpConn(c *webClient, id string) bool { +var ErrUserMismatch = errors.New("user id mismatch") + +func delUpConn(c *webClient, id string, userId string) (string, error) { c.mu.Lock() if c.up == nil { c.mu.Unlock() - return false + return "", os.ErrNotExist } conn := c.up[id] if conn == nil { c.mu.Unlock() - return false + return "", os.ErrNotExist } + if userId != "" && conn.userId != userId { + c.mu.Unlock() + return "", ErrUserMismatch + } + + replace := conn.getReplace(true) + delete(c.up, id) c.mu.Unlock() @@ -264,7 +274,7 @@ func delUpConn(c *webClient, id string) bool { if g != nil { go func(clients []group.Client) { for _, c := range clients { - err := c.PushConn(g, conn.id, nil, nil) + err := c.PushConn(g, conn.id, nil, nil, replace) if err != nil { log.Printf("PushConn: %v", err) } @@ -275,7 +285,7 @@ func delUpConn(c *webClient, id string) bool { } conn.pc.Close() - return true + return conn.replace, nil } func getDownConn(c *webClient, id string) *rtpDownConnection { @@ -355,13 +365,13 @@ func addDownConnHelper(c *webClient, conn *rtpDownConnection, remote conn.Up) er return nil } -func delDownConn(c *webClient, id string) bool { +func delDownConn(c *webClient, id string) error { conn := delDownConnHelper(c, id) if conn != nil { conn.pc.Close() - return true + return nil } - return false + return os.ErrNotExist } func delDownConnHelper(c *webClient, id string) *rtpDownConnection { @@ -429,7 +439,7 @@ func addDownTrack(c *webClient, conn *rtpDownConnection, remoteTrack conn.UpTrac return sender, nil } -func negotiate(c *webClient, down *rtpDownConnection, renegotiate, restartIce bool) error { +func negotiate(c *webClient, down *rtpDownConnection, restartIce bool, replace string) error { if down.pc.SignalingState() == webrtc.SignalingStateHaveLocalOffer { // avoid sending multiple offers back-to-back if restartIce { @@ -470,17 +480,12 @@ func negotiate(c *webClient, down *rtpDownConnection, renegotiate, restartIce bo } } - kind := "" - if renegotiate { - kind = "renegotiate" - } - source, username := down.remote.User() return c.write(clientMessage{ Type: "offer", - Kind: kind, Id: down.id, + Replace: replace, Source: source, Username: username, SDP: down.pc.LocalDescription().SDP, @@ -500,31 +505,21 @@ func sendICE(c *webClient, id string, candidate *webrtc.ICECandidate) error { }) } -func gotOffer(c *webClient, id string, sdp string, renegotiate bool, labels map[string]string) error { - if !renegotiate { - // unless the client indicates that this is a compatible - // renegotiation, tear down the existing connection. - delUpConn(c, id) - } - - up, isnew, err := addUpConn(c, id, labels, sdp) +func gotOffer(c *webClient, id string, sdp string, labels map[string]string, replace string) error { + up, _, err := addUpConn(c, id, labels, sdp) if err != nil { return err } up.userId = c.Id() up.username = c.Username() + up.replace = replace err = up.pc.SetRemoteDescription(webrtc.SessionDescription{ Type: webrtc.SDPTypeOffer, SDP: sdp, }) if err != nil { - if renegotiate && !isnew { - // create a new PC from scratch - log.Printf("SetRemoteDescription(offer): %v", err) - return gotOffer(c, id, sdp, false, labels) - } return err } @@ -679,8 +674,8 @@ func addDownConnTracks(c *webClient, remote conn.Up, tracks []conn.UpTrack) (*rt return down, nil } -func (c *webClient) PushConn(g *group.Group, id string, up conn.Up, tracks []conn.UpTrack) error { - err := c.action(pushConnAction{g, id, up, tracks}) +func (c *webClient) PushConn(g *group.Group, id string, up conn.Up, tracks []conn.UpTrack, replace string) error { + err := c.action(pushConnAction{g, id, up, tracks, replace}) if err != nil { return err } @@ -746,10 +741,11 @@ func StartClient(conn *websocket.Conn) (err error) { } type pushConnAction struct { - group *group.Group - id string - conn conn.Up - tracks []conn.UpTrack + group *group.Group + id string + conn conn.Up + tracks []conn.UpTrack + replace string } type pushConnsAction struct { @@ -811,15 +807,23 @@ func clientLoop(c *webClient, ws *websocket.Conn) error { return nil } if a.conn == nil { - found := delDownConn(c, a.id) - if found { + if a.replace != "" { + err := delDownConn( + c, a.replace, + ) + if err == nil { + c.write(clientMessage{ + Type: "close", + Id: a.replace, + }) + } + } + err := delDownConn(c, a.id) + if err == nil { c.write(clientMessage{ Type: "close", Id: a.id, }) - } else { - log.Printf("Deleting unknown " + - "down connection") } continue } @@ -830,7 +834,9 @@ func clientLoop(c *webClient, ws *websocket.Conn) error { return err } if down != nil { - err = negotiate(c, down, false, false) + err = negotiate( + c, down, false, a.replace, + ) if err != nil { log.Printf( "Negotiation failed: %v", @@ -852,13 +858,17 @@ func clientLoop(c *webClient, ws *websocket.Conn) error { continue } tracks := u.getTracks() + replace := u.getReplace(false) + ts := make([]conn.UpTrack, len(tracks)) for i, t := range tracks { ts[i] = t } - go func(u *rtpUpConnection, ts []conn.UpTrack) { + go func(u *rtpUpConnection, + ts []conn.UpTrack, + replace string) { err := a.client.PushConn( - g, u.id, u, ts, + g, u.id, u, ts, replace, ) if err != nil { log.Printf( @@ -866,11 +876,11 @@ func clientLoop(c *webClient, ws *websocket.Conn) error { err, ) } - }(u, ts) + }(u, ts, replace) } case connectionFailedAction: if down := getDownConn(c, a.id); down != nil { - err := negotiate(c, down, true, true) + err := negotiate(c, down, true, "") if err != nil { return err } @@ -883,7 +893,7 @@ func clientLoop(c *webClient, ws *websocket.Conn) error { go c.PushConn( c.group, down.remote.Id(), down.remote, - tracks, + tracks, "", ) } else if up := getUpConn(c, a.id); up != nil { c.write(clientMessage{ @@ -912,8 +922,12 @@ func clientLoop(c *webClient, ws *websocket.Conn) error { if !c.permissions.Present { up := getUpConns(c) for _, u := range up { - found := delUpConn(c, u.id) - if found { + replace, err := + delUpConn(c, u.id, c.id) + if err == nil { + if replace != "" { + delUpConn(c, replace, c.id) + } failUpConnection( c, u.id, "permission denied", @@ -975,7 +989,7 @@ func leaveGroup(c *webClient) { c.setRequested(map[string]uint32{}) if c.up != nil { for id := range c.up { - delUpConn(c, id) + delUpConn(c, id, c.id) } } @@ -1158,15 +1172,16 @@ func handleClientMessage(c *webClient, m clientMessage) error { return c.setRequested(m.Request) case "offer": if !c.permissions.Present { + if m.Replace != "" { + delUpConn(c, m.Replace, c.id) + } c.write(clientMessage{ Type: "abort", Id: m.Id, }) return c.error(group.UserError("not authorised")) } - err := gotOffer( - c, m.Id, m.SDP, m.Kind == "renegotiate", m.Labels, - ) + err := gotOffer(c, m.Id, m.SDP, m.Labels, m.Replace) if err != nil { log.Printf("gotOffer: %v", err) return failUpConnection(c, m.Id, "negotiation failed") @@ -1184,8 +1199,9 @@ func handleClientMessage(c *webClient, m clientMessage) error { down := getDownConn(c, m.Id) if down.negotiationNeeded > negotiationUnneeded { err := negotiate( - c, down, true, + c, down, down.negotiationNeeded == negotiationRestartIce, + "", ) if err != nil { return failDownConnection( @@ -1196,7 +1212,7 @@ func handleClientMessage(c *webClient, m clientMessage) error { case "renegotiate": down := getDownConn(c, m.Id) if down != nil { - err := negotiate(c, down, true, true) + err := negotiate(c, down, true, "") if err != nil { return failDownConnection( c, m.Id, "renegotiation failed", @@ -1206,14 +1222,22 @@ func handleClientMessage(c *webClient, m clientMessage) error { log.Printf("Trying to renegotiate unknown connection") } case "close": - found := delUpConn(c, m.Id) - if !found { - log.Printf("Deleting unknown up connection %v", m.Id) + replace, err := delUpConn(c, m.Id, c.id) + if err != nil { + log.Printf("Deleting up connection %v: %v", + m.Id, err) + return nil + } + if replace != "" { + _, err := delUpConn(c, replace, c.id) + if err != nil && !os.IsNotExist(err) { + log.Printf("Replace up connection: %v", err) + } } case "abort": - found := delDownConn(c, m.Id) - if !found { - log.Printf("Attempted to abort unknown connection") + err := delDownConn(c, m.Id) + if err != nil { + log.Printf("Abort: %v", err) } c.write(clientMessage{ Type: "close", diff --git a/static/galene.js b/static/galene.js index 3f21145..2dec6de 100644 --- a/static/galene.js +++ b/static/galene.js @@ -340,7 +340,7 @@ function gotClose(code, reason) { function gotDownStream(c) { c.onclose = function(replace) { if(!replace) - delMedia(c.id); + delMedia(c.localId); }; c.onerror = function(e) { console.error(e); @@ -397,10 +397,9 @@ getButtonElement('unpresentbutton').onclick = function(e) { }; function changePresentation() { - let id = findUpMedia('local'); - if(id) { - addLocalMedia(id); - } + let c = findUpMedia('local'); + if(c) + addLocalMedia(c.localId); } /** @@ -624,7 +623,7 @@ function gotUpStats(stats) { * @param {boolean} value */ function setActive(c, value) { - let peer = document.getElementById('peer-' + c.id); + let peer = document.getElementById('peer-' + c.localId); if(value) peer.classList.add('peer-active'); else @@ -773,10 +772,10 @@ async function setMediaChoices(done) { /** - * @param {string} [id] + * @param {string} [localId] */ -function newUpStream(id) { - let c = serverConnection.newUpStream(id); +function newUpStream(localId) { + let c = serverConnection.newUpStream(localId); c.onstatus = function(status) { setMediaStatus(c); }; @@ -1012,9 +1011,9 @@ function isSafari() { } /** - * @param {string} [id] + * @param {string} [localId] */ -async function addLocalMedia(id) { +async function addLocalMedia(localId) { let settings = getSettings(); let audio = settings.audio ? {deviceId: settings.audio} : false; @@ -1039,13 +1038,7 @@ async function addLocalMedia(id) { } } - let old = id && serverConnection.up[id]; - if(!audio && !video) { - if(old) - old.close(); - return; - } - + let old = serverConnection.findByLocalId(localId); if(old && old.onclose) { // make sure that the camera is released before we try to reopen it old.onclose.call(old, true); @@ -1063,7 +1056,7 @@ async function addLocalMedia(id) { setMediaChoices(true); - let c = newUpStream(id); + let c = newUpStream(localId); c.kind = 'local'; c.stream = stream; @@ -1076,21 +1069,21 @@ async function addLocalMedia(id) { stopStream(stream); setFilter(c, null); if(!replace) - delMedia(c.id); + delMedia(c.localId); } } catch(e) { displayWarning(e); c.onclose = replace => { stopStream(c.stream); if(!replace) - delMedia(c.id); + delMedia(c.localId); } } } else { c.onclose = replace => { stopStream(c.stream); if(!replace) - delMedia(c.id); + delMedia(c.localId); } } @@ -1144,7 +1137,7 @@ async function addShareMedia() { c.onclose = replace => { stopStream(stream); if(!replace) - delMedia(c.id); + delMedia(c.localId); } stream.getTracks().forEach(t => { c.pc.addTrack(t, stream); @@ -1180,13 +1173,13 @@ async function addFileMedia(file) { c.onclose = function(replace) { stopStream(c.stream); let media = /** @type{HTMLVideoElement} */ - (document.getElementById('media-' + this.id)); + (document.getElementById('media-' + this.localId)); if(media && media.src) { URL.revokeObjectURL(media.src); media.src = null; } if(!replace) - delMedia(c.id); + delMedia(c.localId); }; stream.onaddtrack = function(e) { @@ -1261,11 +1254,13 @@ function closeUpMediaKind(kind) { /** * @param {string} kind + * @returns {Stream} */ function findUpMedia(kind) { for(let id in serverConnection.up) { - if(serverConnection.up[id].kind === kind) - return id; + let c = serverConnection.up[id] + if(c.kind === kind) + return c; } return null; } @@ -1304,16 +1299,16 @@ function muteLocalTracks(mute) { async function setMedia(c, isUp, mirror, video) { let peersdiv = document.getElementById('peers'); - let div = document.getElementById('peer-' + c.id); + let div = document.getElementById('peer-' + c.localId); if(!div) { div = document.createElement('div'); - div.id = 'peer-' + c.id; + div.id = 'peer-' + c.localId; div.classList.add('peer'); peersdiv.appendChild(div); } let media = /** @type {HTMLVideoElement} */ - (document.getElementById('media-' + c.id)); + (document.getElementById('media-' + c.localId)); if(media) { if(video) { throw new Error("Duplicate video"); @@ -1331,21 +1326,24 @@ async function setMedia(c, isUp, mirror, video) { media.autoplay = true; /** @ts-ignore */ media.playsinline = true; - media.id = 'media-' + c.id; + media.id = 'media-' + c.localId; div.appendChild(media); if(!video) addCustomControls(media, div, c); - if(mirror) - media.classList.add('mirror'); } + if(mirror) + media.classList.add('mirror'); + else + media.classList.remove('mirror'); + if(!video && media.srcObject !== c.stream) media.srcObject = c.stream; - let label = document.getElementById('label-' + c.id); + let label = document.getElementById('label-' + c.localId); if(!label) { label = document.createElement('div'); - label.id = 'label-' + c.id; + label.id = 'label-' + c.localId; label.classList.add('label'); div.appendChild(label); } @@ -1382,7 +1380,7 @@ function cloneHTMLElement(elt) { */ function addCustomControls(media, container, c) { media.controls = false; - let controls = document.getElementById('controls-' + c.id); + let controls = document.getElementById('controls-' + c.localId); if(controls) { console.warn('Attempted to add duplicate controls'); return; @@ -1391,7 +1389,7 @@ function addCustomControls(media, container, c) { let template = document.getElementById('videocontrols-template').firstElementChild; controls = cloneHTMLElement(template); - controls.id = 'controls-' + c.id; + controls.id = 'controls-' + c.localId; let volume = getVideoButton(controls, 'volume'); if(c.kind === 'local') { @@ -1508,16 +1506,16 @@ function registerControlHandlers(media, container) { } /** - * @param {string} id + * @param {string} localId */ -function delMedia(id) { +function delMedia(localId) { let mediadiv = document.getElementById('peers'); - let peer = document.getElementById('peer-' + id); + let peer = document.getElementById('peer-' + localId); if(!peer) throw new Error('Removing unknown media'); let media = /** @type{HTMLVideoElement} */ - (document.getElementById('media-' + id)); + (document.getElementById('media-' + localId)); media.srcObject = null; mediadiv.removeChild(peer); @@ -1534,7 +1532,7 @@ function setMediaStatus(c) { let state = c && c.pc && c.pc.iceConnectionState; let good = state === 'connected' || state === 'completed'; - let media = document.getElementById('media-' + c.id); + let media = document.getElementById('media-' + c.localId); if(!media) { console.warn('Setting status of unknown media.'); return; @@ -1560,7 +1558,7 @@ function setMediaStatus(c) { * @param {string} [fallback] */ function setLabel(c, fallback) { - let label = document.getElementById('label-' + c.id); + let label = document.getElementById('label-' + c.localId); if(!label) return; let l = c.username; diff --git a/static/protocol.js b/static/protocol.js index 5849c48..49d806a 100644 --- a/static/protocol.js +++ b/static/protocol.js @@ -36,15 +36,30 @@ function toHex(array) { return a.reduce((x, y) => x + hex(y), ''); } -/** randomid returns a random string of 32 hex digits (16 bytes). +/** + * newRandomId returns a random string of 32 hex digits (16 bytes). + * * @returns {string} */ -function randomid() { +function newRandomId() { let a = new Uint8Array(16); crypto.getRandomValues(a); return toHex(a); } +let localIdCounter = 0; + +/** + * newLocalId returns a string that is unique in this session. + * + * @returns {string} + */ +function newLocalId() { + let id = `${localIdCounter}` + localIdCounter++; + return id; +} + /** * ServerConnection encapsulates a websocket connection to the server and * all the associated streams. @@ -57,7 +72,7 @@ function ServerConnection() { * @type {string} * @const */ - this.id = randomid(); + this.id = newRandomId(); /** * The group that we have joined, or null if we haven't joined yet. * @@ -168,6 +183,7 @@ function ServerConnection() { * @property {string} type * @property {string} [kind] * @property {string} [id] + * @property {string} [replace] * @property {string} [source] * @property {string} [dest] * @property {string} [username] @@ -259,7 +275,7 @@ ServerConnection.prototype.connect = async function(url) { break; case 'offer': sc.gotOffer(m.id, m.labels, m.source, m.username, - m.sdp, m.kind === 'renegotiate'); + m.sdp, m.replace); break; case 'answer': sc.gotAnswer(m.id, m.sdp); @@ -391,25 +407,51 @@ ServerConnection.prototype.request = function(what) { }; /** - * newUpStream requests the creation of a new up stream. - * - * @param {string} [id] - The id of the stream to create. + * @param {string} localId * @returns {Stream} */ -ServerConnection.prototype.newUpStream = function(id) { - let sc = this; - if(!id) { - id = randomid(); - if(sc.up[id]) - throw new Error('Eek!'); + +ServerConnection.prototype.findByLocalId = function(localId) { + if(!localId) + return null; + + for(let id in serverConnection.up) { + let s = serverConnection.up[id]; + if(s.localId == localId) + return s; } + return null; +} + +/** + * newUpStream requests the creation of a new up stream. + * + * @param {string} [localId] + * - The local id of the stream to create. If a stream already exists with + * the same local id, it is replaced with the new stream. + * @returns {Stream} + */ +ServerConnection.prototype.newUpStream = function(localId) { + let sc = this; + let id = newRandomId(); + if(sc.up[id]) + throw new Error('Eek!'); + let pc = new RTCPeerConnection(sc.rtcConfiguration); if(!pc) throw new Error("Couldn't create peer connection"); - if(sc.up[id]) - sc.up[id].close(); - let c = new Stream(this, id, pc, true); + let oldId = null; + if(localId) { + let old = sc.findByLocalId(localId); + oldId = old && old.id; + if(old) + old.close(true); + } + + let c = new Stream(this, id, localId || newLocalId(), pc, true); + if(oldId) + c.replace = oldId; sc.up[id] = c; pc.onnegotiationneeded = async e => { @@ -518,26 +560,32 @@ ServerConnection.prototype.groupAction = function(kind, message) { * @param {string} source * @param {string} username * @param {string} sdp - * @param {boolean} renegotiate + * @param {string} replace * @function */ -ServerConnection.prototype.gotOffer = async function(id, labels, source, username, sdp, renegotiate) { +ServerConnection.prototype.gotOffer = async function(id, labels, source, username, sdp, replace) { let sc = this; - let c = sc.down[id]; - if(c && !renegotiate) { - // SDP is rather inflexible as to what can be renegotiated. - // Unless the server indicates that this is a renegotiation with - // all parameters unchanged, tear down the existing connection. - c.close(true); - c = null; - } - if(sc.up[id]) throw new Error('Duplicate connection id'); + let oldLocalId = null; + + if(replace) { + let old = sc.down[replace]; + if(old) { + oldLocalId = old.localId; + old.close(true); + } else + console.error("Replacing unknown stream"); + } + + let c = sc.down[id]; + if(c && oldLocalId) + console.error("Replacing duplicate stream"); + if(!c) { let pc = new RTCPeerConnection(sc.rtcConfiguration); - c = new Stream(this, id, pc, false); + c = new Stream(this, id, oldLocalId || newLocalId(), pc, false); sc.down[id] = c; c.pc.onicecandidate = function(e) { @@ -709,11 +757,12 @@ ServerConnection.prototype.gotRemoteIce = async function(id, candidate) { * * @param {ServerConnection} sc * @param {string} id + * @param {string} localId * @param {RTCPeerConnection} pc * * @constructor */ -function Stream(sc, id, pc, up) { +function Stream(sc, id, localId, pc, up) { /** * The associated ServerConnection. * @@ -728,6 +777,13 @@ function Stream(sc, id, pc, up) { * @const */ this.id = id; + /** + * The local id of this stream. + * + * @type {string} + * @const + */ + this.localId = localId; /** * Indicates whether the stream is in the client->server direction. * @@ -779,6 +835,12 @@ function Stream(sc, id, pc, up) { * @type {Object} */ this.labelsByMid = {}; + /** + * The id of the stream that we are currently replacing. + * + * @type {string} + */ + this.replace = null; /** * Indicates whether we have already sent a local description. * @@ -899,7 +961,7 @@ Stream.prototype.close = function(replace) { c.pc.close(); - if(c.up && c.localDescriptionSent) { + if(c.up && !replace && c.localDescriptionSent) { try { c.sc.send({ type: 'close', @@ -1034,10 +1096,12 @@ Stream.prototype.negotiate = async function (restartIce) { username: c.sc.username, kind: this.localDescriptionSent ? 'renegotiate' : '', id: c.id, + replace: this.replace, labels: c.labelsByMid, sdp: c.pc.localDescription.sdp, }); this.localDescriptionSent = true; + this.replace = null; c.flushLocalIceCandidates(); };