Add client-side RTC logic.

It tries to grab the microphone and open a
connection to every other user as soon as the
websocket connection is made. It shows audio
controls for every connection at the bottom
of the page for debugging purposes.

lib/www/client/source/src/lib/rtc.js

@@ -0,0 +1,194 @@
+
+let ws = null;
+let peerId = null;
+let peers = {};
+let stream = null;
+
+function init (socket) {
+  ws = socket;
+  ws.addEventListener("message", (ev) => {
+    try {
+      const payload = JSON.parse(ev.data);
+      if (payload.rtc === true) {
+        // Handle this message
+        handle (payload);
+      }
+    } catch (err) {
+      console.error("Invalid message", ev, err);
+    }
+  });
+}
+
+async function talk () {
+  try {
+    if (!stream) {
+      const constraints = { audio: true, video: false };
+      stream = await navigator.mediaDevices.getUserMedia(constraints);
+      console.log("Grabbed stream", stream);
+    }
+
+    if (peerId && Object.keys(peers).length) {
+      for (const track of stream.getTracks()) {
+        for (const peer in peers) {
+          peers[peer].addTrack(track, stream);
+        }
+      }
+    }
+  } catch (err) {
+    console.error("talk() error", err);
+  }
+}
+
+class PeerConnection {
+
+  constructor (otherPeerId) {
+    this.otherPeerId = otherPeerId;
+    this.makingOffer = false;
+    this.polite = this.otherPeerId > peerId;
+    this.start();
+  }
+
+  send (message) {
+    message.from = peerId;
+    message.to = this.otherPeerId;
+    send(message);
+  }
+
+  async handle (message) {
+    try {
+      let ignoreOffer = false;
+
+      if ("description" in message) {
+        const offerCollision = (message.description.type == "offer") &&
+          (this.makingOffer || this.conn.signalingState != "stable");
+
+        ignoreOffer = !this.polite && offerCollision;
+
+        if (ignoreOffer) {
+          return;
+        }
+
+        await this.conn.setRemoteDescription(message.description);
+        if (message.description.type == "offer") {
+          await this.conn.setLocalDescription();
+          this.send({ description: this.conn.localDescription });
+        }
+      }
+
+      if ("candidate" in message) {
+        try {
+          await this.conn.addIceCandidate(message.candidate);
+        } catch (err) {
+          if (!ignoreOffer) {
+            throw err;
+          }
+        }
+      }
+    } catch (err) {
+      console.error(err);
+    }
+  }
+
+  start () {
+    const config = { iceServers: [] };
+    this.conn = new RTCPeerConnection(config);
+    console.log("Have peer connection", this.conn);
+
+    this.conn.ontrack = ({track, streams}) => {
+      // FIXME Need to remove these elements when done
+      if (!this.remoteAudio) {
+        this.remoteAudio = document.createElement("audio");
+        this.remoteAudio.setAttribute("autoplay", "true");
+        this.remoteAudio.controls = true;
+        document.getElementsByTagName("footer")[0].appendChild(this.remoteAudio);
+        console.log("Added <audio> element", this.remoteAudio);
+      }
+
+      track.onunmute = () => {
+        if (this.remoteAudio.srcObject) {
+          return;
+        }
+        this.remoteAudio.srcObject = streams[0];
+        console.log("unmuted");
+      };
+    };
+
+    this.conn.onnegotiationneeded = async () => {
+      console.log("negotiation needed");
+      try {
+        this.makingOffer = true;
+        await this.conn.setLocalDescription();
+        this.send({ description: this.conn.localDescription });
+      } catch (err) {
+        console.error(err);
+      } finally {
+        this.makingOffer = false;
+      }
+    };
+
+    this.conn.oniceconnectionstatechange = () => {
+      console.log("state change");
+      if (this.conn.iceConnectionState == "failed") {
+        this.conn.restartIce();
+      }
+    }
+
+    this.conn.onicecandidate = async ({candidate}) => {
+      console.log("send candidate", candidate);
+      this.send({candidate});
+    };
+
+    if (stream) {
+      for (const track of stream.getAudioTracks()) {
+        this.addTrack(track, stream);
+      }
+    }
+
+  }
+
+  addTrack (track, stream) {
+    console.log("add track to connection", this.conn, track, stream);
+    if (this.conn) {
+      this.conn.addTrack(track, stream);
+    }
+  }
+
+
+};
+
+
+function send (message) {
+  console.log("Send message", message, "via", ws);
+  if (ws) {
+    message.rtc = true;
+    ws.send(JSON.stringify(message));
+  }
+}
+
+function handle (message) {
+  console.log("Handle message", message);
+
+  if ("peerId" in message) {
+    peerId = message.peerId;
+  }
+
+  if ("otherPeers" in message) {
+    for (const peer of message.otherPeers) {
+      if (peer in peers) continue;
+
+      peers[peer] = new PeerConnection(peer);
+    }
+  }
+
+  if ("newPeer" in message) {
+    peers[message.newPeer] = new PeerConnection(message.newPeer);
+  }
+
+  if ("to" in message && "from" in message && message.to == peerId) {
+    peers[message.from].handle(message);
+  }
+}
+
+export default {
+  init, talk
+};

