Method constructor

packages/platform/node-shared/src/NodeStream.ts:404–434

(
    context: Context.Context<R>,
    stream: Stream.Stream<Uint8Array | string, E, R>
  )

  private fiber: Fiber.Fiber<void, E> | undefined = undefined

  constructor(
    context: Context.Context<R>,
    stream: Stream.Stream<Uint8Array | string, E, R>
  ) {
    super({})
    this.readLatch = Latch.makeUnsafe(false)
    this.fiber = Stream.runForEachArray(stream, (chunk) =>
      this.readLatch.whenOpen(Effect.sync(() => {
        this.readLatch.closeUnsafe()
        for (let i = 0; i < chunk.length; i++) {
          const item = chunk[i]
          if (typeof item === "string") {
            this.push(item, "utf8")
          } else {
            this.push(item)
          }
        }
      }))).pipe(
        this.readLatch.whenOpen,
        Effect.provideContext(context),
        Effect.runFork
      )
    this.fiber.addObserver((exit) => {
      this.fiber = undefined
      if (Exit.isSuccess(exit)) {
        this.push(null)
      } else {
        this.destroy(Cause.squash(exit.cause) as any)
      }
    })
  }

  override _read(_size: number): void {
    this.readLatch.openUnsafe()