Skip to content

Commit 329da15

Browse files
trivikrtrivikr
committed
chore: merge changes in node-http2-handler from #317 
1 parent f9835ae commit 329da15

File tree

7 files changed

+340
-4
lines changed

7 files changed

+340
-4
lines changed

‎packages/node-http-handler/src/index.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1 +1,2 @@
11
export * from "./node-http-handler";
2+
export * from "./node-http2-handler";

‎packages/node-http-handler/src/node-http-handler.ts

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,7 @@
11
import * as https from "https";
22
import * as http from "http";
3-
import { Readable } from "stream";
43
import { buildQueryString } from "@aws-sdk/querystring-builder";
5-
import { HeaderBag, HttpOptions, NodeHttpOptions } from "@aws-sdk/types";
4+
import { HttpOptions, NodeHttpOptions } from "@aws-sdk/types";
65
import { HttpHandler, HttpRequest, HttpResponse } from "@aws-sdk/protocol-http";
76
import { setConnectionTimeout } from "./set-connection-timeout";
87
import { setSocketTimeout } from "./set-socket-timeout";

‎packages/node-http-handler/src/node-http2-handler.spec.ts

Lines changed: 205 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,205 @@
1+
import { NodeHttp2Handler } from "./node-http2-handler";
2+
import { createMockHttp2Server, createResponseFunction } from "./server.mock";
3+
import { AbortController } from "@aws-sdk/abort-controller";
4+
5+
describe("NodeHttp2Handler", () => {
6+
let nodeH2Handler: NodeHttp2Handler;
7+
8+
const protocol = "http:";
9+
const hostname = "localhost";
10+
const port = 45321;
11+
const mockH2Server = createMockHttp2Server().listen(port);
12+
const getMockRequest = () => ({
13+
protocol,
14+
hostname,
15+
port,
16+
method: "GET",
17+
path: "/",
18+
headers: {}
19+
});
20+
21+
const mockResponse = {
22+
statusCode: 200,
23+
headers: {},
24+
body: "test"
25+
};
26+
27+
beforeEach(() => {
28+
nodeH2Handler = new NodeHttp2Handler();
29+
mockH2Server.on("request", createResponseFunction(mockResponse));
30+
});
31+
32+
afterEach(() => {
33+
mockH2Server.removeAllListeners("request");
34+
// @ts-ignore: access private property
35+
const connectionPool = nodeH2Handler.connectionPool;
36+
for (const [, session] of connectionPool) {
37+
session.destroy();
38+
}
39+
connectionPool.clear();
40+
});
41+
42+
afterAll(() => {
43+
mockH2Server.close();
44+
});
45+
46+
describe("connectionPool", () => {
47+
it("is empty on initialization", () => {
48+
// @ts-ignore: access private property
49+
expect(nodeH2Handler.connectionPool.size).toBe(0);
50+
});
51+
52+
it("creates and stores session when request is made", async () => {
53+
await nodeH2Handler.handle(getMockRequest(), {});
54+
55+
// @ts-ignore: access private property
56+
expect(nodeH2Handler.connectionPool.size).toBe(1);
57+
expect(
58+
// @ts-ignore: access private property
59+
nodeH2Handler.connectionPool.get(`${protocol}//${hostname}:${port}`)
60+
).toBeDefined();
61+
});
62+
63+
it("reuses existing session if request is made on same authority again", async () => {
64+
await nodeH2Handler.handle(getMockRequest(), {});
65+
// @ts-ignore: access private property
66+
expect(nodeH2Handler.connectionPool.size).toBe(1);
67+
68+
// @ts-ignore: access private property
69+
const session: ClientHttp2Session = nodeH2Handler.connectionPool.get(
70+
`${protocol}//${hostname}:${port}`
71+
);
72+
const requestSpy = jest.spyOn(session, "request");
73+
74+
await nodeH2Handler.handle(getMockRequest(), {});
75+
// @ts-ignore: access private property
76+
expect(nodeH2Handler.connectionPool.size).toBe(1);
77+
expect(requestSpy.mock.calls.length).toBe(1);
78+
});
79+
80+
it("creates new session if request is made on new authority", async () => {
81+
await nodeH2Handler.handle(getMockRequest(), {});
82+
// @ts-ignore: access private property
83+
expect(nodeH2Handler.connectionPool.size).toBe(1);
84+
85+
const port2 = port + 1;
86+
const mockH2Server2 = createMockHttp2Server().listen(port2);
87+
mockH2Server2.on("request", createResponseFunction(mockResponse));
88+
89+
await nodeH2Handler.handle({ ...getMockRequest(), port: port2 }, {});
90+
// @ts-ignore: access private property
91+
expect(nodeH2Handler.connectionPool.size).toBe(2);
92+
expect(
93+
// @ts-ignore: access private property
94+
nodeH2Handler.connectionPool.get(`${protocol}//${hostname}:${port2}`)
95+
).toBeDefined();
96+
97+
mockH2Server2.close();
98+
});
99+
100+
it("closes and removes session on sessionTimeout", async done => {
101+
const sessionTimeout = 500;
102+
nodeH2Handler = new NodeHttp2Handler({ sessionTimeout });
103+
await nodeH2Handler.handle(getMockRequest(), {});
104+
105+
const authority = `${protocol}//${hostname}:${port}`;
106+
// @ts-ignore: access private property
107+
const session: ClientHttp2Session = nodeH2Handler.connectionPool.get(
108+
authority
109+
);
110+
expect(session.closed).toBe(false);
111+
setTimeout(() => {
112+
expect(session.closed).toBe(true);
113+
// @ts-ignore: access private property
114+
expect(nodeH2Handler.connectionPool.get(authority)).not.toBeDefined();
115+
done();
116+
}, sessionTimeout + 100);
117+
});
118+
});
119+
120+
describe("destroy", () => {
121+
it("destroys sessions and clears connectionPool", async () => {
122+
await nodeH2Handler.handle(getMockRequest(), {});
123+
124+
// @ts-ignore: access private property
125+
const session: ClientHttp2Session = nodeH2Handler.connectionPool.get(
126+
`${protocol}//${hostname}:${port}`
127+
);
128+
129+
// @ts-ignore: access private property
130+
expect(nodeH2Handler.connectionPool.size).toBe(1);
131+
expect(session.destroyed).toBe(false);
132+
nodeH2Handler.destroy();
133+
// @ts-ignore: access private property
134+
expect(nodeH2Handler.connectionPool.size).toBe(0);
135+
expect(session.destroyed).toBe(true);
136+
});
137+
});
138+
139+
describe("abortSignal", () => {
140+
it("will not create session if request already aborted", async () => {
141+
// @ts-ignore: access private property
142+
expect(nodeH2Handler.connectionPool.size).toBe(0);
143+
await expect(
144+
nodeH2Handler.handle(getMockRequest(), {
145+
abortSignal: {
146+
aborted: true
147+
}
148+
})
149+
).rejects.toHaveProperty("name", "AbortError");
150+
// @ts-ignore: access private property
151+
expect(nodeH2Handler.connectionPool.size).toBe(0);
152+
});
153+
154+
it("will not create request on session if request already aborted", async () => {
155+
await nodeH2Handler.handle(getMockRequest(), {});
156+
157+
// @ts-ignore: access private property
158+
const session: ClientHttp2Session = nodeH2Handler.connectionPool.get(
159+
`${protocol}//${hostname}:${port}`
160+
);
161+
const requestSpy = jest.spyOn(session, "request");
162+
163+
await expect(
164+
nodeH2Handler.handle(getMockRequest(), {
165+
abortSignal: {
166+
aborted: true
167+
}
168+
})
169+
).rejects.toHaveProperty("name", "AbortError");
170+
expect(requestSpy.mock.calls.length).toBe(0);
171+
});
172+
173+
it("will close request on session when aborted", async () => {
174+
await nodeH2Handler.handle(getMockRequest(), {});
175+
176+
// @ts-ignore: access private property
177+
const session: ClientHttp2Session = nodeH2Handler.connectionPool.get(
178+
`${protocol}//${hostname}:${port}`
179+
);
180+
const requestSpy = jest.spyOn(session, "request");
181+
182+
const abortController = new AbortController();
183+
// Delay response so that onabort is called earlier
184+
setTimeout(() => {
185+
abortController.abort();
186+
}, 0);
187+
mockH2Server.on(
188+
"request",
189+
async () =>
190+
new Promise(resolve => {
191+
setTimeout(() => {
192+
resolve(createResponseFunction(mockResponse));
193+
}, 1000);
194+
})
195+
);
196+
197+
await expect(
198+
nodeH2Handler.handle(getMockRequest(), {
199+
abortSignal: abortController.signal
200+
})
201+
).rejects.toHaveProperty("name", "AbortError");
202+
expect(requestSpy.mock.calls.length).toBe(1);
203+
});
204+
});
205+
});

