Function fromConn

packages/platform/deno/src/DenoSocket.ts:62–186

(
  open: Effect.Effect<Deno.Conn, Socket.SocketError, RO>
)

 * @since 4.0.0
 */
export const fromConn = <RO>(
  open: Effect.Effect<Deno.Conn, Socket.SocketError, RO>
): Effect.Effect<Socket.Socket, never, Exclude<RO, Scope.Scope>> =>
  Effect.withFiber<Socket.Socket, never, Exclude<RO, Scope.Scope>>((fiber) => {
    let current: {
      readonly conn: Deno.Conn
      readonly writer: WritableStreamDefaultWriter<Uint8Array>
    } | undefined
    let tearingDown = false
    let writeClosed = false
    const latch = Latch.makeUnsafe(false)
    const openServices = fiber.context as Context.Context<RO>

    const run = <R, E, _>(handler: (_: Uint8Array) => Effect.Effect<_, E, R> | void, options?: {
      readonly onOpen?: Effect.Effect<void> | undefined
    }) =>
      Effect.scopedWith(Effect.fnUntraced(function*(scope) {
        const fiberSet = yield* FiberSet.make<any, E | Socket.SocketError>().pipe(
          Scope.provide(scope)
        )
        let conn: Deno.Conn | undefined
        yield* Scope.addFinalizer(
          scope,
          Effect.suspend(() => {
            tearingDown = true
            return conn === undefined ? Effect.void : close(conn)
          })
        )
        conn = yield* Scope.provide(open, scope)
        const reader = conn.readable.getReader()
        const writer = conn.writable.getWriter()
        const runFork = yield* Effect.provideService(FiberSet.runtime(fiberSet)<R>(), Conn, conn)

        current = { conn, writer }
        tearingDown = false
        if (writeClosed) {
          writeClosed = false
          writer.releaseLock()
          yield* closeWrite(conn)
        }
        latch.openUnsafe()

        const read = Effect.tryPromise(
          () => reader.read()
        ).pipe(
          Effect.catchIf(
            (error) => tearingDown && isTeardownError(error.cause),
            () => Effect.succeed({ done: true, value: undefined } as ReadableStreamReadDoneResult<Uint8Array>)
          ),
          Effect.mapError((error) =>
            new Socket.SocketError({
              reason: new Socket.SocketReadError({ cause: error.cause })
            })
          )
        )
        const readLoop: Effect.Effect<void, Socket.SocketError> = Effect.suspend(() =>
          Effect.flatMap(read, ({ done, value }) => {
            if (done) {