| 73 | const openServices = fiber.context as Context.Context<RO> |
| 74 | |
| 75 | const run = <R, E, _>(handler: (_: Uint8Array) => Effect.Effect<_, E, R> | void, options?: { |
| 76 | readonly onOpen?: Effect.Effect<void> | undefined |
| 77 | }) => |
| 78 | Effect.scopedWith(Effect.fnUntraced(function*(scope) { |
| 79 | const fiberSet = yield* FiberSet.make<any, E | Socket.SocketError>().pipe( |
| 80 | Scope.provide(scope) |
| 81 | ) |
| 82 | let conn: Deno.Conn | undefined |
| 83 | yield* Scope.addFinalizer( |
| 84 | scope, |
| 85 | Effect.suspend(() => { |
| 86 | tearingDown = true |
| 87 | return conn === undefined ? Effect.void : close(conn) |
| 88 | }) |
| 89 | ) |
| 90 | conn = yield* Scope.provide(open, scope) |
| 91 | const reader = conn.readable.getReader() |
| 92 | const writer = conn.writable.getWriter() |
| 93 | const runFork = yield* Effect.provideService(FiberSet.runtime(fiberSet)<R>(), Conn, conn) |
| 94 | |
| 95 | current = { conn, writer } |
| 96 | tearingDown = false |
| 97 | if (writeClosed) { |
| 98 | writeClosed = false |
| 99 | writer.releaseLock() |
| 100 | yield* closeWrite(conn) |
| 101 | } |
| 102 | latch.openUnsafe() |
| 103 | |
| 104 | const read = Effect.tryPromise( |
| 105 | () => reader.read() |
| 106 | ).pipe( |
| 107 | Effect.catchIf( |
| 108 | (error) => tearingDown && isTeardownError(error.cause), |
| 109 | () => Effect.succeed({ done: true, value: undefined } as ReadableStreamReadDoneResult<Uint8Array>) |
| 110 | ), |
| 111 | Effect.mapError((error) => |
| 112 | new Socket.SocketError({ |
| 113 | reason: new Socket.SocketReadError({ cause: error.cause }) |
| 114 | }) |
| 115 | ) |
| 116 | ) |
| 117 | const readLoop: Effect.Effect<void, Socket.SocketError> = Effect.suspend(() => |
| 118 | Effect.flatMap(read, ({ done, value }) => { |
| 119 | if (done) { |
| 120 | Deferred.doneUnsafe(fiberSet.deferred, Effect.void) |
| 121 | return Effect.void |
| 122 | } |
| 123 | const result = handler(value) |
| 124 | if (Effect.isEffect(result)) { |
| 125 | runFork(result) |
| 126 | } |
| 127 | return readLoop |
| 128 | }) |
| 129 | ) |
| 130 | yield* FiberSet.run(fiberSet, readLoop) |
| 131 | |
| 132 | if (options?.onOpen) { |