Request as a Highland Stream

request.js

'use strict';

var _ = require('highland');
var http = require('http');
var fs = require('fs');


var reqOptions = {
  'method': 'GET',
  'hostname': 'mockbin.org',
  'path': '/bin/f84a3d66-7255-44d8-8c56-95381f3bd8ff',
  'headers': {
    'accept': 'application/json'
  }
};

var outPath = 'res.json';


downloadToFile(reqOptions, outPath)
  .errors(_.log)
  .done(function () {
    console.log('Done');
  });


function httpJsonRequestStream(options) {
  return _(function (push) {
    var req = http.request(options, function (res) {
      _(res)
        .reduce1(function (a, b) { return a + b; })
        .through(validateRes)
        .errors(push)
        .apply(function (r) {
          try {
            var o = JSON.parse(r);
            push(null, { raw: r, json: o });
            push(null, _.nil);
          } catch (e) {
            push(e);
          }
        });
    });

    req.on('error', push);
    req.end();
  });
}


function downloadToFile(reqOptions, outPath) {
  return httpJsonRequestStream(reqOptions)
    .map(function (o) { return o.raw; })
    .flatMap(_.wrapCallback(function (data, cb) {
      fs.writeFile(outPath, data, cb);
    }));
}


function validateRes(s) {
  return s.consume(function (err, x, push, next) {
    if (x === _.nil) {
      return push(null, _.nil);
    }

    if (err || typeof x !== 'string' || !x.length) {
      return push(err || new Error('Empty response.'));
    }

    push(null, x);
    next();
  });
}