‎packages/node-http-handler/src/node-http2-handler.ts

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
import { connect, constants, ClientHttp2Session } from "http2";
2+
3+
import { buildQueryString } from "@aws-sdk/querystring-builder";
4+
import { HttpOptions, NodeHttp2Options } from "@aws-sdk/types";
5+
import { HttpHandler, HttpRequest, HttpResponse } from "@aws-sdk/protocol-http";
6+
7+
import { writeRequestBody } from "./write-request-body";
8+
import { getTransformedHeaders } from "./get-transformed-headers";
9+
10+
export class NodeHttp2Handler implements HttpHandler {
11+
private readonly connectionPool: Map<string, ClientHttp2Session>;
12+
13+
constructor(private readonly http2Options: NodeHttp2Options = {}) {
14+
this.connectionPool = new Map<string, ClientHttp2Session>();
15+
}
16+
17+
destroy(): void {
18+
for (const [_, http2Session] of this.connectionPool) {
19+
http2Session.destroy();
20+
}
21+
this.connectionPool.clear();
22+
}
23+
24+
handle(
25+
request: HttpRequest,
26+
{ abortSignal }: HttpOptions
27+
): Promise<{ response: HttpResponse }> {
28+
return new Promise((resolve, reject) => {
29+
// if the request was already aborted, prevent doing extra work
30+
if (abortSignal && abortSignal.aborted) {
31+
const abortError = new Error("Request aborted");
32+
abortError.name = "AbortError";
33+
reject(abortError);
34+
return;
35+
}
36+
37+
const { hostname, method, port, protocol, path, query } = request;
38+
const queryString = buildQueryString(query || {});
39+
40+
// create the http2 request
41+
const req = this.getSession(
42+
`${protocol}//${hostname}${port ? `:${port}` : ""}`
43+
).request({
44+
...request.headers,
45+
[constants.HTTP2_HEADER_PATH]: queryString
46+
? `${path}?${queryString}`
47+
: path,
48+
[constants.HTTP2_HEADER_METHOD]: method
49+
});
50+
51+
req.on("response", headers => {
52+
const httpResponse = new HttpResponse({
53+
statusCode: headers[":status"] || -1,
54+
headers: getTransformedHeaders(headers),
55+
body: req
56+
});
57+
resolve({ response: httpResponse });
58+
});
59+
60+
req.on("error", reject);
61+
req.on("frameError", reject);
62+
req.on("aborted", reject);
63+
64+
const { requestTimeout } = this.http2Options;
65+
if (requestTimeout) {
66+
req.setTimeout(requestTimeout, () => {
67+
req.close();
68+
const timeoutError = new Error(
69+
`Stream timed out because of no activity for ${requestTimeout} ms`
70+
);
71+
timeoutError.name = "TimeoutError";
72+
reject(timeoutError);
73+
});
74+
}
75+
76+
if (abortSignal) {
77+
abortSignal.onabort = () => {
78+
req.close();
79+
const abortError = new Error("Request aborted");
80+
abortError.name = "AbortError";
81+
reject(abortError);
82+
};
83+
}
84+
85+
writeRequestBody(req, request);
86+
});
87+
}
88+
89+
private getSession(authority: string): ClientHttp2Session {
90+
const connectionPool = this.connectionPool;
91+
const existingSession = connectionPool.get(authority);
92+
if (existingSession) return existingSession;
93+
94+
const newSession = connect(authority);
95+
connectionPool.set(authority, newSession);
96+
97+
const { sessionTimeout } = this.http2Options;
98+
if (sessionTimeout) {
99+
newSession.setTimeout(sessionTimeout, () => {
100+
newSession.close();
101+
connectionPool.delete(authority);
102+
});
103+
}
104+
return newSession;
105+
}
106+
}

