1/**
2 * Runs the server side of the unstable Effect devtools socket protocol.
3 *
4 * Use this module when an integration needs to accept devtools clients over a
5 * `SocketServer`, decode newline-delimited JSON messages, and handle each
6 * connection with application-specific logic. It does not interpret spans or
7 * metrics itself; it gives handlers a typed surface for the telemetry described
8 * by `DevToolsSchema`.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Effect from "../Effect.ts"
14import * as Ndjson from "../encoding/Ndjson.ts"
15import * as Queue from "../Queue.ts"
16import * as Schema from "../Schema.ts"
17import * as Socket from "../socket/Socket.ts"
18import * as SocketServer from "../socket/SocketServer.ts"
19import * as Stream from "../Stream.ts"
20import * as DevToolsSchema from "./DevToolsSchema.ts"
21
22const RequestSchema = Schema.toCodecJson(DevToolsSchema.Request)
23const ResponseSchema = Schema.toCodecJson(DevToolsSchema.Response)
24
25/**
26 * Handle for a connected devtools client.
27 *
28 * **Details**
29 *
30 * It exposes a queue of non-ping requests received from the socket and a
31 * `send` function for non-pong responses.
32 *
33 * @stability unstable
34 * @category models
35 * @since 4.0.0
36 */
37export interface Client {
38 readonly queue: Queue.Dequeue<DevToolsSchema.Request.WithoutPing>
39 readonly send: (_: DevToolsSchema.Response.WithoutPong) => Effect.Effect<void>
40}
41
42/**
43 * Runs the devtools socket server.
44 *
45 * **Details**
46 *
47 * Each connection is decoded as NDJSON devtools protocol messages, `Ping`
48 * requests are answered with `Pong`, and all other requests are delivered
49 * through the `Client` passed to the handler.
50 *
51 * @stability unstable
52 * @category constructors
53 * @since 4.0.0
54 */
55export const run: <_, E, R>(
56 handle: (client: Client) => Effect.Effect<_, E, R>
57) => Effect.Effect<
58 never,
59 SocketServer.SocketServerError,
60 R | SocketServer.SocketServer
61> = Effect.fnUntraced(function*<R, E, _>(
62 handle: (client: Client) => Effect.Effect<_, E, R>
63) {
64 const server = yield* SocketServer.SocketServer
65
66 return yield* server.run(Effect.fnUntraced(function*(socket) {
67 const responses = yield* Queue.unbounded<DevToolsSchema.Response>()
68 const requests = yield* Queue.unbounded<DevToolsSchema.Request.WithoutPing>()
69
70 const client: Client = {
71 queue: requests,
72 send: (response) => Queue.offer(responses, response).pipe(Effect.asVoid)
73 }
74
75 yield* Stream.fromQueue(responses).pipe(
76 Stream.pipeThroughChannel(
77 Ndjson.duplexSchemaString(Socket.toChannelString(socket), {
78 inputSchema: ResponseSchema,
79 outputSchema: RequestSchema
80 })
81 ),
82 Stream.runForEach((request) =>
83 request._tag === "Ping"
84 ? Queue.offer(responses, { _tag: "Pong" })
85 : Queue.offer(requests, request)
86 ),
87 Effect.ensuring(
88 Queue.shutdown(responses).pipe(
89 Effect.andThen(Queue.shutdown(requests))
90 )
91 ),
92 Effect.forkChild
93 )
94
95 return yield* handle(client)
96 }))
97})