feat: Node.js HTTP/2 Handler

packages/node-http-handler/src/get-transformed-headers.ts

@@ -0,0 +1,17 @@
+import { HeaderBag } from "@aws-sdk/types";
+import { IncomingHttpHeaders } from "http2";
+
+const getTransformedHeaders = (headers: IncomingHttpHeaders) => {
+  const transformedHeaders: HeaderBag = {};
+
+  for (let name of Object.keys(headers)) {
+    let headerValues = <string>headers[name];
+    transformedHeaders[name] = Array.isArray(headerValues)
+      ? headerValues.join(",")
+      : headerValues;
+  }
+
+  return transformedHeaders;
+};
+
+export { getTransformedHeaders };

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

@@ -1 +1,2 @@
 export * from "./node-http-handler";
+export * from "./node-http2-handler";

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

@@ -3,7 +3,6 @@ import * as http from "http";
 import { Readable } from "stream";
 import { buildQueryString } from "@aws-sdk/querystring-builder";
 import {
-  HeaderBag,
   HttpHandler,
   HttpHandlerOptions,
   HttpRequest,
@@ -13,6 +12,7 @@ import {
 import { setConnectionTimeout } from "./set-connection-timeout";
 import { setSocketTimeout } from "./set-socket-timeout";
 import { writeRequestBody } from "./write-request-body";
+import { getTransformedHeaders } from "./get-transformed-headers";
 
 export class NodeHttpHandler implements HttpHandler<Readable, NodeHttpOptions> {
   private readonly httpAgent: http.Agent;
@@ -31,33 +31,9 @@ export class NodeHttpHandler implements HttpHandler<Readable, NodeHttpOptions> {
 
   handle(
     request: HttpRequest<Readable>,
-    options: HttpHandlerOptions
+    { abortSignal }: HttpHandlerOptions
   ): Promise<HttpResponse<Readable>> {
-    // determine which http(s) client to use
-    const isSSL = request.protocol === "https:";
-    const httpClient = isSSL ? https : http;
-
-    let path = request.path;
-    if (request.query) {
-      const queryString = buildQueryString(request.query);
-      if (queryString) {
-        path += `?${queryString}`;
-      }
-    }
-
-    const nodeHttpsOptions: https.RequestOptions = {
-      headers: request.headers,
-      host: request.hostname,
-      method: request.method,
-      path: path,
-      port: request.port,
-      agent: isSSL ? this.httpsAgent : this.httpAgent
-    };
-
     return new Promise((resolve, reject) => {
-      const abortSignal = options && options.abortSignal;
-      const { connectionTimeout, socketTimeout } = this.httpOptions;
-
       // if the request was already aborted, prevent doing extra work
       if (abortSignal && abortSignal.aborted) {
         const abortError = new Error("Request aborted");
@@ -66,21 +42,23 @@ export class NodeHttpHandler implements HttpHandler<Readable, NodeHttpOptions> {
         return;
       }
 
-      // create the http request
-      const req = (httpClient as typeof http).request(nodeHttpsOptions, res => {
-        const httpHeaders = res.headers;
-        const transformedHeaders: HeaderBag = {};
-
-        for (let name of Object.keys(httpHeaders)) {
-          let headerValues = <string>httpHeaders[name];
-          transformedHeaders[name] = Array.isArray(headerValues)
-            ? headerValues.join(",")
-            : headerValues;
-        }
+      // determine which http(s) client to use
+      const isSSL = request.protocol === "https:";
+      const queryString = buildQueryString(request.query || {});
+      const nodeHttpsOptions: https.RequestOptions = {
+        headers: request.headers,
+        host: request.hostname,
+        method: request.method,
+        path: queryString ? `${request.path}?${queryString}` : request.path,
+        port: request.port,
+        agent: isSSL ? this.httpsAgent : this.httpAgent
+      };
 
+      // create the http request
+      const req = (isSSL ? https : http).request(nodeHttpsOptions, res => {
         const httpResponse: HttpResponse<Readable> = {
           statusCode: res.statusCode || -1,
-          headers: transformedHeaders,
+          headers: getTransformedHeaders(res.headers),
           body: res
         };
         resolve(httpResponse);
@@ -89,8 +67,8 @@ export class NodeHttpHandler implements HttpHandler<Readable, NodeHttpOptions> {
       req.on("error", reject);
 
       // wire-up any timeout logic
-      setConnectionTimeout(req, reject, connectionTimeout);
-      setSocketTimeout(req, reject, socketTimeout);
+      setConnectionTimeout(req, reject, this.httpOptions.connectionTimeout);
+      setSocketTimeout(req, reject, this.httpOptions.socketTimeout);
 
       // wire-up abort logic
       if (abortSignal) {

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

@@ -0,0 +1,205 @@
+import { NodeHttp2Handler } from "./node-http2-handler";
+import { createMockHttp2Server, createResponseFunction } from "./server.mock";
+import { AbortController } from "@aws-sdk/abort-controller";
+
+describe("NodeHttp2Handler", () => {
+  let nodeH2Handler: NodeHttp2Handler;
+
+  const protocol = "http:";
+  const hostname = "localhost";
+  const port = 45321;
+  const mockH2Server = createMockHttp2Server().listen(port);
+  const getMockRequest = () => ({
+    protocol,
+    hostname,
+    port,
+    method: "GET",
+    path: "/",
+    headers: {}
+  });
+
+  const mockResponse = {
+    statusCode: 200,
+    headers: {},
+    body: "test"
+  };
+
+  beforeEach(() => {
+    nodeH2Handler = new NodeHttp2Handler();
+    mockH2Server.on("request", createResponseFunction(mockResponse));
+  });
+
+  afterEach(() => {
+    mockH2Server.removeAllListeners("request");
+    // @ts-ignore: access private property
+    const connectionPool = nodeH2Handler.connectionPool;
+    for (const [, session] of connectionPool) {
+      session.destroy();
+    }
+    connectionPool.clear();
+  });
+
+  afterAll(() => {
+    mockH2Server.close();
+  });
+
+  describe("connectionPool", () => {
+    it("is empty on initialization", () => {
+      // @ts-ignore: access private property
+      expect(nodeH2Handler.connectionPool.size).toBe(0);
+    });
+
+    it("creates and stores session when request is made", async () => {
+      await nodeH2Handler.handle(getMockRequest(), {});
+
+      // @ts-ignore: access private property
+      expect(nodeH2Handler.connectionPool.size).toBe(1);
+      expect(
+        // @ts-ignore: access private property
+        nodeH2Handler.connectionPool.get(`${protocol}//${hostname}:${port}`)
+      ).toBeDefined();
+    });
+
+    it("reuses existing session if request is made on same authority again", async () => {
+      await nodeH2Handler.handle(getMockRequest(), {});
+      // @ts-ignore: access private property
+      expect(nodeH2Handler.connectionPool.size).toBe(1);
+
+      // @ts-ignore: access private property
+      const session: ClientHttp2Session = nodeH2Handler.connectionPool.get(
+        `${protocol}//${hostname}:${port}`
+      );
+      const requestSpy = jest.spyOn(session, "request");
+
+      await nodeH2Handler.handle(getMockRequest(), {});
+      // @ts-ignore: access private property
+      expect(nodeH2Handler.connectionPool.size).toBe(1);
+      expect(requestSpy.mock.calls.length).toBe(1);
+    });
+
+    it("creates new session if request is made on new authority", async () => {
+      await nodeH2Handler.handle(getMockRequest(), {});
+      // @ts-ignore: access private property
+      expect(nodeH2Handler.connectionPool.size).toBe(1);
+
+      const port2 = port + 1;
+      const mockH2Server2 = createMockHttp2Server().listen(port2);
+      mockH2Server2.on("request", createResponseFunction(mockResponse));
+
+      await nodeH2Handler.handle({ ...getMockRequest(), port: port2 }, {});
+      // @ts-ignore: access private property
+      expect(nodeH2Handler.connectionPool.size).toBe(2);
+      expect(
+        // @ts-ignore: access private property
+        nodeH2Handler.connectionPool.get(`${protocol}//${hostname}:${port2}`)
+      ).toBeDefined();
+
+      mockH2Server2.close();
+    });
+
+    it("closes and removes session on sessionTimeout", async done => {
+      const sessionTimeout = 500;
+      nodeH2Handler = new NodeHttp2Handler({ sessionTimeout });
+      await nodeH2Handler.handle(getMockRequest(), {});
+
+      const authority = `${protocol}//${hostname}:${port}`;
+      // @ts-ignore: access private property
+      const session: ClientHttp2Session = nodeH2Handler.connectionPool.get(
+        authority
+      );
+      expect(session.closed).toBe(false);
+      setTimeout(() => {
+        expect(session.closed).toBe(true);
+        // @ts-ignore: access private property
+        expect(nodeH2Handler.connectionPool.get(authority)).not.toBeDefined();
+        done();
+      }, sessionTimeout + 100);
+    });
+  });
+
+  describe("destroy", () => {
+    it("destroys sessions and clears connectionPool", async () => {
+      await nodeH2Handler.handle(getMockRequest(), {});
+
+      // @ts-ignore: access private property
+      const session: ClientHttp2Session = nodeH2Handler.connectionPool.get(
+        `${protocol}//${hostname}:${port}`
+      );
+
+      // @ts-ignore: access private property
+      expect(nodeH2Handler.connectionPool.size).toBe(1);
+      expect(session.destroyed).toBe(false);
+      nodeH2Handler.destroy();
+      // @ts-ignore: access private property
+      expect(nodeH2Handler.connectionPool.size).toBe(0);
+      expect(session.destroyed).toBe(true);
+    });
+  });
+
+  describe("abortSignal", () => {
+    it("will not create session if request already aborted", async () => {
+      // @ts-ignore: access private property
+      expect(nodeH2Handler.connectionPool.size).toBe(0);
+      await expect(
+        nodeH2Handler.handle(getMockRequest(), {
+          abortSignal: {
+            aborted: true
+          }
+        })
+      ).rejects.toHaveProperty("name", "AbortError");
+      // @ts-ignore: access private property
+      expect(nodeH2Handler.connectionPool.size).toBe(0);
+    });
+
+    it("will not create request on session if request already aborted", async () => {
+      await nodeH2Handler.handle(getMockRequest(), {});
+
+      // @ts-ignore: access private property
+      const session: ClientHttp2Session = nodeH2Handler.connectionPool.get(
+        `${protocol}//${hostname}:${port}`
+      );
+      const requestSpy = jest.spyOn(session, "request");
+
+      await expect(
+        nodeH2Handler.handle(getMockRequest(), {
+          abortSignal: {
+            aborted: true
+          }
+        })
+      ).rejects.toHaveProperty("name", "AbortError");
+      expect(requestSpy.mock.calls.length).toBe(0);
+    });
+
+    it("will close request on session when aborted", async () => {
+      await nodeH2Handler.handle(getMockRequest(), {});
+
+      // @ts-ignore: access private property
+      const session: ClientHttp2Session = nodeH2Handler.connectionPool.get(
+        `${protocol}//${hostname}:${port}`
+      );
+      const requestSpy = jest.spyOn(session, "request");
+
+      const abortController = new AbortController();
+      // Delay response so that onabort is called earlier
+      setTimeout(() => {
+        abortController.abort();
+      }, 0);
+      mockH2Server.on(
+        "request",
+        async () =>
+          new Promise(resolve => {
+            setTimeout(() => {
+              resolve(createResponseFunction(mockResponse));
+            }, 1000);
+          })
+      );
+
+      await expect(
+        nodeH2Handler.handle(getMockRequest(), {
+          abortSignal: abortController.signal
+        })
+      ).rejects.toHaveProperty("name", "AbortError");
+      expect(requestSpy.mock.calls.length).toBe(1);
+    });
+  });
+});