From 8d60f5158f92353f43a95158ec4b6ecbf0313a22 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gera=20G=2E=20G=C3=BCiles?= Date: Wed, 28 Oct 2026 16:04:01 +0100 Subject: [PATCH] Added AbortSignal cancellation to REQUEST and RESTBuilder --- README.md | 23 ++ builders.js | 7 +- tests/common/request-cancellation.js | 506 +++++++++++++++++++++++++++ tests/run.js | 1 + utils.js | 152 +++++++- 5 files changed, 677 insertions(+), 12 deletions(-) create mode 100644 tests/common/request-cancellation.js diff --git a/README.md b/README.md index fd0777b..5ae1948 100755 --- a/README.md +++ b/README.md @@ -15,6 +15,29 @@ __Total.js framework v5__ is the framework for Node.js platform written in `pure $ npm install total5 ``` +## Cancelling outgoing HTTP requests + +`REQUEST(opt)` accepts an optional standard `AbortSignal` as `opt.signal`. `RESTBuilder` passes it through with the chainable `.signal(signal)` method: + +```js +var controller = new AbortController(); +var pending = REQUEST({ url: 'https://example.com/data', signal: controller.signal }); + +// Or use RESTBuilder: +var builder = RESTBuilder.GET('https://example.com/data').signal(controller.signal); +var response = builder.promise(); + +controller.abort(); +``` + +Cancellation rejects a pending promise, or completes its callback once, with an error named `AbortError` and code `ABORT_ERR`. `error.cause` contains `signal.reason`. The existing callback/promise return values are unchanged. Configure the builder's signal before calling `.exec()`, `.promise()` or `.stream()`. + +An already-aborted signal prevents network I/O. Cancelling during DNS resolution prevents a later connection; it does not claim to cancel the operating system's DNS operation. Cancellation closes the current request and its owned upload/download/decompression streams, including requests made after redirects and HTTPS proxy setup. Shared keep-alive agents are not destroyed. One signal can cancel several requests. + +For `custom: true` / `.stream()`, the callback is delivered once when headers arrive. The signal stays active until that response stream ends or closes. Aborting after headers destroys the stream with `AbortError`; consume its error event as with other Node.js streams. It does not call the already-delivered callback a second time. Completing an ordinary response or closing a custom stream removes its abort listener. Existing `.timeout()` behavior is unchanged; use `AbortSignal.timeout(ms)` if an absolute deadline must also cover the response body. + +Run the focused tests with `node --test tests/common/request-cancellation.js`, or use `npm test` to include them in the framework suite. + ## Social networks Please support the framework on social networks. diff --git a/builders.js b/builders.js index 9dd7c03..0d9877b 100755 --- a/builders.js +++ b/builders.js @@ -670,6 +670,11 @@ RESTP.file = function(name, filename, buffer) { return self; }; +RESTP.signal = function(signal) { + this.options.signal = signal; + return this; +}; + RESTP.timeout = function(number) { this.options.timeout = number; return this; @@ -969,7 +974,7 @@ RESTP.exec = function(callback) { let key; - if (self.$expire && !self.$nocache) { + if (self.$expire && !self.$nocache && !self.options.signal?.aborted) { key = 'restbuilder' + ((self.options.url || '') + (self.options.socketpath || '') + (self.options.path || '') + (self.options.body || '')).hash(true); let data = F.cache.read(key); if (data) { diff --git a/tests/common/request-cancellation.js b/tests/common/request-cancellation.js new file mode 100644 index 0000000..6db7959 --- /dev/null +++ b/tests/common/request-cancellation.js @@ -0,0 +1,506 @@ +'use strict'; + +const Test = require('node:test'); +const Assert = require('node:assert/strict'); +const Http = require('node:http'); +const Net = require('node:net'); +const Events = require('node:events'); +const Zlib = require('node:zlib'); +require('../../index'); + +var server; +var base; +var opened; +var disconnected; +var paths = []; + +function gate(name) { + var resolve; + var promise = new Promise(function(done) { + resolve = done; + }); + return { + resolve: resolve, + wait: function() { + var timer; + return Promise.race([promise, new Promise((resolve, reject) => timer = setTimeout(() => reject(new Error('Test event timed out: ' + name)), 3000))]).finally(() => clearTimeout(timer)); + } + }; +} + +function aborted(err) { + return err instanceof Error && err.name === 'AbortError' && err.code === 'ABORT_ERR'; +} + +function pause(ms) { + return new Promise(resolve => setTimeout(resolve, ms)); +} + +Test.before(async function() { + server = Http.createServer(function(req, res) { + paths.push(req.url); + res.once('close', function() { + disconnected && disconnected.resolve(); + }); + if (req.url === '/slow-ok') { + opened.resolve(); + var delayed = setTimeout(function() { + res.writeHead(200); + res.end('complete'); + }, 50); + res.once('close', () => clearTimeout(delayed)); + return; + } + if (req.url === '/redirect') { + res.writeHead(302, { location: '/hang' }); + return res.end(); + } + if (req.url === '/redirect-dns') { + res.writeHead(302, { location: base.replace('127.0.0.1', 'redirect.test') + '/hang' }); + return res.end(); + } + if (req.url === '/hang' || req.url === '/upload') { + opened && opened.resolve(); + return; + } + if (req.url === '/body' || req.url === '/gzip') { + var gzip = req.url === '/gzip' ? Zlib.createGzip() : null; + res.writeHead(200, { 'content-type': 'text/plain', 'content-encoding': gzip ? 'gzip' : 'identity' }); + if (gzip) + gzip.pipe(res); + var output = gzip || res; + output.write('first chunk'); + if (gzip) + gzip.flush(); + var timer = setInterval(function() { + output.write('next chunk'); + gzip && gzip.flush(); + }, 10); + res.once('close', function() { + clearInterval(timer); + gzip && gzip.destroy(); + }); + return; + } + res.writeHead(200, { 'content-type': 'application/json' }); + res.end('{"success":true}'); + }); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + base = 'http://127.0.0.1:' + server.address().port; +}); + +Test.beforeEach(function() { + opened = gate(); + disconnected = gate(); + paths = []; +}); + +Test.after(async function() { + server.closeAllConnections(); + await new Promise(resolve => server.close(resolve)); + clearInterval(F.internal.interval); +}); + +Test.test('REQUEST keeps existing promise and callback returns; successful signals are detached', async function() { + var control = new AbortController(); + var response = await REQUEST({ url: base + '/ok', signal: control.signal }); + Assert.equal(response.body, '{"success":true}'); + Assert.equal(Events.getEventListeners(control.signal, 'abort').length, 0); + control.abort(); + await new Promise(function(resolve, reject) { + var result = REQUEST({ url: base + '/ok' }, function(err, value) { + err ? reject(err) : resolve(value); + }); + Assert.equal(result, undefined); + }); + Assert.equal((await REQUEST({ url: base + '/ok' })).status, 200); +}); + +Test.test('a signal already aborted rejects without sending or resolving DNS', async function() { + var control = new AbortController(); + var reason = new Error('Stopped by the caller'); + control.abort(reason); + await Assert.rejects(REQUEST({ url: base + '/ok', signal: control.signal }), function(err) { + return aborted(err) && err.cause === reason; + }); + Assert.deepEqual(paths, []); + Assert.equal(Events.getEventListeners(control.signal, 'abort').length, 0); +}); + +Test.test('abort before headers physically closes the request and finishes the callback once', async function() { + var control = new AbortController(); + var calls = 0; + var completed = new Promise(function(resolve, reject) { + REQUEST({ url: base + '/hang', signal: control.signal }, function(err, response) { + calls++; + try { + Assert.ok(aborted(err)); + Assert.equal(response.canceled, true); + resolve(); + } catch (error) { + reject(error); + } + }); + }); + await opened.wait(); + control.abort(); + control.abort(); + await completed; + await disconnected.wait(); + await pause(20); + Assert.equal(calls, 1); + Assert.equal(Events.getEventListeners(control.signal, 'abort').length, 0); +}); + +Test.test('abort during delayed DNS prevents a late connection', async function() { + var original = F.TUtils.resolve; + var control = new AbortController(); + var pending; + F.TUtils.resolve = function(url, callback, options) { + pending = function() { + callback(null, F.TUtils.parseURI(url), options, []); + }; + }; + try { + var request = REQUEST({ url: base.replace('127.0.0.1', 'delayed.test') + '/ok', resolve: true, signal: control.signal }); + var rejected = Assert.rejects(request, aborted); + rejected.catch(NOOP); + Assert.equal(typeof pending, 'function'); + control.abort(); + await rejected; + pending(); + await pause(20); + Assert.deepEqual(paths, []); + } finally { + F.TUtils.resolve = original; + } +}); + +for (var path of ['/body', '/gzip']) { + (function(path) { + Test.test('abort after first decoded data closes ' + path, async function() { + var control = new AbortController(); + var request = REQUEST({ url: base + path, signal: control.signal, ondata: function() { control.abort(); } }); + await Assert.rejects(request, aborted); + await disconnected.wait(); + Assert.equal(Events.getEventListeners(control.signal, 'abort').length, 0); + }); + })(path); +} + +Test.test('custom streams remain abortable after callback delivery without a second callback', async function() { + var control = new AbortController(); + var ready = gate(); + var failure = gate(); + var calls = 0; + REQUEST({ url: base + '/body', custom: true, signal: control.signal }, function(err, response) { + calls++; + Assert.ifError(err); + response.stream.once('error', failure.resolve); + response.stream.once('data', function() { + control.abort(); + }); + ready.resolve(response.stream); + }); + var stream = await ready.wait(); + Assert.ok(aborted(await failure.wait())); + await disconnected.wait(); + Assert.equal(stream.destroyed, true); + Assert.equal(calls, 1); + Assert.equal(Events.getEventListeners(control.signal, 'abort').length, 0); +}); + +Test.test('one signal cancels concurrent requests, including a redirected request', async function() { + var control = new AbortController(); + var first = REQUEST({ url: base + '/hang', signal: control.signal }); + var second = REQUEST({ url: base + '/redirect', signal: control.signal }); + var rejected = Promise.all([Assert.rejects(first, aborted), Assert.rejects(second, aborted)]); + rejected.catch(NOOP); + while (paths.filter(path => path === '/hang').length < 2) + await pause(5); + control.abort(); + await rejected; + Assert.ok(paths.includes('/redirect')); + Assert.equal(Events.getEventListeners(control.signal, 'abort').length, 0); +}); + +Test.test('a multipart upload closes its owned file stream when aborted', async function() { + var original = F.Fs.createReadStream; + var file; + var control = new AbortController(); + F.Fs.createReadStream = function() { + file = new F.Stream.Readable({ read: function() { this.push(Buffer.alloc(1024)); } }); + return file; + }; + try { + var request = REQUEST({ url: base + '/upload', method: 'POST', files: [{ name: 'file', filename: 'audio.wav', path: '/synthetic/audio.wav' }], signal: control.signal }); + var rejected = Assert.rejects(request, aborted); + rejected.catch(NOOP); + await opened.wait(); + control.abort(); + await rejected; + Assert.equal(file.destroyed, true); + await disconnected.wait(); + } finally { + F.Fs.createReadStream = original; + } +}); + +Test.test('cancellation closes an HTTPS request during TLS handshake', async function() { + var accepted = gate('accepted'); + var closed = gate('closed'); + var peer; + var tls = Net.createServer(function(socket) { + peer = socket; + socket.on('data', NOOP); + socket.once('close', closed.resolve); + accepted.resolve(); + }); + await new Promise(resolve => tls.listen(0, '127.0.0.1', resolve)); + try { + var control = new AbortController(); + var request = REQUEST({ url: 'https://127.0.0.1:' + tls.address().port, signal: control.signal }); + var rejected = Assert.rejects(request, aborted); + rejected.catch(NOOP); + await accepted.wait(); + control.abort(); + await rejected; + await closed.wait(); + } finally { + peer && peer.destroy(); + await new Promise(resolve => tls.close(resolve)); + } +}); + +Test.test('abort closes a pending HTTPS proxy CONNECT tunnel', async function() { + var accepted = gate('accepted'); + var closed = gate('closed'); + var peer; + var proxy = Http.createServer(); + proxy.on('connect', function(req, socket) { + peer = socket; + socket.on('data', NOOP); + socket.once('end', () => socket.end()); + socket.on('error', NOOP); + socket.once('close', closed.resolve); + accepted.resolve(); + }); + await new Promise(resolve => proxy.listen(0, '127.0.0.1', resolve)); + try { + var control = new AbortController(); + var request = REQUEST({ url: 'https://destination.test/ok', proxy: 'http://127.0.0.1:' + proxy.address().port, signal: control.signal }); + var rejected = Assert.rejects(request, aborted); + rejected.catch(NOOP); + await accepted.wait(); + control.abort(); + await rejected; + await closed.wait(); + } finally { + peer && peer.destroy(); + await new Promise(resolve => proxy.close(resolve)); + } +}); + +Test.test('signal listeners are detached on network failure and native timeout', async function() { + var control = new AbortController(); + await Assert.rejects(REQUEST({ url: base + '/hang', timeout: 30, signal: control.signal })); + Assert.equal(Events.getEventListeners(control.signal, 'abort').length, 0); + var stopped = Http.createServer(); + await new Promise(resolve => stopped.listen(0, '127.0.0.1', resolve)); + var port = stopped.address().port; + await new Promise(resolve => stopped.close(resolve)); + await Assert.rejects(REQUEST({ url: 'http://127.0.0.1:' + port, signal: control.signal })); + Assert.equal(Events.getEventListeners(control.signal, 'abort').length, 0); +}); + +Test.test('custom stream completion releases the signal without closing shared connections', async function() { + var control = new AbortController(); + var response = await REQUEST({ url: base + '/ok', custom: true, signal: control.signal, keepalive: true }); + var size = 0; + for await (var chunk of response.stream) + size += chunk.length; + Assert.ok(size > 0); + Assert.equal(Events.getEventListeners(control.signal, 'abort').length, 0); + control.abort(); + Assert.equal((await REQUEST({ url: base + '/ok', keepalive: true })).status, 200); +}); + +Test.test('abort from inside the stream callback does not deliver that callback twice', async function() { + var control = new AbortController(); + var calls = 0; + var completed = gate(); + REQUEST({ url: base + '/body', custom: true, signal: control.signal }, function(err, response) { + calls++; + Assert.ifError(err); + response.stream.once('error', completed.resolve); + control.abort(); + }); + Assert.ok(aborted(await completed.wait())); + await disconnected.wait(); + Assert.equal(calls, 1); +}); + +Test.test('abort during redirect DNS does not open the destination', async function() { + var original = F.TUtils.resolve; + var control = new AbortController(); + var waiting = gate(); + var pending; + F.TUtils.resolve = function(url, callback, options) { + if (typeof url === 'string') { + options.uri.hostname = '127.0.0.1'; + callback(null, { host: '127.0.0.1' }, options, []); + } else { + pending = () => callback(null, url, options, []); + waiting.resolve(); + } + }; + try { + var request = REQUEST({ url: base.replace('127.0.0.1', 'origin.test') + '/redirect-dns', resolve: true, signal: control.signal }); + var rejected = Assert.rejects(request, aborted); + rejected.catch(NOOP); + await waiting.wait(); + control.abort(); + await rejected; + pending(); + await pause(20); + Assert.deepEqual(paths, ['/redirect-dns']); + } finally { + F.TUtils.resolve = original; + } +}); + +Test.test('multipart cancellation also stops a remote file download', async function() { + var control = new AbortController(); + var request = REQUEST({ url: base + '/upload', method: 'POST', files: [{ name: 'file', filename: 'audio.wav', url: base + '/body' }], signal: control.signal }); + var rejected = Assert.rejects(request, aborted); + rejected.catch(NOOP); + while (!paths.includes('/body')) + await pause(5); + control.abort(); + await rejected; + Assert.equal(Events.getEventListeners(control.signal, 'abort').length, 0); +}); + +Test.test('RESTBuilder.signal is chainable and cancels promise mode before headers', async function() { + var control = new AbortController(); + var builder = RESTBuilder.GET(base + '/hang'); + Assert.equal(builder.signal(control.signal), builder); + var request = builder.promise(); + var rejected = Assert.rejects(request, aborted); + rejected.catch(NOOP); + await opened.wait(); + control.abort(); + await rejected; + await disconnected.wait(); +}); + +Test.test('RESTBuilder callbacks receive one AbortError and preserve response metadata', async function() { + var control = new AbortController(); + var calls = 0; + var completed = gate(); + RESTBuilder.GET(base + '/hang').signal(control.signal).exec(function(err, response, output) { + calls++; + completed.resolve({ err: err, output: output }); + }); + await opened.wait(); + control.abort(); + var result = await completed.wait(); + Assert.ok(aborted(result.err)); + Assert.equal(result.output.status, 0); + await disconnected.wait(); + Assert.equal(calls, 1); +}); + +Test.test('RESTBuilder streams remain cancelable after receiving headers', async function() { + var control = new AbortController(); + var completed = gate(); + var failure = gate(); + RESTBuilder.GET(base + '/body').signal(control.signal).stream(function(err, output) { + Assert.ifError(err); + output.stream.once('error', failure.resolve); + output.stream.once('data', function() { control.abort(); }); + completed.resolve(); + }); + await completed.wait(); + Assert.ok(aborted(await failure.wait())); + await disconnected.wait(); +}); + +Test.test('invalid URL, body and request options do not leak signal listeners', async function() { + for (var options of [{ url: 'invalid url' }, { url: base, method: 'POST', body: {} }, { url: base, method: 'INVALID METHOD' }]) { + var control = new AbortController(); + await Assert.rejects(REQUEST({ ...options, signal: control.signal })); + Assert.equal(Events.getEventListeners(control.signal, 'abort').length, 0); + } +}); + +Test.test('aborting one request leaves another active request on the shared keep-alive agent intact', async function() { + var control = new AbortController(); + var first = REQUEST({ url: base + '/hang', keepalive: true, signal: control.signal }); + var rejected = Assert.rejects(first, aborted); + rejected.catch(NOOP); + await opened.wait(); + opened = gate('second keep-alive request'); + var second = REQUEST({ url: base + '/slow-ok', keepalive: true }); + second.catch(NOOP); + await opened.wait(); + control.abort(); + await rejected; + Assert.equal((await second).body, 'complete'); +}); + +Test.test('abort closes TLS negotiation after an HTTPS proxy CONNECT succeeds', async function() { + var accepted = gate('accepted'); + var closed = gate('closed'); + var peer; + var proxy = Http.createServer(); + proxy.on('connect', function(req, socket) { + peer = socket; + socket.on('data', NOOP); + socket.once('end', () => socket.end()); + socket.on('error', NOOP); + socket.once('close', closed.resolve); + socket.once('data', accepted.resolve); + socket.write('HTTP/1.1 200 Connection Established\r\n\r\n'); + }); + await new Promise(resolve => proxy.listen(0, '127.0.0.1', resolve)); + try { + var control = new AbortController(); + var request = REQUEST({ url: 'https://destination.test/ok', proxy: 'http://127.0.0.1:' + proxy.address().port, signal: control.signal }); + var rejected = Assert.rejects(request, aborted); + rejected.catch(NOOP); + await accepted.wait(); + control.abort(); + await rejected; + await closed.wait(); + } finally { + peer && peer.destroy(); + await new Promise(resolve => proxy.close(resolve)); + } +}); + + +Test.test('AbortSignal.timeout also cancels after the first response byte', async function() { + var signal = AbortSignal.timeout(30); + await Assert.rejects(REQUEST({ url: base + '/body', signal: signal }), function(err) { + return aborted(err) && err.cause?.name === 'TimeoutError'; + }); + await disconnected.wait(); + Assert.equal(Events.getEventListeners(signal, 'abort').length, 0); +}); + +Test.test('RESTBuilder does not return a cached response for an already-aborted signal', async function() { + var original = F.cache.read; + var before = paths.length; + F.cache.read = function() { + return { value: { status: 200, value: { cached: true } } }; + }; + try { + var control = new AbortController(); + control.abort(); + await Assert.rejects(RESTBuilder.GET(base + '/ok').cache('1 minute').signal(control.signal).promise(), aborted); + Assert.equal(paths.length, before); + } finally { + F.cache.read = original; + } +}); diff --git a/tests/run.js b/tests/run.js index 7606e32..3fde13c 100644 --- a/tests/run.js +++ b/tests/run.js @@ -3,6 +3,7 @@ require('../index'); var tests = []; tests.push('aimodel.js'); +tests.push('common/request-cancellation.js'); // tests.push('bundles/run.js'); // tests.push('common/utils.js'); // tests.push('tms/index.js'); diff --git a/utils.js b/utils.js index 6f675f9..cbd3503 100755 --- a/utils.js +++ b/utils.js @@ -590,6 +590,74 @@ function parseProxy(p) { return F.temporary.utils[key] = obj; } +function request_signal_cleanup(options) { + if (options.signal) { + options.signal.removeEventListener('abort', options.abort); + options.signal = null; + } + if (options.signalstream) { + options.signalstream.removeListener('end', options.signalend); + options.signalstream.removeListener('close', options.signalend); + options.signalstream = null; + } + options.request = options.proxyrequest = options.proxysocket = options.uploadstream = options.decoder = options.resstream = null; +} + +function request_signal_abort(options) { + if (!options.signal) + return; + var callback = options.callback; + var err = new Error('The operation was aborted'); + err.name = 'AbortError'; + err.code = 'ABORT_ERR'; + err.cause = options.signal.reason; + + options.callback = null; + options.canceled = options.iserror = true; + options.response.canceled = true; + if (!options.signaldelivered) + options.response.status = 0; + if (options.timeoutid) { + clearTimeout(options.timeoutid); + options.timeoutid = null; + } + + // Destroy only this operation's transports, never a shared keep-alive agent. + for (var stream of [options.request, options.proxyrequest, options.proxysocket, options.uploadstream, options.decoder, options.resstream]) { + if (stream && !stream.destroyed) { + stream.once('error', NOOP); + stream.destroy(err); + } + } + request_signal_cleanup(options); + callback && callback(err, options.response); +} + +function request_signal_init(options, signal) { + options.signal = signal; + var handler = options.callback; + options.callback = function(err, response) { + if (options.signaldelivered) + return; + options.signaldelivered = true; + if (!err && response.stream && options.signal) { + options.signalstream = response.stream; + options.signalend = () => request_signal_cleanup(options); + response.stream.once('end', options.signalend); + response.stream.once('close', options.signalend); + } else + request_signal_cleanup(options); + handler(err, response); + }; + options.abort = () => request_signal_abort(options); + if (options.signal.aborted) { + options.abort(); + return true; + } + options.signal.addEventListener('abort', options.abort, { once: true }); + return false; +} + function _request(opt, callback) { const options = { length: 0, timeout: opt.timeout == false || opt.timeout == 0 ? 0 : (opt.timeout || 8000), encoding: opt.encoding || 'utf8', callback: callback || opt.callback || NOOP, post: true, redirect: 0 }; @@ -628,6 +696,9 @@ function _request(opt, callback) { options.response = opt.response ? opt.response : {}; options.response.body = ''; + if (opt.signal?.aborted && request_signal_init(options, opt.signal)) + return; + options.iserror = false; if (opt.resolve || opt.dnscache) @@ -724,7 +795,7 @@ function _request(opt, callback) { if ((opt.unixsocket && !uri.socketPath) || (!opt.unixsocket && (!uri.hostname || !uri.host))) { options.response.canceled = true; - opt.callback && opt.callback('Invalid hostname', options.response); + options.callback && options.callback('Invalid hostname', options.response); return; } @@ -774,9 +845,12 @@ function _request(opt, callback) { uri.agent = typeof(options.keepalive) === 'boolean' ? KeepAliveHttps : options.keepalive instanceof F.Https.Agent ? options.keepalive : new F.Https.Agent(options.keepalive); } else uri.agent = typeof(options.keepalive) === 'boolean' ? KeepAlive : options.keepalive instanceof F.Http.Agent ? options.keepalive : new F.Http.Agent(options.keepalive); - } else + } else if (!proxy || uri.protocol !== 'https:') uri.agent = null; + if (opt.signal && request_signal_init(options, opt.signal)) + return; + if (proxy) request_call(uri, options); else if (options.resolve) @@ -817,6 +891,10 @@ const PAP = ProxyAgent.prototype; PAP.createConnection = function(pending) { var self = this; self.createSocket(pending, function(socket) { + if (pending.request.destroyed || self.options?.canceled) { + socket.destroy(); + return; + } pending.request.onSocket(socket); }); }; @@ -835,10 +913,19 @@ PAP.createSocket = function(options, callback) { PROXYOPTIONS.headers['Proxy-Authorization'] = proxy._auth; const req = self.request(PROXYOPTIONS); + if (self.options.signal) + self.options.proxyrequest = req; req.setTimeout(10000); req.on('response', proxyagent_response); req.on('connect', function(res, socket) { + if (!self.options || self.options.canceled) { + socket.destroy(); + return; + } + if (self.options.signal) + self.options.proxysocket = socket; + if (res.statusCode === 200) { socket.$req = req; callback(socket); @@ -846,7 +933,6 @@ PAP.createSocket = function(options, callback) { var err = new Error('Proxy could not be established (maybe a problem in auth), code: ' + res.statusCode); err.code = 'ECONNRESET'; req.destroy && req.destroy(); - req = null; self.requests = null; self.options = null; } @@ -857,7 +943,6 @@ PAP.createSocket = function(options, callback) { const e = new Error('Request Proxy "proxy {0} --> target {1}": {2}'.format(PROXYOPTIONS.host + ':' + proxy.port, PROXYOPTIONS.path, err.toString())); e.code = err.code; req.destroy && req.destroy(); - req = null; self.requests = null; self.options = null; }); @@ -876,6 +961,10 @@ PAP.addRequest = function(req, options) { function createSecureSocket(options, callback) { var self = this; PAP.createSocket.call(self, options, function(socket) { + if (!self.options || self.options.canceled) { + socket.destroy(); + return; + } PROXYTLS.servername = self.options.uri.hostname; PROXYTLS.headers = self.options.uri.headers; PROXYTLS.socket = socket; @@ -886,6 +975,9 @@ function createSecureSocket(options, callback) { function request_call(uri, options) { + if (options.canceled) + return; + var opt; if (options.proxy && !options.proxy.tls) { @@ -905,10 +997,29 @@ function request_call(uri, options) { opt = uri; const connection = uri.protocol === 'https:' ? F.Https : F.Http; - const req = opt.method === 'GET' ? connection.get(opt, request_response) : connection.request(opt, request_response); + var req; + + try { + req = opt.method === 'GET' ? connection.get(opt, request_response) : connection.request(opt, request_response); + } catch (err) { + if (!options.signal) + throw err; + var callback = options.callback; + options.callback = null; + callback && callback(err, options.response); + return; + } req.$options = options; req.$uri = uri; + if (options.signal) + options.request = req; + + if (options.canceled) { + req.on('error', NOOP); + req.destroy(); + return; + } if (!options.callback) { req.on('error', NOOP); @@ -945,7 +1056,7 @@ function request_call(uri, options) { try { req.write((options.first ? '' : NEWLINE) + '--' + options.boundary + NEWLINE + 'Content-Disposition: form-data; name="' + key + '"' + NEWLINE + NEWLINE + value); } catch (e) { - request_process_error.apply(req, e); + request_process_error.call(req, e); break; } if (options.first) @@ -958,7 +1069,7 @@ function request_call(uri, options) { }); } else { if (options.opt.compress) - F.Zlib[options.opt.compress](options.body, (err, buffer) => req.end(buffer)); + F.Zlib[options.opt.compress](options.body, (err, buffer) => !options.canceled && req.end(buffer)); else req.end(options.body); } @@ -1032,7 +1143,7 @@ function request_writefile(req, options, file, next) { try { req.write(filedata); } catch (e) { - request_process_error.apply(req, e); + request_process_error.call(req, e); } next(); } else if (isurl) { @@ -1041,12 +1152,20 @@ function request_writefile(req, options, file, next) { opt.method = 'GET'; opt.custom = true; opt.url = filedata; + opt.signal = options.opt.signal; opt.callback = function(err, response) { + if (options.canceled) { + response && response.stream && response.stream.destroy(); + next(); + return; + } if (response && response.stream) { + if (options.signal) + options.uploadstream = response.stream; response.stream.on('close', next); response.stream.pipe(req, STREAMPIPE); } else { - err && request_process_error.apply(req, err); + err && request_process_error.call(req, err); next(); } }; @@ -1054,6 +1173,8 @@ function request_writefile(req, options, file, next) { } else { F.stats.performance.open++; var stream = F.Fs.createReadStream(file.path || file.filename); + if (options.signal) + options.uploadstream = stream; stream.once('close', next); stream.pipe(req, STREAMPIPE); } @@ -1064,6 +1185,14 @@ function request_response(res) { const options = this.$options; const uri = this.$uri; + if (options.canceled) { + res.on('error', NOOP); + res.destroy(); + return; + } + if (options.signal) + options.resstream = res; + res._buffer = null; res._bufferlength = 0; @@ -1253,8 +1382,6 @@ function request_response(res) { return; } - options.timeoutid && res.once('data', request_process_ok); - var encoding = res.headers['content-encoding'] || ''; if (encoding) encoding = encoding.split(',')[0]; @@ -1268,8 +1395,11 @@ function request_response(res) { options.callback && options.callback(null, options.response); options.callback = null; } else { + options.timeoutid && res.once('data', request_process_ok); if (COMPRESS[encoding]) { var zlib = encoding === 'gzip' ? F.Zlib.createGunzip() : F.Zlib.createInflate(); + if (options.signal) + options.decoder = zlib; zlib._buffer = res.buffer; zlib.headers = res.headers; zlib.statusCode = res.statusCode;