@jshaftoe/infra/src/AsyncQueue.js

//@flow

import autoBind from 'auto-bind'

import Module from './Module'

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


const DEFAULT_MAX_QUEUE_SIZE = -1

type QueueConfig = {|
  name?: ?string,
  maxSize?: ?number,
|}

export class QueueClosedError extends Error {
  constructor(queueName: ?string) {
    super(queueName ? `Queue '${queueName}' is closed` : 'Queue closed')
  }
}

export default class AsyncQueue<T> {
  
  name: ?string

  _items: Array<T>
  _closed: boolean
  _maxSize: number

  _enqueueWaitingList: Array<{ res: () => void, rej: (err: Error) => void }>
  _dequeueWaitingList: Array<{ res: (T) => void, rej: (err: Error) => void }>
  _closeWaitingList: Array<{ res: () => void, rej: (err: Error) => void }>


  constructor(config?: QueueConfig) {

    // config
    config = config || { name: undefined, maxSize: undefined }
    this.name = config.name
    this._maxSize = config.maxSize || DEFAULT_MAX_QUEUE_SIZE

    // state
    this._items = []
    this._closed = false

    this._enqueueWaitingList = []
    this._dequeueWaitingList = []
    this._closeWaitingList = []

    autoBind(this)
  }


  async enqueue(newItem: T): Promise<void> {

    if (this._closed) {
      throw new Error(`Queue is closed`)
    }

    // wait for space to enqueue
    if (this._maxSize > 0 && this._items.length >= this._maxSize) {
      await new Promise((res, rej) => this._enqueueWaitingList.push({ res, rej }))  
    }

    //log(`Enqueued '${newItem}' to queue '${this.name}'`)    
    
    // release oldest waiting dequeue
    if (this._dequeueWaitingList.length) {
      this._dequeueWaitingList.shift().res(newItem)
    } else {
      this._items.push(newItem)
    }

    this.name && noteCount(`${this.name}.enqueue`, 1)
    this.name && noteGauge(`${this.name}.items`, this._items.length)
  }

  async dequeue(): Promise<T> {
    let item: T
    if (this._items.length === 0) { // if there are NO queued items
     
      if (this._closed) {
        throw new QueueClosedError(this.name)
      }
      item = await new Promise((res, rej) => this._dequeueWaitingList.push({ res, rej }))  
      
    } else { // if there are queued items
      
      item = this._items.shift() 

      // release oldest waiting enqueue
      if (this._enqueueWaitingList.length) {
        this._enqueueWaitingList.shift().res()
      }  

    }

    this.name && noteCount(`${this.name}.dequeue`, 1)
    this.name && noteGauge(`${this.name}.items`, this._items.length)

    //log(`Dequeued '${item}' from queue '${this.name}'`)

    // release all waiting closes
    if (this._items.length === 0) {
      this._closeWaitingList.forEach(cbs => cbs.res())
    }

    return item   
  }

  dequeueAllAvailable(): Array<T> {
    const items = this._items.splice(0)

    this.name && noteCount(`${this.name}.dequeue`, items.length)
    this.name && noteGauge(`${this.name}.items`, this._items.length)

    if (this._enqueueWaitingList.length > 0) {
      const enqueuesToRelease = Math.min(this._enqueueWaitingList.length, this._maxSize)
      for (let index = 0; index < enqueuesToRelease; index++) {
        this._enqueueWaitingList.shift().res()
      }
    }

    // release all waiting closes
    this._closeWaitingList.forEach(cbs => cbs.res())

    return items
  }

  async close(): Promise<void> {
    this._closed = true

    // fail all waiting enqueued
    this._enqueueWaitingList.forEach(cbs => cbs.rej(new Error('Queue closed')))

    // wait for all existing items to be dequeued
    if (this._items.length > 0) {
      await new Promise((res, rej)  => this._closeWaitingList.push({ res, rej }))  
    }

    // release all remaining waiting dequeues
    this._dequeueWaitingList.forEach(cbs => cbs.rej(new QueueClosedError(this.name)))
  }

  async enqueueRange (newItems: Array<T>) {
    for (const newItem of newItems) {
      await this.enqueue(newItem)
    }
  }

  async dequeueRange(count: number): Promise<Array<T>> {
    const results = []
    while (results.length < count) {
      try {
        const item = await this.dequeue()
        results.push(item)
      } catch (err) {
        if (err instanceof QueueClosedError)
          break
        throw err
      }
    }
    return results
  }

  getLength = (): number => {
    return this._items.length
  }
}