Class SemaphoreImpl

packages/effect/src/Semaphore.ts:225–322

  })

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> {
    return internal.suspend(() => {
      if (this.free < n) return internal.succeed(false)
      this.taken += n
      return internal.succeed(true)
    })
  }

  releaseUnsafe(fiber: Fiber<any, any>, n: number): number {
    this.taken -= n
    if (this.waiters.size > 0) {
      fiber.currentDispatcher.scheduleTask(() => {
        for (const observer of this.waiters) {
          if (this.free <= 0) break
          observer()
        }
      }, 0)
    }
    return this.free
  }

  resize(permits: number) {
    return core.withFiber((fiber) => {
      this.permits = permits
      if (this.free < 0) return internal.void
      this.releaseUnsafe(fiber, 0)
      return internal.void
    })
  }

  release(n: number): Effect.Effect<number> {
    return core.withFiber((fiber) => internal.succeed(this.releaseUnsafe(fiber, n)))
  }