Function doBuffer

packages/effect/src/internal/channel.ts:97–124

(
      empty: InElem,
      isEmpty: Predicate<InElem>,
      ref: Ref.Ref<InElem>
    )

): Channel.Channel<InElem, InElem, InErr, InErr, InDone, InDone> =>
  core.suspend(() => {
    const doBuffer = <InErr, InElem, InDone>(
      empty: InElem,
      isEmpty: Predicate<InElem>,
      ref: Ref.Ref<InElem>
    ): Channel.Channel<InElem, InElem, InErr, InErr, InDone, InDone> =>
      unwrap(
        Ref.modify(ref, (inElem) =>
          isEmpty(inElem) ?
            [
              core.readWith({
                onInput: (input: InElem) =>
                  core.flatMap(
                    core.write(input),
                    () => doBuffer<InErr, InElem, InDone>(empty, isEmpty, ref)
                  ),
                onFailure: (error: InErr) => core.fail(error),
                onDone: (done: InDone) => core.succeedNow(done)
              }),
              inElem
            ] as const :
            [
              core.flatMap(
                core.write(inElem),
                () => doBuffer<InErr, InElem, InDone>(empty, isEmpty, ref)
              ),
              empty
            ] as const)
      )
    return doBuffer(options.empty, options.isEmpty, options.ref)
  })