<commit_before>var request = require('request')
var Rx = require('rx')
var _ = require('lodash')
var fs = require('fs')

var requestBody = function (params) {
  var responseHeaders
  return Rx.Observable.create(function (observer) {
    request(params)
      .on('data', buffer => observer.onNext({buffer, headers: responseHeaders}))
      .on('response', x => responseHeaders = x.headers)
      .on('complete', x => observer.onCompleted(x))
      .on('error', x => observer.onError(x))
  })
}
var requestHead = Rx.Observable.fromNodeCallback(request.head, null, _.identity)
var fsOpen = Rx.Observable.fromNodeCallback(fs.open)
var fsWrite = Rx.Observable.fromNodeCallback(fs.write)
var fsTruncate = Rx.Observable.fromNodeCallback(fs.truncate)
var fsRename = Rx.Observable.fromNodeCallback(fs.rename)
var immutable = function (obj) {
  var observer
  const stream = Rx.Observable.create(function (_observer) {
    observer = _observer
  })
  return {stream, update: (cb) => observer.onNext(obj = cb(obj))}
}
module.exports = {
  requestBody,
  requestHead,
  fsOpen,
  fsWrite,
  fsTruncate,
  fsRename
}
<commit_msg>pass complete response as stream<commit_after>var request = require('request')
var Rx = require('rx')
var _ = require('lodash')
var fs = require('fs')

var requestBody = function (params) {
  var response
  return Rx.Observable.create(function (observer) {
    request(params)
      .on('data', buffer => observer.onNext({buffer, response}))
      .on('response', _response => response = _response)
      .on('complete', x => observer.onCompleted(x))
      .on('error', x => observer.onError(x))
  })
}
var requestHead = Rx.Observable.fromNodeCallback(request.head, null, _.identity)
var fsOpen = Rx.Observable.fromNodeCallback(fs.open)
var fsWrite = Rx.Observable.fromNodeCallback(fs.write)
var fsTruncate = Rx.Observable.fromNodeCallback(fs.truncate)
var fsRename = Rx.Observable.fromNodeCallback(fs.rename)
var immutable = function (obj) {
  var observer
  const stream = Rx.Observable.create(function (_observer) {
    observer = _observer
  })
  return {stream, update: (cb) => observer.onNext(obj = cb(obj))}
}
module.exports = {
  requestBody,
  requestHead,
  fsOpen,
  fsWrite,
  fsTruncate,
  fsRename
}

{'fraction_non_alphanumeric': 0.13895114298520841, 'fraction_numerical': 0.0, 'mean_word_length': 4.566084788029925, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '624168', 'n_tokens_mistral': 740, 'n_tokens_neox': 715, 'n_words': 159}