Function streamEffect

packages/effect/src/unstable/rpc/RpcServer.ts:396–444

(
    client: Client,
    request: Request<Rpcs>,
    stream:
      | Stream.Stream<any, any>
      | Effect.Effect<Queue.Dequeue<any, any>, any, Scope.Scope>
  )

  }

  const streamEffect = (
    client: Client,
    request: Request<Rpcs>,
    stream:
      | Stream.Stream<any, any>
      | Effect.Effect<Queue.Dequeue<any, any>, any, Scope.Scope>
  ) => {
    let latch = client.latches.get(request.id)
    if (supportsAck && !latch) {
      latch = Latch.makeUnsafe(false)
      client.latches.set(request.id, latch)
    }
    if (Effect.isEffect(stream)) {
      return stream.pipe(
        Effect.flatMap((queue) =>
          Effect.whileLoop({
            while: constTrue,
            body: constant(
              Effect.flatMap(Queue.takeAll(queue), (values) => {
                const write = options.onFromServer({
                  _tag: "Chunk",
                  clientId: client.id,
                  requestId: request.id,
                  values
                })
                if (!latch) return write
                latch.closeUnsafe()
                return Effect.flatMap(write, () => latch.await)
              })
            ),
            step: constVoid
          })
        ),
        Pull.catchDone(() => Effect.void),
        Effect.scoped
      )
    }
    return Stream.runForEachArray(stream, (values) => {
      const write = options.onFromServer({
        _tag: "Chunk",
        clientId: client.id,
        requestId: request.id,
        values
      })
      if (!latch) return write
      latch.closeUnsafe()
      return Effect.andThen(write, latch.await)
    })
  }

  const sendDefect = (client: Client, defect: unknown) =>
    Effect.suspend(() => {