rxjs audio

audio.js

import Rx from 'rx';

const eventNames = [
    'canplay', 'play', 'pause', 'ended', 'error',
    'timeupdate', 'progress', 'seeking', 'seeked', 'loadedMetaData'
  ];

export default class Audio {
  constructor() {
    this.subjects = {
      src: new Rx.ReplaySubject(1),
      playing: new Rx.BehaviorSubject(false),
      loading: new Rx.BehaviorSubject(false),
      ready: new Rx.AsyncSubject(),
      readyToLoad: new Rx.AsyncSubject(),
      load: new Rx.Subject(),
      playPause: new Rx.Subject(),
      play: new Rx.Subject(),
      pause: new Rx.Subject(),
    };
    this.interface = new Audio5js({
      swf_path: '/ext/audio5js.swf',
      ready: () => {
        this.subjects.ready.onNext();
        this.subjects.ready.onCompleted();
      }
    });
    this.events = {}
    eventNames.forEach((event) => {
      this.events[event] = Rx.Observable.fromEvent(this.interface, event);
    });
    this.events.error.subscribe((err) => console.log(err));

    this.pause().subscribe(() => {
      this.interface.pause();
    });

    this.play().subscribe(() => {
      this.interface.play();
    });

    this.load().subscribe((src) => {
      this.interface.load(src);
    });

    // Handle safari mobile by defering to first user interaction
    Rx.Observable.just().subscribe(this.subjects.readyToLoad);

    // get ground truth from interface
    Rx.Observable.merge(
      this.events.play.map(true),
      this.events.pause.map(false)
    ).subscribe(this.subjects.playing);

    Rx.Observable.merge(
      this.load().map(() => true),
      this.loadedOrError().map(() => false)
    ).subscribe(this.subjects.loading);

    let playPause$ = this.subjects.playPause.withLatestFrom(
      this.subjects.loading,
      this.subjects.playing,
      (playPause, loading, playing) => {
        if(loading && playing) {
          return Rx.Observable.empty();
        }
        else {
          if(playing) {
            return Rx.Observable.just('pause')
          }
          else {
            return Rx.Observable.just('play')
          }
        }
      }
    )
    .mergeAll();

    playPause$.filter((act) => act === 'play').subscribe(this.subjects.play);
    playPause$.filter((act) => act === 'pause').subscribe(this.subjects.pause);
  }
  // internal
  error() {
    return this.events.error.flatMap((err) => Rx.Observable.throw(err))
  }
  loadedOrError() {
    // first error or canplay event after load
    return this.load().flatMap(() => {
      return Rx.Observable.amb(
        this.error().catch(() => false).take(1),
        this.events.canplay.take(1)
      );
    });
  }
  load() {
    return Rx.Observable.combineLatest(
      this.subjects.src,
      this.subjects.readyToLoad,
      (src, ready) => src
    );
  }
  // input
  src() {
    return this.subjects.src;
  }
  pause() {
    return this.subjects.pause;
  }
  play() {
    return this.subjects.play;
  }
  playPause() {
    return this.subjects.playPause;
  }
  // output
  ended() {
    return this.events.ended;
  }
  loading() {
    return this.subjects.loading;
  }
  playing() {
    return this.subjects.playing;
  }
  state() {
    return Rx.Observable.combineLatest(
      this.loading(),
      this.playing(),
      (loading, playing) => ({loading, playing})
    );
  }
}