RxJS vs vanilla comparison

<!DOCTYPE html>
<html>

  <head>
    <script data-require="rxjs@5.0.1" data-semver="5.0.1" src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.1/Rx.min.js"></script>
    <script data-require="axios@0.15.3" data-semver="0.15.3" src="https://cdnjs.cloudflare.com/ajax/libs/axios/0.15.3/axios.min.js"></script>
    <script src="script.js"></script>
  </head>

  <body>
    Open the console and go to script.js
  </body>

</html>

console.clear();
window.onload = init;



// Vanilla version
function createOnlineEmitter() {
  let cbs = []; //array of registered callbacks for the event
  let unsub; //function for removing the main event listener

  //this is the main event listener that gets registered with window.online/offline event
  const mainListener = (isOnline) => {
    //call all the subscribed callbacks
    cbs.forEach(cb => cb(isOnline));
  };

  const registerMainListener = () => {
    const boundOnline = mainListener.bind(null, true);
    const boundOffline = mainListener.bind(null, false);
    window.addEventListener('online', boundOnline);
    window.addEventListener('offline', boundOffline);
    //return unsubcribe functionality in a closure
    return function unsubscribe() {
      window.removeEventListener('online', boundOnline);
      window.removeEventListener('offline', boundOffline);
    };
  };

  const addCb = (cb) => {
    cbs.push(cb);
    //register main listener only once
    //use existence of `unsub` as indicator if main event listener is added or not
    if(!unsub) {
      unsub = registerMainListener();
    }
  };
  
  const removeCb = (cb) => {
    const index = cbs.indexOf(cb);
    if(index > -1) {
      cbs.splice(index, 1);
    }
    //if no callbacks left, remove main event listener
    if(cbs.length === 0 && unsub) {
      unsub();
      unsub = null;
    }
  };

  return function initOnlineEmitter(cb) {
    addCb(cb);
    //call it async with the initial val
    setTimeout(() => {
      cb(navigator.onLine);
    });
    //return unsubscribe function to caller
    return removeCb.bind(null, cb);
  };
}


// RxJS version
function createOnline$() {
  //merge several events into one
  return Rx.Observable.merge(
    //use .map() to transform the returned Event type into a true/false value
  	Rx.Observable.fromEvent(window, 'offline').map(() => false),
  	Rx.Observable.fromEvent(window, 'online').map(() => true),
    //start the stream with the current online status
  	Rx.Observable.create(sub => {
  		sub.next(navigator.onLine);
  		sub.complete(); //this one only emits once, so now we end it
  	})
  );
}


function init() {
  
  // implement vanilla
  const onlineEmitter = createOnlineEmitter();
  let unsub = onlineEmitter(isOnline => console.log(isOnline));
  //unsub();
  
  // implement RxJS
  const online$ = createOnline$();
  const subscription = online$.subscribe(isOnline => console.log(isOnline));
  //subscription.unsubscribe();
  
  //compose new observable from RxJS version
  //only make the network request when we're online
  //the request will simply get queued up until then 
  function requestWhenOnline(ajaxPromiseFn) {
    return online$
      .filter(online => online)
      //we only reach this point when we're online
      .switchMap(ajaxPromiseFn) //instead of emitting true/false, emit the value of this function
      .take(1) //this ensures the observable ends after we've gotten a response
      .toPromise(); //convert it all to a promise
  }
  
  requestWhenOnline(() => axios.get('https://jsonplaceholder.typicode.com/posts/1'))
    .then(console.log)
    .catch(console.error);

}

/* Styles go here */