import listen = Deno.listen;import Conn = Deno.Conn;import Reader = Deno.Reader;import { BufReader, BufWriter } from "https://deno.land/std@v0.7.0/io/bufio.ts";import { defer, Deferred, promiseInterrupter } from "./promises.ts";import { initServeOptions, readRequest } from "./serveio.ts";import { createResponder, ServerResponder } from "./responder.ts";
export type ClientRequest = { url: string; method: string; headers?: Headers; body?: Uint8Array | Reader;};
export type ServerResponse = { status: number; headers?: Headers; body?: Uint8Array | Reader;};
export type IncomingHttpRequest = { url: string; method: string; proto: string; headers: Headers; body?: Reader; trailers?: Headers; keepAlive?: KeepAlive; finalize: () => Promise<void>;};
export type KeepAlive = { timeout: number; max: number;};
export type ServerRequest = IncomingHttpRequest & { conn: Conn; bufWriter: BufWriter; bufReader: BufReader;} & ServerResponder;
export type IncomingHttpResponse = { proto: string; status: number; statusText: string; headers: Headers; body?: Reader; trailers?: Headers; finalize: () => Promise<void>;};
export type ClientResponse = IncomingHttpResponse & { conn: Conn; bufWriter: BufWriter; bufReader: BufReader;};
export type ServeOptions = { cancel?: Promise<void>; keepAliveTimeout?: number; readTimeout?: number;};
export async function* serve( addr: string, opts: ServeOptions = {}): AsyncIterableIterator<ServerRequest> { opts = initServeOptions(opts); const listener = listen("tcp", addr); let onRequestDeferred = defer(); let requestQueue: ServerRequest[] = []; const yieldingPromises: WeakMap<ServerRequest, Deferred> = new WeakMap(); const throwIfCancelled = promiseInterrupter({ cancel: opts.cancel }); const enqueueRequest = (req: ServerRequest): Promise<void> => { requestQueue.push(req); const d = defer(); yieldingPromises.set(req, d); onRequestDeferred.resolve(); return d.promise; }; let closed = false; const closeListener = () => { if (!closed) { listener.close(); closed = true; } }; const acceptRoutine = () => { if (closed) return; listener .accept() .then(conn => { handleKeepAliveConn(conn, enqueueRequest, opts); acceptRoutine(); }) .catch(closeListener); }; acceptRoutine(); while (true) { try { await throwIfCancelled(onRequestDeferred.promise); onRequestDeferred = defer(); const list = requestQueue; requestQueue = []; for (const req of list) { const d = yieldingPromises.get(req); try { yield req; d.resolve(); } catch (e) { d.reject(); } finally { yieldingPromises.delete(req); } } } catch (unused) { break; } } closeListener();}
export function listenAndServe( addr: string, handler: (req: ServerRequest) => Promise<void>, opts: ServeOptions = {}): void { opts = initServeOptions(opts); const listener = listen("tcp", addr); const throwIfCancelled = promiseInterrupter({ cancel: opts.cancel }); let closed = false; const closeListener = () => { if (!closed) { listener.close(); closed = true; } }; const acceptRoutine = () => { if (closed) return; throwIfCancelled(listener.accept()) .then(conn => { handleKeepAliveConn(conn, handler, opts); acceptRoutine(); }) .catch(closeListener); }; acceptRoutine();}
function handleKeepAliveConn( conn: Conn, handler: (req: ServerRequest) => Promise<void>, opts: ServeOptions = {}) { const bufReader = new BufReader(conn); const bufWriter = new BufWriter(conn); const originalOpts = opts; scheduleReadRequest({ keepAliveTimeout: opts.readTimeout, readTimeout: opts.readTimeout, cancel: opts.cancel });
function scheduleReadRequest(opts: ServeOptions) { processRequest(opts) .then(scheduleReadRequest) .catch(() => conn.close()); }
async function processRequest(opts: ServeOptions): Promise<ServeOptions> { const req = await readRequest(bufReader, opts); const nextReq = Object.assign( req, { bufWriter, bufReader, conn }, createResponder(bufWriter) ); await handler(nextReq); await req.finalize(); let keepAliveTimeout = originalOpts.keepAliveTimeout; if (req.keepAlive && req.keepAlive.max <= 0) { throw "EOF"; } if (req.headers.get("connection") === "close") { throw "EOF"; } if (req.keepAlive && Number.isInteger(req.keepAlive.timeout)) { keepAliveTimeout = Math.min( keepAliveTimeout, req.keepAlive.timeout * 1000 ); } return { keepAliveTimeout, readTimeout: opts.readTimeout, cancel: opts.cancel }; }}