302 lines
9.5 KiB
TypeScript
302 lines
9.5 KiB
TypeScript
/**
|
|
* Node socket server adapters for Effect's unstable socket server API.
|
|
*
|
|
* This module turns `node:net` TCP or Unix-domain servers and `ws` WebSocket
|
|
* servers into scoped `SocketServer.SocketServer` services. Use the TCP
|
|
* constructors when handlers should receive a `Socket.Socket` backed by a Node
|
|
* `net.Socket`; use the WebSocket constructors when handlers should receive a
|
|
* `Socket.Socket` backed by `ws` and have access to the per-connection
|
|
* WebSocket and `IncomingMessage` services.
|
|
*
|
|
* @since 4.0.0
|
|
*/
|
|
import type { Cause } from "effect/Cause"
|
|
import * as Context from "effect/Context"
|
|
import * as Deferred from "effect/Deferred"
|
|
import * as Effect from "effect/Effect"
|
|
import * as Exit from "effect/Exit"
|
|
import * as Fiber from "effect/Fiber"
|
|
import { pipe } from "effect/Function"
|
|
import * as Function from "effect/Function"
|
|
import * as Layer from "effect/Layer"
|
|
import * as References from "effect/References"
|
|
import * as Scope from "effect/Scope"
|
|
import * as Socket from "effect/unstable/socket/Socket"
|
|
import * as SocketServer from "effect/unstable/socket/SocketServer"
|
|
import type * as Http from "node:http"
|
|
import * as Net from "node:net"
|
|
import * as NodeSocket from "./NodeSocket.ts"
|
|
import { NodeWS } from "./NodeSocket.ts"
|
|
|
|
/**
|
|
* Service tag for the Node `IncomingMessage` associated with the current
|
|
* WebSocket server connection.
|
|
*
|
|
* @category services
|
|
* @since 4.0.0
|
|
*/
|
|
export class IncomingMessage extends Context.Service<
|
|
IncomingMessage,
|
|
Http.IncomingMessage
|
|
>()("@effect/platform-node-shared/NodeSocketServer/IncomingMessage") {}
|
|
|
|
/**
|
|
* Creates a scoped TCP `SocketServer` from a Node `net.Server`, starts
|
|
* listening with the supplied options, queues pending connections until `run`
|
|
* is called, and closes the server when the scope ends.
|
|
*
|
|
* @category constructors
|
|
* @since 4.0.0
|
|
*/
|
|
export const make = Effect.fnUntraced(function*(
|
|
options: Net.ServerOpts & Net.ListenOptions
|
|
) {
|
|
const errorDeferred = Deferred.makeUnsafe<never, Error>()
|
|
const pending = new Set<Net.Socket>()
|
|
function defaultOnConnection(conn: Net.Socket) {
|
|
pending.add(conn)
|
|
const remove = () => {
|
|
pending.delete(conn)
|
|
}
|
|
conn.on("close", remove)
|
|
conn.on("error", remove)
|
|
}
|
|
let onConnection = defaultOnConnection
|
|
// oxlint-disable-next-line prefer-const
|
|
let server: Net.Server | undefined
|
|
yield* Effect.addFinalizer(() =>
|
|
Effect.callback<void>((resume) => {
|
|
server?.close(() => resume(Effect.void))
|
|
})
|
|
)
|
|
server = Net.createServer(options, (conn) => onConnection(conn))
|
|
server.on("error", (err) => Deferred.doneUnsafe(errorDeferred, Exit.fail(err)))
|
|
|
|
yield* Effect.callback<void, SocketServer.SocketServerError>((resume) => {
|
|
server.listen(options, () => resume(Effect.void))
|
|
}).pipe(
|
|
Effect.raceFirst(Effect.mapError(Deferred.await(errorDeferred), (err) =>
|
|
new SocketServer.SocketServerError({
|
|
reason: new SocketServer.SocketServerOpenError({
|
|
cause: err
|
|
})
|
|
})))
|
|
)
|
|
|
|
const run = Effect.fnUntraced(function*<R, E, _>(handler: (socket: Socket.Socket) => Effect.Effect<_, E, R>) {
|
|
const scope = yield* Scope.make()
|
|
const services = Context.omit(Scope.Scope)(yield* Effect.context<R>()) as Context.Context<R>
|
|
const trackFiber = Fiber.runIn(scope)
|
|
const prevOnConnection = onConnection
|
|
onConnection = function(conn: Net.Socket) {
|
|
let error: Error | undefined
|
|
conn.on("error", (err) => {
|
|
error = err
|
|
})
|
|
pipe(
|
|
NodeSocket.fromDuplex(
|
|
Effect.acquireRelease(
|
|
Effect.suspend((): Effect.Effect<Net.Socket, Socket.SocketError> => {
|
|
if (error) {
|
|
return Effect.fail(
|
|
new Socket.SocketError({
|
|
reason: new Socket.SocketOpenError({
|
|
kind: "Unknown",
|
|
cause: error
|
|
})
|
|
})
|
|
)
|
|
} else if (conn.closed) {
|
|
return Effect.fail(
|
|
new Socket.SocketError({
|
|
reason: new Socket.SocketCloseError({ code: 1000 })
|
|
})
|
|
)
|
|
}
|
|
return Effect.succeed(conn)
|
|
}),
|
|
(conn) =>
|
|
Effect.sync(() => {
|
|
if (conn.closed === false) {
|
|
conn.destroySoon()
|
|
}
|
|
})
|
|
)
|
|
),
|
|
Effect.flatMap(handler),
|
|
Effect.catchCause(reportUnhandledError),
|
|
Effect.runForkWith(Context.add(services, NodeSocket.NetSocket, conn)),
|
|
trackFiber
|
|
)
|
|
}
|
|
pending.forEach((conn) => {
|
|
conn.removeAllListeners("error")
|
|
conn.removeAllListeners("close")
|
|
onConnection(conn)
|
|
})
|
|
pending.clear()
|
|
return yield* Effect.callback<never>((_resume) => {
|
|
return Effect.suspend(() => {
|
|
onConnection = prevOnConnection
|
|
return Scope.close(scope, Exit.void)
|
|
})
|
|
})
|
|
})
|
|
|
|
const address = server.address()!
|
|
return SocketServer.SocketServer.of({
|
|
address: typeof address === "string" ?
|
|
{
|
|
_tag: "UnixAddress",
|
|
path: address
|
|
} :
|
|
{
|
|
_tag: "TcpAddress",
|
|
hostname: address.address,
|
|
port: address.port
|
|
},
|
|
run
|
|
})
|
|
})
|
|
|
|
/**
|
|
* Provides a TCP `SocketServer` by creating and managing a scoped Node
|
|
* `net.Server` with the supplied server and listen options.
|
|
*
|
|
* @category layers
|
|
* @since 4.0.0
|
|
*/
|
|
export const layer: (
|
|
options: Net.ServerOpts & Net.ListenOptions
|
|
) => Layer.Layer<
|
|
SocketServer.SocketServer,
|
|
SocketServer.SocketServerError
|
|
> = Function.flow(make, Layer.effect(SocketServer.SocketServer))
|
|
|
|
/**
|
|
* Creates a scoped WebSocket `SocketServer` backed by the `ws` package,
|
|
* providing the WebSocket and its Node `IncomingMessage` to connection
|
|
* handlers and closing the server when the scope ends.
|
|
*
|
|
* @category constructors
|
|
* @since 4.0.0
|
|
*/
|
|
export const makeWebSocket: (
|
|
options: NodeWS.ServerOptions<typeof NodeWS.WebSocket, typeof Http.IncomingMessage>
|
|
) => Effect.Effect<
|
|
SocketServer.SocketServer["Service"],
|
|
SocketServer.SocketServerError,
|
|
Scope.Scope
|
|
> = Effect.fnUntraced(function*(
|
|
options: NodeWS.ServerOptions
|
|
) {
|
|
const server = yield* Effect.acquireRelease(
|
|
Effect.sync(() => new NodeWS.WebSocketServer(options)),
|
|
(server) =>
|
|
Effect.callback<void>((resume) => {
|
|
server.close(() => resume(Effect.void))
|
|
})
|
|
)
|
|
const pendingConnections = new Set<readonly [globalThis.WebSocket, Http.IncomingMessage]>()
|
|
function defaultHandler(conn: globalThis.WebSocket, req: Http.IncomingMessage) {
|
|
const entry = [conn, req] as const
|
|
pendingConnections.add(entry)
|
|
conn.addEventListener("close", () => {
|
|
pendingConnections.delete(entry)
|
|
})
|
|
}
|
|
let onConnection = defaultHandler
|
|
server.on("connection", (conn, req) => onConnection(conn as any, req))
|
|
|
|
yield* Effect.callback<void, SocketServer.SocketServerError>((resume) => {
|
|
server.once("error", (error) => {
|
|
resume(Effect.fail(
|
|
new SocketServer.SocketServerError({
|
|
reason: new SocketServer.SocketServerOpenError({
|
|
cause: error
|
|
})
|
|
})
|
|
))
|
|
})
|
|
server.once("listening", () => {
|
|
resume(Effect.void)
|
|
})
|
|
})
|
|
|
|
const run = Effect.fnUntraced(function*<R, E, _>(handler: (socket: Socket.Socket) => Effect.Effect<_, E, R>) {
|
|
const scope = yield* Scope.make()
|
|
const services = Context.omit(Scope.Scope)(yield* Effect.context<R>()) as Context.Context<R>
|
|
const trackFiber = Fiber.runIn(scope)
|
|
const prevOnConnection = onConnection
|
|
onConnection = function(conn: globalThis.WebSocket, req: Http.IncomingMessage) {
|
|
const map = new Map(services.mapUnsafe)
|
|
map.set(IncomingMessage.key, req)
|
|
map.set(Socket.WebSocket.key, conn as any)
|
|
pipe(
|
|
Socket.fromWebSocket(
|
|
Effect.acquireRelease(
|
|
Effect.succeed(conn),
|
|
(conn) =>
|
|
Effect.sync(() => {
|
|
conn.close()
|
|
})
|
|
)
|
|
),
|
|
Effect.flatMap(handler),
|
|
Effect.catchCause(reportUnhandledError),
|
|
Effect.runForkWith(Context.makeUnsafe(map)),
|
|
trackFiber
|
|
)
|
|
}
|
|
for (const [conn, req] of pendingConnections) {
|
|
onConnection(conn, req)
|
|
}
|
|
pendingConnections.clear()
|
|
return yield* Effect.callback<never>((_resume) => {
|
|
return Effect.sync(() => {
|
|
onConnection = prevOnConnection
|
|
})
|
|
}).pipe(
|
|
Effect.ensuring(Scope.close(scope, Exit.void))
|
|
)
|
|
})
|
|
|
|
const address = server.address()!
|
|
return SocketServer.SocketServer.of({
|
|
address: typeof address === "string" ?
|
|
{
|
|
_tag: "UnixAddress",
|
|
path: address
|
|
} :
|
|
{
|
|
_tag: "TcpAddress",
|
|
hostname: address.address,
|
|
port: address.port
|
|
},
|
|
run
|
|
})
|
|
})
|
|
|
|
/**
|
|
* Provides a WebSocket `SocketServer` backed by the `ws` package and managed
|
|
* with the supplied server options.
|
|
*
|
|
* @category layers
|
|
* @since 4.0.0
|
|
*/
|
|
export const layerWebSocket: (
|
|
options: NodeSocket.NodeWS.ServerOptions<typeof NodeSocket.NodeWS.WebSocket, typeof Http.IncomingMessage>
|
|
) => Layer.Layer<
|
|
SocketServer.SocketServer,
|
|
SocketServer.SocketServerError
|
|
> = Function.flow(makeWebSocket, Layer.effect(SocketServer.SocketServer))
|
|
|
|
const reportUnhandledError = <E>(cause: Cause<E>) =>
|
|
Effect.withFiber<void>((fiber) => {
|
|
const unhandledLogLevel = fiber.getRef(References.UnhandledLogLevel)
|
|
if (unhandledLogLevel) {
|
|
return Effect.logWithLevel(unhandledLogLevel)(cause, "Unhandled error in SocketServer")
|
|
}
|
|
return Effect.void
|
|
})
|