Class SemaphoreImpl

repos/effect/packages/effect/src/Semaphore.ts:192–294

 * // Only 3 tasks can run concurrently
 * const program = Effect.all([
 *   task(1),
 *   task(2),
 *   task(3),
 *   task(4),
 *   task(5)
 * ], { concurrency: "unbounded" })
 *
 * await Effect.runPromise(program) // => [1, 2, 3, 4, 5]
 * ```
 *
 * @category constructors
 * @since 4.0.0
 */
export const makeUnsafe = (permits: number): Semaphore => new SemaphoreImpl(permits)

const waitForPermits = <A, E, R>(
  self: SemaphoreImpl,
  n: number,
  effect: Effect.Effect<A, E, R>
): Effect.Effect<A, E, R> =>
  internal.callback((resume) => {
    if (self.free >= n) return resume(effect)
    const observer = () => {
      if (self.free < n) return
      self.waiters.delete(observer)
      resume(effect)
    }
    self.waiters.add(observer)
    return internal.sync(() => {
      self.waiters.delete(observer)
    })
  })

class SemaphoreImpl implements Semaphore {
  public waiters = new Set<() => void>()
  public taken = 0
  public permits: number

  constructor(permits: number) {
    this.permits = permits
  }

  get free() {
    return this.permits - this.taken
  }

  take(n: number): Effect.Effect<number> {
    const take: Effect.Effect<number> = internal.suspend(() => {
      if (this.free < n) {
        return waitForPermits(this, n, take)
      }
      this.taken += n
      return internal.succeed(n)
    })
    return take
  }

  takeIfAvailable(n: number): Effect.Effect<boolean> {