11import { once } from 'node:events'
22import { request as requestHTTP } from 'node:http'
33import type { IncomingMessage , RequestOptions } from 'node:http'
4+ import { connect } from 'node:http2'
5+ import type { IncomingHttpHeaders as IncomingHttp2Headers } from 'node:http2'
6+ import { Server as HttpsServer , request as requestHTTPS } from 'node:https'
47import type { AddressInfo } from 'node:net'
5- import { Readable } from 'node:stream'
68import { newHeadersFromIncoming } from '../../src/request'
79import { GlobalResponse } from '../../src/response'
810import type { ServerType } from '../../src/types'
@@ -15,39 +17,174 @@ export type ServerRequestInit = Omit<
1517 path : string
1618}
1719
18- export const requestServer = async (
19- server : ServerType ,
20- init : ServerRequestInit
21- ) : Promise < Response > => {
22- const address = server . address ( ) as AddressInfo | null
20+ export type ServerRequestHooks = {
21+ onHeaders ?: ( headers : Headers , status : number ) => void
22+ onChunk ?: ( chunk : Buffer , index : number ) => void
23+ }
24+
25+ export type Http2RequestInit = {
26+ headers ?: Record < string , string >
27+ method ?: string
28+ path : string
29+ }
30+
31+ const ensureListening = async ( server : ServerType ) => {
32+ let address = server . address ( ) as AddressInfo | null
33+ let startedByHelper = false
34+ if ( ! address ) {
35+ server . listen ( 0 , '127.0.0.1' )
36+ await once ( server , 'listening' )
37+ startedByHelper = true
38+ address = server . address ( ) as AddressInfo | null
39+ }
2340 if ( ! address ) {
2441 throw new Error ( 'Server is not listening on a TCP address' )
2542 }
2643
44+ return {
45+ address,
46+ // a no-op if the server was already listening before the request, its
47+ // owner is responsible for closing it in that case
48+ close : async ( ) => {
49+ if ( startedByHelper ) {
50+ await new Promise < void > ( ( resolve ) => server . close ( ( ) => resolve ( ) ) )
51+ }
52+ } ,
53+ }
54+ }
55+
56+ const sendRequest = ( server : ServerType , address : AddressInfo , init : ServerRequestInit ) => {
2757 const { body, path, ...options } = init
28- const request = requestHTTP ( {
58+ const common = {
2959 ...options ,
3060 agent : false ,
3161 hostname : address . address ,
3262 path,
3363 port : address . port ,
34- } )
64+ }
65+ const request =
66+ server instanceof HttpsServer
67+ ? requestHTTPS ( { ...common , rejectUnauthorized : false } )
68+ : requestHTTP ( common )
3569 request . end ( body )
70+ return request
71+ }
3672
73+ const receiveResponse = async ( request : ReturnType < typeof sendRequest > ) => {
3774 const [ incoming ] = ( await once ( request , 'response' ) ) as [ IncomingMessage ]
3875 const status = incoming . statusCode
3976 if ( ! status ) {
4077 throw new Error ( 'Server response did not include a status code' )
4178 }
79+ return { incoming, status }
80+ }
4281
43- const responseBody =
44- options . method ?. toUpperCase ( ) === 'HEAD' || [ 101 , 204 , 205 , 304 ] . includes ( status )
45- ? null
46- : ( Readable . toWeb ( incoming ) as ReadableStream < Uint8Array > )
82+ export const requestServer = async (
83+ server : ServerType ,
84+ init : ServerRequestInit
85+ ) : Promise < Response > => {
86+ const { address, close } = await ensureListening ( server )
4787
48- return new GlobalResponse ( responseBody , {
49- headers : newHeadersFromIncoming ( incoming ) ,
50- status,
51- statusText : incoming . statusMessage ,
52- } )
88+ // the request may fail mid-flight (e.g. aborted via `init.signal`), so the
89+ // helper-started server is always fully closed before returning or
90+ // surfacing an error
91+ try {
92+ const { incoming, status } = await receiveResponse ( sendRequest ( server , address , init ) )
93+
94+ const hasBody = init . method ?. toUpperCase ( ) !== 'HEAD' && ! [ 101 , 204 , 205 , 304 ] . includes ( status )
95+
96+ let responseBody : BodyInit | null = null
97+ if ( hasBody ) {
98+ // buffer the body so a helper-started server can be fully closed before
99+ // returning, otherwise a subsequent request could hit a half-closed server
100+ const chunks : Buffer [ ] = [ ]
101+ for await ( const chunk of incoming ) {
102+ chunks . push ( chunk )
103+ }
104+ responseBody = new Uint8Array ( Buffer . concat ( chunks ) )
105+ }
106+
107+ return new GlobalResponse ( responseBody , {
108+ headers : newHeadersFromIncoming ( incoming ) ,
109+ status,
110+ statusText : incoming . statusMessage ,
111+ } )
112+ } finally {
113+ await close ( )
114+ }
115+ }
116+
117+ /**
118+ * Same as `requestServer`, but exposes the raw body chunks for tests that
119+ * assert streaming behavior: `onHeaders` fires before any body data arrives
120+ * and `onChunk` fires per received chunk, so a test can synchronize with the
121+ * server mid-response. Chunk boundaries are preserved by the chunked
122+ * transfer-encoding framing.
123+ */
124+ export const requestServerChunked = async (
125+ server : ServerType ,
126+ init : ServerRequestInit ,
127+ hooks : ServerRequestHooks = { }
128+ ) : Promise < { chunks : Buffer [ ] ; response : Response } > => {
129+ const { address, close } = await ensureListening ( server )
130+
131+ try {
132+ const { incoming, status } = await receiveResponse ( sendRequest ( server , address , init ) )
133+ const headers = newHeadersFromIncoming ( incoming )
134+ hooks . onHeaders ?.( headers , status )
135+
136+ const chunks : Buffer [ ] = [ ]
137+ incoming . on ( 'data' , ( chunk : Buffer ) => {
138+ hooks . onChunk ?.( chunk , chunks . length )
139+ chunks . push ( chunk )
140+ } )
141+ await once ( incoming , 'end' )
142+
143+ return {
144+ chunks,
145+ response : new GlobalResponse ( new Uint8Array ( Buffer . concat ( chunks ) ) , {
146+ headers,
147+ status,
148+ statusText : incoming . statusMessage ,
149+ } ) ,
150+ }
151+ } finally {
152+ await close ( )
153+ }
154+ }
155+
156+ export const requestServerHttp2 = async (
157+ server : ServerType ,
158+ init : Http2RequestInit
159+ ) : Promise < Response > => {
160+ const { address, close } = await ensureListening ( server )
161+ const client = connect ( `http://${ address . address } ${ address . port } )
162+ // stream errors reject `once()` below, a session error would otherwise crash
163+ client . once ( 'error' , ( ) => { } )
164+
165+ try {
166+ const stream = client . request ( {
167+ ':method' : init . method ?? 'GET' ,
168+ ':path' : init . path ,
169+ ...init . headers ,
170+ } )
171+ stream . end ( )
172+
173+ const [ incomingHeaders , , rawHeaders ] = ( await once ( stream , 'response' ) ) as [
174+ IncomingHttp2Headers ,
175+ number ,
176+ string [ ] ,
177+ ]
178+ const status = Number ( incomingHeaders [ ':status' ] )
179+ const headers = newHeadersFromIncoming ( { rawHeaders } )
180+
181+ const chunks : Buffer [ ] = [ ]
182+ stream . on ( 'data' , ( chunk : Buffer ) => chunks . push ( chunk ) )
183+ await once ( stream , 'end' )
184+
185+ return new GlobalResponse ( new Uint8Array ( Buffer . concat ( chunks ) ) , { headers, status } )
186+ } finally {
187+ client . close ( )
188+ await close ( )
189+ }
53190}
0 commit comments