‎packages/node-http-handler/src/server.mock.ts

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import {
88
createServer as createHttpsServer,
99
Server as HttpsServer
1010
} from "https";
11+
import { createServer as createHttp2Server, Http2Server } from "http2";
1112
import { readFileSync } from "fs";
1213
import { join } from "path";
1314
import { Readable } from "stream";
@@ -54,3 +55,8 @@ export function createMockHttpServer(): HttpServer {
5455
const server = createHttpServer();
5556
return server;
5657
}
58+
59+
export function createMockHttp2Server(): Http2Server {
60+
const server = createHttp2Server();
61+
return server;
62+
}

‎packages/node-http-handler/src/write-request-body.ts

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,10 @@
11
import { ClientRequest } from "http";
2+
import { ClientHttp2Stream } from "http2";
23
import { Readable } from "stream";
34
import { HttpRequest } from "@aws-sdk/types";
45

56
export function writeRequestBody(
6-
httpRequest: ClientRequest,
7+
httpRequest: ClientRequest | ClientHttp2Stream,
78
request: HttpRequest<Readable>
89
) {
910
const expect = request.headers["Expect"] || request.headers["expect"];
@@ -17,7 +18,7 @@ export function writeRequestBody(
1718
}
1819

1920
function writeBody(
20-
httpRequest: ClientRequest,
21+
httpRequest: ClientRequest | ClientHttp2Stream,
2122
body?: string | ArrayBuffer | ArrayBufferView | Readable
2223
) {
2324
if (body instanceof Readable) {

‎packages/types/src/http.ts

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -143,3 +143,21 @@ export interface NodeHttpOptions {
143143
*/
144144
socketTimeout?: number;
145145
}
146+
147+
/**
148+
* Represents the http2 options that can be passed to a node http2 client.
149+
*/
150+
export interface NodeHttp2Options extends HttpOptions {
151+
/**
152+
* The maximum time in milliseconds that a stream may remain idle before it
153+
* is closed.
154+
*/
155+
requestTimeout?: number;
156+
157+
/**
158+
* The maximum time in milliseconds that a session or socket may remain idle
159+
* before it is closed.
160+
* https://nodejs.org/docs/latest-v12.x/api/http2.html#http2_http2session_and_sockets
161+
*/
162+
sessionTimeout?: number;
163+
}

0 commit comments

Comments
 (0)