🎉 Initial commit

Signed-off-by: kei-g <km.8k6ce+github@gmail.com>

Author: kei-g

.editorconfig

@@ -0,0 +1,17 @@
+# EditorConfig is awesome: https://EditorConfig.org
+
+# top-most EditorConfig file
+root = true
+
+[*]
+charset = utf-8
+end_of_line = lf
+indent_size = 2
+indent_style = space
+insert_final_newline = true
+tab_width = 2
+trim_trailing_whitespace = true
+
+[*.md]
+max_line_length = off
+trim_trailing_whitespace = false

.eslintrc.json

@@ -0,0 +1,40 @@
+{
+  "env": {
+    "browser": true,
+    "es2021": true,
+    "node": true
+  },
+  "extends": [
+    "eslint:recommended",
+    "plugin:@typescript-eslint/recommended"
+  ],
+  "parser": "@typescript-eslint/parser",
+  "parserOptions": {
+    "ecmaVersion": 12,
+    "sourceType": "module"
+  },
+  "plugins": [
+    "@typescript-eslint"
+  ],
+  "rules": {
+    "indent": [
+      "error",
+      2,
+      {
+        "SwitchCase": 1
+      }
+    ],
+    "linebreak-style": [
+      "error",
+      "unix"
+    ],
+    "quotes": [
+      "error",
+      "single"
+    ],
+    "semi": [
+      "error",
+      "never"
+    ]
+  }
+}

.gitignore

@@ -0,0 +1,7 @@
+**/.nyc_output/
+**/.vscode/
+**/build/
+**/coverage/
+**/lib/
+**/node_modules/
+**/package-lock.json

.mocharc.json

@@ -0,0 +1,8 @@
+{
+  "exit": true,
+  "extension": [
+    "ts"
+  ],
+  "require": "ts-node/register",
+  "timeout": 5000
+}

.travis.yml

@@ -0,0 +1,10 @@
+cache: npm
+language: node_js
+node_js:
+  - 14
+  - 15
+  - 16
+notifications:
+  email: false
+script:
+  - npm test

LICENSE

@@ -0,0 +1,29 @@
+BSD 3-Clause License
+
+Copyright (c) 2021-, kei-g
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice, this
+   list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution.
+
+3. Neither the name of the copyright holder nor the names of its
+   contributors may be used to endorse or promote products derived from
+   this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

README.md

