Reorganize, remove Coffeescript/Gulp/Lodash/string
parent
6b485859b7
commit
9a82b222ad
@ -0,0 +1,5 @@
|
||||
{
|
||||
"presets": [
|
||||
[ "@babel/preset-env" ]
|
||||
]
|
||||
}
|
||||
@ -0,0 +1 @@
|
||||
node 0.10
|
||||
@ -1,28 +0,0 @@
|
||||
var gulp = require('gulp');
|
||||
|
||||
/* CoffeeScript compile deps */
|
||||
var path = require('path');
|
||||
var gutil = require('gulp-util');
|
||||
var concat = require('gulp-concat');
|
||||
var rename = require('gulp-rename');
|
||||
var coffee = require('gulp-coffee');
|
||||
var cache = require('gulp-cached');
|
||||
var remember = require('gulp-remember');
|
||||
var plumber = require('gulp-plumber');
|
||||
|
||||
var source = ["lib/**/*.coffee", "index.coffee"]
|
||||
|
||||
gulp.task('coffee', function() {
|
||||
return gulp.src(source, {base: "."})
|
||||
.pipe(plumber())
|
||||
.pipe(cache("coffee"))
|
||||
.pipe(coffee({bare: true}).on('error', gutil.log)).on('data', gutil.log)
|
||||
.pipe(remember("coffee"))
|
||||
.pipe(gulp.dest("."));
|
||||
});
|
||||
|
||||
gulp.task('watch', function () {
|
||||
gulp.watch(source, ['coffee']);
|
||||
});
|
||||
|
||||
gulp.task('default', ['coffee', 'watch']);
|
||||
@ -1 +0,0 @@
|
||||
module.exports = require "./lib/bhttp"
|
||||
@ -1 +1,3 @@
|
||||
"use strict";
|
||||
|
||||
module.exports = require("./lib/bhttp");
|
||||
|
||||
@ -1,654 +0,0 @@
|
||||
# FIXME: Force-lowercase user-supplied headers before merging them into the request?
|
||||
# FIXME: Deep-merge query-string arguments between URL and argument?
|
||||
# FIXME: Named arrays for multipart/form-data?
|
||||
# FIXME: Are arrays of streams in `data` correctly recognized as being streams?
|
||||
|
||||
# Core modules
|
||||
urlUtil = require "url"
|
||||
querystring = require "querystring"
|
||||
stream = require "stream"
|
||||
http = require "http"
|
||||
https = require "https"
|
||||
util = require "util"
|
||||
|
||||
# Utility modules
|
||||
Promise = require "bluebird"
|
||||
_ = require "lodash"
|
||||
S = require "string"
|
||||
formFixArray = require "form-fix-array"
|
||||
errors = require "errors"
|
||||
debug = require("debug")
|
||||
debugRequest = debug("bhttp:request")
|
||||
debugResponse = debug("bhttp:response")
|
||||
extend = require "extend"
|
||||
devNull = require "dev-null"
|
||||
|
||||
# Other third-party modules
|
||||
formData = require "form-data2"
|
||||
concatStream = require "concat-stream"
|
||||
toughCookie = require "tough-cookie"
|
||||
streamLength = require "stream-length"
|
||||
sink = require "through2-sink"
|
||||
spy = require "through2-spy"
|
||||
|
||||
# For the version in the user agent, etc.
|
||||
packageConfig = require "../package.json"
|
||||
|
||||
bhttpErrors = {}
|
||||
|
||||
# Error types
|
||||
|
||||
errors.create
|
||||
name: "bhttpError"
|
||||
scope: bhttpErrors
|
||||
|
||||
errors.create
|
||||
name: "ConflictingOptionsError"
|
||||
parents: bhttpErrors.bhttpError
|
||||
scope: bhttpErrors
|
||||
|
||||
errors.create
|
||||
name: "UnsupportedProtocolError"
|
||||
parents: bhttpErrors.bhttpError
|
||||
scope: bhttpErrors
|
||||
|
||||
errors.create
|
||||
name: "RedirectError"
|
||||
parents: bhttpErrors.bhttpError
|
||||
scope: bhttpErrors
|
||||
|
||||
errors.create
|
||||
name: "MultipartError"
|
||||
parents: bhttpErrors.bhttpError
|
||||
scope: bhttpErrors
|
||||
|
||||
errors.create
|
||||
name: "ConnectionTimeoutError"
|
||||
parents: bhttpErrors.bhttpError
|
||||
scope: bhttpErrors
|
||||
|
||||
errors.create
|
||||
name: "ResponseTimeoutError"
|
||||
parents: bhttpErrors.bhttpError
|
||||
scope: bhttpErrors
|
||||
|
||||
# Utility functions
|
||||
|
||||
ofTypes = (obj, types) ->
|
||||
match = false
|
||||
for type in types
|
||||
match = match or obj instanceof type
|
||||
return match
|
||||
|
||||
addErrorData = (err, request, response, requestState) ->
|
||||
err.request = request
|
||||
err.response = response
|
||||
err.requestState = requestState
|
||||
return err
|
||||
|
||||
isStream = (obj) ->
|
||||
obj? and (ofTypes(obj, [stream.Readable, stream.Duplex, stream.Transform]) or obj.hasOwnProperty("_bhttpStreamWrapper"))
|
||||
|
||||
# Middleware
|
||||
# NOTE: requestState is an object that signifies the current state of the overall request; eg. for a response involving one or more redirects, it will hold a 'redirect history'.
|
||||
prepareSession = (request, response, requestState) ->
|
||||
debugRequest "preparing session"
|
||||
Promise.try ->
|
||||
if requestState.sessionOptions?
|
||||
# Request options take priority over session options
|
||||
request.options = _.merge _.clone(requestState.sessionOptions), request.options
|
||||
|
||||
# Create a headers parameter if it doesn't exist yet - we'll need to add some stuff to this later on
|
||||
# FIXME: We may need to do a deep-clone of other mutable options later on as well; otherwise, when getting a redirect in a session with pre-defined options, the contents may not be correctly cleared after following the redirect.
|
||||
if request.options.headers?
|
||||
request.options.headers = _.clone(request.options.headers, true)
|
||||
else
|
||||
request.options.headers = {}
|
||||
|
||||
# If we have a cookie jar, start out by setting the cookie string.
|
||||
if request.options.cookieJar?
|
||||
Promise.try ->
|
||||
# Move the cookieJar to the request object, the http/https module doesn't need it.
|
||||
request.cookieJar = request.options.cookieJar
|
||||
delete request.options.cookieJar
|
||||
|
||||
# Get the current cookie string for the URL
|
||||
request.cookieJar.get request.url
|
||||
.then (cookieString) ->
|
||||
debugRequest "sending cookie string: %s", cookieString
|
||||
request.options.headers["cookie"] = cookieString
|
||||
Promise.resolve [request, response, requestState]
|
||||
else
|
||||
Promise.resolve [request, response, requestState]
|
||||
|
||||
prepareDefaults = (request, response, requestState) ->
|
||||
debugRequest "preparing defaults"
|
||||
Promise.try ->
|
||||
# These are the options that we need for response processing, but don't need to be passed on to the http/https module.
|
||||
request.responseOptions =
|
||||
discardResponse: request.options.discardResponse ? false
|
||||
keepRedirectResponses: request.options.keepRedirectResponses ? false
|
||||
followRedirects: request.options.followRedirects ? true
|
||||
noDecode: request.options.noDecode ? false
|
||||
decodeJSON: request.options.decodeJSON ? false
|
||||
stream: request.options.stream ? false
|
||||
justPrepare: request.options.justPrepare ? false
|
||||
redirectLimit: request.options.redirectLimit ? 10
|
||||
onDownloadProgress: request.options.onDownloadProgress
|
||||
responseTimeout: request.options.responseTimeout
|
||||
|
||||
# Whether chunked transfer encoding for multipart/form-data payloads is acceptable. This is likely to break quietly on a lot of servers.
|
||||
request.options.allowChunkedMultipart ?= false
|
||||
|
||||
# Whether we should always use multipart/form-data for payloads, even if querystring-encoding would be a possibility.
|
||||
request.options.forceMultipart ?= false
|
||||
|
||||
# If no custom user-agent is defined, set our own
|
||||
request.options.headers["user-agent"] ?= "bhttp/#{packageConfig.version}"
|
||||
|
||||
# Normalize the request method to lowercase.
|
||||
request.options.method = request.options.method.toLowerCase()
|
||||
|
||||
Promise.resolve [request, response, requestState]
|
||||
|
||||
prepareUrl = (request, response, requestState) ->
|
||||
debugRequest "preparing URL"
|
||||
Promise.try ->
|
||||
# Parse the specified URL, and use the resulting information to build a complete `options` object
|
||||
urlOptions = urlUtil.parse request.url, true
|
||||
|
||||
_.extend request.options, {hostname: urlOptions.hostname, port: urlOptions.port}
|
||||
request.options.path = urlUtil.format {pathname: urlOptions.pathname, query: request.options.query ? urlOptions.query}
|
||||
request.protocol = S(urlOptions.protocol).chompRight(":").toString()
|
||||
|
||||
Promise.resolve [request, response, requestState]
|
||||
|
||||
prepareProtocol = (request, response, requestState) ->
|
||||
debugRequest "preparing protocol"
|
||||
Promise.try ->
|
||||
request.protocolModule = switch request.protocol
|
||||
when "http" then http
|
||||
when "https" then https # CAUTION / FIXME: Node will silently ignore SSL settings without a custom agent!
|
||||
else null
|
||||
|
||||
if not request.protocolModule?
|
||||
return Promise.reject() new bhttpErrors.UnsupportedProtocolError "The protocol specified (#{protocol}) is not currently supported by this module."
|
||||
|
||||
request.options.port ?= switch request.protocol
|
||||
when "http" then 80
|
||||
when "https" then 443
|
||||
|
||||
Promise.resolve [request, response, requestState]
|
||||
|
||||
prepareOptions = (request, response, requestState) ->
|
||||
debugRequest "preparing options"
|
||||
Promise.try ->
|
||||
# Do some sanity checks - there are a number of options that cannot be used together
|
||||
if (request.options.formFields? or request.options.files?) and (request.options.inputStream? or request.options.inputBuffer?)
|
||||
return Promise.reject addErrorData(new bhttpErrors.ConflictingOptionsError("You cannot define both formFields/files and a raw inputStream or inputBuffer."), request, response, requestState)
|
||||
|
||||
if request.options.encodeJSON and (request.options.inputStream? or request.options.inputBuffer?)
|
||||
return Promise.reject addErrorData(new bhttpErrors.ConflictingOptionsError("You cannot use both encodeJSON and a raw inputStream or inputBuffer.", undefined, "If you meant to JSON-encode the stream, you will currently have to do so manually."), request, response, requestState)
|
||||
|
||||
# If the user plans on streaming the response, we need to disable the agent entirely - otherwise the streams will block the pool.
|
||||
if request.responseOptions.stream
|
||||
request.options.agent ?= false
|
||||
|
||||
Promise.resolve [request, response, requestState]
|
||||
|
||||
preparePayload = (request, response, requestState) ->
|
||||
debugRequest "preparing payload"
|
||||
Promise.try ->
|
||||
# Persist the download progress event handler on the request object, if there is one.
|
||||
request.onUploadProgress = request.options.onUploadProgress
|
||||
|
||||
# If a 'files' parameter is present, then we will send the form data as multipart data - it's most likely binary data.
|
||||
multipart = request.options.forceMultipart or request.options.files?
|
||||
|
||||
# Similarly, if any of the formFields values are either a Stream or a Buffer, we will assume that the form should be sent as multipart.
|
||||
multipart = multipart or _.any request.options.formFields, (item) ->
|
||||
item instanceof Buffer or isStream(item)
|
||||
|
||||
# Really, 'files' and 'formFields' are the same thing - they mostly have different names for 1) clarity and 2) multipart detection. We combine them here.
|
||||
_.extend request.options.formFields, request.options.files
|
||||
|
||||
# For a last sanity check, we want to know whether there are any Stream objects in our form data *at all* - these can't be used when encodeJSON is enabled.
|
||||
containsStreams = _.any request.options.formFields, (item) -> isStream(item)
|
||||
|
||||
if request.options.encodeJSON and containsStreams
|
||||
return Promise.reject() new bhttpErrors.ConflictingOptionsError "Sending a JSON-encoded payload containing data from a stream is not currently supported.", undefined, "Either don't use encodeJSON, or read your stream into a string or Buffer."
|
||||
|
||||
if request.options.method not in ["get", "head", "delete"]
|
||||
# Prepare the payload, and set the appropriate headers.
|
||||
if (request.options.encodeJSON or request.options.formFields?) and not multipart
|
||||
# We know the payload and its size in advance.
|
||||
debugRequest "got url-encodable form-data"
|
||||
|
||||
if request.options.encodeJSON
|
||||
debugRequest "... but encodeJSON was set, so we will send JSON instead"
|
||||
request.options.headers["content-type"] = "application/json"
|
||||
request.payload = JSON.stringify request.options.formFields ? null
|
||||
else if not _.isEmpty request.options.formFields
|
||||
# The `querystring` module copies the key name verbatim, even if the value is actually an array. Things like PHP don't understand this, and expect every array-containing key to be suffixed with []. We'll just append that ourselves, then.
|
||||
request.options.headers["content-type"] = "application/x-www-form-urlencoded"
|
||||
request.payload = querystring.stringify formFixArray(request.options.formFields)
|
||||
else
|
||||
request.payload = ""
|
||||
|
||||
request.options.headers["content-length"] = request.payload.length
|
||||
|
||||
return Promise.resolve()
|
||||
else if request.options.formFields? and multipart
|
||||
# This is going to be multipart data, and we'll let `form-data` set the headers for us.
|
||||
debugRequest "got multipart form-data"
|
||||
formDataObject = new formData()
|
||||
|
||||
for fieldName, fieldValue of formFixArray(request.options.formFields)
|
||||
if not _.isArray fieldValue
|
||||
fieldValue = [fieldValue]
|
||||
|
||||
for valueElement in fieldValue
|
||||
if valueElement._bhttpStreamWrapper?
|
||||
streamOptions = valueElement.options
|
||||
valueElement = valueElement.stream
|
||||
else
|
||||
streamOptions = {}
|
||||
|
||||
formDataObject.append fieldName, valueElement, streamOptions
|
||||
|
||||
request.payloadStream = formDataObject
|
||||
|
||||
Promise.try ->
|
||||
formDataObject.getHeaders()
|
||||
.then (headers) ->
|
||||
if headers["content-transfer-encoding"] == "chunked" and not request.options.allowChunkedMultipart
|
||||
Promise.reject addErrorData(new MultipartError("Most servers do not support chunked transfer encoding for multipart/form-data payloads, and we could not determine the length of all the input streams. See the documentation for more information."), request, response, requestState)
|
||||
else
|
||||
_.extend request.options.headers, headers
|
||||
Promise.resolve()
|
||||
else if request.options.inputStream?
|
||||
# A raw inputStream was provided, just leave it be.
|
||||
debugRequest "got inputStream"
|
||||
Promise.try ->
|
||||
request.payloadStream = request.options.inputStream
|
||||
|
||||
if request.payloadStream._bhttpStreamWrapper? and (request.payloadStream.options.contentLength? or request.payloadStream.options.knownLength?)
|
||||
Promise.resolve(request.payloadStream.options.contentLength ? request.payloadStream.options.knownLength)
|
||||
else
|
||||
streamLength request.options.inputStream
|
||||
.then (length) ->
|
||||
debugRequest "length for inputStream is %s", length
|
||||
request.options.headers["content-length"] = length
|
||||
.catch (err) ->
|
||||
debugRequest "unable to determine inputStream length, switching to chunked transfer encoding"
|
||||
request.options.headers["content-transfer-encoding"] = "chunked"
|
||||
else if request.options.inputBuffer?
|
||||
# A raw inputBuffer was provided, just leave it be (but make sure it's an actual Buffer).
|
||||
debugRequest "got inputBuffer"
|
||||
if typeof request.options.inputBuffer == "string"
|
||||
request.payload = new Buffer(request.options.inputBuffer) # Input string should be utf-8!
|
||||
else
|
||||
request.payload = request.options.inputBuffer
|
||||
|
||||
debugRequest "length for inputBuffer is %s", request.payload.length
|
||||
request.options.headers["content-length"] = request.payload.length
|
||||
|
||||
return Promise.resolve()
|
||||
else
|
||||
# No payload specified.
|
||||
return Promise.resolve()
|
||||
else
|
||||
# GET, HEAD and DELETE should not have a payload. While technically not prohibited by the spec, it's also not specified, and we'd rather not upset poorly-compliant webservers.
|
||||
# FIXME: Should this throw an Error?
|
||||
return Promise.resolve()
|
||||
.then ->
|
||||
Promise.resolve [request, response, requestState]
|
||||
|
||||
prepareCleanup = (request, response, requestState) ->
|
||||
debugRequest "preparing cleanup"
|
||||
Promise.try ->
|
||||
# Remove the options that we're not going to pass on to the actual http/https library.
|
||||
delete request.options[key] for key in ["query", "formFields", "files", "encodeJSON", "inputStream", "inputBuffer", "discardResponse", "keepRedirectResponses", "followRedirects", "noDecode", "decodeJSON", "allowChunkedMultipart", "forceMultipart", "onUploadProgress", "onDownloadProgress"]
|
||||
|
||||
# Lo-Dash apparently has no `map` equivalent for object keys...?
|
||||
fixedHeaders = {}
|
||||
for key, value of request.options.headers
|
||||
fixedHeaders[key.toLowerCase()] = value
|
||||
request.options.headers = fixedHeaders
|
||||
|
||||
Promise.resolve [request, response, requestState]
|
||||
|
||||
# The guts of the module
|
||||
|
||||
prepareRequest = (request, response, requestState) ->
|
||||
debugRequest "preparing request"
|
||||
# FIXME: Mock httpd for testing functionality.
|
||||
Promise.try ->
|
||||
middlewareFunctions = [
|
||||
prepareSession
|
||||
prepareDefaults
|
||||
prepareUrl
|
||||
prepareProtocol
|
||||
prepareOptions
|
||||
preparePayload
|
||||
prepareCleanup
|
||||
]
|
||||
|
||||
promiseChain = Promise.resolve [request, response, requestState]
|
||||
|
||||
middlewareFunctions.forEach (middleware) -> # We must use the functional construct here, to avoid losing references
|
||||
promiseChain = promiseChain.spread (_request, _response, _requestState) ->
|
||||
middleware(_request, _response, _requestState)
|
||||
|
||||
return promiseChain
|
||||
|
||||
makeRequest = (request, response, requestState) ->
|
||||
debugRequest "making %s request to %s", request.options.method.toUpperCase(), request.url
|
||||
Promise.try ->
|
||||
# Instantiate a regular HTTP/HTTPS request
|
||||
req = request.protocolModule.request request.options
|
||||
|
||||
timeoutTimer = null
|
||||
|
||||
new Promise (resolve, reject) ->
|
||||
# Connection timeout handling, if one is set.
|
||||
if request.responseOptions.responseTimeout?
|
||||
debugRequest "setting response timeout timer to #{request.responseOptions.responseTimeout}ms..."
|
||||
req.on "socket", (socket) ->
|
||||
timeoutHandler = ->
|
||||
debugRequest "a response timeout occurred!"
|
||||
req.abort()
|
||||
reject addErrorData(new bhttpErrors.ResponseTimeoutError("The response timed out."))
|
||||
|
||||
timeoutTimer = setTimeout(timeoutHandler, request.responseOptions.responseTimeout)
|
||||
|
||||
# Set up the upload progress monitoring.
|
||||
totalBytes = request.options.headers["content-length"]
|
||||
completedBytes = 0
|
||||
|
||||
progressStream = spy (chunk) ->
|
||||
completedBytes += chunk.length
|
||||
req.emit "progress", completedBytes, totalBytes
|
||||
|
||||
if request.onUploadProgress?
|
||||
req.on "progress", (completedBytes, totalBytes) ->
|
||||
request.onUploadProgress(completedBytes, totalBytes, req)
|
||||
|
||||
# This is where we write our payload or stream to the request, and the actual request is made.
|
||||
if request.payload?
|
||||
# The entire payload is a single Buffer. We'll still pretend that it's a stream for our progress events, though, to provide a consistent API.
|
||||
debugRequest "sending payload"
|
||||
req.emit "progress", request.payload.length, request.payload.length
|
||||
req.write request.payload
|
||||
req.end()
|
||||
else if request.payloadStream?
|
||||
# The payload is a stream.
|
||||
debugRequest "piping payloadStream"
|
||||
if request.payloadStream._bhttpStreamWrapper?
|
||||
request.payloadStream.stream
|
||||
.pipe progressStream
|
||||
.pipe req
|
||||
else
|
||||
request.payloadStream
|
||||
.pipe progressStream
|
||||
.pipe req
|
||||
else
|
||||
# For GET, HEAD, DELETE, etc. there is no payload, but we still need to call end() to complete the request.
|
||||
debugRequest "closing request without payload"
|
||||
req.end()
|
||||
|
||||
# In case something goes wrong during this process, somehow...
|
||||
req.on "error", (err) ->
|
||||
if err.code == "ETIMEDOUT"
|
||||
debugRequest "a connection timeout occurred!"
|
||||
reject addErrorData(new bhttpErrors.ConnectionTimeoutError("The connection timed out."))
|
||||
else
|
||||
reject err
|
||||
|
||||
req.on "response", (res) ->
|
||||
if timeoutTimer?
|
||||
debugResponse "got response in time, clearing response timeout timer"
|
||||
clearTimeout(timeoutTimer)
|
||||
resolve res
|
||||
.then (response) ->
|
||||
Promise.resolve [request, response, requestState]
|
||||
|
||||
processResponse = (request, response, requestState) ->
|
||||
debugResponse "processing response, got status code %s", response.statusCode
|
||||
|
||||
# When we receive the response, we'll buffer it up and/or decode it, depending on what the user specified, and resolve the returned Promise. If the user just wants the raw stream, we resolve immediately after receiving a response.
|
||||
|
||||
Promise.try ->
|
||||
# First, if a cookie jar is set and we received one or more cookies from the server, we should store them in our cookieJar.
|
||||
if request.cookieJar? and response.headers["set-cookie"]?
|
||||
promises = for cookieHeader in response.headers["set-cookie"]
|
||||
debugResponse "storing cookie: %s", cookieHeader
|
||||
request.cookieJar.set cookieHeader, request.url
|
||||
Promise.all promises
|
||||
else
|
||||
Promise.resolve()
|
||||
.then ->
|
||||
# Now the actual response processing.
|
||||
response.request = request
|
||||
response.requestState = requestState
|
||||
response.redirectHistory = requestState.redirectHistory
|
||||
|
||||
if response.statusCode in [301, 302, 303, 307] and request.responseOptions.followRedirects
|
||||
if requestState.redirectHistory.length >= (request.responseOptions.redirectLimit - 1)
|
||||
return Promise.reject addErrorData(new bhttpErrors.RedirectError("The maximum amount of redirects ({request.responseOptions.redirectLimit}) was reached."))
|
||||
|
||||
# 301: For GET and HEAD, redirect unchanged. For POST, PUT, PATCH, DELETE, "ask user" (in our case: throw an error.)
|
||||
# 302: Redirect, change method to GET.
|
||||
# 303: Redirect, change method to GET.
|
||||
# 307: Redirect, retain method. Make same request again.
|
||||
switch response.statusCode
|
||||
when 301
|
||||
switch request.options.method
|
||||
when "get", "head"
|
||||
return redirectUnchanged request, response, requestState
|
||||
when "post", "put", "patch", "delete"
|
||||
return Promise.reject addErrorData(new bhttpErrors.RedirectError("Encountered a 301 redirect for POST, PUT, PATCH or DELETE. RFC says we can't automatically continue."), request, response, requestState)
|
||||
else
|
||||
return Promise.reject addErrorData(new bhttpErrors.RedirectError("Encountered a 301 redirect, but not sure how to proceed for the #{request.options.method.toUpperCase()} method."))
|
||||
when 302, 303
|
||||
return redirectGet request, response, requestState
|
||||
when 307
|
||||
if request.containsStreams and request.options.method not in ["get", "head"]
|
||||
return Promise.reject addErrorData(new bhttpErrors.RedirectError("Encountered a 307 redirect for POST, PUT or DELETE, but your payload contained (single-use) streams. We therefore can't automatically follow the redirect."), request, response, requestState)
|
||||
else
|
||||
return redirectUnchanged request, response, requestState
|
||||
else if request.responseOptions.discardResponse
|
||||
response.pipe(devNull()) # Drain the response stream
|
||||
Promise.resolve response
|
||||
else
|
||||
totalBytes = response.headers["content-length"]
|
||||
if totalBytes? # Otherwise `undefined` will turn into `NaN`, and we don't want that.
|
||||
totalBytes = parseInt(totalBytes)
|
||||
completedBytes = 0
|
||||
|
||||
progressStream = sink (chunk) ->
|
||||
completedBytes += chunk.length
|
||||
response.emit "progress", completedBytes, totalBytes
|
||||
|
||||
if request.responseOptions.onDownloadProgress?
|
||||
response.on "progress", (completedBytes, totalBytes) ->
|
||||
request.responseOptions.onDownloadProgress(completedBytes, totalBytes, response)
|
||||
|
||||
new Promise (resolve, reject) ->
|
||||
# This is a very, very dirty hack - however, using .pipe followed by .pause breaks in Node.js v0.10.35 with "Cannot switch to old mode now". Our solution is to monkeypatch the `on` and `resume` methods to attach the progress event handler as soon as something else is attached to the response stream (or when it is drained). This way, a user can also pipe the response in a later tick, without the stream draining prematurely.
|
||||
_resume = response.resume.bind(response)
|
||||
_on = response.on.bind(response)
|
||||
_progressStreamAttached = false
|
||||
|
||||
attachProgressStream = ->
|
||||
# To keep this from sending us into an infinite loop.
|
||||
if not _progressStreamAttached
|
||||
debugResponse "attaching progress stream"
|
||||
_progressStreamAttached = true
|
||||
response.pipe(progressStream)
|
||||
|
||||
response.on = (eventName, handler) ->
|
||||
debugResponse "'on' called, #{eventName}"
|
||||
if eventName == "data" or eventName == "readable"
|
||||
attachProgressStream()
|
||||
_on(eventName, handler)
|
||||
|
||||
response.resume = ->
|
||||
attachProgressStream()
|
||||
_resume()
|
||||
|
||||
# Continue with the regular response processing.
|
||||
if request.responseOptions.stream
|
||||
resolve response
|
||||
else
|
||||
response.on "error", (err) ->
|
||||
reject err
|
||||
|
||||
response.pipe concatStream (body) ->
|
||||
# FIXME: Separate module for header parsing?
|
||||
if request.responseOptions.decodeJSON or ((response.headers["content-type"] ? "").split(";")[0] == "application/json" and not request.responseOptions.noDecode)
|
||||
try
|
||||
response.body = JSON.parse body
|
||||
catch err
|
||||
reject err
|
||||
else
|
||||
response.body = body
|
||||
|
||||
resolve response
|
||||
|
||||
.then (response) ->
|
||||
Promise.resolve [request, response, requestState]
|
||||
|
||||
# Some wrappers
|
||||
|
||||
doPayloadRequest = (url, data, options, callback) ->
|
||||
# A wrapper that processes the second argument to .post, .put, .patch shorthand API methods.
|
||||
# FIXME: Treat a {} for data as a null? Otherwise {} combined with inputBuffer/inputStream will error.
|
||||
if isStream(data)
|
||||
options.inputStream = data
|
||||
else if ofTypes(data, [Buffer]) or typeof data == "string"
|
||||
options.inputBuffer = data
|
||||
else
|
||||
options.formFields = data
|
||||
|
||||
@request url, options, callback
|
||||
|
||||
redirectGet = (request, response, requestState) ->
|
||||
debugResponse "following forced-GET redirect to %s", response.headers["location"]
|
||||
Promise.try ->
|
||||
options = _.clone(requestState.originalOptions)
|
||||
options.method = "get"
|
||||
|
||||
delete options[key] for key in ["inputBuffer", "inputStream", "files", "formFields"]
|
||||
|
||||
doRedirect request, response, requestState, options
|
||||
|
||||
redirectUnchanged = (request, response, requestState) ->
|
||||
debugResponse "following same-method redirect to %s", response.headers["location"]
|
||||
Promise.try ->
|
||||
options = _.clone(requestState.originalOptions)
|
||||
doRedirect request, response, requestState, options
|
||||
|
||||
doRedirect = (request, response, requestState, newOptions) ->
|
||||
Promise.try ->
|
||||
if not request.responseOptions.keepRedirectResponses
|
||||
response.pipe(devNull()) # Let the response stream drain out...
|
||||
|
||||
requestState.redirectHistory.push response
|
||||
bhttpAPI._doRequest urlUtil.resolve(request.url, response.headers["location"]), newOptions, requestState
|
||||
|
||||
createCookieJar = (jar) ->
|
||||
# Creates a cookie jar wrapper with a simplified API.
|
||||
return {
|
||||
set: (cookie, url) ->
|
||||
new Promise (resolve, reject) =>
|
||||
@jar.setCookie cookie, url, (err, cookie) ->
|
||||
if err then reject(err) else resolve(cookie)
|
||||
get: (url) ->
|
||||
new Promise (resolve, reject) =>
|
||||
@jar.getCookieString url, (err, cookies) ->
|
||||
if err then reject(err) else resolve(cookies)
|
||||
jar: jar
|
||||
}
|
||||
|
||||
# The exposed API
|
||||
|
||||
bhttpAPI =
|
||||
head: (url, options = {}, callback) ->
|
||||
options.method = "head"
|
||||
@request url, options, callback
|
||||
get: (url, options = {}, callback) ->
|
||||
options.method = "get"
|
||||
@request url, options, callback
|
||||
post: (url, data, options = {}, callback) ->
|
||||
options.method = "post"
|
||||
doPayloadRequest.bind(this) url, data, options, callback
|
||||
put: (url, data, options = {}, callback) ->
|
||||
options.method = "put"
|
||||
doPayloadRequest.bind(this) url, data, options, callback
|
||||
patch: (url, data, options = {}, callback) ->
|
||||
options.method = "patch"
|
||||
doPayloadRequest.bind(this) url, data, options, callback
|
||||
delete: (url, options = {}, callback) ->
|
||||
options.method = "delete"
|
||||
@request url, options, callback
|
||||
request: (url, options = {}, callback) ->
|
||||
@_doRequest(url, options).nodeify(callback)
|
||||
_doRequest: (url, options, requestState) ->
|
||||
# This is split from the `request` method, so that the user doesn't have to pass in `undefined` for the `requestState` when they want to specify a `callback`.
|
||||
Promise.try =>
|
||||
request = {url: url, options: _.clone(options)}
|
||||
response = null
|
||||
requestState ?= {originalOptions: _.clone(options), redirectHistory: []}
|
||||
requestState.sessionOptions ?= @_sessionOptions ? {}
|
||||
|
||||
prepareRequest request, response, requestState
|
||||
.spread (request, response, requestState) =>
|
||||
if request.responseOptions.justPrepare
|
||||
Promise.resolve [request, response, requestState]
|
||||
else
|
||||
Promise.try ->
|
||||
bhttpAPI.executeRequest request, response, requestState
|
||||
.spread (request, response, requestState) ->
|
||||
# The user likely only wants the response.
|
||||
Promise.resolve response
|
||||
executeRequest: (request, response, requestState) ->
|
||||
# Executes a pre-configured request.
|
||||
Promise.try ->
|
||||
makeRequest request, response, requestState
|
||||
.spread (request, response, requestState) ->
|
||||
processResponse request, response, requestState
|
||||
session: (options) ->
|
||||
options ?= {}
|
||||
options = _.clone options
|
||||
session = {}
|
||||
|
||||
for key, value of this
|
||||
if value instanceof Function
|
||||
value = value.bind(session)
|
||||
session[key] = value
|
||||
|
||||
if not options.cookieJar?
|
||||
options.cookieJar = createCookieJar(new toughCookie.CookieJar())
|
||||
else if options.cookieJar == false
|
||||
delete options.cookieJar
|
||||
else
|
||||
# Assume we've gotten a cookie jar.
|
||||
options.cookieJar = createCookieJar(options.cookieJar)
|
||||
|
||||
session._sessionOptions = options
|
||||
|
||||
return session
|
||||
wrapStream: (stream, options) ->
|
||||
# This is a method for wrapping a stream in an object that also contains metadata.
|
||||
return {
|
||||
_bhttpStreamWrapper: true
|
||||
stream: stream
|
||||
options: options
|
||||
}
|
||||
|
||||
extend(bhttpAPI, bhttpErrors)
|
||||
|
||||
module.exports = bhttpAPI
|
||||
|
||||
# That's all, folks!
|
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,768 @@
|
||||
"use strict";
|
||||
|
||||
// FIXME: Force-lowercase user-supplied headers before merging them into the request?
|
||||
// FIXME: Deep-merge query-string arguments between URL and argument?
|
||||
// FIXME: Named arrays for multipart/form-data?
|
||||
// FIXME: Are arrays of streams in `data` correctly recognized as being streams?
|
||||
|
||||
// Core modules
|
||||
const urlUtil = require("url");
|
||||
const querystring = require("querystring");
|
||||
const stream = require("stream");
|
||||
const http = require("http");
|
||||
const https = require("https");
|
||||
|
||||
// Utility modules
|
||||
const Promise = require("bluebird");
|
||||
const formFixArray = require("form-fix-array");
|
||||
const errors = require("errors");
|
||||
const debug = require("debug");
|
||||
const debugRequest = debug("bhttp:request");
|
||||
const debugResponse = debug("bhttp:response");
|
||||
const extend = require("extend");
|
||||
const devNull = require("dev-null");
|
||||
const deepClone = require("lodash.clonedeep");
|
||||
const deepMerge = require("lodash.merge");
|
||||
|
||||
// Other third-party modules
|
||||
const formData = require("form-data2");
|
||||
const concatStream = require("concat-stream");
|
||||
const toughCookie = require("tough-cookie");
|
||||
const streamLength = require("stream-length");
|
||||
const sink = require("through2-sink");
|
||||
const spy = require("through2-spy");
|
||||
|
||||
// For the version in the user agent, etc.
|
||||
const packageConfig = require("../package.json");
|
||||
|
||||
const bhttpErrors = {};
|
||||
|
||||
// Error types
|
||||
|
||||
errors.create({
|
||||
name: "bhttpError",
|
||||
scope: bhttpErrors
|
||||
});
|
||||
|
||||
errors.create({
|
||||
name: "ConflictingOptionsError",
|
||||
parents: bhttpErrors.bhttpError,
|
||||
scope: bhttpErrors
|
||||
});
|
||||
|
||||
errors.create({
|
||||
name: "UnsupportedProtocolError",
|
||||
parents: bhttpErrors.bhttpError,
|
||||
scope: bhttpErrors
|
||||
});
|
||||
|
||||
errors.create({
|
||||
name: "RedirectError",
|
||||
parents: bhttpErrors.bhttpError,
|
||||
scope: bhttpErrors
|
||||
});
|
||||
|
||||
errors.create({
|
||||
name: "MultipartError",
|
||||
parents: bhttpErrors.bhttpError,
|
||||
scope: bhttpErrors
|
||||
});
|
||||
|
||||
errors.create({
|
||||
name: "ConnectionTimeoutError",
|
||||
parents: bhttpErrors.bhttpError,
|
||||
scope: bhttpErrors
|
||||
});
|
||||
|
||||
errors.create({
|
||||
name: "ResponseTimeoutError",
|
||||
parents: bhttpErrors.bhttpError,
|
||||
scope: bhttpErrors
|
||||
});
|
||||
|
||||
// Utility functions
|
||||
|
||||
function shallowClone(object) {
|
||||
return Object.assign({}, object);
|
||||
}
|
||||
|
||||
const ofTypes = function(obj, types) {
|
||||
let match = false;
|
||||
for (let type of types) {
|
||||
match = match || obj instanceof type;
|
||||
}
|
||||
return match;
|
||||
};
|
||||
|
||||
const addErrorData = function(err, request, response, requestState) {
|
||||
err.request = request;
|
||||
err.response = response;
|
||||
err.requestState = requestState;
|
||||
return err;
|
||||
};
|
||||
|
||||
const isStream = obj => (obj != null) && (ofTypes(obj, [stream.Readable, stream.Duplex, stream.Transform]) || obj.hasOwnProperty("_bhttpStreamWrapper"));
|
||||
|
||||
// Middleware
|
||||
// NOTE: requestState is an object that signifies the current state of the overall request; eg. for a response involving one or more redirects, it will hold a 'redirect history'.
|
||||
const prepareSession = function(request, response, requestState) {
|
||||
debugRequest("preparing session");
|
||||
return Promise.try(function() {
|
||||
if (requestState.sessionOptions != null) {
|
||||
// Request options take priority over session options
|
||||
request.options = deepMerge(shallowClone(requestState.sessionOptions), request.options);
|
||||
}
|
||||
|
||||
// Create a headers parameter if it doesn't exist yet - we'll need to add some stuff to this later on
|
||||
// FIXME: We may need to do a deep-clone of other mutable options later on as well; otherwise, when getting a redirect in a session with pre-defined options, the contents may not be correctly cleared after following the redirect.
|
||||
if (request.options.headers != null) {
|
||||
request.options.headers = deepClone(request.options.headers);
|
||||
} else {
|
||||
request.options.headers = {};
|
||||
}
|
||||
|
||||
// If we have a cookie jar, start out by setting the cookie string.
|
||||
if (request.options.cookieJar != null) {
|
||||
return Promise.try(function() {
|
||||
// Move the cookieJar to the request object, the http/https module doesn't need it.
|
||||
request.cookieJar = request.options.cookieJar;
|
||||
delete request.options.cookieJar;
|
||||
|
||||
// Get the current cookie string for the URL
|
||||
return request.cookieJar.get(request.url);}).then(function(cookieString) {
|
||||
debugRequest("sending cookie string: %s", cookieString);
|
||||
request.options.headers["cookie"] = cookieString;
|
||||
return Promise.resolve([request, response, requestState]);});
|
||||
} else {
|
||||
return Promise.resolve([request, response, requestState]);
|
||||
}});
|
||||
};
|
||||
|
||||
const prepareDefaults = function(request, response, requestState) {
|
||||
debugRequest("preparing defaults");
|
||||
return Promise.try(function() {
|
||||
// These are the options that we need for response processing, but don't need to be passed on to the http/https module.
|
||||
request.responseOptions = {
|
||||
discardResponse: request.options.discardResponse != null ? request.options.discardResponse : false,
|
||||
keepRedirectResponses: request.options.keepRedirectResponses != null ? request.options.keepRedirectResponses : false,
|
||||
followRedirects: request.options.followRedirects != null ? request.options.followRedirects : true,
|
||||
noDecode: request.options.noDecode != null ? request.options.noDecode : false,
|
||||
decodeJSON: request.options.decodeJSON != null ? request.options.decodeJSON : false,
|
||||
stream: request.options.stream != null ? request.options.stream : false,
|
||||
justPrepare: request.options.justPrepare != null ? request.options.justPrepare : false,
|
||||
redirectLimit: request.options.redirectLimit != null ? request.options.redirectLimit : 10,
|
||||
onDownloadProgress: request.options.onDownloadProgress,
|
||||
responseTimeout: request.options.responseTimeout
|
||||
};
|
||||
|
||||
// Whether chunked transfer encoding for multipart/form-data payloads is acceptable. This is likely to break quietly on a lot of servers.
|
||||
if (request.options.allowChunkedMultipart == null) { request.options.allowChunkedMultipart = false; }
|
||||
|
||||
// Whether we should always use multipart/form-data for payloads, even if querystring-encoding would be a possibility.
|
||||
if (request.options.forceMultipart == null) { request.options.forceMultipart = false; }
|
||||
|
||||
// If no custom user-agent is defined, set our own
|
||||
if (request.options.headers["user-agent"] == null) { request.options.headers["user-agent"] = `bhttp/${packageConfig.version}`; }
|
||||
|
||||
// Normalize the request method to lowercase.
|
||||
request.options.method = request.options.method.toLowerCase();
|
||||
|
||||
return Promise.resolve([request, response, requestState]);});
|
||||
};
|
||||
|
||||
const prepareUrl = function(request, response, requestState) {
|
||||
debugRequest("preparing URL");
|
||||
return Promise.try(function() {
|
||||
// Parse the specified URL, and use the resulting information to build a complete `options` object
|
||||
const urlOptions = urlUtil.parse(request.url, true);
|
||||
|
||||
Object.assign(request.options, {hostname: urlOptions.hostname, port: urlOptions.port});
|
||||
request.options.path = urlUtil.format({pathname: urlOptions.pathname, query: request.options.query != null ? request.options.query : urlOptions.query});
|
||||
request.protocol = urlOptions.protocol.replace(/:$/, "");
|
||||
|
||||
return Promise.resolve([request, response, requestState]);});
|
||||
};
|
||||
|
||||
const prepareProtocol = function(request, response, requestState) {
|
||||
debugRequest("preparing protocol");
|
||||
return Promise.try(function() {
|
||||
request.protocolModule = (() => { switch (request.protocol) {
|
||||
case "http": return http;
|
||||
case "https": return https; // CAUTION / FIXME: Node will silently ignore SSL settings without a custom agent!
|
||||
default: return null;
|
||||
} })();
|
||||
|
||||
if ((request.protocolModule == null)) {
|
||||
return Promise.reject(new bhttpErrors.UnsupportedProtocolError(`The protocol specified (${request.protocol}) is not currently supported by this module.`));
|
||||
}
|
||||
|
||||
if (request.options.port == null) { request.options.port = (() => { switch (request.protocol) {
|
||||
case "http": return 80;
|
||||
case "https": return 443;
|
||||
} })(); }
|
||||
|
||||
return Promise.resolve([request, response, requestState]);});
|
||||
};
|
||||
|
||||
const prepareOptions = function(request, response, requestState) {
|
||||
debugRequest("preparing options");
|
||||
return Promise.try(function() {
|
||||
// Do some sanity checks - there are a number of options that cannot be used together
|
||||
if (((request.options.formFields != null) || (request.options.files != null)) && ((request.options.inputStream != null) || (request.options.inputBuffer != null))) {
|
||||
return Promise.reject(addErrorData(new bhttpErrors.ConflictingOptionsError("You cannot define both formFields/files and a raw inputStream or inputBuffer."), request, response, requestState));
|
||||
}
|
||||
|
||||
if (request.options.encodeJSON && ((request.options.inputStream != null) || (request.options.inputBuffer != null))) {
|
||||
return Promise.reject(addErrorData(new bhttpErrors.ConflictingOptionsError("You cannot use both encodeJSON and a raw inputStream or inputBuffer.", undefined, "If you meant to JSON-encode the stream, you will currently have to do so manually."), request, response, requestState));
|
||||
}
|
||||
|
||||
// If the user plans on streaming the response, we need to disable the agent entirely - otherwise the streams will block the pool.
|
||||
if (request.responseOptions.stream) {
|
||||
if (request.options.agent == null) { request.options.agent = false; }
|
||||
}
|
||||
|
||||
return Promise.resolve([request, response, requestState]);});
|
||||
};
|
||||
|
||||
const preparePayload = function(request, response, requestState) {
|
||||
debugRequest("preparing payload");
|
||||
return Promise.try(function() {
|
||||
// Persist the download progress event handler on the request object, if there is one.
|
||||
request.onUploadProgress = request.options.onUploadProgress;
|
||||
|
||||
// If a 'files' parameter is present, then we will send the form data as multipart data - it's most likely binary data.
|
||||
let multipart = request.options.forceMultipart || (request.options.files != null);
|
||||
|
||||
// Similarly, if any of the formFields values are either a Stream or a Buffer, we will assume that the form should be sent as multipart.
|
||||
multipart = multipart || Object.values(request.options.formFields).some((item) => item instanceof Buffer || isStream(item));
|
||||
|
||||
// Really, 'files' and 'formFields' are the same thing - they mostly have different names for 1) clarity and 2) multipart detection. We combine them here.
|
||||
Object.assign(request.options.formFields, request.options.files);
|
||||
|
||||
// For a last sanity check, we want to know whether there are any Stream objects in our form data *at all* - these can't be used when encodeJSON is enabled.
|
||||
const containsStreams = Object.values(request.options.formFields).some((item) => isStream(item));
|
||||
|
||||
if (request.options.encodeJSON && containsStreams) {
|
||||
return Promise.reject(new bhttpErrors.ConflictingOptionsError("Sending a JSON-encoded payload containing data from a stream is not currently supported.", undefined, "Either don't use encodeJSON, or read your stream into a string or Buffer."));
|
||||
}
|
||||
|
||||
if (!["get", "head", "delete"].includes(request.options.method)) {
|
||||
// Prepare the payload, and set the appropriate headers.
|
||||
if ((request.options.encodeJSON || (request.options.formFields != null)) && !multipart) {
|
||||
// We know the payload and its size in advance.
|
||||
debugRequest("got url-encodable form-data");
|
||||
|
||||
if (request.options.encodeJSON) {
|
||||
debugRequest("... but encodeJSON was set, so we will send JSON instead");
|
||||
request.options.headers["content-type"] = "application/json";
|
||||
request.payload = JSON.stringify(request.options.formFields != null ? request.options.formFields : null);
|
||||
} else if (Object.keys(request.options.formFields).length > 0) {
|
||||
// The `querystring` module copies the key name verbatim, even if the value is actually an array. Things like PHP don't understand this, and expect every array-containing key to be suffixed with []. We'll just append that ourselves, then.
|
||||
request.options.headers["content-type"] = "application/x-www-form-urlencoded";
|
||||
request.payload = querystring.stringify(formFixArray(request.options.formFields));
|
||||
} else {
|
||||
request.payload = "";
|
||||
}
|
||||
|
||||
request.options.headers["content-length"] = request.payload.length;
|
||||
|
||||
return Promise.resolve();
|
||||
} else if ((request.options.formFields != null) && multipart) {
|
||||
// This is going to be multipart data, and we'll let `form-data` set the headers for us.
|
||||
debugRequest("got multipart form-data");
|
||||
const formDataObject = new formData();
|
||||
|
||||
const object = formFixArray(request.options.formFields);
|
||||
for (let fieldName in object) {
|
||||
let fieldValue = object[fieldName];
|
||||
if (!Array.isArray(fieldValue)) {
|
||||
fieldValue = [fieldValue];
|
||||
}
|
||||
|
||||
for (let valueElement of fieldValue) {
|
||||
var streamOptions;
|
||||
if (valueElement._bhttpStreamWrapper != null) {
|
||||
streamOptions = valueElement.options;
|
||||
valueElement = valueElement.stream;
|
||||
} else {
|
||||
streamOptions = {};
|
||||
}
|
||||
|
||||
formDataObject.append(fieldName, valueElement, streamOptions);
|
||||
}
|
||||
}
|
||||
|
||||
request.payloadStream = formDataObject;
|
||||
|
||||
return Promise.try(() => formDataObject.getHeaders()).then(function(headers) {
|
||||
if ((headers["content-transfer-encoding"] === "chunked") && !request.options.allowChunkedMultipart) {
|
||||
return Promise.reject(addErrorData(new bhttpErrors.MultipartError("Most servers do not support chunked transfer encoding for multipart/form-data payloads, and we could not determine the length of all the input streams. See the documentation for more information."), request, response, requestState));
|
||||
} else {
|
||||
Object.assign(request.options.headers, headers);
|
||||
return Promise.resolve();
|
||||
}
|
||||
});
|
||||
} else if (request.options.inputStream != null) {
|
||||
// A raw inputStream was provided, just leave it be.
|
||||
debugRequest("got inputStream");
|
||||
return Promise.try(function() {
|
||||
request.payloadStream = request.options.inputStream;
|
||||
|
||||
if ((request.payloadStream._bhttpStreamWrapper != null) && ((request.payloadStream.options.contentLength != null) || (request.payloadStream.options.knownLength != null))) {
|
||||
return Promise.resolve(request.payloadStream.options.contentLength != null ? request.payloadStream.options.contentLength : request.payloadStream.options.knownLength);
|
||||
} else {
|
||||
return streamLength(request.options.inputStream);
|
||||
}
|
||||
}).then(function(length) {
|
||||
debugRequest("length for inputStream is %s", length);
|
||||
request.options.headers["content-length"] = length;
|
||||
}).catch(function(_error) {
|
||||
debugRequest("unable to determine inputStream length, switching to chunked transfer encoding");
|
||||
request.options.headers["content-transfer-encoding"] = "chunked";
|
||||
});
|
||||
} else if (request.options.inputBuffer != null) {
|
||||
// A raw inputBuffer was provided, just leave it be (but make sure it's an actual Buffer).
|
||||
debugRequest("got inputBuffer");
|
||||
if (typeof request.options.inputBuffer === "string") {
|
||||
request.payload = new Buffer(request.options.inputBuffer); // Input string should be utf-8!
|
||||
} else {
|
||||
request.payload = request.options.inputBuffer;
|
||||
}
|
||||
|
||||
debugRequest("length for inputBuffer is %s", request.payload.length);
|
||||
request.options.headers["content-length"] = request.payload.length;
|
||||
|
||||
return Promise.resolve();
|
||||
} else {
|
||||
// No payload specified.
|
||||
return Promise.resolve();
|
||||
}
|
||||
} else {
|
||||
// GET, HEAD and DELETE should not have a payload. While technically not prohibited by the spec, it's also not specified, and we'd rather not upset poorly-compliant webservers.
|
||||
// FIXME: Should this throw an Error?
|
||||
return Promise.resolve();
|
||||
}}).then(() => Promise.resolve([request, response, requestState]));
|
||||
};
|
||||
|
||||
const prepareCleanup = function(request, response, requestState) {
|
||||
debugRequest("preparing cleanup");
|
||||
return Promise.try(function() {
|
||||
// Remove the options that we're not going to pass on to the actual http/https library.
|
||||
let key;
|
||||
for (key of ["query", "formFields", "files", "encodeJSON", "inputStream", "inputBuffer", "discardResponse", "keepRedirectResponses", "followRedirects", "noDecode", "decodeJSON", "allowChunkedMultipart", "forceMultipart", "onUploadProgress", "onDownloadProgress"]) { delete request.options[key]; }
|
||||
|
||||
// Lo-Dash apparently has no `map` equivalent for object keys...?
|
||||
const fixedHeaders = {};
|
||||
for (key in request.options.headers) {
|
||||
const value = request.options.headers[key];
|
||||
fixedHeaders[key.toLowerCase()] = value;
|
||||
}
|
||||
request.options.headers = fixedHeaders;
|
||||
|
||||
return Promise.resolve([request, response, requestState]);});
|
||||
};
|
||||
|
||||
// The guts of the module
|
||||
|
||||
const prepareRequest = function(request, response, requestState) {
|
||||
debugRequest("preparing request");
|
||||
// FIXME: Mock httpd for testing functionality.
|
||||
return Promise.try(function() {
|
||||
const middlewareFunctions = [
|
||||
prepareSession,
|
||||
prepareDefaults,
|
||||
prepareUrl,
|
||||
prepareProtocol,
|
||||
prepareOptions,
|
||||
preparePayload,
|
||||
prepareCleanup
|
||||
];
|
||||
|
||||
let promiseChain = Promise.resolve([request, response, requestState]);
|
||||
|
||||
middlewareFunctions.forEach((middleware) => {
|
||||
// We must use the functional construct here, to avoid losing references
|
||||
promiseChain = promiseChain.spread((_request, _response, _requestState) => middleware(_request, _response, _requestState));
|
||||
});
|
||||
|
||||
return promiseChain;
|
||||
});
|
||||
};
|
||||
|
||||
const makeRequest = function(request, response, requestState) {
|
||||
debugRequest("making %s request to %s", request.options.method.toUpperCase(), request.url);
|
||||
return Promise.try(function() {
|
||||
// Instantiate a regular HTTP/HTTPS request
|
||||
const req = request.protocolModule.request(request.options);
|
||||
|
||||
let timeoutTimer = null;
|
||||
|
||||
return new Promise(function(resolve, reject) {
|
||||
// Connection timeout handling, if one is set.
|
||||
if (request.responseOptions.responseTimeout != null) {
|
||||
debugRequest(`setting response timeout timer to ${request.responseOptions.responseTimeout}ms...`);
|
||||
req.on("socket", function(_socket) {
|
||||
const timeoutHandler = function() {
|
||||
debugRequest("a response timeout occurred!");
|
||||
req.abort();
|
||||
return reject(addErrorData(new bhttpErrors.ResponseTimeoutError("The response timed out.")));
|
||||
};
|
||||
|
||||
timeoutTimer = setTimeout(timeoutHandler, request.responseOptions.responseTimeout);
|
||||
});
|
||||
}
|
||||
|
||||
// Set up the upload progress monitoring.
|
||||
const totalBytes = request.options.headers["content-length"];
|
||||
let completedBytes = 0;
|
||||
|
||||
const progressStream = spy(function(chunk) {
|
||||
completedBytes += chunk.length;
|
||||
return req.emit("progress", completedBytes, totalBytes);
|
||||
});
|
||||
|
||||
if (request.onUploadProgress != null) {
|
||||
req.on("progress", (completedBytes, totalBytes) => request.onUploadProgress(completedBytes, totalBytes, req));
|
||||
}
|
||||
|
||||
// This is where we write our payload or stream to the request, and the actual request is made.
|
||||
if (request.payload != null) {
|
||||
// The entire payload is a single Buffer. We'll still pretend that it's a stream for our progress events, though, to provide a consistent API.
|
||||
debugRequest("sending payload");
|
||||
req.emit("progress", request.payload.length, request.payload.length);
|
||||
req.write(request.payload);
|
||||
req.end();
|
||||
} else if (request.payloadStream != null) {
|
||||
// The payload is a stream.
|
||||
debugRequest("piping payloadStream");
|
||||
if (request.payloadStream._bhttpStreamWrapper != null) {
|
||||
request.payloadStream.stream
|
||||
.pipe(progressStream)
|
||||
.pipe(req);
|
||||
} else {
|
||||
request.payloadStream
|
||||
.pipe(progressStream)
|
||||
.pipe(req);
|
||||
}
|
||||
} else {
|
||||
// For GET, HEAD, DELETE, etc. there is no payload, but we still need to call end() to complete the request.
|
||||
debugRequest("closing request without payload");
|
||||
req.end();
|
||||
}
|
||||
|
||||
// In case something goes wrong during this process, somehow...
|
||||
req.on("error", function(err) {
|
||||
if (err.code === "ETIMEDOUT") {
|
||||
debugRequest("a connection timeout occurred!");
|
||||
return reject(addErrorData(new bhttpErrors.ConnectionTimeoutError("The connection timed out.")));
|
||||
} else {
|
||||
return reject(err);
|
||||
}
|
||||
});
|
||||
|
||||
return req.on("response", function(res) {
|
||||
if (timeoutTimer != null) {
|
||||
debugResponse("got response in time, clearing response timeout timer");
|
||||
clearTimeout(timeoutTimer);
|
||||
}
|
||||
return resolve(res);
|
||||
});
|
||||
});}).then(response => Promise.resolve([request, response, requestState]));
|
||||
};
|
||||
|
||||
const processResponse = function(request, response, requestState) {
|
||||
debugResponse("processing response, got status code %s", response.statusCode);
|
||||
|
||||
// When we receive the response, we'll buffer it up and/or decode it, depending on what the user specified, and resolve the returned Promise. If the user just wants the raw stream, we resolve immediately after receiving a response.
|
||||
|
||||
return Promise.try(function() {
|
||||
// First, if a cookie jar is set and we received one or more cookies from the server, we should store them in our cookieJar.
|
||||
if ((request.cookieJar != null) && (response.headers["set-cookie"] != null)) {
|
||||
const promises = (() => {
|
||||
const result = [];
|
||||
for (let cookieHeader of response.headers["set-cookie"]) {
|
||||
debugResponse("storing cookie: %s", cookieHeader);
|
||||
result.push(request.cookieJar.set(cookieHeader, request.url));
|
||||
}
|
||||
return result;
|
||||
})();
|
||||
return Promise.all(promises);
|
||||
} else {
|
||||
return Promise.resolve();
|
||||
}}).then(function() {
|
||||
// Now the actual response processing.
|
||||
response.request = request;
|
||||
response.requestState = requestState;
|
||||
response.redirectHistory = requestState.redirectHistory;
|
||||
|
||||
if ([301, 302, 303, 307].includes(response.statusCode) && request.responseOptions.followRedirects) {
|
||||
if (requestState.redirectHistory.length >= (request.responseOptions.redirectLimit - 1)) {
|
||||
return Promise.reject(addErrorData(new bhttpErrors.RedirectError("The maximum amount of redirects ({request.responseOptions.redirectLimit}) was reached.")));
|
||||
}
|
||||
|
||||
// 301: For GET and HEAD, redirect unchanged. For POST, PUT, PATCH, DELETE, "ask user" (in our case: throw an error.)
|
||||
// 302: Redirect, change method to GET.
|
||||
// 303: Redirect, change method to GET.
|
||||
// 307: Redirect, retain method. Make same request again.
|
||||
switch (response.statusCode) {
|
||||
case 301:
|
||||
switch (request.options.method) {
|
||||
case "get": case "head":
|
||||
return redirectUnchanged(request, response, requestState);
|
||||
case "post": case "put": case "patch": case "delete":
|
||||
return Promise.reject(addErrorData(new bhttpErrors.RedirectError("Encountered a 301 redirect for POST, PUT, PATCH or DELETE. RFC says we can't automatically continue."), request, response, requestState));
|
||||
default:
|
||||
return Promise.reject(addErrorData(new bhttpErrors.RedirectError(`Encountered a 301 redirect, but not sure how to proceed for the ${request.options.method.toUpperCase()} method.`)));
|
||||
}
|
||||
case 302: case 303:
|
||||
return redirectGet(request, response, requestState);
|
||||
case 307:
|
||||
if (request.containsStreams && !["get", "head"].includes(request.options.method)) {
|
||||
return Promise.reject(addErrorData(new bhttpErrors.RedirectError("Encountered a 307 redirect for POST, PUT or DELETE, but your payload contained (single-use) streams. We therefore can't automatically follow the redirect."), request, response, requestState));
|
||||
} else {
|
||||
return redirectUnchanged(request, response, requestState);
|
||||
}
|
||||
}
|
||||
} else if (request.responseOptions.discardResponse) {
|
||||
response.pipe(devNull()); // Drain the response stream
|
||||
return Promise.resolve(response);
|
||||
} else {
|
||||
let totalBytes = response.headers["content-length"];
|
||||
if (totalBytes != null) { // Otherwise `undefined` will turn into `NaN`, and we don't want that.
|
||||
totalBytes = parseInt(totalBytes);
|
||||
}
|
||||
let completedBytes = 0;
|
||||
|
||||
const progressStream = sink(function(chunk) {
|
||||
completedBytes += chunk.length;
|
||||
return response.emit("progress", completedBytes, totalBytes);
|
||||
});
|
||||
|
||||
if (request.responseOptions.onDownloadProgress != null) {
|
||||
response.on("progress", (completedBytes, totalBytes) => request.responseOptions.onDownloadProgress(completedBytes, totalBytes, response));
|
||||
}
|
||||
|
||||
return new Promise(function(resolve, reject) {
|
||||
// This is a very, very dirty hack - however, using .pipe followed by .pause breaks in Node.js v0.10.35 with "Cannot switch to old mode now". Our solution is to monkeypatch the `on` and `resume` methods to attach the progress event handler as soon as something else is attached to the response stream (or when it is drained). This way, a user can also pipe the response in a later tick, without the stream draining prematurely.
|
||||
const _resume = response.resume.bind(response);
|
||||
const _on = response.on.bind(response);
|
||||
let _progressStreamAttached = false;
|
||||
|
||||
const attachProgressStream = function() {
|
||||
// To keep this from sending us into an infinite loop.
|
||||
if (!_progressStreamAttached) {
|
||||
debugResponse("attaching progress stream");
|
||||
_progressStreamAttached = true;
|
||||
return response.pipe(progressStream);
|
||||
}
|
||||
};
|
||||
|
||||
response.on = function(eventName, handler) {
|
||||
debugResponse(`'on' called, ${eventName}`);
|
||||
if ((eventName === "data") || (eventName === "readable")) {
|
||||
attachProgressStream();
|
||||
}
|
||||
return _on(eventName, handler);
|
||||
};
|
||||
|
||||
response.resume = function() {
|
||||
attachProgressStream();
|
||||
return _resume();
|
||||
};
|
||||
|
||||
// Continue with the regular response processing.
|
||||
if (request.responseOptions.stream) {
|
||||
return resolve(response);
|
||||
} else {
|
||||
response.on("error", err => reject(err));
|
||||
|
||||
return response.pipe(concatStream(function(body) {
|
||||
// FIXME: Separate module for header parsing?
|
||||
if (request.responseOptions.decodeJSON || (((response.headers["content-type"] != null ? response.headers["content-type"] : "").split(";")[0] === "application/json") && !request.responseOptions.noDecode)) {
|
||||
try {
|
||||
response.body = JSON.parse(body);
|
||||
} catch (err) {
|
||||
reject(err);
|
||||
}
|
||||
} else {
|
||||
response.body = body;
|
||||
}
|
||||
|
||||
return resolve(response);
|
||||
})
|
||||
);
|
||||
}
|
||||
});
|
||||
}}).then(response => Promise.resolve([request, response, requestState]));
|
||||
};
|
||||
|
||||
// Some wrappers
|
||||
|
||||
const doPayloadRequest = function(url, data, options, callback) {
|
||||
// A wrapper that processes the second argument to .post, .put, .patch shorthand API methods.
|
||||
// FIXME: Treat a {} for data as a null? Otherwise {} combined with inputBuffer/inputStream will error.
|
||||
if (isStream(data)) {
|
||||
options.inputStream = data;
|
||||
} else if (ofTypes(data, [Buffer]) || (typeof data === "string")) {
|
||||
options.inputBuffer = data;
|
||||
} else {
|
||||
options.formFields = data;
|
||||
}
|
||||
|
||||
return this.request(url, options, callback);
|
||||
};
|
||||
|
||||
var redirectGet = function(request, response, requestState) {
|
||||
debugResponse("following forced-GET redirect to %s", response.headers["location"]);
|
||||
return Promise.try(function() {
|
||||
const options = shallowClone(requestState.originalOptions);
|
||||
options.method = "get";
|
||||
|
||||
for (let key of ["inputBuffer", "inputStream", "files", "formFields"]) { delete options[key]; }
|
||||
|
||||
return doRedirect(request, response, requestState, options);
|
||||
});
|
||||
};
|
||||
|
||||
var redirectUnchanged = function(request, response, requestState) {
|
||||
debugResponse("following same-method redirect to %s", response.headers["location"]);
|
||||
return Promise.try(function() {
|
||||
const options = shallowClone(requestState.originalOptions);
|
||||
return doRedirect(request, response, requestState, options);
|
||||
});
|
||||
};
|
||||
|
||||
var doRedirect = (request, response, requestState, newOptions) => Promise.try(function() {
|
||||
if (!request.responseOptions.keepRedirectResponses) {
|
||||
response.pipe(devNull()); // Let the response stream drain out...
|
||||
}
|
||||
|
||||
requestState.redirectHistory.push(response);
|
||||
return bhttpAPI._doRequest(urlUtil.resolve(request.url, response.headers["location"]), newOptions, requestState);
|
||||
});
|
||||
|
||||
const createCookieJar = function(jar) {
|
||||
// Creates a cookie jar wrapper with a simplified API.
|
||||
return {
|
||||
set(cookie, url) {
|
||||
return new Promise((resolve, reject) => {
|
||||
return this.jar.setCookie(cookie, url, function(err, cookie) {
|
||||
if (err) { return reject(err); } else { return resolve(cookie); }
|
||||
});
|
||||
});
|
||||
},
|
||||
get(url) {
|
||||
return new Promise((resolve, reject) => {
|
||||
return this.jar.getCookieString(url, function(err, cookies) {
|
||||
if (err) { return reject(err); } else { return resolve(cookies); }
|
||||
});
|
||||
});
|
||||
},
|
||||
jar
|
||||
};
|
||||
};
|
||||
|
||||
// The exposed API
|
||||
|
||||
var bhttpAPI = {
|
||||
head(url, options, callback) {
|
||||
if (options == null) { options = {}; }
|
||||
options.method = "head";
|
||||
return this.request(url, options, callback);
|
||||
},
|
||||
get(url, options, callback) {
|
||||
if (options == null) { options = {}; }
|
||||
options.method = "get";
|
||||
return this.request(url, options, callback);
|
||||
},
|
||||
post(url, data, options, callback) {
|
||||
if (options == null) { options = {}; }
|
||||
options.method = "post";
|
||||
return doPayloadRequest.bind(this)(url, data, options, callback);
|
||||
},
|
||||
put(url, data, options, callback) {
|
||||
if (options == null) { options = {}; }
|
||||
options.method = "put";
|
||||
return doPayloadRequest.bind(this)(url, data, options, callback);
|
||||
},
|
||||
patch(url, data, options, callback) {
|
||||
if (options == null) { options = {}; }
|
||||
options.method = "patch";
|
||||
return doPayloadRequest.bind(this)(url, data, options, callback);
|
||||
},
|
||||
delete(url, options, callback) {
|
||||
if (options == null) { options = {}; }
|
||||
options.method = "delete";
|
||||
return this.request(url, options, callback);
|
||||
},
|
||||
request(url, options, callback) {
|
||||
if (options == null) { options = {}; }
|
||||
return this._doRequest(url, options).nodeify(callback);
|
||||
},
|
||||
_doRequest(url, options, requestState) {
|
||||
// This is split from the `request` method, so that the user doesn't have to pass in `undefined` for the `requestState` when they want to specify a `callback`.
|
||||
return Promise.try(() => {
|
||||
const request = {url, options: shallowClone(options)};
|
||||
const response = null;
|
||||
if (requestState == null) { requestState = {originalOptions: shallowClone(options), redirectHistory: []}; }
|
||||
if (requestState.sessionOptions == null) { requestState.sessionOptions = this._sessionOptions != null ? this._sessionOptions : {}; }
|
||||
|
||||
return prepareRequest(request, response, requestState);
|
||||
}).spread((request, response, requestState) => {
|
||||
if (request.responseOptions.justPrepare) {
|
||||
return Promise.resolve([request, response, requestState]);
|
||||
} else {
|
||||
return Promise.try(() => {
|
||||
return bhttpAPI.executeRequest(request, response, requestState);
|
||||
}).spread((request, response, _requestState) => {
|
||||
// The user likely only wants the response.
|
||||
return Promise.resolve(response);
|
||||
});
|
||||
}
|
||||
});
|
||||
},
|
||||
executeRequest(request, response, requestState) {
|
||||
// Executes a pre-configured request.
|
||||
return Promise.try(() => makeRequest(request, response, requestState)).spread((request, response, requestState) => processResponse(request, response, requestState));
|
||||
},
|
||||
session(options) {
|
||||
if (options == null) { options = {}; }
|
||||
options = shallowClone(options);
|
||||
const session = {};
|
||||
|
||||
for (let key in this) {
|
||||
let value = this[key];
|
||||
if (value instanceof Function) {
|
||||
value = value.bind(session);
|
||||
}
|
||||
session[key] = value;
|
||||
}
|
||||
|
||||
if ((options.cookieJar == null)) {
|
||||
options.cookieJar = createCookieJar(new toughCookie.CookieJar());
|
||||
} else if (options.cookieJar === false) {
|
||||
delete options.cookieJar;
|
||||
} else {
|
||||
// Assume we've gotten a cookie jar.
|
||||
options.cookieJar = createCookieJar(options.cookieJar);
|
||||
}
|
||||
|
||||
session._sessionOptions = options;
|
||||
|
||||
return session;
|
||||
},
|
||||
wrapStream(stream, options) {
|
||||
// This is a method for wrapping a stream in an object that also contains metadata.
|
||||
return {
|
||||
_bhttpStreamWrapper: true,
|
||||
stream,
|
||||
options
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
extend(bhttpAPI, bhttpErrors);
|
||||
|
||||
module.exports = bhttpAPI;
|
||||
|
||||
// That's all, folks!
|
||||
@ -0,0 +1,14 @@
|
||||
fs = require "fs"
|
||||
stream = require "stream"
|
||||
|
||||
transform = new stream.Transform(highWaterMark: 200 * 1024)
|
||||
transform._transform = (chunk, encoding, callback) ->
|
||||
console.log "Got chunk, length", chunk.length
|
||||
@push chunk
|
||||
callback()
|
||||
|
||||
console.log "Starting stream..."
|
||||
fs.createReadStream "./lower.txt"
|
||||
.pipe transform
|
||||
.on "end", ->
|
||||
console.log "Done!"
|
||||
@ -0,0 +1,36 @@
|
||||
var net = require('net');
|
||||
var stream = require("stream");
|
||||
var util = require('util');
|
||||
|
||||
function MyTransformStream(options) {
|
||||
stream.Transform.call(this, options);
|
||||
}
|
||||
util.inherits(MyTransformStream, stream.Transform);
|
||||
|
||||
MyTransformStream.prototype._transform = function _transform(chunk, encoding, callback) {
|
||||
console.log("Got chunk, length", chunk.length);
|
||||
this.push(chunk);
|
||||
return callback();
|
||||
};
|
||||
|
||||
var transformStream = new MyTransformStream({
|
||||
highWaterMark: 200 * 102
|
||||
});
|
||||
|
||||
transformStream.on('end', function onEnd() {
|
||||
console.log('Got end on transform stream!');
|
||||
});
|
||||
|
||||
// Start reading on the transform stream so that
|
||||
// the pipeline is kept flowing
|
||||
//transformStream.on('data', function() {});
|
||||
|
||||
var server = net.createServer(function onConnection(socket) {
|
||||
// Make sure that libuv's loop is not kept alive by the server's handle
|
||||
server.unref();
|
||||
socket.pipe(transformStream);
|
||||
});
|
||||
|
||||
server.listen(4242, function() {
|
||||
console.log('server listening...');
|
||||
});
|
||||
Loading…
Reference in New Issue