src/streamer/streamer.js

/**
 * @file Streamer
 * @author Alexander Rose <alexander.rose@weirdbyte.de>
 * @private
 */

import { DecompressorRegistry } from '../globals.js'
import { uint8ToString, defaults } from '../utils.js'

class Streamer {
  constructor (src, params) {
    const p = params || {}

    this.compressed = defaults(p.compressed, false)
    this.binary = defaults(p.binary, false)
    this.json = defaults(p.json, false)
    this.xml = defaults(p.xml, false)

    this.src = src
    this.chunkSize = 1024 * 1024 * 10
    this.newline = '\n'

    this.__pointer = 0
    this.__partialLine = ''

    if (this.__srcName) {
      this[ this.__srcName ] = src
    }
  }

  get type () { return '' }

  get __srcName () { return undefined }

  isBinary () {
    return this.binary || this.compressed
  }

  onload () {}

  onprogress () {}

  onerror () {}

  read () {
    return new Promise((resolve, reject) => {
      try {
        this._read(data => {
          const decompressFn = DecompressorRegistry.get(this.compressed)

          if (this.compressed && decompressFn) {
            this.data = decompressFn(data)
          } else {
            if ((this.binary || this.compressed) && data instanceof ArrayBuffer) {
              data = new Uint8Array(data)
            }
            this.data = data
          }

          resolve(this.data)
        })
      } catch (e) {
        reject(e)
      }
    })
  }

  _read (callback) {
    // overwrite this method when this.src does not contain the data
    callback(this.src)
  }

  _chunk (start, end) {
    end = Math.min(this.data.length, end)

    if (start === 0 && this.data.length === end) {
      return this.data
    } else {
      if (this.isBinary()) {
        return this.data.subarray(start, end)
      } else {
        return this.data.substring(start, end)
      }
    }
  }

  chunk (start) {
    const end = start + this.chunkSize

    return this._chunk(start, end)
  }

  peekLines (m) {
    const data = this.data
    const n = data.length

    // FIXME does not work for multi-char newline
    const newline = this.isBinary() ? this.newline.charCodeAt(0) : this.newline

    let i
    let count = 0
    for (i = 0; i < n; ++i) {
      if (data[ i ] === newline) ++count
      if (count === m) break
    }

    const chunk = this._chunk(0, i + 1)
    const d = this.chunkToLines(chunk, '', i > n)

    return d.lines
  }

  chunkCount () {
    return Math.floor(this.data.length / this.chunkSize) + 1
  }

  asText () {
    return this.isBinary() ? uint8ToString(this.data) : this.data
  }

  chunkToLines (chunk, partialLine, isLast) {
    const newline = this.newline

    if (!this.isBinary() && chunk.length === this.data.length) {
      return {
        lines: chunk.split(newline),
        partialLine: ''
      }
    }

    let lines = []
    const str = this.isBinary() ? uint8ToString(chunk) : chunk
    const idx = str.lastIndexOf(newline)

    if (idx === -1) {
      partialLine += str
    } else {
      const str2 = partialLine + str.substr(0, idx)
      lines = lines.concat(str2.split(newline))

      if (idx === str.length - newline.length) {
        partialLine = ''
      } else {
        partialLine = str.substr(idx + newline.length)
      }
    }

    if (isLast && partialLine !== '') {
      lines.push(partialLine)
    }

    return {
      lines: lines,
      partialLine: partialLine
    }
  }

  nextChunk () {
    const start = this.__pointer

    if (start > this.data.length) {
      return undefined
    }

    this.__pointer += this.chunkSize
    return this.chunk(start)
  }

  nextChunkOfLines () {
    const chunk = this.nextChunk()

    if (chunk === undefined) {
      return undefined
    }

    const isLast = this.__pointer > this.data.length
    const d = this.chunkToLines(chunk, this.__partialLine, isLast)

    this.__partialLine = d.partialLine

    return d.lines
  }

  eachChunk (callback) {
    const chunkSize = this.chunkSize
    const n = this.data.length
    const chunkCount = this.chunkCount()

    for (let i = 0; i < n; i += chunkSize) {
      const chunk = this.chunk(i)
      const chunkNo = Math.round(i / chunkSize)

      callback(chunk, chunkNo, chunkCount)
    }
  }

  eachChunkOfLines (callback) {
    this.eachChunk((chunk, chunkNo, chunkCount) => {
      const isLast = chunkNo === chunkCount + 1
      const d = this.chunkToLines(chunk, this.__partialLine, isLast)

      this.__partialLine = d.partialLine

      callback(d.lines, chunkNo, chunkCount)
    })
  }

  dispose () {
    delete this.src

    if (this.__srcName) {
      delete this[ this.__srcName ]
    }
  }
}

export default Streamer