Function streamEffect

packages/rpc/src/RpcServer.ts:356–403

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

  }

  const streamEffect = (
    client: Client,
    request: Request<Rpcs>,
    stream: Stream.Stream<any, any> | Effect.Effect<Mailbox.ReadonlyMailbox<any, any>, any, Scope.Scope>
  ) => {
    let latch = client.latches.get(request.id)
    if (supportsAck && !latch) {
      latch = Effect.unsafeMakeLatch(false)
      client.latches.set(request.id, latch)
    }
    if (Effect.isEffect(stream)) {
      let done = false
      return stream.pipe(
        Effect.flatMap((mailbox) =>
          Effect.whileLoop({
            while: () => !done,
            body: constant(Effect.flatMap(mailbox.takeAll, ([chunk, done_]) => {
              done = done_
              if (!Chunk.isNonEmpty(chunk)) return Effect.void
              const write = options.onFromServer({
                _tag: "Chunk",
                clientId: client.id,
                requestId: request.id,
                values: Chunk.toReadonlyArray(chunk)
              })
              if (!latch) return write
              latch.unsafeClose()
              return Effect.zipRight(write, latch.await)
            })),
            step: constVoid
          })
        ),
        Effect.scoped
      )
    }
    return Stream.runForEachChunk(stream, (chunk) => {
      if (!Chunk.isNonEmpty(chunk)) return Effect.void
      const write = options.onFromServer({
        _tag: "Chunk",
        clientId: client.id,
        requestId: request.id,
        values: Chunk.toReadonlyArray(chunk)
      })
      if (!latch) return write
      latch.unsafeClose()
      return Effect.zipRight(write, latch.await)
    })
  }

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