@jshaftoe/infra/src/asyncSyncUtils.js

//@flow

import Module from './Module'

const { MODULE_NAME, log, warn, error } = new Module(__filename) // eslint-disable-line no-unused-vars

export class OneTimeBroadcastEvent {

  _waitingList: Array<() => void>
  signaled: boolean


  constructor(signaledInitially: boolean) {
    this._waitingList = []
    this.signaled = signaledInitially
  }


  wait = async (): Promise<void> => {
    if (this.signaled)
      return
    await new Promise(res => this._waitingList.push(res))
  }

  signal = (): void => {
    this._waitingList.forEach(res => res())
    this.signaled = true
  }

  reset = (): void => {
    this.signaled = false
  }
}

export class Semaphore {

  _waitingList: Array<() => void>
  _maxCapacity: number
  _name: ?string

  takenCount: number

  constructor(maxCapacity: number, name: ?string) {
    this._maxCapacity = maxCapacity    
    this._waitingList = []
    this._name = name
    this._name && log(`Semaphore was created`, { name: this._name, maxCapacity: maxCapacity })
    this.takenCount = 0
  }


  enter = async (): Promise<void> => {
    if (this.takenCount >= this._maxCapacity) {
      this._name && log(`Semaphore is awaited`, { name: this._name, takenCount: this.takenCount })
      await new Promise(res => this._waitingList.push(res))
      this._name && log(`Semaphore awaiting completed`, { name: this._name, takenCount: this.takenCount })
    }
    this.takenCount++
    this._name && log(`Semaphore was entered`, { name: this._name, takenCount: this.takenCount })
  }

  exit = (): void => {
    this._waitingList.length && this._waitingList.shift()()
    if (this.takenCount > 0)
      this.takenCount--
    this._name && log(`Semaphore was exited`, { name: this._name, takenCount: this.takenCount })
  }

  async run<T>(asyncFunc: () => Promise<T>): Promise<T> {
    try {
      await this.enter()
      return await asyncFunc()
    } finally {
      this.exit()
    }
  }

  async runConditional(asyncFunc: () => Promise<void>, condition: () => boolean): Promise<void> {
    if (!condition || condition()) {
      try {
        await this.enter()
        if (!condition || condition()) {
          await asyncFunc()
        }
      } finally {
        this.exit()
      }
    }
  }
}

export class CriticalSection extends Semaphore {
  constructor() {
    super(1)
  }
}