Line data Source code
1 : import { getProxyErrorData, RawProxyError } from '../channel_common.js';
2 : import { HttpClient, HttpFetchResult } from './http_client.js';
3 : import { Logger, LoggableException } from '../logger/logger.js';
4 : import { HEADER_NAME_BATCH_BYTES, HEADER_NAME_BATCH_EVENT, HEADER_NAME_BATCH_TIMEOUT, HEADER_NAME_ENDPOINT, HEADER_NAME_ERROR, HEADER_NAME_METHOD, HEADER_NAME_PATH, HEADER_NAME_READ_TIMEOUT, HEADER_NAME_RESPONSE_STARTED, HEADER_NAME_SEARCH_PARAMS, HEADER_NAME_STREAM_ID } from '../native_proxy_headers.js';
5 : import { nativeProxyFetch } from '../electron_native_fetch.js';
6 :
7 1 : export enum NativeHttpServerStreamBatchEvent {
8 1 : StreamFailed = "StreamFailed",
9 1 : StreamFinished = "StreamFinished",
10 1 : FlushAfterClose = "FlushAfterClose",
11 1 : FlushAfterTimeout = "FlushAfterTimeout",
12 1 : FlushAfterBytes = "FlushAfterBytes",
13 1 : }
14 :
15 : export interface NativeHttpProxyConfig {
16 : /// The endpoint URL
17 : proxyEndpoint: URL;
18 : };
19 :
20 1 : export class NativeHttpError extends LoggableException {
21 : /// The data
22 1 : data: Record<string, string>;
23 :
24 1 : constructor(o: RawProxyError) {
25 1 : const data = getProxyErrorData(o);
26 1 : const message = o.message || 'Native HTTP proxy request failed';
27 1 : super(data.error ? `${message}: ${data.error}` : message, data, 'native_http_client');
28 1 : this.data = data;
29 1 : }
30 1 : }
31 :
32 1 : export class NativeHttpServerStream implements HttpFetchResult {
33 : /// The endpoint
34 1 : endpoint: NativeHttpProxyConfig;
35 : /// The stream id
36 1 : streamId: number | null;
37 : /// The headers
38 1 : headers: Headers;
39 : /// The status
40 1 : status: number;
41 : /// The status text
42 1 : statusText: string;
43 : /// The logger
44 1 : logger: Logger;
45 : /// The text decoder for decoding utf8
46 1 : textDecoder: TextDecoder;
47 : /// Body bytes received while loading the upstream response headers
48 1 : initialBody: ArrayBuffer | null;
49 : /// Whether more response batches need to be fetched
50 1 : fetchNext: boolean;
51 : /// The abort signal of the request
52 1 : signal: AbortSignal | null;
53 :
54 : /// Constructor
55 1 : constructor(endpoint: NativeHttpProxyConfig, streamId: number | null, headers: Headers, status: number, statusText: string, logger: Logger, signal: AbortSignal | null = null) {
56 1 : this.headers = headers;
57 1 : this.status = status;
58 1 : this.statusText = statusText;
59 1 : this.endpoint = endpoint;
60 1 : this.streamId = streamId;
61 1 : this.logger = logger;
62 1 : this.textDecoder = new TextDecoder();
63 1 : this.initialBody = null;
64 1 : this.fetchNext = streamId != null;
65 1 : this.signal = signal;
66 : // An abort drops the stream in the proxy, which also ends the upstream request
67 1 : signal?.addEventListener('abort', () => void this.drop(), { once: true });
68 1 : }
69 :
70 : /// Drop a stream that has more batches, the proxy forgets finished streams by itself
71 1 : async drop(): Promise<void> {
72 1 : if (this.streamId == null || !this.fetchNext) return;
73 1 : this.fetchNext = false;
74 1 : const url = new URL(this.endpoint.proxyEndpoint);
75 1 : url.pathname = `/http/stream/${this.streamId}`;
76 1 : try {
77 1 : await nativeProxyFetch(new Request(url, { method: 'DELETE' }));
78 1 : } catch (error) {
79 0 : this.logger.warn("Failed to drop an aborted http stream", { streamId: this.streamId.toString(), error: String(error) }, "native_http_client");
80 0 : }
81 1 : }
82 :
83 1 : private updateFetchState(batchEvent: string | null): void {
84 1 : switch (batchEvent) {
85 0 : case NativeHttpServerStreamBatchEvent.StreamFailed:
86 1 : case NativeHttpServerStreamBatchEvent.StreamFinished:
87 1 : case NativeHttpServerStreamBatchEvent.FlushAfterClose:
88 1 : this.fetchNext = false;
89 1 : break;
90 0 : case NativeHttpServerStreamBatchEvent.FlushAfterTimeout:
91 1 : case NativeHttpServerStreamBatchEvent.FlushAfterBytes:
92 1 : this.fetchNext = true;
93 : break;
94 1 : }
95 1 : }
96 :
97 1 : private async readNextBatch(): Promise<{ response: Response; buffer: ArrayBuffer }> {
98 1 : this.signal?.throwIfAborted();
99 1 : const url = new URL(this.endpoint.proxyEndpoint);
100 1 : url.pathname = `/http/stream/${this.streamId}`;
101 1 : const headers = new Headers();
102 1 : headers.set(HEADER_NAME_BATCH_BYTES, "4000000"); // 4 MB
103 1 : headers.set(HEADER_NAME_BATCH_TIMEOUT, "1000");
104 1 : headers.set(HEADER_NAME_READ_TIMEOUT, "10000");
105 :
106 1 : const response = await nativeProxyFetch(new Request(url, {
107 1 : method: 'GET',
108 1 : headers,
109 1 : }));
110 1 : this.signal?.throwIfAborted();
111 1 : if (response.headers.get(HEADER_NAME_ERROR) ?? false) {
112 1 : const proxyError = await response.json() as RawProxyError;
113 1 : throw new NativeHttpError(proxyError);
114 : }
115 :
116 1 : const batchEvent = response.headers.get(HEADER_NAME_BATCH_EVENT);
117 1 : this.logger.debug("Received fetch response", { "event": batchEvent }, "native_http_client");
118 1 : this.updateFetchState(batchEvent);
119 1 : return { response, buffer: await response.arrayBuffer() };
120 1 : }
121 :
122 1 : async initialize(): Promise<void> {
123 1 : if (this.streamId == null) {
124 0 : return;
125 : }
126 1 : const chunks = [];
127 1 : let totalChunkBytes = 0;
128 1 : while (this.fetchNext) {
129 1 : const { response, buffer } = await this.readNextBatch();
130 1 : chunks.push(buffer);
131 1 : totalChunkBytes += buffer.byteLength;
132 1 : if (response.headers.has(HEADER_NAME_RESPONSE_STARTED)) {
133 1 : this.headers = response.headers;
134 1 : this.status = response.status;
135 1 : this.statusText = response.statusText;
136 1 : if (response.status >= 400) {
137 1 : this.logger.warn('Native HTTP upstream error response', {
138 1 : status: response.status.toString(),
139 1 : streamId: this.streamId?.toString(),
140 1 : contentType: response.headers.get('content-type'),
141 1 : hyperdbStatusPresent: response.headers.has('x-hyperdb-status').toString(),
142 1 : }, 'native_http_client');
143 : }
144 1 : break;
145 1 : }
146 1 : }
147 :
148 1 : const combined = new Uint8Array(totalChunkBytes);
149 1 : let offset = 0;
150 1 : for (const chunk of chunks) {
151 1 : combined.set(new Uint8Array(chunk), offset);
152 1 : offset += chunk.byteLength;
153 1 : }
154 1 : this.initialBody = combined.buffer;
155 1 : }
156 :
157 1 : async json(): Promise<any> {
158 1 : const buffer = await this.arrayBuffer();
159 1 : const text = this.textDecoder.decode(buffer);
160 1 : if (text == "") {
161 0 : this.logger.debug(`Response body is empty`, {});
162 0 : return {};
163 0 : } else {
164 1 : return JSON.parse(text);
165 : }
166 1 : }
167 :
168 0 : async text(): Promise<any> {
169 0 : const buffer = await this.arrayBuffer();
170 0 : return this.textDecoder.decode(buffer);
171 0 : }
172 :
173 : /// Get the response as array buffer
174 1 : async arrayBuffer(): Promise<ArrayBuffer> {
175 1 : if (this.streamId == null) {
176 0 : return new ArrayBuffer(0);
177 : }
178 :
179 1 : const chunks = [];
180 1 : let totalChunkBytes = 0;
181 1 : if (this.initialBody != null) {
182 1 : chunks.push(this.initialBody);
183 1 : totalChunkBytes += this.initialBody.byteLength;
184 1 : this.initialBody = null;
185 1 : }
186 :
187 : // Fetch all the chunks
188 1 : while (this.fetchNext) {
189 1 : const { buffer } = await this.readNextBatch();
190 0 : chunks.push(buffer)
191 0 : totalChunkBytes += buffer.byteLength;
192 0 : }
193 :
194 :
195 : // Combine buffers
196 1 : const combined = new Uint8Array(new ArrayBuffer(totalChunkBytes));
197 1 : let combinedWriter = 0;
198 1 : for (const chunk of chunks) {
199 1 : combined.set(new Uint8Array(chunk), combinedWriter);
200 1 : combinedWriter += chunk.byteLength;
201 1 : }
202 :
203 1 : return combined.buffer;
204 1 : }
205 1 : }
206 :
207 1 : export class NativeHttpClient implements HttpClient {
208 : /// The logger
209 1 : logger: Logger;
210 : /// The endpoint
211 1 : endpoint: NativeHttpProxyConfig;
212 : /// The text encoder
213 1 : encoder: TextEncoder;
214 :
215 : /// Constructor
216 1 : constructor(proxy: NativeHttpProxyConfig, logger: Logger) {
217 1 : this.logger = logger;
218 1 : this.endpoint = proxy;
219 1 : this.encoder = new TextEncoder();
220 1 : }
221 :
222 1 : public async fetch(input: URL | Request | string, init?: RequestInit): Promise<HttpFetchResult> {
223 1 : const inputUrl = input instanceof URL ? input : new URL(input instanceof Request ? input.url : input);
224 1 : if (inputUrl.protocol === 'app:') {
225 0 : return await fetch(input, init);
226 : }
227 1 : const signal = init?.signal ?? null;
228 1 : signal?.throwIfAborted();
229 1 : const url = new URL(this.endpoint.proxyEndpoint);
230 1 : url.pathname = `/http/streams`;
231 1 : const remote = `${inputUrl.protocol}//${inputUrl.host}`;
232 :
233 1 : const headers = new Headers(init?.headers);
234 1 : headers.set(HEADER_NAME_METHOD, init?.method ?? "GET");
235 1 : headers.set(HEADER_NAME_ENDPOINT, remote);
236 1 : headers.set(HEADER_NAME_PATH, inputUrl.pathname);
237 1 : headers.set(HEADER_NAME_SEARCH_PARAMS, inputUrl.searchParams.toString());
238 1 : headers.set(HEADER_NAME_BATCH_TIMEOUT, "1000");
239 1 : headers.set(HEADER_NAME_READ_TIMEOUT, "10000");
240 :
241 1 : this.logger.debug(`Fetching http stream`, { "remote": remote, "path": inputUrl.toString() }, "native_http_client");
242 :
243 1 : const request = new Request(url, {
244 1 : method: 'POST',
245 1 : headers,
246 1 : body: init?.body
247 1 : });
248 1 : const response = await nativeProxyFetch(request);
249 :
250 : // Parse the stream id
251 1 : let streamId: number | null = null;
252 1 : if (response.status == 200) {
253 1 : const streamIdText = response.headers.get(HEADER_NAME_STREAM_ID);
254 1 : if (streamIdText == null) {
255 0 : this.logger.error("Fetch returned with status 200 but did not include a stream id", {}, "native_http_client");
256 0 : throw new Error("Missing stream id");
257 0 : }
258 1 : streamId = Number.parseInt(streamIdText);
259 :
260 1 : const stream = new NativeHttpServerStream(this.endpoint, streamId, response.headers, response.status, response.statusText, this.logger, signal);
261 : // The request may have been aborted while the stream started
262 1 : if (signal?.aborted) await stream.drop();
263 1 : await stream.initialize();
264 1 : return stream;
265 1 : } else {
266 1 : if (response.headers.get(HEADER_NAME_ERROR) ?? false) {
267 1 : throw new NativeHttpError(await response.json() as RawProxyError);
268 : }
269 0 : return new NativeHttpServerStream(this.endpoint, streamId, response.headers, response.status, response.statusText, this.logger);
270 0 : }
271 1 : }
272 1 : }
|
