From 5dbe93878036086dcdc8e9730c4b9facbbd0cfd0 Mon Sep 17 00:00:00 2001 From: Noah Petherbridge Date: Thu, 26 Jan 2023 22:54:02 -0800 Subject: [PATCH] Initial WebRTC Code * WebRTC pees exchange local/remote descriptions ("offer" and "answer") * They don't seem to exchange ICE candidates yet * Some back and forth happens but the final WebRTC stream connection isn't established yet. --- pkg/handlers.go | 68 ++++++++++++-- pkg/messages.go | 22 ++++- pkg/pages.go | 17 ++-- pkg/websocket.go | 81 ++++++++++------- web/static/js/BareRTC.js | 189 +++++++++++++++++++++++++++++++++++++++ web/templates/chat.html | 3 +- 6 files changed, 333 insertions(+), 47 deletions(-) diff --git a/pkg/handlers.go b/pkg/handlers.go index 146cc18..66b86be 100644 --- a/pkg/handlers.go +++ b/pkg/handlers.go @@ -11,7 +11,7 @@ import ( func (s *Server) OnLogin(sub *Subscriber, msg Message) { // Ensure the username is unique, or rename it. var duplicate bool - for other := range s.IterSubscribers() { + for _, other := range s.IterSubscribers() { if other.ID != sub.ID && other.Username == msg.Username { duplicate = true break @@ -48,11 +48,7 @@ func (s *Server) OnLogin(sub *Subscriber, msg Message) { func (s *Server) OnMessage(sub *Subscriber, msg Message) { log.Info("[%s] %s", sub.Username, msg.Message) if sub.Username == "" { - sub.SendJSON(Message{ - Action: ActionMessage, - Username: "ChatServer", - Message: "You must log in first.", - }) + sub.ChatServer("You must log in first.") return } @@ -75,3 +71,63 @@ func (s *Server) OnMe(sub *Subscriber, msg Message) { // Sync the WhoList to everybody. s.SendWhoList() } + +// OnOpen is a client wanting to start WebRTC with another, e.g. to see their camera. +func (s *Server) OnOpen(sub *Subscriber, msg Message) { + // Look up the other subscriber. + other, err := s.GetSubscriber(msg.Username) + if err != nil { + sub.ChatServer(err.Error()) + return + } + + // Make up a WebRTC shared secret and send it to both of them. + secret := RandomString(16) + log.Info("WebRTC: %s opens %s with secret %s", sub.Username, other.Username, secret) + + // Ring the target of this request and give them the secret. + other.SendJSON(Message{ + Action: ActionRing, + Username: sub.Username, + OpenSecret: secret, + }) + + // To the caller, echo back the Open along with the secret. + sub.SendJSON(Message{ + Action: ActionOpen, + Username: other.Username, + OpenSecret: secret, + }) +} + +// OnCandidate handles WebRTC candidate signaling. +func (s *Server) OnCandidate(sub *Subscriber, msg Message) { + // Look up the other subscriber. + other, err := s.GetSubscriber(msg.Username) + if err != nil { + sub.ChatServer(err.Error()) + return + } + + other.SendJSON(Message{ + Action: ActionCandidate, + Username: sub.Username, + Candidate: msg.Candidate, + }) +} + +// OnSDP handles WebRTC sdp signaling. +func (s *Server) OnSDP(sub *Subscriber, msg Message) { + // Look up the other subscriber. + other, err := s.GetSubscriber(msg.Username) + if err != nil { + sub.ChatServer(err.Error()) + return + } + + other.SendJSON(Message{ + Action: ActionSDP, + Username: sub.Username, + Description: msg.Description, + }) +} diff --git a/pkg/messages.go b/pkg/messages.go index d290527..932e0ff 100644 --- a/pkg/messages.go +++ b/pkg/messages.go @@ -2,14 +2,21 @@ package barertc type Message struct { Action string `json:"action,omitempty"` - Username string `json:"username"` - Message string `json:"message"` + Username string `json:"username,omitempty"` + Message string `json:"message",omitempty` // WhoList for `who` actions - WhoList []WhoList `json:"whoList"` + WhoList []WhoList `json:"whoList,omitempty"` // Sent on `me` actions along with Username - VideoActive bool `json:"videoActive"` // user tells us their cam status + VideoActive bool `json:"videoActive,omitempty"` // user tells us their cam status + + // Sent on `open` actions along with the (other) Username. + OpenSecret string `json:"openSecret,omitempty"` + + // Parameters sent on WebRTC signaling messages. + Candidate string `json:"candidate,omitempty"` // candidate + Description string `json:"description,omitempty"` // sdp } const ( @@ -19,10 +26,17 @@ const ( // Actions sent by server or client ActionMessage = "message" // post a message to the room ActionMe = "me" // user self-info sent by FE or BE + ActionOpen = "open" // user wants to view a webcam (open WebRTC) + ActionRing = "ring" // receiver of a WebRTC open request // Actions sent by server only ActionWhoList = "who" // server pushes the Who List ActionPresence = "presence" // a user joined or left the room + ActionError = "error" // ChatServer errors + + // WebRTC signaling messages. + ActionCandidate = "candidate" + ActionSDP = "sdp" ) // WhoList is a member entry in the chat room. diff --git a/pkg/pages.go b/pkg/pages.go index 82a7c3a..bf7f240 100644 --- a/pkg/pages.go +++ b/pkg/pages.go @@ -16,12 +16,7 @@ func IndexPage() http.HandlerFunc { tmpl.Funcs(template.FuncMap{ // Cache busting random string for JS and CSS dependency. "CacheHash": func() string { - const charset = "abcdefghijklmnopqrstuvwxyz" - var result = make([]byte, 8) - for i := 0; i < 8; i++ { - result[i] = charset[rand.Intn(len(charset))] - } - return string(result) + return RandomString(8) }, }) tmpl, err := tmpl.ParseFiles("web/templates/chat.html") @@ -34,3 +29,13 @@ func IndexPage() http.HandlerFunc { tmpl.ExecuteTemplate(w, "index", nil) }) } + +// RandomString returns a random string of any length. +func RandomString(n int) string { + const charset = "abcdefghijklmnopqrstuvwxyz" + var result = make([]byte, n) + for i := 0; i < n; i++ { + result[i] = charset[rand.Intn(len(charset))] + } + return string(result) +} diff --git a/pkg/websocket.go b/pkg/websocket.go index ca64450..7b98afa 100644 --- a/pkg/websocket.go +++ b/pkg/websocket.go @@ -3,6 +3,7 @@ package barertc import ( "context" "encoding/json" + "errors" "fmt" "net/http" "time" @@ -61,12 +62,14 @@ func (sub *Subscriber) ReadLoop(s *Server) { s.OnMessage(sub, msg) case ActionMe: s.OnMe(sub, msg) + case ActionOpen: + s.OnOpen(sub, msg) + case ActionCandidate: + s.OnCandidate(sub, msg) + case ActionSDP: + s.OnSDP(sub, msg) default: - sub.SendJSON(Message{ - Action: ActionMessage, - Username: "ChatServer", - Message: "Unsupported message type.", - }) + sub.ChatServer("Unsupported message type.") } } }() @@ -91,6 +94,15 @@ func (sub *Subscriber) SendMe() { }) } +// ChatServer is a convenience function to deliver a ChatServer error to the client. +func (sub *Subscriber) ChatServer(message string, v ...interface{}) { + sub.SendJSON(Message{ + Action: ActionError, + Username: "ChatServer", + Message: fmt.Sprintf(message, v...), + }) +} + // WebSocket handles the /ws websocket connection. func (s *Server) WebSocket() http.HandlerFunc { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -152,6 +164,18 @@ func (s *Server) AddSubscriber(sub *Subscriber) { s.subscribersMu.Unlock() } +// GetSubscriber by username. +func (s *Server) GetSubscriber(username string) (*Subscriber, error) { + s.subscribersMu.RLock() + defer s.subscribersMu.RUnlock() + for _, sub := range s.IterSubscribers(true) { + if sub.Username == username { + return sub, nil + } + } + return nil, errors.New("not found") +} + // DeleteSubscriber removes a subscriber from the server. func (s *Server) DeleteSubscriber(sub *Subscriber) { log.Error("DeleteSubscriber: %s", sub.Username) @@ -162,31 +186,24 @@ func (s *Server) DeleteSubscriber(sub *Subscriber) { // IterSubscribers loops over the subscriber list with a read lock. If the // caller already holds a lock, pass the optional `true` parameter for isLocked. -func (s *Server) IterSubscribers(isLocked ...bool) chan *Subscriber { - var out = make(chan *Subscriber) - go func() { - log.Debug("IterSubscribers START..") +func (s *Server) IterSubscribers(isLocked ...bool) []*Subscriber { + log.Debug("IterSubscribers START..") - var result = []*Subscriber{} + var result = []*Subscriber{} - // Has the caller already taken the read lock or do we get it? - if locked := len(isLocked) > 0 && isLocked[0]; !locked { - log.Debug("Taking the lock") - s.subscribersMu.RLock() - defer s.subscribersMu.RUnlock() - } + // Has the caller already taken the read lock or do we get it? + if locked := len(isLocked) > 0 && isLocked[0]; !locked { + log.Debug("Taking the lock") + s.subscribersMu.RLock() + defer s.subscribersMu.RUnlock() + } - for sub := range s.subscribers { - result = append(result, sub) - } + for sub := range s.subscribers { + result = append(result, sub) + } - for _, r := range result { - out <- r - } - close(out) - log.Debug("IterSubscribers STOP!") - }() - return out + log.Debug("IterSubscribers STOP..") + return result } // Broadcast a message to the chat room. @@ -194,7 +211,7 @@ func (s *Server) Broadcast(msg Message) { log.Debug("Broadcast: %+v", msg) s.subscribersMu.RLock() defer s.subscribersMu.RUnlock() - for sub := range s.IterSubscribers(true) { + for _, sub := range s.IterSubscribers(true) { sub.SendJSON(Message{ Action: msg.Action, Username: msg.Username, @@ -205,15 +222,19 @@ func (s *Server) Broadcast(msg Message) { // SendWhoList broadcasts the connected members to everybody in the room. func (s *Server) SendWhoList() { - var users = []WhoList{} - for sub := range s.IterSubscribers() { + var ( + users = []WhoList{} + subscribers = s.IterSubscribers() + ) + + for _, sub := range subscribers { users = append(users, WhoList{ Username: sub.Username, VideoActive: sub.VideoActive, }) } - for sub := range s.IterSubscribers() { + for _, sub := range subscribers { sub.SendJSON(Message{ Action: ActionWhoList, WhoList: users, diff --git a/web/static/js/BareRTC.js b/web/static/js/BareRTC.js index 5eb7ad2..79bdafa 100644 --- a/web/static/js/BareRTC.js +++ b/web/static/js/BareRTC.js @@ -1,5 +1,13 @@ console.log("BareRTC!"); +// WebRTC configuration. +const configuration = { + iceServers: [{ + urls: 'stun:stun.l.google.com:19302' + }] +}; + + const app = Vue.createApp({ delimiters: ['[[', ']]'], data() { @@ -27,6 +35,15 @@ const app = Vue.createApp({ stream: null, // MediaStream object }, + // WebRTC sessions with other users. + WebRTC: { + // Streams per username. + streams: {}, + + // RTCPeerConnections per username. + pc: {}, + }, + // Chat history. history: [], historyScrollbox: null, @@ -55,6 +72,10 @@ const app = Vue.createApp({ this.dial(); }, + /** + * Chat API Methods (WebSocket packets sent/received) + */ + sendMessage() { if (!this.message) { return; @@ -87,11 +108,36 @@ const app = Vue.createApp({ // in our choice of username. if (this.username != msg.username) { this.ChatServer(`Your username has been changed to ${msg.username}.`); + this.username = msg.username; } this.ChatClient(`User sync from backend: ${JSON.stringify(msg)}`); }, + // Send a video request to access a user's camera. + sendOpen(username) { + this.ws.conn.send(JSON.stringify({ + action: "open", + username: username, + })); + }, + onOpen(msg) { + // Response for the opener to begin WebRTC connection. + const secret = msg.openSecret; + console.log("OPEN: connect to %s with secret %s", msg.username, secret); + this.ChatClient(`Connecting to stream for ${msg.username}.`); + + this.startWebRTC(msg.username, true); + }, + onRing(msg) { + // Message for the receiver to begin WebRTC connection. + const secret = msg.openSecret; + console.log("RING: connection from %s with secret %s", msg.username, secret); + this.ChatServer(`${msg.username} has opened your camera.`); + + this.startWebRTC(msg.username, false); + }, + // Handle messages sent in chat. onMessage(msg) { this.pushHistory({ @@ -102,6 +148,7 @@ const app = Vue.createApp({ // Dial the WebSocket connection. dial() { + console.log("Dialing WebSocket..."); const conn = new WebSocket(`ws://${location.host}/ws`); conn.addEventListener("close", ev => { @@ -152,6 +199,24 @@ const app = Vue.createApp({ message: msg.message, }); break; + case "ring": + this.onRing(msg); + break; + case "open": + this.onOpen(msg); + break; + case "candidate": + this.onCandidate(msg); + break; + case "sdp": + this.onSDP(msg); + break; + case "error": + this.pushHistory({ + username: msg.username || 'Internal Server Error', + message: msg.message, + isChatServer: true, + }); default: console.error("Unexpected action: %s", JSON.stringify(msg)); } @@ -160,6 +225,120 @@ const app = Vue.createApp({ this.ws.conn = conn; }, + /** + * WebRTC concerns. + */ + + // Start WebRTC with the other username. + startWebRTC(username, isOfferer) { + this.ChatClient(`Begin WebRTC with ${username}.`); + let pc = new RTCPeerConnection(configuration); + this.WebRTC.pc[username] = pc; + + // 'onicecandidate' notifies us whenever an ICE agent needs to deliver a + // message to the other peer through the signaling server. + pc.onicecandidate = event => { + this.ChatClient("On ICE Candidate called"); + console.log(event); + console.log(event.candidate); + if (event.candidate) { + this.ChatClient(`Send ICE candidate: ${event.candidate}`); + this.ws.conn.send(JSON.stringify({ + action: "candidate", + username: username, + candidate: event.candidate, + })); + } + }; + + // If the user is offerer let the 'negotiationneeded' event create the offer. + if (isOfferer) { + this.ChatClient("Sending offer:"); + pc.onnegotiationneeded = () => { + this.ChatClient("Negotiation Needed, creating WebRTC offer."); + pc.createOffer().then(this.localDescCreated(pc, username)).catch(this.ChatClient); + }; + } + + // When a remote stream arrives. + pc.ontrack = event => { + const stream = event.streams[0]; + + // Do we already have it? + this.ChatClient(`Received a video stream from ${username}.`); + if (this.WebRTC.streams[username] == undefined || + this.WebRTC.streams[username].id !== stream.id) { + this.WebRTC.streams[username] = stream; + } + }; + + // If we were already broadcasting video, send our stream to + // the connecting user. + if (!isOfferer && this.webcam.active) { + this.ChatClient(`Sharing our video stream to ${username}.`); + this.webcam.stream.getTracks().forEach(track => pc.addTrack(track, this.webcam.stream)); + } + + // If we are the offerer, begin the connection. + if (isOfferer) { + pc.createOffer().then(this.localDescCreated(pc, username)).catch(this.ChatClient); + } + }, + + // Common handler function for + localDescCreated(pc, username) { + return (desc) => { + this.ChatClient(`setLocalDescription ${JSON.stringify(desc)}`); + pc.setLocalDescription( + new RTCSessionDescription(desc), + () => { + this.ws.conn.send(JSON.stringify({ + action: "sdp", + username: username, + description: JSON.stringify(pc.localDescription), + })); + }, + console.error, + ) + }; + }, + + // Handle inbound WebRTC signaling messages proxied by the websocket. + onCandidate(msg) { + if (this.WebRTC.pc[msg.username] == undefined) return; + let pc = this.WebRTC.pc[msg.username]; + + // Add the new ICE candidate. + console.log("Add ICE candidate: %s", msg.candidate); + this.ChatClient(`Received an ICE candidate from ${username}: ${msg.candidate}`); + pc.addIceCandidate( + new RTCIceCandidate( + msg.candidate, + () => {}, + console.error, + ) + ); + }, + onSDP(msg) { + if (this.WebRTC.pc[msg.username] == undefined) return; + let pc = this.WebRTC.pc[msg.username]; + let description = JSON.parse(msg.description); + + // Add the new ICE candidate. + console.log("Set description: %s", description); + this.ChatClient(`Received a Remote Description from ${msg.username}: ${msg.description}.`); + pc.setRemoteDescription(new RTCSessionDescription(description), () => { + // When receiving an offer let's answer it. + if (pc.remoteDescription.type === 'offer') { + pc.createAnswer().then(this.localDescCreated(pc, msg.username)).catch(this.ChatClient); + } + }, console.error); + }, + + /** + * Front-end web app concerns. + */ + // Start broadcasting my webcam. startVideo() { if (this.webcam.busy) return; @@ -182,6 +361,16 @@ const app = Vue.createApp({ }) }, + // Begin connecting to someone else's webcam. + openVideo(user) { + if (user.username === this.username) { + this.ChatClient("You can already see your own webcam."); + return; + } + + this.sendOpen(user.username); + }, + // Stop broadcasting. stopVideo() { this.webcam.elem.srcObject = null; diff --git a/web/templates/chat.html b/web/templates/chat.html index 3083143..a0aaca6 100644 --- a/web/templates/chat.html +++ b/web/templates/chat.html @@ -223,7 +223,8 @@
[[ u.username ]]