-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(child-process,workerd,web-worker): use new module transport (con…
…nect, send) (#140)
- Loading branch information
Showing
20 changed files
with
322 additions
and
223 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| import assert from "node:assert"; | ||
| import type { ModuleRunnerTransport } from "vite/module-runner"; | ||
|
|
||
| export function createSSEClientTransport(url: string): ModuleRunnerTransport { | ||
| let sseClient: SSEClient; | ||
|
|
||
| return { | ||
| async connect(handlers) { | ||
| sseClient = await createSSEClient(url, handlers); | ||
| }, | ||
| async send(payload) { | ||
| assert(sseClient); | ||
| sseClient.send(payload); | ||
| }, | ||
| timeout: 2000, | ||
| }; | ||
| } | ||
|
|
||
| type SSEClient = Awaited<ReturnType<typeof createSSEClient>>; | ||
|
|
||
| async function createSSEClient( | ||
| url: string, | ||
| handlers: { | ||
| onMessage: (payload: any) => void; | ||
| onDisconnection: () => void; | ||
| }, | ||
| ) { | ||
| const response = await fetch(url); | ||
| assert(response.ok); | ||
| const clientId = response.headers.get("x-client-id"); | ||
| assert(clientId); | ||
| assert(response.body); | ||
| response.body | ||
| .pipeThrough(new TextDecoderStream()) | ||
| .pipeThrough(splitTransform("\n\n")) | ||
| .pipeTo( | ||
| new WritableStream({ | ||
| write(chunk) { | ||
| // console.log("[client.response]", chunk); | ||
| if (chunk.startsWith("data: ")) { | ||
| const payload = JSON.parse(chunk.slice("data: ".length)); | ||
| handlers.onMessage(payload); | ||
| } | ||
| }, | ||
| abort(e) { | ||
| console.log("[client.abort]", e); | ||
| }, | ||
| close() { | ||
| console.log("[client.close]"); | ||
| handlers.onDisconnection(); | ||
| }, | ||
| }), | ||
| ) | ||
| .catch((e) => { | ||
| console.log("[client.pipeTo.catch]", e); | ||
| }); | ||
|
|
||
| return { | ||
| send: async (payload: unknown) => { | ||
| const response = await fetch(url, { | ||
| method: "POST", | ||
| body: JSON.stringify(payload), | ||
| headers: { | ||
| "x-client-id": clientId, | ||
| }, | ||
| }); | ||
| assert(response.ok); | ||
| const result = await response.json(); | ||
| return result; | ||
| }, | ||
| }; | ||
| } | ||
|
|
||
| export function splitTransform(sep: string): TransformStream<string, string> { | ||
| let pending = ""; | ||
| return new TransformStream({ | ||
| transform(chunk, controller) { | ||
| while (true) { | ||
| const i = chunk.indexOf(sep); | ||
| if (i >= 0) { | ||
| pending += chunk.slice(0, i); | ||
| controller.enqueue(pending); | ||
| pending = ""; | ||
| chunk = chunk.slice(i + sep.length); | ||
| continue; | ||
| } | ||
| pending += chunk; | ||
| break; | ||
| } | ||
| }, | ||
| }); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,119 @@ | ||
| import assert from "node:assert"; | ||
| import type { HotChannel, HotChannelListener, HotPayload } from "vite"; | ||
|
|
||
| export function createSSEServerTransport(): HotChannel { | ||
| interface SSEClientProxy { | ||
| send(payload: HotPayload): void; | ||
| close(): void; | ||
| } | ||
|
|
||
| const clientMap = new Map<string, SSEClientProxy>(); | ||
| const listenerManager = createListenerManager(); | ||
|
|
||
| async function handler(request: Request): Promise<Response | undefined> { | ||
| const url = new URL(request.url); | ||
| if (url.pathname === "/sse") { | ||
| // handle `send` | ||
| const senderId = request.headers.get("x-client-id"); | ||
| if (senderId) { | ||
| const client = clientMap.get(senderId); | ||
| assert(client); | ||
| const payload = await request.json(); | ||
| listenerManager.dispatch(payload, client); | ||
| return Response.json({ ok: true }); | ||
| } | ||
| // otherwise handle `connect` | ||
| let controller: ReadableStreamDefaultController<string>; | ||
| const stream = new ReadableStream<string>({ | ||
| start: (controller_) => { | ||
| controller = controller_; | ||
| controller.enqueue(`:ping\n\n`); | ||
| }, | ||
| cancel() { | ||
| clientMap.delete(clientId); | ||
| }, | ||
| }); | ||
| const pingInterval = setInterval(() => { | ||
| controller.enqueue(`:ping\n\n`); | ||
| }, 10_000); | ||
| const clientId = Math.random().toString(36).slice(2); | ||
| const client: SSEClientProxy = { | ||
| send(payload) { | ||
| controller.enqueue(`data: ${JSON.stringify(payload)}\n\n`); | ||
| }, | ||
| close() { | ||
| clearInterval(pingInterval); | ||
| controller.close(); | ||
| }, | ||
| }; | ||
| clientMap.set(clientId, client); | ||
| return new Response(stream, { | ||
| headers: { | ||
| "x-client-id": clientId, | ||
| "content-type": "text/event-stream", | ||
| "cache-control": "no-cache", | ||
| connection: "keep-alive", | ||
| }, | ||
| }); | ||
| } | ||
| return undefined; | ||
| } | ||
|
|
||
| const channel: HotChannel = { | ||
| listen() {}, | ||
| close() { | ||
| for (const client of clientMap.values()) { | ||
| client.close(); | ||
| } | ||
| }, | ||
| on: listenerManager.on, | ||
| off: listenerManager.off, | ||
| send: (payload) => { | ||
| for (const client of clientMap.values()) { | ||
| client.send(payload); | ||
| } | ||
| }, | ||
| // expose SSE handler via hot.api | ||
| api: { | ||
| type: "sse", | ||
| handler, | ||
| }, | ||
| }; | ||
|
|
||
| return channel; | ||
| } | ||
|
|
||
| // wrapper to simplify listener management | ||
| function createListenerManager(): Pick<HotChannel, "on" | "off"> & { | ||
| dispatch: ( | ||
| payload: HotPayload, | ||
| client: { send: (payload: HotPayload) => void }, | ||
| ) => void; | ||
| } { | ||
| const listerMap: Record<string, Set<HotChannelListener>> = {}; | ||
| const getListerMap = (e: string) => (listerMap[e] ??= new Set()); | ||
|
|
||
| return { | ||
| on(event: string, listener: HotChannelListener) { | ||
| // console.log("[channel.on]", event, listener); | ||
| if (event === "connection") { | ||
| return; | ||
| } | ||
| getListerMap(event).add(listener); | ||
| }, | ||
| off(event, listener: any) { | ||
| // console.log("[channel.off]", event, listener); | ||
| if (event === "connection") { | ||
| return; | ||
| } | ||
| getListerMap(event).delete(listener); | ||
| }, | ||
| dispatch(payload, client) { | ||
| if (payload.type === "custom") { | ||
| for (const lister of getListerMap(payload.event)) { | ||
| lister(payload.data, client); | ||
| } | ||
| } | ||
| }, | ||
| }; | ||
| } |
Oops, something went wrong.