@@ -0,0 +1,3 @@
+# async-iterable-queue [![License](https://img.shields.io/github/license/kei-g/async-iterable-queue)](https://opensource.org/licenses/BSD-3-Clause) [![Libraries.io dependency status for latest release](https://img.shields.io/librariesio/release/npm/async-iterable-queue)](https://npmjs.com/package/async-iterable-queue?activeTab=dependencies) [![Travis CI](https://img.shields.io/travis/com/kei-g/async-iterable-queue?logo=travis&style=flat)](https://www.travis-ci.com/github/kei-g/async-iterable-queue) [![npm](https://img.shields.io/npm/v/async-iterable-queue&style=flat)](https://npmjs.com/package/async-iterable-queue)
+
+[![npms.io (maintenance)](https://img.shields.io/npms-io/maintenance-score/async-iterable-queue)](https://npms.io/search?q=async-iterable-queue) [![npms.io (quality)](https://img.shields.io/npms-io/quality-score/async-iterable-queue)](https://npms.io/search?q=async-iterable-queue)

async-iterable-queue.ts

@@ -0,0 +1,231 @@
+import { EventEmitter } from 'stream'
+
+/**
+ * 非同期反復可能な先入れ先出し型の待ち行列への非同期反復子
+ */
+class AIQAsyncIterator<T> implements AsyncIterator<T> {
+  /**
+   * 事象発生器への参照
+   */
+  readonly #emitter: EventEmitter
+
+  /**
+   * 反復結果解決関数の配列への参照
+   */
+  readonly #resolvers: IteratorResultResolver<T>[]
+
+  /**
+   * コンストラクタ
+   * @param emitter 事象発生器への参照
+   * @param resolvers 反復結果解決関数の配列への参照
+   */
+  constructor(emitter: EventEmitter, resolvers: IteratorResultResolver<T>[]) {
+    this.#emitter = emitter
+    this.#resolvers = resolvers
+  }
+
+  /**
+   * 次の要素を返す
+   * @returns 次の要素
+   */
+  next(): Promise<IteratorResult<T>> {
+    return new Promise((resolve: IteratorResultResolver<T>) => (
+      this.#resolvers.push(resolve),
+      this.#emitter.emit('deq')
+    ))
+  }
+}
+
+/**
+ * 非同期反復可能な先入れ先出し型の待ち行列の状態を表す型
+ */
+type AIQState = 'ending' | 'finished'
+
+/**
+ * 非同期反復可能な先入れ先出し型の待ち行列
+ */
+export class AsyncIterableQueue<T> implements AsyncIterable<T> {
+  /**
+   * 事象発生器
+   */
+  readonly #emitter = new EventEmitter()
+
+  /**
+   * 先入れ先出し型の待ち行列
+   */
+  readonly #queue = [] as Terminatable<T>[]
+
+  /**
+   * 反復結果解決関数の配列
+   */
+  readonly #resolvers = [] as IteratorResultResolver<T>[]
+
+  /**
+   * この待ち行列の現在の状態
+   */
+  #state?: AIQState
+
+  /**
+   * コンストラクタ
+   */
+  constructor() {
+    const resolveAsync = createAsyncResolver({
+      finish: () => {
+        const state = this.#state
+        this.#state = 'finished'
+        return state
+      },
+      resolvers: this.#resolvers,
+    })
+    this.#emitter.on('deq', async () => {
+      while (this.#queue.length && this.#resolvers.length)
+        await resolveAsync(this.#queue.shift())
+    })
+    this.#emitter.on('enq', async (value: Terminatable<T>) =>
+      this.#resolvers.length ?
+        await resolveAsync(value) :
+        this.#queue.push(value)
+    )
+  }
+
+  /**
+   * この待ち行列への要素の追加を終了する
+   * @param cb 終端が読み取られた後に呼ばれるコールバック関数
+   */
+  end(cb?: NoParameterCallback): Promise<void> {
+    const state = this.#state
+    if (state)
+      throw new Error(state)
+    this.#state = 'ending'
+    return new Promise((resolve: Resolver<void>) => (
+      this.#emitter.emit('enq', new Terminator(cb)),
+      resolve()
+    ))
+  }
+
+  /**
+   * この待ち行列の末尾に要素を追加する
+   * @param value 要素の値
+   */
+  push(value: T): Promise<void> {
+    const state = this.#state
+    if (state)
+      throw new Error(state)
+    return new Promise((resolve: Resolver<void>) => (
+      this.#emitter.emit('enq', value),
+      resolve()
+    ))
+  }
+
+  /**
+   * 非同期反復子を返す
+   * @returns 非同期反復子
+   */
+  [Symbol.asyncIterator](): AsyncIterator<T> {
+    return new AIQAsyncIterator(this.#emitter, this.#resolvers)
+  }
+}
+
+/**
+ * 反復結果解決関数を非同期的に処理する関数を作成する際のパラメータの型
+ */
+type AsyncResolverCreateParameter<T> = {
+  /**
+   * 非同期反復可能な先入れ先出し型の待ち行列を終了状態にする
+   * @returns 以前の状態を返す
+   */
+  finish(): AIQState
+
+  /**
+   * 反復結果解決関数の配列への参照を取得する
+   */
+  get resolvers(): IteratorResultResolver<T>[]
+}
+
+/**
+ * 反復結果解決関数型
+ */
+type IteratorResultResolver<T> = Resolver<IteratorResult<T>>
+
+/**
+ * 引数無しコールバック関数型
+ */
+type NoParameterCallback = () => PromiseLike<void> | void
+
+/**
+ * 解決関数型
+ */
+type Resolver<T> = SingleParameterAction<T>
+
+/**
+ * 引数1個の関数型
+ */
+type SingleParameterAction<T> = (arg: T) => void
+
+/**
+ * 終端
+ */
+class Terminator {
+  /**
+   * コンストラクタ
+   * @param cb コールバック関数
+   */
+  constructor(private readonly cb?: NoParameterCallback) {
+  }
+
+  /**
+   * コールバック関数を呼び出す
+   */
+  call(): Promise<void> {
+    return new Promise((
+      resolve: Resolver<void>,
+      reject: SingleParameterAction<unknown>,
+    ) => {
+      if (this.cb)
+        try {
+          const result = this.cb()
+          if (result instanceof Promise)
+            result.catch(reject).then(resolve)
+          else
+            resolve()
+        }
+        catch (err: unknown) {
+          reject(err)
+        }
+      else
+        resolve()
+    })
+  }
+}
+
+/**
+ * 終端可能型
+ */
+type Terminatable<T> = Terminator | T
+
+/**
+ * 反復結果解決関数を非同期的に処理する関数を作成する
+ * @param param パラメータ
+ * @returns 反復結果解決関数を非同期的に処理する関数を返す
+ */
+const createAsyncResolver = <T>(param: AsyncResolverCreateParameter<T>) => {
+  const resolveAsync = (result: IteratorResult<T>) =>
+    new Promise((callback: Resolver<void>) => {
+      const resolver = param.resolvers.shift()
+      resolver(result)
+      callback()
+    })
+  return async (value: Terminatable<T>) => {
+    if (value instanceof Terminator) {
+      const state = param.finish()
+      await resolveAsync({ done: true } as IteratorResult<T>)
+      if (state === 'ending')
+        await value.call()
+    }
+    else
+      await resolveAsync({
+        done: false,
+        value,
+      })
+  }
+}