Function take

packages/effect/src/internal/stream/handoff.ts:139–163

(self: Handoff<A>)

/** @internal */
export const take = <A>(self: Handoff<A>): Effect.Effect<A> =>
  Effect.flatMap(Deferred.make<void>(), (deferred) =>
    Effect.flatten(
      Ref.modify(self.ref, (state) =>
        pipe(
          state,
          handoffStateMatch(
            (notifyConsumer) =>
              [
                Effect.flatMap(
                  Deferred.await(notifyConsumer),
                  () => take(self)
                ),
                state
              ] as const,
            (value, notifyProducer) => [
              Effect.as(
                Deferred.succeed(notifyProducer, void 0),
                value
              ),
              handoffStateEmpty<A>(deferred)
            ]
          )
        ))
    ))

/** @internal */
export const poll = <A>(self: Handoff<A>): Effect.Effect<Option.Option<A>> =>