fix: Settle `consumeBody` promise when the response closes prematurely (#768)

The stream.finished function is used consolidate the terminal cases.
The writable option must be set to false, since the body stream is
a Duplex stream, but our terminal cases only apply to the Readable side.

src/body.js

@@ -5,7 +5,7 @@
  * Body interface provides common methods for Request and Response
  */
 
-import Stream, {PassThrough} from 'stream';
+import Stream, {finished, PassThrough} from 'stream';
 import {types} from 'util';
 
 import Blob from 'fetch-blob';
@@ -207,18 +207,6 @@ function consumeBody() {
 			}, this.timeout);
 		}
 
-		// Handle stream errors
-		body.on('error', err => {
-			if (isAbortError(err)) {
-				// If the request was aborted, reject with this Error
-				abort = true;
-				reject(err);
-			} else {
-				// Other errors, such as incorrect content-encoding
-				reject(new FetchError(`Invalid response body while trying to fetch ${this.url}: ${err.message}`, 'system', err));
-			}
-		});
-
 		body.on('data', chunk => {
 			if (abort || chunk === null) {
 				return;
@@ -234,18 +222,28 @@ function consumeBody() {
 			accum.push(chunk);
 		});
 
-		body.on('end', () => {
-			if (abort) {
-				return;
-			}
-
+		finished(body, {writable: false}, err => {
 			clearTimeout(resTimeout);
+			if (err) {
+				if (isAbortError(err)) {
+					// If the request was aborted, reject with this Error
+					abort = true;
+					reject(err);
+				} else {
+					// Other errors, such as incorrect content-encoding
+					reject(new FetchError(`Invalid response body while trying to fetch ${this.url}: ${err.message}`, 'system', err));
+				}
+			} else {
+				if (abort) {
+					return;
+				}
 
-			try {
-				resolve(Buffer.concat(accum, accumBytes));
-			} catch (error) {
-				// Handle streams that have accumulated too much data (issue #414)
-				reject(new FetchError(`Could not create Buffer from response body for ${this.url}: ${error.message}`, 'system', error));
+				try {
+					resolve(Buffer.concat(accum, accumBytes));
+				} catch (error) {
+					// Handle streams that have accumulated too much data (issue #414)
+					reject(new FetchError(`Could not create Buffer from response body for ${this.url}: ${error.message}`, 'system', error));
+				}
 			}
 		});
 	});

test/main.js

@@ -595,6 +595,16 @@ describe('node-fetch', () => {
 			.and.have.property('code', 'ECONNRESET');
 	});
 
+	it('should handle network-error partial response', () => {
+		const url = `${base}error/premature`;
+		return fetch(url).then(res => {
+			expect(res.status).to.equal(200);
+			expect(res.ok).to.be.true;
+			return expect(res.text()).to.eventually.be.rejectedWith(Error)
+				.and.have.property('message').includes('Premature close');
+		});
+	});
+
 	it('should handle DNS-error response', () => {
 		const url = 'http://domain.invalid';
 		return expect(fetch(url)).to.eventually.be.rejected

test/utils/server.js

@@ -302,6 +302,14 @@ export default class TestServer {
 			res.destroy();
 		}
 
+		if (p === '/error/premature') {
+			res.writeHead(200, {'content-length': 50});
+			res.write('foo');
+			setTimeout(() => {
+				res.destroy();
+			}, 100);
+		}
+
 		if (p === '/error/json') {
 			res.statusCode = 200;
 			res.setHeader('Content-Type', 'application/json');