Catch body errors before next tick happens

source/create.ts

@@ -1,3 +1,4 @@
+import {Readable} from 'stream';
 import {Merge} from 'type-fest';
 import is from '@sindresorhus/is';
 import asPromise, {createRejection} from './as-promise';
@@ -99,7 +100,29 @@ const aliases: readonly HTTPAlias[] = [
 	'delete'
 ];
 
-export const defaultHandler: HandlerFunction = (options, next) => next(options);
+export const defaultHandler: HandlerFunction = (options, next) => {
+	const result = next(options);
+
+	if (is.nodeStream(options.body)) {
+		const errorHandler = (error: Error): void => {
+			if (options.isStream) {
+				(result as ProxyStream).destroy(error);
+				return;
+			}
+
+			(result as CancelableRequest<unknown>).cancel(error.message);
+		};
+
+		options.body.once('error', errorHandler);
+
+		// @ts-ignore Each member of the union type '...' has signatures, but none of those signatures are compatible with each other
+		result.on('request', () => {
+			(options.body as Readable).off('error', errorHandler);
+		});
+	}
+
+	return result;
+};
 
 const create = (defaults: Defaults): Got => {
 	// Proxy properties from next handlers
@@ -167,10 +190,7 @@ const create = (defaults: Defaults): Got => {
 		}
 
 		handlers = handlers.filter(handler => handler !== defaultHandler);
-
-		if (handlers.length === 0) {
-			handlers.push(defaultHandler);
-		}
+		handlers.push(defaultHandler);
 
 		return create({
 			options: mergeOptions(...optionsArray) as DefaultOptions,

test/post.ts

@@ -1,7 +1,10 @@
 import {promisify} from 'util';
 import stream = require('stream');
+import fs = require('fs');
 import test from 'ava';
 import {Handler} from 'express';
+import delay = require('delay');
+import getStream = require('get-stream');
 import toReadableStream = require('to-readable-stream');
 import got from '../source';
 import withServer from './helpers/with-server';
@@ -271,3 +274,29 @@ test('DELETE method sends plain objects as JSON', withServer, async (t, server,
 	});
 	t.deepEqual(body, {such: 'wow'});
 });
+
+test('catches body errors before calling pipeline() - promise', withServer, async (t, server, got) => {
+	server.post('/', defaultEndpoint);
+
+	await t.throwsAsync(got.post({
+		body: fs.createReadStream('./file-that-does-not-exist.txt')
+	}), {
+		message: /ENOENT: no such file or directory/
+	});
+
+	// Wait for unhandled errors
+	await delay(100);
+});
+
+test('catches body errors before calling pipeline() - stream', withServer, async (t, server, got) => {
+	server.post('/', defaultEndpoint);
+
+	await t.throwsAsync(getStream(got.stream.post({
+		body: fs.createReadStream('./file-that-does-not-exist.txt')
+	})), {
+		message: /ENOENT: no such file or directory/
+	});
+
+	// Wait for unhandled errors
+	await delay(100);
+});