| 106 | * @since 4.0.0 |
| 107 | */ |
| 108 | export const fromDuplex = <RO>( |
| 109 | open: Effect.Effect<Duplex, Socket.SocketError, RO>, |
| 110 | options?: { |
| 111 | readonly openTimeout?: Duration.Input | undefined |
| 112 | } |
| 113 | ): Effect.Effect<Socket.Socket, never, Exclude<RO, Scope.Scope>> => |
| 114 | Effect.withFiber<Socket.Socket, never, Exclude<RO, Scope.Scope>>((fiber) => { |
| 115 | let currentSocket: Duplex | undefined |
| 116 | const latch = Latch.makeUnsafe(false) |
| 117 | const openServices = fiber.context as Context.Context<RO> |
| 118 | |
| 119 | const run = <R, E, _>(handler: (_: Uint8Array) => Effect.Effect<_, E, R> | void, opts?: { |
| 120 | readonly onOpen?: Effect.Effect<void> | undefined |
| 121 | }) => |
| 122 | Effect.scopedWith(Effect.fnUntraced(function*(scope) { |
| 123 | const fiberSet = yield* FiberSet.make<any, E | Socket.SocketError>().pipe( |
| 124 | Scope.provide(scope) |
| 125 | ) |
| 126 | let conn: Duplex | undefined = undefined |
| 127 | yield* Scope.addFinalizer( |
| 128 | scope, |
| 129 | Effect.sync(() => { |
| 130 | if (!conn) return |
| 131 | conn.off("data", onData) |
| 132 | conn.off("end", onEnd) |
| 133 | conn.off("error", onError) |
| 134 | conn.off("close", onClose) |
| 135 | }) |
| 136 | ) |
| 137 | conn = yield* Scope.provide(open, scope).pipe( |
| 138 | options?.openTimeout ? |
| 139 | Effect.timeoutOrElse({ |
| 140 | duration: options.openTimeout, |
| 141 | orElse: () => |
| 142 | Effect.fail( |
| 143 | new Socket.SocketError({ |
| 144 | reason: new Socket.SocketOpenError({ kind: "Timeout", cause: new Error("Connection timed out") }) |
| 145 | }) |
| 146 | ) |
| 147 | }) : |
| 148 | identity |
| 149 | ) |
| 150 | conn.on("end", onEnd) |
| 151 | conn.on("error", onError) |
| 152 | conn.on("close", onClose) |
| 153 | const run = yield* Effect.provideService(FiberSet.runtime(fiberSet)<R>(), NetSocket, conn as Net.Socket) |
| 154 | conn.on("data", onData) |
| 155 | |
| 156 | currentSocket = conn |
| 157 | latch.openUnsafe() |
| 158 | if (opts?.onOpen) { |
| 159 | yield* opts.onOpen |
| 160 | } |
| 161 | |
| 162 | return yield* FiberSet.join(fiberSet) |
| 163 | |
| 164 | function onData(chunk: Uint8Array) { |
| 165 | const result = handler(chunk) |