lib/www/client/source/src/main.js

@@ -6,6 +6,8 @@ import vuetify from './plugins/vuetify'
 import vueDebounce from 'vue-debounce'
 import { mapMutations } from 'vuex';
 
+import rtc from './lib/rtc';
+
 
 Vue.config.productionTip = false
 
@@ -51,12 +53,18 @@ new Vue({
 
       this.ws.addEventListener("message", (ev) => {
         const msg = JSON.parse(ev.data);
-        this.setServerEvent(msg);
+        if (msg.rtc === true) {
+          // Handle WebRTC message
+        } else {
+          this.setServerEvent(msg);
+        }
       });
 
       this.ws.addEventListener("open", (ev) => {
         console.log("WebSocket connection open", ev);
         this.setServerConnectionState(true);
+        rtc.init(this.ws);
+        rtc.talk();
       });
 
       this.ws.addEventListener("close", (ev) => {

lib/www/server/ws/index.js

@@ -1,6 +1,7 @@
 const ws = require('ws');
 const URL = require('url');
 const db = require('./db');
+const ptt = require('./ptt');
 
 function start (server, pingInterval=30000) {
 	
@@ -16,7 +17,9 @@ function start (server, pingInterval=30000) {
 	wsServer.on('connection', socket => {
 		socket.alive = true;
 		socket.on('pong', function () { this.alive = true; })
-		socket.on('message', message => console.log(message));
+		socket.on('message', (message) => ptt.handler(socket, message));
+		socket.on('close', () => ptt.removeConnection(socket));
+		ptt.addConnection(socket);
 	});
 
 	server.on('upgrade', (request, socket, head) => {

lib/www/server/ws/ptt.js

@@ -0,0 +1,62 @@
+
+
+const connections = [];
+
+function broadcast (message, sender) {
+	message.rtc = true;
+	const body = JSON.stringify(message);
+	for (const socket of connections) {
+		if (socket != sender) {
+			socket.send(body);
+		}
+	}
+}
+
+function unicast (message, socket) {
+	message.rtc = true;
+	socket.send(JSON.stringify(message));
+}
+
+function addConnection (socket) {
+	if (!connections.includes(socket)) {
+		const peerId = Math.round(Math.random()*1000000000);
+		const otherPeers = connections.map(c => c.peerId);
+		broadcast({newPeer: peerId}, socket);
+		unicast({peerId, otherPeers}, socket);
+		socket.peerId = peerId;
+		connections.push(socket);
+	}
+}
+
+function removeConnection (socket) {
+	const pos = connections.indexOf(socket);
+
+	if (pos != -1) {
+		connections.splice(pos, 1);
+	}
+}
+
+function handler (socket, message) {
+	try {
+		const payload = JSON.parse(message);
+		if (payload.rtc === true) {
+			console.log("RTC Message", payload);
+			if ("to" in payload) {
+				const dest = connections.find(c => c.peerId == payload.to);
+				if (dest) {
+					unicast(payload, dest);
+				}
+			} else {
+				broadcast(payload, socket);
+			}
+		}
+	} catch (err) {
+		console.error("Invalid message", message, err);
+	}
+}
+
+module.exports = {
+	addConnection,
+	removeConnection,
+	handler
+};