diff --git a/lib/internal/http2/core.js b/lib/internal/http2/core.js index c2ade7942951bb..6ce633092bca4b 100644 --- a/lib/internal/http2/core.js +++ b/lib/internal/http2/core.js @@ -60,6 +60,8 @@ const { owner_symbol, }, } = require('internal/async_hooks'); +const { AsyncResource } = require('async_hooks'); + const { AbortError, aggregateTwoErrors, @@ -241,6 +243,7 @@ const kPendingRequestCalls = Symbol('kPendingRequestCalls'); const kProceed = Symbol('proceed'); const kProtocol = Symbol('protocol'); const kRemoteSettings = Symbol('remote-settings'); +const kRequestAsyncResource = Symbol('requestAsyncResource'); const kSelectPadding = Symbol('select-padding'); const kSentHeaders = Symbol('sent-headers'); const kSentTrailers = Symbol('sent-trailers'); @@ -408,7 +411,11 @@ function onSessionHeaders(handle, id, cat, flags, headers, sensitiveHeaders) { originSet.delete(stream[kOrigin]); } debugStream(id, type, "emitting stream '%s' event", event); - process.nextTick(emit, stream, event, obj, flags, headers); + const reqAsync = stream[kRequestAsyncResource]; + if (reqAsync) + reqAsync.runInAsyncScope(process.nextTick, null, emit, stream, event, obj, flags, headers); + else + process.nextTick(emit, stream, event, obj, flags, headers); } if (endOfStream) { stream.push(null); @@ -1797,6 +1804,8 @@ class ClientHttp2Session extends Http2Session { stream[kSentHeaders] = headers; stream[kOrigin] = `${headers[HTTP2_HEADER_SCHEME]}://` + `${getAuthority(headers)}`; + const reqAsync = new AsyncResource('PendingRequest'); + stream[kRequestAsyncResource] = reqAsync; // Close the writable side of the stream if options.endStream is set. if (options.endStream) @@ -1819,7 +1828,7 @@ class ClientHttp2Session extends Http2Session { } } - const onConnect = requestOnConnect.bind(stream, headersList, options); + const onConnect = reqAsync.bind(requestOnConnect.bind(stream, headersList, options)); if (this.connecting) { if (this[kPendingRequestCalls] !== null) { this[kPendingRequestCalls].push(onConnect); diff --git a/test/parallel/test-http2-async-local-storage.js b/test/parallel/test-http2-async-local-storage.js new file mode 100644 index 00000000000000..699285221f847e --- /dev/null +++ b/test/parallel/test-http2-async-local-storage.js @@ -0,0 +1,55 @@ +'use strict'; + +const common = require('../common'); +if (!common.hasCrypto) + common.skip('missing crypto'); +const assert = require('assert'); +const http2 = require('http2'); +const async_hooks = require('async_hooks'); + +const storage = new async_hooks.AsyncLocalStorage(); + +const { + HTTP2_HEADER_CONTENT_TYPE, + HTTP2_HEADER_PATH, + HTTP2_HEADER_STATUS, +} = http2.constants; + +const server = http2.createServer(); +server.on('stream', (stream) => { + stream.respond({ + [HTTP2_HEADER_CONTENT_TYPE]: 'text/plain; charset=utf-8', + [HTTP2_HEADER_STATUS]: 200 + }); + stream.on('error', common.mustNotCall()); + stream.end('data'); +}); + +server.listen(0, async () => { + const client = storage.run({ id: 0 }, () => http2.connect(`http://localhost:${server.address().port}`)); + + async function doReq(id) { + const req = client.request({ [HTTP2_HEADER_PATH]: '/' }); + + req.on('response', common.mustCall((headers) => { + assert.strictEqual(headers[HTTP2_HEADER_STATUS], 200); + assert.strictEqual(id, storage.getStore().id); + })); + req.on('data', common.mustCall((data) => { + assert.strictEqual(data.toString(), 'data'); + assert.strictEqual(id, storage.getStore().id); + })); + req.on('end', common.mustCall(() => { + assert.strictEqual(id, storage.getStore().id); + server.close(); + client.close(); + })); + } + + function doReqWith(id) { + storage.run({ id }, () => doReq(id)); + } + + doReqWith(1); + doReqWith(2); +});