255 lines
8.7 KiB
TypeScript
255 lines
8.7 KiB
TypeScript
|
import * as plugins from '../../plugins.js';
|
||
|
|
import type { IHttpRouteContext } from '../../core/models/route-context.js';
|
||
|
|
import type { ILogger } from './models/types.js';
|
||
|
|
import type { IMetricsTracker } from './request-handler.js';
|
||
|
|
|
||
|
|
/**
|
||
|
|
* HTTP/2 Request Handler Helper - handles HTTP/2 streams with specific destinations
|
||
|
|
* This is a helper class for the main RequestHandler
|
||
|
|
*/
|
||
|
|
export class Http2RequestHandler {
|
||
|
|
/**
|
||
|
|
* Handle HTTP/2 stream with direct HTTP/2 backend
|
||
|
|
*/
|
||
|
|
public static async handleHttp2WithHttp2Destination(
|
||
|
|
stream: plugins.http2.ServerHttp2Stream,
|
||
|
|
headers: plugins.http2.IncomingHttpHeaders,
|
||
|
|
destination: { host: string, port: number },
|
||
|
|
routeContext: IHttpRouteContext,
|
||
|
|
sessions: Map<string, plugins.http2.ClientHttp2Session>,
|
||
|
|
logger: ILogger,
|
||
|
|
metricsTracker?: IMetricsTracker | null
|
||
|
|
): Promise<void> {
|
||
|
|
const key = `${destination.host}:${destination.port}`;
|
||
|
|
|
||
|
|
// Get or create a client HTTP/2 session
|
||
|
|
let session = sessions.get(key);
|
||
|
|
if (!session || session.closed || (session as any).destroyed) {
|
||
|
|
try {
|
||
|
|
// Connect to the backend HTTP/2 server
|
||
|
|
session = plugins.http2.connect(`http://${destination.host}:${destination.port}`);
|
||
|
|
sessions.set(key, session);
|
||
|
|
|
||
|
|
// Handle session errors and cleanup
|
||
|
|
session.on('error', (err) => {
|
||
|
|
logger.error(`HTTP/2 session error to ${key}: ${err.message}`);
|
||
|
|
sessions.delete(key);
|
||
|
|
});
|
||
|
|
|
||
|
|
session.on('close', () => {
|
||
|
|
logger.debug(`HTTP/2 session closed to ${key}`);
|
||
|
|
sessions.delete(key);
|
||
|
|
});
|
||
|
|
} catch (err) {
|
||
|
|
logger.error(`Failed to establish HTTP/2 session to ${key}: ${err.message}`);
|
||
|
|
stream.respond({ ':status': 502 });
|
||
|
|
stream.end('Bad Gateway: Failed to establish connection to backend');
|
||
|
|
if (metricsTracker) metricsTracker.incrementFailedRequests();
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
try {
|
||
|
|
// Build headers for backend HTTP/2 request
|
||
|
|
const h2Headers: Record<string, any> = {
|
||
|
|
':method': headers[':method'],
|
||
|
|
':path': headers[':path'],
|
||
|
|
':authority': `${destination.host}:${destination.port}`
|
||
|
|
};
|
||
|
|
|
||
|
|
// Copy other headers, excluding pseudo-headers
|
||
|
|
for (const [key, value] of Object.entries(headers)) {
|
||
|
|
if (!key.startsWith(':') && typeof value === 'string') {
|
||
|
|
h2Headers[key] = value;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
logger.debug(
|
||
|
|
`Proxying HTTP/2 request to ${destination.host}:${destination.port}${headers[':path']}`,
|
||
|
|
{ method: headers[':method'] }
|
||
|
|
);
|
||
|
|
|
||
|
|
// Create HTTP/2 request stream to the backend
|
||
|
|
const h2Stream = session.request(h2Headers);
|
||
|
|
|
||
|
|
// Pipe client stream to backend stream
|
||
|
|
stream.pipe(h2Stream);
|
||
|
|
|
||
|
|
// Handle responses from the backend
|
||
|
|
h2Stream.on('response', (responseHeaders) => {
|
||
|
|
// Map status and headers to client response
|
||
|
|
const resp: Record<string, any> = {
|
||
|
|
':status': responseHeaders[':status'] as number
|
||
|
|
};
|
||
|
|
|
||
|
|
// Copy non-pseudo headers
|
||
|
|
for (const [key, value] of Object.entries(responseHeaders)) {
|
||
|
|
if (!key.startsWith(':') && value !== undefined) {
|
||
|
|
resp[key] = value;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// Send headers to client
|
||
|
|
stream.respond(resp);
|
||
|
|
|
||
|
|
// Pipe backend response to client
|
||
|
|
h2Stream.pipe(stream);
|
||
|
|
|
||
|
|
// Track successful requests
|
||
|
|
stream.on('end', () => {
|
||
|
|
if (metricsTracker) metricsTracker.incrementRequestsServed();
|
||
|
|
logger.debug(
|
||
|
|
`HTTP/2 request completed: ${headers[':method']} ${headers[':path']} ${responseHeaders[':status']}`,
|
||
|
|
{ method: headers[':method'], status: responseHeaders[':status'] }
|
||
|
|
);
|
||
|
|
});
|
||
|
|
});
|
||
|
|
|
||
|
|
// Handle backend errors
|
||
|
|
h2Stream.on('error', (err) => {
|
||
|
|
logger.error(`HTTP/2 stream error: ${err.message}`);
|
||
|
|
|
||
|
|
// Only send error response if headers haven't been sent
|
||
|
|
if (!stream.headersSent) {
|
||
|
|
stream.respond({ ':status': 502 });
|
||
|
|
stream.end(`Bad Gateway: ${err.message}`);
|
||
|
|
} else {
|
||
|
|
stream.end();
|
||
|
|
}
|
||
|
|
|
||
|
|
if (metricsTracker) metricsTracker.incrementFailedRequests();
|
||
|
|
});
|
||
|
|
|
||
|
|
// Handle client stream errors
|
||
|
|
stream.on('error', (err) => {
|
||
|
|
logger.debug(`Client HTTP/2 stream error: ${err.message}`);
|
||
|
|
h2Stream.destroy();
|
||
|
|
if (metricsTracker) metricsTracker.incrementFailedRequests();
|
||
|
|
});
|
||
|
|
|
||
|
|
} catch (err: any) {
|
||
|
|
logger.error(`Error handling HTTP/2 request: ${err.message}`);
|
||
|
|
|
||
|
|
// Only send error response if headers haven't been sent
|
||
|
|
if (!stream.headersSent) {
|
||
|
|
stream.respond({ ':status': 500 });
|
||
|
|
stream.end('Internal Server Error');
|
||
|
|
} else {
|
||
|
|
stream.end();
|
||
|
|
}
|
||
|
|
|
||
|
|
if (metricsTracker) metricsTracker.incrementFailedRequests();
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* Handle HTTP/2 stream with HTTP/1 backend
|
||
|
|
*/
|
||
|
|
public static async handleHttp2WithHttp1Destination(
|
||
|
|
stream: plugins.http2.ServerHttp2Stream,
|
||
|
|
headers: plugins.http2.IncomingHttpHeaders,
|
||
|
|
destination: { host: string, port: number },
|
||
|
|
routeContext: IHttpRouteContext,
|
||
|
|
logger: ILogger,
|
||
|
|
metricsTracker?: IMetricsTracker | null
|
||
|
|
): Promise<void> {
|
||
|
|
try {
|
||
|
|
// Build headers for HTTP/1 proxy request, excluding HTTP/2 pseudo-headers
|
||
|
|
const outboundHeaders: Record<string, string> = {};
|
||
|
|
for (const [key, value] of Object.entries(headers)) {
|
||
|
|
if (typeof key === 'string' && typeof value === 'string' && !key.startsWith(':')) {
|
||
|
|
outboundHeaders[key] = value;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// Always rewrite host header to match target
|
||
|
|
outboundHeaders.host = `${destination.host}:${destination.port}`;
|
||
|
|
|
||
|
|
logger.debug(
|
||
|
|
`Proxying HTTP/2 request to HTTP/1 backend ${destination.host}:${destination.port}${headers[':path']}`,
|
||
|
|
{ method: headers[':method'] }
|
||
|
|
);
|
||
|
|
|
||
|
|
// Create HTTP/1 proxy request
|
||
|
|
const proxyReq = plugins.http.request(
|
||
|
|
{
|
||
|
|
hostname: destination.host,
|
||
|
|
port: destination.port,
|
||
|
|
path: headers[':path'] as string,
|
||
|
|
method: headers[':method'] as string,
|
||
|
|
headers: outboundHeaders
|
||
|
|
},
|
||
|
|
(proxyRes) => {
|
||
|
|
// Map status and headers back to HTTP/2
|
||
|
|
const responseHeaders: Record<string, number | string | string[]> = {
|
||
|
|
':status': proxyRes.statusCode || 500
|
||
|
|
};
|
||
|
|
|
||
|
|
// Copy headers from HTTP/1 response to HTTP/2 response
|
||
|
|
for (const [key, value] of Object.entries(proxyRes.headers)) {
|
||
|
|
if (value !== undefined) {
|
||
|
|
responseHeaders[key] = value as string | string[];
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// Send headers to client
|
||
|
|
stream.respond(responseHeaders);
|
||
|
|
|
||
|
|
// Pipe HTTP/1 response to HTTP/2 stream
|
||
|
|
proxyRes.pipe(stream);
|
||
|
|
|
||
|
|
// Clean up when client disconnects
|
||
|
|
stream.on('close', () => proxyReq.destroy());
|
||
|
|
stream.on('error', () => proxyReq.destroy());
|
||
|
|
|
||
|
|
// Track successful requests
|
||
|
|
stream.on('end', () => {
|
||
|
|
if (metricsTracker) metricsTracker.incrementRequestsServed();
|
||
|
|
logger.debug(
|
||
|
|
`HTTP/2 to HTTP/1 request completed: ${headers[':method']} ${headers[':path']} ${proxyRes.statusCode}`,
|
||
|
|
{ method: headers[':method'], status: proxyRes.statusCode }
|
||
|
|
);
|
||
|
|
});
|
||
|
|
}
|
||
|
|
);
|
||
|
|
|
||
|
|
// Handle proxy request errors
|
||
|
|
proxyReq.on('error', (err) => {
|
||
|
|
logger.error(`HTTP/1 proxy error: ${err.message}`);
|
||
|
|
|
||
|
|
// Only send error response if headers haven't been sent
|
||
|
|
if (!stream.headersSent) {
|
||
|
|
stream.respond({ ':status': 502 });
|
||
|
|
stream.end(`Bad Gateway: ${err.message}`);
|
||
|
|
} else {
|
||
|
|
stream.end();
|
||
|
|
}
|
||
|
|
|
||
|
|
if (metricsTracker) metricsTracker.incrementFailedRequests();
|
||
|
|
});
|
||
|
|
|
||
|
|
// Pipe client stream to proxy request
|
||
|
|
stream.pipe(proxyReq);
|
||
|
|
|
||
|
|
// Handle client stream errors
|
||
|
|
stream.on('error', (err) => {
|
||
|
|
logger.debug(`Client HTTP/2 stream error: ${err.message}`);
|
||
|
|
proxyReq.destroy();
|
||
|
|
if (metricsTracker) metricsTracker.incrementFailedRequests();
|
||
|
|
});
|
||
|
|
|
||
|
|
} catch (err: any) {
|
||
|
|
logger.error(`Error handling HTTP/2 to HTTP/1 request: ${err.message}`);
|
||
|
|
|
||
|
|
// Only send error response if headers haven't been sent
|
||
|
|
if (!stream.headersSent) {
|
||
|
|
stream.respond({ ':status': 500 });
|
||
|
|
stream.end('Internal Server Error');
|
||
|
|
} else {
|
||
|
|
stream.end();
|
||
|
|
}
|
||
|
|
|
||
|
|
if (metricsTracker) metricsTracker.incrementFailedRequests();
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|