Function run

repos/effect/packages/platform-node-shared/src/NodeSocket.ts:119–201

(handler: (_: Uint8Array) => Effect.Effect<_, E, R> | void, opts?: {
      readonly onOpen?: Effect.Effect<void> | undefined
    })

    const openServices = fiber.context as Context.Context<RO>

    const run = <R, E, _>(handler: (_: Uint8Array) => Effect.Effect<_, E, R> | void, opts?: {
      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: Duplex | undefined = undefined
        yield* Scope.addFinalizer(
          scope,
          Effect.sync(() => {
            if (!conn) return
            conn.off("data", onData)
            conn.off("end", onEnd)
            conn.off("error", onError)
            conn.off("close", onClose)
          })
        )
        conn = yield* Scope.provide(open, scope).pipe(
          options?.openTimeout ?
            Effect.timeoutOrElse({
              duration: options.openTimeout,
              orElse: () =>
                Effect.fail(
                  new Socket.SocketError({
                    reason: new Socket.SocketOpenError({ kind: "Timeout", cause: new Error("Connection timed out") })
                  })
                )
            }) :
            identity
        )
        conn.on("end", onEnd)
        conn.on("error", onError)
        conn.on("close", onClose)
        const run = yield* Effect.provideService(FiberSet.runtime(fiberSet)<R>(), NetSocket, conn as Net.Socket)
        conn.on("data", onData)

        currentSocket = conn
        latch.openUnsafe()
        if (opts?.onOpen) {
          yield* opts.onOpen
        }

        return yield* FiberSet.join(fiberSet)

        function onData(chunk: Uint8Array) {
          const result = handler(chunk)
          if (Effect.isEffect(result)) {
            run(result)
          }
        }
        function onEnd() {
          Deferred.doneUnsafe(fiberSet.deferred, Effect.void)
        }
        function onError(cause: Error) {
          Deferred.doneUnsafe(
            fiberSet.deferred,
            Effect.fail(