refactoring

Author: cnuss

cli/bootstrap.ts

@@ -1,6 +1,6 @@
 import { log } from "../src/internal/log";
 import packageJson from "../package.json";
-import { bootstrap } from "../src/internal/bootstrap";
+import { run } from "../src";
 
 (async () => {
   if (process.argv.includes("--version")) {
@@ -11,7 +11,7 @@ import { bootstrap } from "../src/internal/bootstrap";
   log("Starting bootstrap", { env: JSON.stringify(process.env) });
 
   try {
-    await bootstrap();
+    await run();
   } catch (e) {
     if (e instanceof Error) {
       console.error(e.message);
@@ -20,5 +20,6 @@ import { bootstrap } from "../src/internal/bootstrap";
     }
     process.exit(1);
   }
+
   log("Bootstrap complete");
 })();

src/index.ts

@@ -1,18 +1,50 @@
-import { endpointSpawn, endpointProxy } from "./internal/proxy";
-import { nextEvent, respondToEvent } from "./internal/runtime";
+import { pollForEvents } from "./internal/events";
+import { endpointSpawn, endpointProxy } from "./internal/endpoints";
+import { log } from "./internal/log";
+import { getRuntimeEvent, postRuntimeEventResponse } from "./internal/runtime";
 import {
-  LambdaEvent,
+  RuntimeEvent,
   EndpointExecRequest,
   EndpointProxyRequest,
   EndpointResponse,
 } from "./internal/types";
 
+const { _HANDLER, AWS_LAMBDA_RUNTIME_API } = process.env;
+
+export const run = async (): Promise<void> => {
+  if (!AWS_LAMBDA_RUNTIME_API) {
+    throw new Error("No AWS_LAMBDA_RUNTIME_API specified");
+  }
+
+  if (!_HANDLER) {
+    throw new Error("No handler specified");
+  }
+
+  log("Bootstraping", { _HANDLER, AWS_LAMBDA_RUNTIME_API });
+
+  const { childProcess, bin, endpoint } = await endpointSpawn(
+    _HANDLER,
+    process.env
+  );
+
+  try {
+    log("Polling for events", { bin, endpoint });
+    await pollForEvents(AWS_LAMBDA_RUNTIME_API, bin, endpoint);
+  } catch (e) {
+    if (childProcess) {
+      log("Killing child process", { pid: childProcess.pid });
+      childProcess.kill();
+    }
+    throw e;
+  }
+};
+
 export {
   endpointSpawn,
   endpointProxy,
-  nextEvent,
-  respondToEvent,
-  LambdaEvent,
+  getRuntimeEvent,
+  postRuntimeEventResponse,
+  RuntimeEvent,
   EndpointExecRequest,
   EndpointProxyRequest,
   EndpointResponse,

src/internal/bootstrap.ts

@@ -9,9 +9,24 @@ import {
   EndpointExecRequest,
   EndpointProxyRequest,
   EndpointResponse,
+  SpawnResult,
 } from "./types";
 import { info, log } from "./log";
 import { ChildProcess, spawn } from "child_process";
+import { APIGatewayProxyEventV2 } from "aws-lambda";
+
+function wsify(url?: URL): URL | undefined {
+  if (!url) return undefined;
+
+  const wsUrl = new URL(url.toString());
+  wsUrl.protocol = url.protocol.replace("http", "ws");
+
+  if (!wsUrl.protocol.startsWith("ws")) {
+    return undefined;
+  }
+
+  return wsUrl;
+}
 
 function convertHeaders(
   headers: RawAxiosResponseHeaders | AxiosResponseHeaders
@@ -78,12 +93,9 @@ const waitForEndpoint = async (
 
 export const endpointSpawn = async (
   handler: string,
-  env?: NodeJS.ProcessEnv
-): Promise<{
-  childProcess?: ChildProcess;
-  bin?: string;
-  endpoint?: URL;
-}> => {
+  env?: NodeJS.ProcessEnv,
+  detached: boolean = true
+): Promise<SpawnResult> => {
   // handler is in the format of
   // - `{some-bin}@http://localhost:{the-bins-port} (will start some-bin, and forward requests to the http server)
   // - `http://localhost:{some-port}` (will forward the request to the http server)
@@ -109,20 +121,26 @@ export const endpointSpawn = async (
 
     const cmds = bin.split(" ");
     childProcess = spawn(cmds[0], cmds.slice(1), {
-      detached: true,
+      detached,
       stdio: "inherit",
       env: env,
     });
 
-    // TODO Decide if we should do this...
-    childProcess.unref();
+    if (detached) {
+      childProcess.unref();
+    }
 
     log("Started child process", { cmds, pid: childProcess.pid });
   }
 
   endpoint = endpoint ? new URL(endpoint) : undefined;
 
-  return { childProcess, bin, endpoint };
+  return {
+    childProcess,
+    bin,
+    endpoint,
+    wsEndpoint: wsify(endpoint),
+  };
 };
 
 export const endpointExec = async ({
@@ -133,12 +151,21 @@ export const endpointExec = async ({
 }: EndpointExecRequest): Promise<EndpointResponse> => {
   const { execa } = await import("execa");
 
-  const { stdout } = await execa({
+  const timeout = deadline - Date.now();
+
+  info(`Running: \`${bin}\``);
+
+  const subprocess = execa({
     stderr: ["inherit"],
   })`${bin} ${event}`;
 
-  // TODO: handle deadline
-  info("TODO: need to handle deadline", { deadline });
+  setTimeout(() => {
+    subprocess.kill(
+      Error(`${bin} took longer than ${timeout} milliseconds to start.`)
+    );
+  }, timeout);
+
+  const { stdout } = await subprocess;
 
   const payload = JSON.parse(stdout);
 
@@ -154,25 +181,36 @@ export const endpointProxy = async ({
   event,
   deadline,
 }: EndpointProxyRequest): Promise<EndpointResponse> => {
+  const rawEvent = JSON.parse(event) as Partial<APIGatewayProxyEventV2>;
+
   const {
     requestContext,
     rawPath,
     rawQueryString,
     headers: rawHeaders,
     body: rawBody,
     isBase64Encoded,
-  } = event;
+  } = rawEvent;
+
+  if (!requestContext) {
+    throw new Error("No request context found in event");
+  }
+
   const method = requestContext.http.method;
 
   log("Waiting for endpoint to start", { endpoint, deadline });
   const { timeout } = await waitForEndpoint(endpoint, deadline);
 
   if (!timeout) {
     throw new Error(
-      `${endpoint.toString()} took longer than ${deadline} milliseconds to start.`
+      `${endpoint.toString()} took longer than ${timeout} milliseconds to start.`
     );
   }
 
+  if (!rawPath) {
+    throw new Error("No path found in event");
+  }
+
   const url = new URL(rawPath, endpoint);
   if (rawQueryString) {
     url.search = new URLSearchParams(rawQueryString).toString();
@@ -181,7 +219,7 @@ export const endpointProxy = async ({
   const decodedBody =
     isBase64Encoded && rawBody ? Buffer.from(rawBody, "base64") : rawBody;
 
-  log("Proxying request", { url, method, rawHeaders, decodedBody, timeout });
+  log("Proxying request", { url, method, rawHeaders, timeout });
 
   let response: AxiosResponse<any, any> | undefined = undefined;
   try {
@@ -207,7 +245,7 @@ export const endpointProxy = async ({
 
   const { data: rawData, headers: rawResponseHeaders } = response;
 
-  log("Proxy request complete", { url, method, rawResponseHeaders, rawData });
+  log("Proxy request complete", { url, method, rawResponseHeaders });
 
   return {
     requestId,

src/internal/proxy.ts renamed to src/internal/endpoints.ts

@@ -1,16 +1,16 @@
-import { EndpointProxyRequest, EndpointExecRequest } from "./types";
-import { endpointExec, endpointProxy } from "./proxy";
 import { log } from "./log";
-import { nextEvent, respondToEvent } from "./runtime";
+import { EndpointProxyRequest, EndpointExecRequest } from "./types";
+import { endpointExec, endpointProxy } from "./endpoints";
+import { getRuntimeEvent, postRuntimeEventResponse } from "./runtime";
 
-export const routeEvents = async (
+export const pollForEvents = async (
   runtimeApi: string,
   bin?: string,
   endpoint?: URL
 ): Promise<void> => {
   log("Waiting for next event from Lambda Runtime API", { runtimeApi });
 
-  const { requestId, event, deadline } = await nextEvent(runtimeApi);
+  const { requestId, event, deadline } = await getRuntimeEvent(runtimeApi);
 
   let payload: any | undefined = undefined;
 
@@ -53,9 +53,9 @@ Expected format: {bin}@{endpoint} or {bin} or {endpoint}:
     );
   }
 
-  await respondToEvent(runtimeApi, requestId, payload);
+  await postRuntimeEventResponse(runtimeApi, requestId, payload);
 
   log("Response sent to Lambda Runtime API", { runtimeApi, requestId });
 
-  return routeEvents(runtimeApi, bin, endpoint);
+  return pollForEvents(runtimeApi, bin, endpoint);
 };

src/internal/routing.ts renamed to src/internal/events.ts

@@ -1,8 +1,10 @@
 import axios from "axios";
-import { LambdaEvent } from "./types";
+import { RuntimeEvent } from "./types";
 
-export const nextEvent = async (runtimeApi: string): Promise<LambdaEvent> => {
-  const { headers, data } = await axios.get(
+export const getRuntimeEvent = async (
+  runtimeApi: string
+): Promise<RuntimeEvent> => {
+  const { headers, data: event } = await axios.get(
     `http://${runtimeApi}/2018-06-01/runtime/invocation/next`,
     {
       // block indefinitely until a response is received
@@ -19,12 +21,14 @@ export const nextEvent = async (runtimeApi: string): Promise<LambdaEvent> => {
 
   const deadline = Number.parseInt(headers["lambda-runtime-deadline-ms"]);
 
-  const event = JSON.parse(data);
+  if (!event || typeof event !== "string") {
+    throw new Error("No event found in response data");
+  }
 
   return { requestId, event, deadline };
 };
 
-export const respondToEvent = async (
+export const postRuntimeEventResponse = async (
   runtimeApi: string,
   requestId: string,
   payload: any

src/internal/runtime.ts

@@ -1,22 +1,30 @@
 import { APIGatewayProxyResult } from "aws-lambda";
+import { ChildProcess } from "child_process";
 
-export type LambdaEvent = {
+export type SpawnResult = {
+  childProcess?: ChildProcess;
+  bin?: string;
+  endpoint?: URL;
+  wsEndpoint?: URL;
+};
+
+export type RuntimeEvent = {
   requestId: string;
-  event: any;
+  event: string;
   deadline: number;
 };
 
 export type EndpointExecRequest = {
   requestId: string;
   bin: string;
-  event: any;
+  event: string;
   deadline: number;
 };
 
 export type EndpointProxyRequest = {
   requestId: string;
   endpoint: URL;
-  event: any;
+  event: string;
   deadline: number;
 };