| 384 | * @since 4.0.0 |
| 385 | */ |
| 386 | export const toChannelMap = <IE, A>( |
| 387 | self: Socket, |
| 388 | f: (data: Uint8Array | string) => A |
| 389 | ): Channel.Channel< |
| 390 | NonEmptyReadonlyArray<A>, |
| 391 | SocketError | IE, |
| 392 | void, |
| 393 | NonEmptyReadonlyArray<Uint8Array | string | CloseEvent>, |
| 394 | IE |
| 395 | > => |
| 396 | Channel.fromTransform(Effect.fnUntraced(function*(upstream, scope) { |
| 397 | const queue = yield* Queue.make<A, SocketError | IE | Cause.Done>() |
| 398 | |
| 399 | const writeScope = yield* Scope.fork(scope) |
| 400 | const write = yield* Scope.provide(self.writer, writeScope) |
| 401 | |
| 402 | let chunk: NonEmptyReadonlyArray<Uint8Array | string | CloseEvent> | undefined |
| 403 | let index = 0 |
| 404 | const writeChunk = Effect.whileLoop({ |
| 405 | while: () => index < chunk!.length, |
| 406 | body: () => write(chunk![index++]), |
| 407 | step: constVoid |
| 408 | }) |
| 409 | |
| 410 | yield* upstream.pipe( |
| 411 | Effect.flatMap((arr) => { |
| 412 | if (arr.length === 1) return write(arr[0]) |
| 413 | chunk = arr |
| 414 | index = 0 |
| 415 | return writeChunk |
| 416 | }), |
| 417 | Effect.forever({ disableYield: true }), |
| 418 | Effect.catchCauseFilter( |
| 419 | Pull.filterNoDone, |
| 420 | (cause) => Queue.failCause(queue, cause) |
| 421 | ), |
| 422 | Effect.ensuring(Scope.close(writeScope, Exit.void)), |
| 423 | Effect.forkIn(scope) |
| 424 | ) |
| 425 | |
| 426 | yield* self.runRaw((data) => { |
| 427 | Queue.offerUnsafe(queue, f(data)) |
| 428 | }).pipe( |
| 429 | Queue.into(queue), |
| 430 | Effect.forkIn(scope) |
| 431 | ) |
| 432 | |
| 433 | // @effect-diagnostics-next-line returnEffectInGen:off |
| 434 | return Queue.takeAll(queue) |
| 435 | })) |
| 436 | |
| 437 | /** |
| 438 | * Converts a `Socket` into a binary `Channel`, encoding incoming string frames |