diff options
Diffstat (limited to 'node_modules/stream-http/lib/response.js')
-rw-r--r-- | node_modules/stream-http/lib/response.js | 56 |
1 files changed, 49 insertions, 7 deletions
diff --git a/node_modules/stream-http/lib/response.js b/node_modules/stream-http/lib/response.js index e5d057d24..17d1fb716 100644 --- a/node_modules/stream-http/lib/response.js +++ b/node_modules/stream-http/lib/response.js @@ -10,7 +10,7 @@ var rStates = exports.readyStates = { DONE: 4 } -var IncomingMessage = exports.IncomingMessage = function (xhr, response, mode) { +var IncomingMessage = exports.IncomingMessage = function (xhr, response, mode, fetchTimer) { var self = this stream.Readable.call(self) @@ -35,30 +35,64 @@ var IncomingMessage = exports.IncomingMessage = function (xhr, response, mode) { self.statusCode = response.status self.statusMessage = response.statusText - response.headers.forEach(function(header, key){ + response.headers.forEach(function (header, key){ self.headers[key.toLowerCase()] = header self.rawHeaders.push(key, header) }) + if (capability.writableStream) { + var writable = new WritableStream({ + write: function (chunk) { + return new Promise(function (resolve, reject) { + if (self._destroyed) { + reject() + } else if(self.push(new Buffer(chunk))) { + resolve() + } else { + self._resumeFetch = resolve + } + }) + }, + close: function () { + global.clearTimeout(fetchTimer) + if (!self._destroyed) + self.push(null) + }, + abort: function (err) { + if (!self._destroyed) + self.emit('error', err) + } + }) - // TODO: this doesn't respect backpressure. Once WritableStream is available, this can be fixed + try { + response.body.pipeTo(writable).catch(function (err) { + global.clearTimeout(fetchTimer) + if (!self._destroyed) + self.emit('error', err) + }) + return + } catch (e) {} // pipeTo method isn't defined. Can't find a better way to feature test this + } + // fallback for when writableStream or pipeTo aren't available var reader = response.body.getReader() function read () { reader.read().then(function (result) { if (self._destroyed) return if (result.done) { + global.clearTimeout(fetchTimer) self.push(null) return } self.push(new Buffer(result.value)) read() - }).catch(function(err) { - self.emit('error', err) + }).catch(function (err) { + global.clearTimeout(fetchTimer) + if (!self._destroyed) + self.emit('error', err) }) } read() - } else { self._xhr = xhr self._pos = 0 @@ -102,7 +136,15 @@ var IncomingMessage = exports.IncomingMessage = function (xhr, response, mode) { inherits(IncomingMessage, stream.Readable) -IncomingMessage.prototype._read = function () {} +IncomingMessage.prototype._read = function () { + var self = this + + var resolve = self._resumeFetch + if (resolve) { + self._resumeFetch = null + resolve() + } +} IncomingMessage.prototype._onXHRProgress = function () { var self = this |