From cec9d0ff683498609095d675c0e1cecddc43975f Mon Sep 17 00:00:00 2001 From: kei-g Date: Sun, 22 Aug 2021 11:15:15 +0900 Subject: [PATCH] :tada: Initial commit Signed-off-by: kei-g --- .editorconfig | 17 +++ .eslintrc.json | 40 ++++++ .gitignore | 7 + .mocharc.json | 8 ++ .travis.yml | 10 ++ LICENSE | 29 ++++ README.md | 3 + async-iterable-queue.ts | 231 ++++++++++++++++++++++++++++++ package.json | 56 ++++++++ test/async-iterable-queue.spec.ts | 98 +++++++++++++ tsconfig.json | 27 ++++ 11 files changed, 526 insertions(+) create mode 100644 .editorconfig create mode 100644 .eslintrc.json create mode 100644 .gitignore create mode 100644 .mocharc.json create mode 100644 .travis.yml create mode 100644 LICENSE create mode 100644 README.md create mode 100644 async-iterable-queue.ts create mode 100644 package.json create mode 100644 test/async-iterable-queue.spec.ts create mode 100644 tsconfig.json diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 0000000..9a8fa00 --- /dev/null +++ b/.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 diff --git a/.eslintrc.json b/.eslintrc.json new file mode 100644 index 0000000..c3528eb --- /dev/null +++ b/.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" + ] + } +} diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..cd90a06 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +**/.nyc_output/ +**/.vscode/ +**/build/ +**/coverage/ +**/lib/ +**/node_modules/ +**/package-lock.json diff --git a/.mocharc.json b/.mocharc.json new file mode 100644 index 0000000..c00b1c1 --- /dev/null +++ b/.mocharc.json @@ -0,0 +1,8 @@ +{ + "exit": true, + "extension": [ + "ts" + ], + "require": "ts-node/register", + "timeout": 5000 +} diff --git a/.travis.yml b/.travis.yml new file mode 100644 index 0000000..065a72d --- /dev/null +++ b/.travis.yml @@ -0,0 +1,10 @@ +cache: npm +language: node_js +node_js: + - 14 + - 15 + - 16 +notifications: + email: false +script: + - npm test diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..e14ba17 --- /dev/null +++ b/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. diff --git a/README.md b/README.md new file mode 100644 index 0000000..936b4f7 --- /dev/null +++ b/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) diff --git a/async-iterable-queue.ts b/async-iterable-queue.ts new file mode 100644 index 0000000..94430fd --- /dev/null +++ b/async-iterable-queue.ts @@ -0,0 +1,231 @@ +import { EventEmitter } from 'stream' + +/** + * 非同期反復可能な先入れ先出し型の待ち行列への非同期反復子 + */ +class AIQAsyncIterator implements AsyncIterator { + /** + * 事象発生器への参照 + */ + readonly #emitter: EventEmitter + + /** + * 反復結果解決関数の配列への参照 + */ + readonly #resolvers: IteratorResultResolver[] + + /** + * コンストラクタ + * @param emitter 事象発生器への参照 + * @param resolvers 反復結果解決関数の配列への参照 + */ + constructor(emitter: EventEmitter, resolvers: IteratorResultResolver[]) { + this.#emitter = emitter + this.#resolvers = resolvers + } + + /** + * 次の要素を返す + * @returns 次の要素 + */ + next(): Promise> { + return new Promise((resolve: IteratorResultResolver) => ( + this.#resolvers.push(resolve), + this.#emitter.emit('deq') + )) + } +} + +/** + * 非同期反復可能な先入れ先出し型の待ち行列の状態を表す型 + */ +type AIQState = 'ending' | 'finished' + +/** + * 非同期反復可能な先入れ先出し型の待ち行列 + */ +export class AsyncIterableQueue implements AsyncIterable { + /** + * 事象発生器 + */ + readonly #emitter = new EventEmitter() + + /** + * 先入れ先出し型の待ち行列 + */ + readonly #queue = [] as Terminatable[] + + /** + * 反復結果解決関数の配列 + */ + readonly #resolvers = [] as IteratorResultResolver[] + + /** + * この待ち行列の現在の状態 + */ + #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) => + this.#resolvers.length ? + await resolveAsync(value) : + this.#queue.push(value) + ) + } + + /** + * この待ち行列への要素の追加を終了する + * @param cb 終端が読み取られた後に呼ばれるコールバック関数 + */ + end(cb?: NoParameterCallback): Promise { + const state = this.#state + if (state) + throw new Error(state) + this.#state = 'ending' + return new Promise((resolve: Resolver) => ( + this.#emitter.emit('enq', new Terminator(cb)), + resolve() + )) + } + + /** + * この待ち行列の末尾に要素を追加する + * @param value 要素の値 + */ + push(value: T): Promise { + const state = this.#state + if (state) + throw new Error(state) + return new Promise((resolve: Resolver) => ( + this.#emitter.emit('enq', value), + resolve() + )) + } + + /** + * 非同期反復子を返す + * @returns 非同期反復子 + */ + [Symbol.asyncIterator](): AsyncIterator { + return new AIQAsyncIterator(this.#emitter, this.#resolvers) + } +} + +/** + * 反復結果解決関数を非同期的に処理する関数を作成する際のパラメータの型 + */ +type AsyncResolverCreateParameter = { + /** + * 非同期反復可能な先入れ先出し型の待ち行列を終了状態にする + * @returns 以前の状態を返す + */ + finish(): AIQState + + /** + * 反復結果解決関数の配列への参照を取得する + */ + get resolvers(): IteratorResultResolver[] +} + +/** + * 反復結果解決関数型 + */ +type IteratorResultResolver = Resolver> + +/** + * 引数無しコールバック関数型 + */ +type NoParameterCallback = () => PromiseLike | void + +/** + * 解決関数型 + */ +type Resolver = SingleParameterAction + +/** + * 引数1個の関数型 + */ +type SingleParameterAction = (arg: T) => void + +/** + * 終端 + */ +class Terminator { + /** + * コンストラクタ + * @param cb コールバック関数 + */ + constructor(private readonly cb?: NoParameterCallback) { + } + + /** + * コールバック関数を呼び出す + */ + call(): Promise { + return new Promise(( + resolve: Resolver, + reject: SingleParameterAction, + ) => { + 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 = Terminator | T + +/** + * 反復結果解決関数を非同期的に処理する関数を作成する + * @param param パラメータ + * @returns 反復結果解決関数を非同期的に処理する関数を返す + */ +const createAsyncResolver = (param: AsyncResolverCreateParameter) => { + const resolveAsync = (result: IteratorResult) => + new Promise((callback: Resolver) => { + const resolver = param.resolvers.shift() + resolver(result) + callback() + }) + return async (value: Terminatable) => { + if (value instanceof Terminator) { + const state = param.finish() + await resolveAsync({ done: true } as IteratorResult) + if (state === 'ending') + await value.call() + } + else + await resolveAsync({ + done: false, + value, + }) + } +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..6193f98 --- /dev/null +++ b/package.json @@ -0,0 +1,56 @@ +{ + "author": { + "email": "km.8k6ce+github@gmail.com", + "name": "kei-g" + }, + "bugs": { + "url": "https://github.com/kei-g/async-iterable-queue/issues" + }, + "categories": [ + "Other" + ], + "description": "Async Iterable Queue", + "devDependencies": { + "@types/chai": "^4.2.21", + "@types/mocha": "^9.0.0", + "@types/node": "^16.7.1", + "@typescript-eslint/eslint-plugin": "^4.29.2", + "@typescript-eslint/parser": "^4.29.2", + "chai": "^4.3.4", + "eslint": "^7.32.0", + "mocha": "^9.1.0", + "nyc": "^15.1.0", + "terser": "^5.7.1", + "ts-node": "^10.2.1", + "typescript": "^4.3.5", + "uuid": "^8.3.2" + }, + "files": [ + "lib" + ], + "homepage": "https://github.com/kei-g/async-iterable-queue", + "name": "async-iterable-queue", + "nyc": { + "branches": 93, + "functions": 100, + "lines": 100, + "statements": 100 + }, + "publishConfig": { + "access": "public" + }, + "repository": { + "type": "git", + "url": "https://github.com/kei-g/async-iterable-queue.git" + }, + "scripts": { + "build": "npm run lint && tsc && ([ -d lib ] || mkdir lib) && cp build/async-iterable-queue.d.ts lib/ && terser build/async-iterable-queue.js -c -m -o lib/async-iterable-queue.js --toplevel", + "clean": "rm -fr .nyc_output/ build/ coverage/ lib/", + "cover": "nyc --check-coverage -r html -r text _mocha", + "lint": "eslint *.ts", + "prepublishOnly": "npm run build", + "test": "npm run lint && npm run cover" + }, + "types": "lib/async-iterable-queue.d.ts", + "version": "1.0.0" +} diff --git a/test/async-iterable-queue.spec.ts b/test/async-iterable-queue.spec.ts new file mode 100644 index 0000000..412dd8b --- /dev/null +++ b/test/async-iterable-queue.spec.ts @@ -0,0 +1,98 @@ +import { AsyncIterableQueue } from '../async-iterable-queue' +import { describe, it } from 'mocha' +import { expect } from 'chai' +import { throws } from 'assert' + +const source = [ + Math.LN2, + Math.LN10, + Math.E, + Math.PI, +] + +const popAsync = async (q: AsyncIterableQueue) => { + let index = 0 + for await (const value of q) + expect(value).equals(source[index++]) +} + +describe('failure mission', async () => { + it('call \'end\' twice', async () => { + const q = new AsyncIterableQueue() + const pushAsync = async () => { + for (const value of source) + await q.push(value) + await q.end() + throws(() => q.end()) + } + await Promise.all([popAsync(q), pushAsync()]) + }) + it('\'push\' after \'end\'', async () => { + const q = new AsyncIterableQueue() + const pushAsync = async () => { + for (const value of source) + await q.push(value) + await q.end() + throws(() => q.push(Math.SQRT2)) + } + await Promise.all([popAsync(q), pushAsync()]) + }) + it('with callback which throws an error', async () => { + const q = new AsyncIterableQueue() + const pushAsync = async () => { + for (const value of source) + await q.push(value) + await q.end(() => { + throw new Error() + }) + } + await Promise.all([popAsync(q), pushAsync()]) + }) +}) + +describe('successful mission', async () => { + it('parallel push', async () => { + const q = new AsyncIterableQueue() + const pushAsync = async () => { + await Promise.all(source.map((value: number) => q.push(value))) + await q.end() + } + await Promise.all([popAsync(q), pushAsync()]) + }) + it('with async callback', async () => { + const q = new AsyncIterableQueue() + const pushAsync = async () => { + for (const value of source) + await q.push(value) + let complete = false + await q.end(() => new Promise((resolve: () => void) => { + complete = true + resolve() + })) + expect(complete).equals(true) + } + await Promise.all([popAsync(q), pushAsync()]) + }) + it('with callback', async () => { + const q = new AsyncIterableQueue() + const pushAsync = async () => { + for (const value of source) + await q.push(value) + let complete = false + await q.end(() => { + complete = true + }) + expect(complete).equals(true) + } + await Promise.all([popAsync(q), pushAsync()]) + }) + it('without callback', async () => { + const q = new AsyncIterableQueue() + const pushAsync = async () => { + for (const value of source) + await q.push(value) + await q.end() + } + await Promise.all([popAsync(q), pushAsync()]) + }) +}) diff --git a/tsconfig.json b/tsconfig.json new file mode 100644 index 0000000..ddbbd29 --- /dev/null +++ b/tsconfig.json @@ -0,0 +1,27 @@ +{ + "compilerOptions": { + "alwaysStrict": true, + "charset": "utf8", + "declaration": true, + "downlevelIteration": true, + "emitBOM": false, + "emitDeclarationOnly": false, + "emitDecoratorMetadata": false, + "experimentalDecorators": false, + "lib": [ + "ES2015", + ], + "module": "CommonJS", + "moduleResolution": "node", + "newLine": "lf", + "outDir": "build", + "pretty": false, + "removeComments": true, + "strict": true, + "strictNullChecks": false, + "target": "ES6", + }, + "files": [ + "async-iterable-queue.ts", + ], +}