Reactive WebRTC conference

Observables

Traditional JavaScript uses ad-hoc callbacks for all kinds of completion notifications. For example the following code initializes the application when all resources on a page have been loaded:

window.addEventListener('load', initApplication);

But that code contains a subtle race condition — if the document is loaded before the event listener has been registered then the application will never be initialized.

A work-around is to first check if the document is ready and if not add the listener:

if (document.readyState === 'complete') {
    initApplication();
} else {
    window.addEventListener('load', initApplication);
}

An alternative approach involves decoupling the potentially asynchronous event (loading the page) from application initialization and using Promises to represent the asynchronous event:

var documentReady = (function() {
    if (document.readyState === 'complete') {
        return Promise.resolve();
    }
    return new Promise(function(resolve) {
        window.addEventListener('load', resolve);
    });
}());

Then the clients of this API do not need to be aware of the document’s loading status:

function initApplication() {
    print('Init app!');
}

documentReady.then(initApplication);

Several DOM APIs that initially used callbacks have already started to adopt promises (e.g. Web Crypto, Web MIDI and WebRTC). This is an approach recommended to specification authors by the W3C TAG.

The fact that once a Promise is resolved its status and value never changes makes Promises good for representing events that happen only once — loading an image or a remote resource. But Promise is inappropriate for events that happen multiple times — like mouse movements or messages from a remote server. One of the solutions to handling multiple values in an asynchronous way is the Observable.

There is a similarity between synchronous concepts of a regular function returning a single value and a generator returning multiple values and the asynchronous concepts of promises that resolve to a single value and observables that model asynchronous streams of multiple values.

function print(msg) {
    var special = false;
    if (typeof msg === 'undefined') {
        msg = 'undefined';
        special = true;
    }
    if (msg === null) {
        msg = 'null';
        special = true;
    }
    var li = document.createElement('LI');
    li.textContent = msg;
    li.className = special ? 'primitive' : '';
    output.appendChild(li);
}

function clear() {
    output.innerHTML = '';
}
clear();

A simple value is singular and synchronous:

var value = 17;

print(value);

Promise represents a value that is returned asynchronously. Once a promise is resolved the value is never changed:

var value = new Promise(function(resolve) {
    setTimeout(function() {
        resolve(42);
    }, 1000);
});

value.then(print);

An iterable (e.g. an array or a generator) represents multiple synchronous values:

var source = [ { clientX: 10 }, { clientX: 20 } ];

source.map(function(e) { return e.clientX; })
	.filter(function(x) { return x % 2 === 0; })
	.forEach(print);

Observables deliver multiple values asynchronously.

The example below captures all mouse movements within two seconds and prints their X coordinates:

var source = Observable.fromEvent(document, 'mousemove')
	.takeUntil(Observable.timeout(2000));

source.map(function(e) { return e.clientX; })
    .filter(function(x) { return x % 2 === 0; })
    .forEach(print);

An observable has a similar interface to an array but the party in control is different. In arrays and iterables the consumer pulls values out of the iterable. Observables on the other hand push values into all registered subscribers.

Signaling

WebRTC allows creating direct connections between peers but to setup these connections an intermediary is needed — the signaling channel.

The signaling channel is used to exchange metadata about media or data channels (session descriptions) and network connection (ICE candidates). The exact method for signaling is not mandated by the WebRTC specification and can be anything (e.g. XMPP, WebSockets, SIP, …).

When a direct connection between peers has been estabilished the signaling channel is no longer needed.

This signaling channel in this article uses WebSockets and exposes messages property that emits messages received by the WebSocket.

Sending data to a WebSocket connection can be achieved using the next function.

function WebSocketSignalingChannel(address) {
    this.socket = new WebSocket(address);
    var socketObservable = Observable.fromWebSocket(this.socket);
    this.messages = socketObservable.map(function(event) {
        return JSON.parse(event.data);
    });
}

WebSocketSignalingChannel.prototype.next = function(object) {
    this.socket.send(JSON.stringify(object));
};

To allow multiple conversations at once each conference is assigned an identifier. Room identifier is either taken from the current page’s query string or generated as a random hex-encoded 10 byte sequence using ES6 Uint8Array.prototype.map function.

function getRoom() {
    var room = location.search.substring(1);
    if (room.length !== 20) {
        room = Array.prototype.map.call(crypto.getRandomValues(new Uint8Array(10)), function(b) {
            return b.toString(16);
        }).map(function pad(b) {
            return ('00' + b).slice(-2);
        }).join('');
        history.replaceState(null, '', '?' + room + '#conference');
    }
    return room;
}

Initialization of the WebSocket signaling channel uses current scheme and host.

