mse-rx.js

import Rx from 'rx'

const audioUrls = [
  '/assets/cmajseginit.mp4',
  '/assets/cmajseg1.m4s',
  '/assets/cmajseg2.m4s',
  '/assets/cmajseg3.m4s',
]

/**
 * An audio source observable that fetches and emits audio buffers
 */
function getSourceStream(urls) {
  return Rx.Observable.create(observer => {

    /**
     * Recursively fetches and appends audio buffers to a given source
     * buffer.
     */
    function fetchNextChunk(chunkUrls) {
      fetchArrayBuffer(chunkUrls[0]).then(buffer => {

        // Emit the buffer
        observer.onNext(buffer)

        if (chunkUrls.length >= 2) {
          fetchNextChunk(chunkUrls.slice(1))
        }
        else {

          // Emit the end event
          observer.onCompleted()
        }
      })
    }

    fetchNextChunk(urls)
  })
}

function createDownsampler(takeEvery = 1000) {
  return function downsample(buffer) {
    const samples = new Uint8Array(buffer)

    const numSamples = Math.ceil(samples.length / takeEvery)
    const downsampledBuffer = new Uint8Array(numSamples)

    for (let i = 0; i < numSamples; i++) {
      downsampledBuffer[i] = samples[i * takeEvery]
    }

    return downsampledBuffer
  }
}

/**
 * Returns an array buffer with whatever is at the given URL.
 */
function fetchArrayBuffer(url) {
  return new Promise((resolve) => {
    const xhr = new XMLHttpRequest()
    xhr.open('get', url)
    xhr.responseType = 'arraybuffer'
    xhr.addEventListener('readystatechange', () => {
      if (xhr.readyState === 4) {
        resolve(xhr.response)
      }
    })
    xhr.send()
  })
}

function run() {

  // Create a stream that emits buffer data
  //
  // Note: The $ suffix means it is a stream
  const source$ = getSourceStream(audioUrls)

  // Create a new stream that downsamples the source audio buffers
  const downsampledSource$ = source$
    .skip(1) // Skip init segment
    .map(createDownsampler())

  downsampledSource$.subscribe(
    downsampledChunk => console.log('new downsampled chunk', downsampledChunk),
    err => console.error(err),
    () => console.log('we are done')
  )
}