request-rx.js

request-rx.js

'use 6to5';

const rx = require('rx');
const request = require('request');
const fs = require('fs');

let wrapMethodInRx = (method) => {
  return function(...args) {
    return rx.Observable.create((subj) => {
      // Push the callback as the last parameter
      args.push((err, resp, body) => {
        if (err) {
          subj.onError(err);
          return;
        }

        if (resp.statusCode >= 400) {
          subj.onError(new Error(`Request failed: ${resp.statusCode}\n${body}`));
          return;
        }

        subj.onNext({response: resp, body: body});
        subj.onCompleted();
      });

      try {
        method(...args);
      } catch (e) {
        subj.onError(e);
      }

      return rx.Disposable.empty;
    });
  };
};

let requestRx = wrapMethodInRx(request);
requestRx.get = wrapMethodInRx(request.get);
requestRx.post = wrapMethodInRx(request.post);
requestRx.patch = wrapMethodInRx(request.patch);
requestRx.put = wrapMethodInRx(request.put);
requestRx.del = wrapMethodInRx(request.del);

requestRx.pipe = (url, stream) => {
  return rx.Observable.create((subj) => {
    try {
      request.get(url)
        .on('response', (resp) => {
          if (resp.statusCode > 399) subj.onError(new Error("Failed request: " + resp.statusCode));
        })
        .on('error', (err) => subj.onError(err))
        .on('end', () => { subj.onNext(true); subj.onCompleted(); })
        .pipe(stream);
    } catch (e) {
      subj.onError(e);
    }
  });
};


let isHttpUrl = (pathOrUrl) => pathOrUrl.match(/^http/i);

// Public: Fetches a file or URL, then returns its content as an Observable
//
// pathOrUrl - Either a file path or an HTTP URL
//
// Returns: An Observable which will yield a single value and complete, the contents
// of the given path or URL.
requestRx.fetchFileOrUrl = (pathOrUrl) => {
  if (!isHttpUrl(pathOrUrl)) {
    try {
      return rx.Observable.return(fs.readFileSync(pathOrUrl, { encoding: 'utf8' }));
    } catch (e) {
      return rx.Observable.throw(e);
    }
  }

  return requestRx(pathOrUrl).map((x) => x.body);
};