queue.js

"use strict"

import MessagePack from "what-the-pack"
import pako from "pako"
import uuidv4 from "uuid/v4"

const UintBuffer = MessagePack.Buffer || Buffer

const MESSAGE_BUFFER_SIZE = 2 ** 25 // 32MB
const MESSAGE_SLICE_SIZE = 2 ** 10 // 1KB

const { encode, decode } = MessagePack.initialize(MESSAGE_BUFFER_SIZE)

const buffer = {}

onmessage = (event) => {
  if (!event || !event.data) {
    return
  }

  if (event.data.method == "send" || event.data.method == "broadcast") {
    let message = event.data.message
    const compressed = !(
      message == undefined ||
      message instanceof String ||
      typeof message == "string" ||
      message instanceof Uint8Array ||
      message.message instanceof Uint8Array ||
      message.message == undefined
    )
    const uuid = uuidv4()

    //console.log("send in", message)

    message = encode(message)

    if (compressed) {
      message = pako.deflate(message)
    }

    for (
      let offset = 0;
      offset < message.length;
      offset += MESSAGE_SLICE_SIZE
    ) {
      event.data.message = {
        uuid,
        message: message.subarray(offset, offset + MESSAGE_SLICE_SIZE),
        slice: offset / MESSAGE_SLICE_SIZE,
        length: Math.ceil(message.length / MESSAGE_SLICE_SIZE),
        compressed,
      }

      //console.log("send out", event.data)

      self.postMessage(event.data)
    }
  } else if (event.data.method == "received") {
    let message = event.data.message.message

    //console.log("receive in", message)

    if (event.data.message.length > 1) {
      let messages = buffer[event.data.uid]
      if (!messages) {
        messages = {}
        buffer[event.data.uid] = messages
      }

      let slices = messages[event.data.message.uuid]
      if (!slices) {
        slices = []
        messages[event.data.message.uuid] = slices
      }

      slices.push(event.data.message)

      if (slices.length < slices[slices.length - 1].length) {
        return
      }

      message = new Uint8Array(
        slices.reduce((acc, s) => acc + s.message.length, 0),
      )

      slices.sort((a, b) => a.slice - b.slice)

      let offset = 0

      for (const slice of slices) {
        message.set(slice.message, offset)
        offset += slice.message.length
      }

      delete messages[event.data.message.uuid]
    }

    if (event.data.message.compressed) {
      message = pako.inflate(Uint8Array.from(message))
    }

    if (!UintBuffer.isBuffer(message)) {
      message = UintBuffer.from(message)
    }

    message = decode(message)

    event.data.message = message

    //console.log("receive out", event.data)

    self.postMessage(event.data)
  }
}