function getSignalingChannel(room) {
    var url = location.protocol.replace('http', 'ws') + '//' + location.host + '/rooms/' + room;
    return new WebSocketSignalingChannel(url);
}

Session description exchange

Signaling is used to transfer session descriptions and ICE candidates.

There are two kinds of session descriptions exchanged by peers:

• offer is created by the initiator,
• answer is created by the second party and sent back to the initiator.

Additionally each peer has two slots for session descriptions: local description for the description of itself and remote description for the description of the other peer.

RTCPeer wrapper object handles incoming messages acting as both peers depending on the message received. The join message is emitted by the signaling channel when another peer joins the channel. The second peer never sees the join message of the first peer but it is instead presented with an offer.

RTCPeer.prototype.handleMessage = function(body) {
    var peer = this.peer, type = body.type;

    if (type === 'join') {
        // peer 1
        return peer.createOffer().then(this.setLocal);
    } else if (type === 'offer') {
        // peer 2
        peer.setRemoteDescription(new RTCSessionDescription(body));
        return peer.createAnswer().then(this.setLocal);
    } else if (type === 'answer') {
        // peer 1
        peer.setRemoteDescription(new RTCSessionDescription(body));
    } else if (type === 'candidate') {
        // both peers
        peer.addIceCandidate(new RTCIceCandidate(body));
    }

    return Promise.resolve({ target: self });
};

The following function returns a promise resolving to an answer that needs to be sent back to another peer. Replies are created by mapping incoming messages into responses.

function getReplies(messages, peers) {
    return messages.map(function(message) {
        return peers.get(message.from).handleMessage(message.body);
    }).unwrapPromises().filter(function(reply) {
        return reply.response;
    }).map(function(reply) {
        return {
            to: reply.target.id,
            body: reply.response
        };
    });
}

This function accepts two parameters:

1. messages — observable emitting commands (a signaling channel),
2. peers — lazy map that for each key retrieves an RTCPeer object creating new one if it is not already created.

The unwrapPromises function turns a stream of promises into a stream of values that these promises resolve to.

ICE and streams

ICE allows both peers to find the best communication path between them even if both of them are behind NAT. Each possible way to connect peers is known as an ICE candidate. To generate candidates two kinds of servers are used:

• STUN — to determine the public IP and detect if peers are behind a NAT,
• TURN — acts as a proxy or a relay server if no direct connection can be made.

The implementation uses simple object wrapper around WebRTC RTCPeerConnections which adds two observable properties:

• streams represent remote streams that have been added to the peer,
• iceCandidates a list of ICE candidates used to connect two peers together.

function RTCPeer(id, peer) {
    this.id = id;
    this.peer = peer;

    var iceEvents = Observable.fromEvent(peer, 'icecandidate');
    this.iceCandidates = iceEvents.filter(function(event) {
        return event.candidate;
    }).map(function(event) {
        return {
            target: this,
            candidate: event.candidate
        };
    }, this);

    var addStreamEvents = Observable.fromEvent(peer, 'addstream');
    this.streams = addStreamEvents.map(function(event) {
        return {
            target: this,
            stream: event.stream
        };
    }, this);

    this.setLocal = this.setLocal.bind(this);
}

RTCPeer.prototype.setLocal = function(description) {
    return this.peer.setLocalDescription(description).then(function() {
        return { target: this, response: description };
    }.bind(this));
};

An observable that emits RTCPeer objects is converted into a stream of ICE candidates. Because mapping peers into their ICE candiates produces a nested observable (stream of streams of candidates) a merge function is used to flatten the structure.

function getCandidates(objects) {
    return objects.mergeMap(function(peer) {
        return peer.iceCandidates;
    }).map(function(event) {
        var candidate = JSON.parse(JSON.stringify(event.candidate));
        candidate.type = 'candidate';
        return {
            to: event.target.id,
            body: candidate
        };
    });
}

Remote streams are obtained by mapping created RTCPeer objects into streams.

function getRemoteStreams(objects) {
    return objects.mergeMap(function(peer) {
        return peer.streams;
    }).map(function(event) {
        return event.stream;
    });
}

Peers

Peers are stored using a small utility object — the LazyMap. A lazy map is a map that produces values for keys on demand. This map additionally provides an observable that emits produced objects.

function LazyMap(create) {
    var map = new Map, generators = [];
    this.get = function(key) {
        if (map.has(key)) {
            return map.get(key);
        }
        var value = create(key);
        map.set(key, value);
        generators.forEach(function(generator) {
            generator.next(value);
        });
        return value;
    };
    this.objects = new Observable(function observe(generator) {
        generators.push(generator)
        return generator;
    });
}

