Function makeUpgradeHandler

packages/platform-node/src/internal/httpServer.ts:161–209

(
  lazyWss: Effect.Effect<WS.WebSocketServer>,
  httpApp: App.Default<E, R>,
  middleware?: Middleware.HttpMiddleware
)

/** @internal */
export const makeUpgradeHandler = <R, E>(
  lazyWss: Effect.Effect<WS.WebSocketServer>,
  httpApp: App.Default<E, R>,
  middleware?: Middleware.HttpMiddleware
) => {
  const handledApp = App.toHandled(httpApp, handleResponse, middleware)
  return Effect.map(FiberSet.makeRuntime<R>(), (runFork) =>
    function handler(
      nodeRequest: Http.IncomingMessage,
      socket: Duplex,
      head: Buffer
    ) {
      let nodeResponse_: Http.ServerResponse | undefined = undefined
      const nodeResponse = () => {
        if (nodeResponse_ === undefined) {
          nodeResponse_ = new Http.ServerResponse(nodeRequest)
          nodeResponse_.assignSocket(socket as any)
          nodeResponse_.on("finish", () => {
            socket.end()
          })
        }
        return nodeResponse_
      }
      const upgradeEffect = Socket.fromWebSocket(Effect.flatMap(
        lazyWss,
        (wss) =>
          Effect.acquireRelease(
            Effect.async<globalThis.WebSocket>((resume) =>
              wss.handleUpgrade(nodeRequest, socket, head, (ws) => {
                resume(Effect.succeed(ws as any))
              })
            ),
            (ws) => Effect.sync(() => ws.close())
          )
      ))
      const fiber = runFork(
        Effect.provideService(
          handledApp,
          ServerRequest.HttpServerRequest,
          new ServerRequestImpl(nodeRequest, nodeResponse, upgradeEffect)
        )
      )
      socket.on("close", () => {
        if (!socket.writableEnded) {
          fiber.unsafeInterruptAsFork(Error.clientAbortFiberId)
        }
      })
    })
}

class ServerRequestImpl extends HttpIncomingMessageImpl<Error.RequestError> implements ServerRequest.HttpServerRequest {
  readonly [ServerRequest.TypeId]: ServerRequest.TypeId