Using RxJS for WebRTC icecandidate handling via WebSocket signaling server

webrtc.js

// Create WebRTC peer and setup event handlers
let peer = new RTCPeerConnection(iceConfig)

// Subject for the websocket signalling server
let socketSubject = Observable.webSocket({
  // deserialize each binary message received
  resultSelector: e => deserialize(e.data)
})
// Filter for only icecandidate messages
.filter(msg => msg && msg.header && msg.header.event === 'icecandidate' && msg.body && msg.body.candidate)

// Observable for the local icecandidate stream
Observable.create(obs => {
  // This is called whenever the local peer has an icecandidate ready to be transmitted to the remote peer
  peer.onicecandidate = (e) => {
    // A null candidate means there are no more candidates
    if (!e.candidate) return obs.complete()
    obs.next(e.candidate.candidate)
  }
})
// Prepare message wrapper around candidate and serialize to send to remote peer
.map(candidate => serialize({ event: 'icecandidate' }, { candidate: 'a=' + candidate }))
.subscribe({
  // Send each local icecandidate message to the remote peer via the WebSocket Subject
  next: msg => socketSubject.next(msg)
})

// Subscribe to socket subject
let socketSubscription = socketSubject.subscribe({
  next: msg => {
    // Add icecandidate from remote peer to local peer
    peer.addIceCandidate(new RTCIceCandidate({
      candidate: msg.body.candidate.substr(2)
    }))
  }
})

// Triggered when remote peer creates a data channel
peer.ondatachannel = e => {
  socketSubscription.unsubscribe()
}