index.js

/* global Y */
"use strict"

import LioWebRTC from "liowebrtc"

function extend(Y) {
  class WebRTC extends Y.AbstractConnector {
    constructor(y, options) {
      if (options === undefined) {
        throw new Error("Options must not be undefined!")
      }

      options.role = "slave"
      super(y, options)
      this.webrtcOptions = options

      this.queue = new Worker("js/queue.js")
      this.queue.onmessage = (event) => {
        const method = event.data.method

        if (method == "send") {
          const { uid, channel, message } = event.data

          // y-js db transactions can send messages after a peer has disconnected
          if (
            (channel == "y-js" && !this.peers.has(uid)) ||
            !this.webrtc.getPeerById(uid)
          ) {
            return
          }

          this.webrtc.whisper(this.webrtc.getPeerById(uid), channel, message)
        } else if (method == "broadcast") {
          const { channel, message } = event.data

          return this.webrtc.shout(channel, message)
        } else if (method == "received") {
          const { type, message, peer } = event.data

          if (type === "tw-ml") {
            // Handshakes can only be sent and received directly
            if (message === "tw") {
              // Response message in the handshake
              this.queue.postMessage({
                method: "send",
                uid: peer.id,
                channel: "tw-ml",
                message: "ml",
              })
            } else if (message == "ml") {
              // Handshake completed
              this.checkAndInsertPeer(peer.id)
            }
          } else {
            this.checkAndInsertPeer(peer.id)

            const health = this.peers.get(peer.id)
            health.heartbeat = Date.now()
            health.failures = -1

            if (type === "y-js") {
              this.checkAndInsertPeer(peer.id)
              this.receiveMessage(peer.id, message)
            }
          }
        }
      }

      if (options.onUserEvent) {
        this.onUserEvent(options.onUserEvent)
      }
      this.initialiseConnection()

      window.addEventListener("unload", () => {
        this.y.destroy()
      })
    }

    initialiseConnection() {
      this.webrtc = new LioWebRTC({
        url: this.webrtcOptions.url,
        dataOnly: true,
        /*network: {
          minPeers: 4,
          maxPeers: 8,
        },*/
      })

      this.peers = new Map()

      this.webrtc.on("ready", () => {
        this.webrtc.joinRoom(this.webrtcOptions.room)
      })

      this.webrtc.on("joinedRoom", () => {
        this.checkAndEnsureUser()
      })

      this.webrtc.on("leftRoom", () => {
        console.log("TODO: LEFT ROOM")
      })

      this.webrtc.on("channelError", (a, b, c, d) => console.log(a, b, c, d))

      this.webrtc.on("channelOpen", (dataChannel, peer) => {
        this.checkAndEnsureUser()

        // Start a handshake to ensure both sides are able to use the channel
        function handshake(peer) {
          const _peer = this.webrtc.getPeerById(peer.id)

          if (!_peer || _peer !== peer) {
            return
          }

          if (this.peers.has(peer.id)) {
            return
          }

          // Initial message in the handshake
          this.queue.postMessage({
            method: "send",
            uid: peer.id,
            channel: "tw-ml",
            message: "tw",
          })

          setTimeout(handshake.bind(this, peer), 500)
        }

        setTimeout(handshake.bind(this, peer), 100)
      })

      this.webrtc.on("receivedPeerData", (type, message, peer) => {
        this.checkAndEnsureUser()

        // Message could have been forwarded but yjs only needs to know about directly connected peers
        this.queue.postMessage({
          method: "received",
          type,
          message,
          peer: { id: peer.forwardedBy ? peer.forwardedBy.id : peer.id },
        })
      })

      this.webrtc.on("channelClose", (dataChannel, peer) => {
        this.checkAndEnsureUser()
        this.checkAndRemovePeer(peer.id)
      })
    }

    // Ensure that y-js is up to date on the user's id
    checkAndEnsureUser() {
      const id = this.webrtc.getMyId()

      if (this.y.db.userId === id) {
        return
      }

      for (const f of this.userEventListeners) {
        f({ action: "userID", id: id })
      }

      this.setUserId(id)
    }

    // Ensure that y-js knows that the peer has joined
    checkAndInsertPeer(uid) {
      if (this.peers.has(uid)) {
        return
      }

      const health = {
        heartbeat: Date.now(),
        failures: 0,
      }
      //health.cb = setInterval(this.heartbeat.bind(this, this.webrtc.getPeerById(uid), health), 500);

      this.peers.set(uid, health)

      this.userJoined(uid, "master")
    }

    heartbeat(peer, health) {
      const _peer = this.webrtc.getPeerById(peer.id)

      if (!_peer || _peer !== peer || !this.peers.has(peer.id)) {
        clearInterval(health.cb)

        return
      }

      if (health.failures++ / 2 >= 5) {
        clearInterval(health.cb)

        return peer.end(true)
      } else if (health.failures > 0) {
        console.log(`Missed heartbeat ${health.failures} from ${peer.id}`)
      }

      this.queue.postMessage({
        method: "send",
        uid: peer.id,
        channel: "heartbeat",
      })
    }

    // Ensure that y-js knows that the peer has left
    checkAndRemovePeer(uid) {
      if (!this.peers.has(uid)) {
        return
      }

      this.peers.delete(uid)

      this.userLeft(uid)
    }

    connectToPeer(/*uid*/) {
      // currently deprecated
    }

    disconnect() {
      this.queue.terminate()

      this.webrtc.quit()

      super.disconnect()
    }

    reconnect() {
      this.initialiseConnection()

      super.reconnect()
    }

    send(uid, message) {
      this.queue.postMessage({ method: "send", channel: "y-js", uid, message })
    }

    broadcast(message) {
      this.queue.postMessage({ method: "broadcast", channel: "y-js", message })
    }

    isDisconnected() {
      return false
    }
  }

  Y.extend("webrtc", WebRTC)
}

export default extend
if (typeof Y !== "undefined") {
  extend(Y)
}