The function below creates RTCPeer objects with peer connections on demand with specified local stream. The configuration object contains addresses for STUN and TURN servers.

function createMap(stream) {
    var configuration = {
        iceServers: [{
            url: 'stun:stun.l.google.com:19302'
        }, {
            url: 'stun:stun.services.mozilla.com'
        }, {
            url: 'turn:turn.anyfirewall.com:443?transport=tcp',
            credential: 'webrtc',
            username: 'webrtc'
        }]
    };
    return new LazyMap(function(id) {
        var peer = new RTCPeerConnection(configuration);
        peer.addStream(stream);
        return new RTCPeer(id, peer);
    });
}

Conference

if (!Uint8Array.prototype.map) {
    Uint8Array.prototype.map = Array.prototype.map;
}

document.createElement = document.ownerDocument.createElement.bind(document.ownerDocument);

document.createElement = document.ownerDocument.createElement.bind(document.ownerDocument);

Media to be shared are selected through a simple form. Each radio button specifies constraints to be used for selecting media source either as:

• data-source attribute in JSON format,
• value property that points to a global function that returns a promise with constraints.

<input type="radio" name="source"
    data-source='{"audio":true,"video":false}'> Audio only

<input type="radio" name="source" checked
    data-source='{"audio":false,"video":true}'> Video only

<input type="radio" name="source"
    data-source='{"audio":true,"video":true}'> Audio + Video

<input type="radio" name="source"
    value="sourceChromeScreen"> Screensharing in Chrome

<input type="radio" name="source"
    data-source='{"audio":false,"video":{"mediaSource":"screen"}}'> Screensharing in Firefox

<input type="radio" name="source"
    data-source='{"audio":false,"video":true,"fake":true}'> Fake media in Firefox

<input type="button" value="Connect">

Selected radio is turned into a promise that resolves to constraints.

var form = document.querySelector('form');
form.hidden = true;

var source = Array.from(form.source).find(function(source) {
    return source.checked;
});

var constraints;

if (source.dataset.source) {
    constraints = Promise.resolve(JSON.parse(source.dataset.source));
} else {
    constraints = window[source.value]();
}

Constraints are passed to getUserMedia that exchanges them for the MediaStream object and invokes connect.

var mediaDevices = navigator.mediaDevices;
var getUserMedia = mediaDevices.getUserMedia.bind(mediaDevices);

constraints.then(getUserMedia).then(connect);

The function that joins all the pieces — connect passes messages and streams from created objects into ui. At the end two observables — one for candidates and one for message replies are merged together and sent into the signaling channel.

function connect(stream) {
    var peers = createMap(stream);

    var signalingChannel = getSignalingChannel(getRoom());

    // pass events to UI object
    var ui = withUI(UI(document));
    ui.addLocalStream(stream);
    signalingChannel.messages.forEach(ui.updateStatus);
    getRemoteStreams(peers.objects).forEach(ui.addRemoteStream);

    // get responses
    var candidates = getCandidates(peers.objects);
    var replies = getReplies(signalingChannel.messages, peers);

    candidates.merge(replies).observe(signalingChannel);
}

Serverless mode

By substituting WebSocket singaling channel object with a different object that has the same interface it is possible to build a completely serverless WebRTC conference. Instead WebSocket messages users need to exchange the the communication data manually (for example via IM or e-mail).

The following object uses two text fields for outgoing (send) and incoming (receive) messages and two buttons — one to interpret received data and one to create offer and initiate exchange.

function StaticSignalingChannel(send, receive, receiveButton, offerButton) {
    this.objects = [];
    this.sendInput = send;
    this.messages = new Observable(function(generator) {
        var done = false,
            decoratedGenerator =
            decorate(generator, function() {
                done = true
            });
        receiveButton.addEventListener('click', function(e) {
            e.preventDefault();
            var messages = JSON.parse(receive.value);
            messages.forEach(function(message) {
                if (done) return;
                decoratedGenerator.next(message);
            }, this);
        }.bind(this));
        receiveButton.disabled = false;
        offerButton.addEventListener('click', function(e) {
            e.preventDefault();
            decoratedGenerator.next({
                body: {
                    type: 'join'
                }
            });
        }.bind(this));
        offerButton.disabled = false;

        return decoratedGenerator;
    });
    this.next = function(message) {
        this.objects.push(message);
        send.value = JSON.stringify(this.objects);
    };
    this.throw = console.error.bind(console);
}

Initialization of this object does not require the room identifier.

function getSignalingChannel() {
    var $ = document.querySelector.bind(document);
    return new StaticSignalingChannel(
        $('#sendInput'), $('#receiveInput'),
        $('#receiveBtn'), $('#offerBtn'));
}