mirror of
https://github.com/vatesfr/xen-orchestra.git
synced 2026-11-09 17:06:13 -06:00
feat(generator): create a toolbox (#8295)
This commit is contained in:
committed by
GitHub
parent
4b8cdf8043
commit
9723075253
58
@vates/generator-toolbox/.USAGE.md
Normal file
58
@vates/generator-toolbox/.USAGE.md
Normal file
@@ -0,0 +1,58 @@
|
||||
A toolbox to ease the use of generator
|
||||
|
||||
### Timeout
|
||||
|
||||
wrap a source async generator to have it throw an error when timeout is reached
|
||||
timeout is a positive number in milliseconds
|
||||
|
||||
```js
|
||||
import { Timeout } from '@vates/generator-toolbox'
|
||||
|
||||
const wrappedGenerator = new Timeout(sourceGenerator, timeout)
|
||||
```
|
||||
|
||||
### Throttle
|
||||
|
||||
wrap a source async generator to have it respect a max speed ( in bytes per seconds).
|
||||
speed is either a strictly positive number or a function returning a strictly positive number. A speed change will be used for the next emitted packet.
|
||||
|
||||
The source generator must yield object with a length property.
|
||||
|
||||
Optimized for small yields regarding to the speed, since it won't split incoming packet.
|
||||
|
||||
If the generator reached the max speed it will be paused, limiting memory consumption.
|
||||
|
||||
```js
|
||||
import { Throttle } from '@vates/generator-toolbox'
|
||||
|
||||
const wrappedGenerator = new Throttle(sourceGenerator, speed)
|
||||
```
|
||||
|
||||
### Synchronized
|
||||
|
||||
Fork a generator. The rules ares:
|
||||
|
||||
- if the source returns, all the forks returns
|
||||
- if the forks errors, all the forks errors with the same error
|
||||
- if a the fork return , it is stopped, but the generator continue with the other
|
||||
- if a the fork error , it is stopped, but the generator continue with the other
|
||||
- if all the fork return , the source is stopped
|
||||
- if all the fork error , the source is errored with the last error
|
||||
- the source start producing a packet when the fastest forked ask for it
|
||||
- the source forks get the packet only when all the forks asked for it, no buffer stores in memory
|
||||
|
||||
```ts
|
||||
import { Synchronized } from '@vates/generator-toolbox'
|
||||
|
||||
async function consume(generator: AsyncGenerator) {
|
||||
for await (const val of generator) {
|
||||
console.log({ val })
|
||||
}
|
||||
}
|
||||
const forker = new Synchronized(generator)
|
||||
const first = forker.fork('first')
|
||||
const second = forker.fork('second')
|
||||
await Promise.all([consume(first), consume(second)])
|
||||
```
|
||||
|
||||
Note: you can stop early a generator by calling `generator.return()`, and you can stop in in error by calling `generator.throw(error)`
|
||||
1
@vates/generator-toolbox/.npmignore
Symbolic link
1
@vates/generator-toolbox/.npmignore
Symbolic link
@@ -0,0 +1 @@
|
||||
../../scripts/npmignore
|
||||
89
@vates/generator-toolbox/README.md
Normal file
89
@vates/generator-toolbox/README.md
Normal file
@@ -0,0 +1,89 @@
|
||||
<!-- DO NOT EDIT MANUALLY, THIS FILE HAS BEEN GENERATED -->
|
||||
|
||||
# @vates/generator-toolbox
|
||||
|
||||
[](https://npmjs.org/package/@vates/generator-toolbox)  [](https://bundlephobia.com/result?p=@vates/generator-toolbox) [](https://npmjs.org/package/@vates/generator-toolbox)
|
||||
|
||||
## Install
|
||||
|
||||
Installation of the [npm package](https://npmjs.org/package/@vates/generator-toolbox):
|
||||
|
||||
```sh
|
||||
npm install --save @vates/generator-toolbox
|
||||
```
|
||||
|
||||
## Usage
|
||||
|
||||
A toolbox to ease the use of generator
|
||||
|
||||
### Timeout
|
||||
|
||||
wrap a source async generator to have it throw an error when timeout is reached
|
||||
timeout is a positive number in milliseconds
|
||||
|
||||
```js
|
||||
import { Timeout } from '@vates/generator-toolbox'
|
||||
|
||||
const wrappedGenerator = new Timeout(sourceGenerator, timeout)
|
||||
```
|
||||
|
||||
### Throttle
|
||||
|
||||
wrap a source async generator to have it respect a max speed ( in bytes per seconds).
|
||||
speed is either a strictly positive number or a function returning a strictly positive number. A speed change will be used for the next emitted packet.
|
||||
|
||||
The source generator must yield object with a length property.
|
||||
|
||||
Optimized for small yields regarding to the speed, since it won't split incoming packet.
|
||||
|
||||
If the generator reached the max speed it will be paused, limiting memory consumption.
|
||||
|
||||
```js
|
||||
import { Throttle } from '@vates/generator-toolbox'
|
||||
|
||||
const wrappedGenerator = new Throttle(sourceGenerator, speed)
|
||||
```
|
||||
|
||||
### Synchronized
|
||||
|
||||
Fork a generator. The rules ares:
|
||||
|
||||
- if the source returns, all the forks returns
|
||||
- if the forks errors, all the forks errors with the same error
|
||||
- if a the fork return , it is stopped, but the generator continue with the other
|
||||
- if a the fork error , it is stopped, but the generator continue with the other
|
||||
- if all the fork return , the source is stopped
|
||||
- if all the fork error , the source is errored with the last error
|
||||
- the source start producing a packet when the fastest forked ask for it
|
||||
- the source forks get the packet only when all the forks asked for it, no buffer stores in memory
|
||||
|
||||
```ts
|
||||
import { Synchronized } from '@vates/generator-toolbox'
|
||||
|
||||
async function consume(generator: AsyncGenerator) {
|
||||
for await (const val of generator) {
|
||||
console.log({val})
|
||||
}
|
||||
}
|
||||
const forker = new Synchronized(generator)
|
||||
const first = forker.fork('first')
|
||||
const second = forker.fork('second')
|
||||
await Promise.all([consume(first), consume(second)])
|
||||
```
|
||||
|
||||
Note: you can stop early a generator by calling `generator.return()`, and you can stop in in error by calling `generator.throw(error)`
|
||||
|
||||
## Contributions
|
||||
|
||||
Contributions are _very_ welcomed, either on the documentation or on
|
||||
the code.
|
||||
|
||||
You may:
|
||||
|
||||
- report any [issue](https://github.com/vatesfr/xen-orchestra/issues)
|
||||
you've encountered;
|
||||
- fork and create a pull request.
|
||||
|
||||
## License
|
||||
|
||||
[MIT](https://spdx.org/licenses/MIT) © [Vates SAS](https://vates.fr)
|
||||
6
@vates/generator-toolbox/eslint.config.mjs
Normal file
6
@vates/generator-toolbox/eslint.config.mjs
Normal file
@@ -0,0 +1,6 @@
|
||||
// @ts-check
|
||||
|
||||
import eslint from '@eslint/js'
|
||||
import tseslint from 'typescript-eslint'
|
||||
|
||||
export default tseslint.config(eslint.configs.recommended, tseslint.configs.recommended)
|
||||
33
@vates/generator-toolbox/package.json
Normal file
33
@vates/generator-toolbox/package.json
Normal file
@@ -0,0 +1,33 @@
|
||||
{
|
||||
"name": "@vates/generator-toolbox",
|
||||
"version": "0.0.0",
|
||||
"main": "dist/index.mjs",
|
||||
"license": "MIT",
|
||||
"private": false,
|
||||
"type": "module",
|
||||
"devDependencies": {
|
||||
"@eslint/js": "^9.19.0",
|
||||
"@types/node": "^20.6",
|
||||
"typescript": "~5.6",
|
||||
"typescript-eslint": "^8.23.0"
|
||||
},
|
||||
"scripts": {
|
||||
"build": "tsc",
|
||||
"dev": "tsc --watch",
|
||||
"test": "tsc && node --test **/*.test.mjs"
|
||||
},
|
||||
"homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@vates/generator-toolbox",
|
||||
"bugs": "https://github.com/vatesfr/xen-orchestra/issues",
|
||||
"repository": {
|
||||
"directory": "@vates/generator-toolbox",
|
||||
"type": "git",
|
||||
"url": "https://github.com/vatesfr/xen-orchestra.git"
|
||||
},
|
||||
"author": {
|
||||
"name": "Vates SAS",
|
||||
"url": "https://vates.fr"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=20.18"
|
||||
}
|
||||
}
|
||||
3
@vates/generator-toolbox/src/index.mts
Normal file
3
@vates/generator-toolbox/src/index.mts
Normal file
@@ -0,0 +1,3 @@
|
||||
export { Synchronized } from './synchronized.mjs'
|
||||
export { Throttle } from './throttle.mjs'
|
||||
export { Timeout } from './timeout.mjs'
|
||||
148
@vates/generator-toolbox/src/synchronized.mts
Normal file
148
@vates/generator-toolbox/src/synchronized.mts
Normal file
@@ -0,0 +1,148 @@
|
||||
import assert from 'node:assert'
|
||||
|
||||
export class Synchronized<T, TReturn, TNext> {
|
||||
#source: AsyncGenerator<T, TReturn, TNext>
|
||||
#forks = new Map<string, Forked<T, TReturn, TNext>>()
|
||||
#removedForks = new Set<string>()
|
||||
#waitingForks = new Set<string>()
|
||||
#started = false
|
||||
|
||||
#nextValueForksReady?: {
|
||||
promise: Promise<IteratorResult<T>>
|
||||
forksWaitingReject: (error: Error) => void
|
||||
forksWaitingResolve: () => void
|
||||
}
|
||||
|
||||
constructor(source: AsyncGenerator<T, TReturn, TNext>) {
|
||||
this.#source = source
|
||||
}
|
||||
|
||||
fork(uid: string): AsyncGenerator {
|
||||
assert.strictEqual(this.#started, false, `can't create a fork after consuming the data`)
|
||||
const fork = new Forked<T, TReturn, TNext>(this, uid)
|
||||
this.#forks.set(uid, fork)
|
||||
return fork
|
||||
}
|
||||
|
||||
async #resolveWhenAllForksReady(): Promise<IteratorResult<T>> {
|
||||
if (!this.#nextValueForksReady) {
|
||||
throw new Error('Can t wait forks if there are noone waiting')
|
||||
}
|
||||
const { promise, forksWaitingResolve } = this.#nextValueForksReady
|
||||
if (this.#waitingForks.size === this.#forks.size) {
|
||||
// reset value
|
||||
this.#waitingForks.clear()
|
||||
this.#nextValueForksReady = undefined
|
||||
forksWaitingResolve() // for the other forks waiting
|
||||
}
|
||||
return promise
|
||||
}
|
||||
|
||||
async next(uid: string): Promise<IteratorResult<T>> {
|
||||
if (this.#removedForks.has(uid)) {
|
||||
return { done: true, value: undefined }
|
||||
}
|
||||
if (!this.#forks.has(uid)) {
|
||||
throw new Error(`trying to advance fork ${uid} that is not a fork of this one`)
|
||||
}
|
||||
|
||||
if (this.#waitingForks.has(uid)) {
|
||||
throw new Error(`Fork ${uid} is already waiting`)
|
||||
}
|
||||
|
||||
this.#started = true
|
||||
if (this.#nextValueForksReady === undefined) {
|
||||
let forksWaitingResolve = () => {}
|
||||
let forksWaitingReject: (reason?: Error) => void = () => {}
|
||||
const next = this.#source.next().catch(async error => {
|
||||
const e = new Error(`Error in the source generator ${error.message}`, { cause: error })
|
||||
forksWaitingReject(e)
|
||||
// source has failed, kill everything, and stop the forks
|
||||
for (const uid of [...this.#forks.keys()]) {
|
||||
await this.remove(uid, error)
|
||||
}
|
||||
})
|
||||
const promise = Promise.all([
|
||||
next,
|
||||
new Promise((_resolve, _reject) => {
|
||||
forksWaitingResolve = () => _resolve(undefined)
|
||||
forksWaitingReject = _reject
|
||||
}),
|
||||
]).then(([_]) => _ as IteratorResult<T>)
|
||||
|
||||
this.#nextValueForksReady = { promise, forksWaitingResolve, forksWaitingReject }
|
||||
}
|
||||
this.#waitingForks.add(uid)
|
||||
return this.#resolveWhenAllForksReady()
|
||||
}
|
||||
|
||||
async remove(uid: string, error?: Error): Promise<IteratorResult<T>> {
|
||||
const fork = this.#forks.get(uid)
|
||||
if (fork === undefined) {
|
||||
if (this.#removedForks.has(uid)) {
|
||||
// already removed
|
||||
return { done: true, value: undefined }
|
||||
}
|
||||
throw new Error(`trying to remove fork wih uid ${uid} that is not a fork of this one`)
|
||||
}
|
||||
this.#forks.delete(uid)
|
||||
this.#waitingForks.delete(uid)
|
||||
this.#removedForks.add(uid)
|
||||
try {
|
||||
if (error === undefined) {
|
||||
await fork.return()
|
||||
} else {
|
||||
await fork.throw(error)
|
||||
}
|
||||
} catch (cleaningError) {
|
||||
console.error('Error while cleaning the forked', {
|
||||
cleaningError,
|
||||
sourceError: error,
|
||||
})
|
||||
}
|
||||
|
||||
if (this.#forks.size === 0) {
|
||||
if (error === undefined) {
|
||||
await this.#source.return(undefined as TReturn)
|
||||
} else {
|
||||
await this.#source.throw(error)
|
||||
}
|
||||
// Reject any pending forks waiting for the next value
|
||||
if (this.#nextValueForksReady) {
|
||||
this.#nextValueForksReady.forksWaitingReject(new Error('Source generator terminated.', { cause: error }))
|
||||
this.#nextValueForksReady = undefined
|
||||
}
|
||||
// clear state
|
||||
this.#removedForks.clear()
|
||||
this.#waitingForks.clear()
|
||||
} else {
|
||||
// this fork was maybe blocking the others
|
||||
if (this.#nextValueForksReady) {
|
||||
await this.#resolveWhenAllForksReady()
|
||||
}
|
||||
}
|
||||
return { done: true, value: undefined }
|
||||
}
|
||||
}
|
||||
|
||||
class Forked<T, TReturn, TNext> implements AsyncGenerator<T, TReturn, TNext> {
|
||||
#parent: Synchronized<T, TReturn, TNext>
|
||||
#uid: string
|
||||
constructor(parent: Synchronized<T, TReturn, TNext>, uid: string) {
|
||||
this.#parent = parent
|
||||
this.#uid = uid
|
||||
}
|
||||
next(): Promise<IteratorResult<T>> {
|
||||
return this.#parent.next(this.#uid)
|
||||
}
|
||||
async return(): Promise<IteratorResult<T>> {
|
||||
return this.#parent.remove(this.#uid)
|
||||
}
|
||||
async throw(e: Error): Promise<IteratorResult<T>> {
|
||||
return this.#parent.remove(this.#uid, e)
|
||||
}
|
||||
|
||||
[Symbol.asyncIterator](): AsyncGenerator<T> {
|
||||
return this
|
||||
}
|
||||
}
|
||||
142
@vates/generator-toolbox/src/synchronized.test.mts
Normal file
142
@vates/generator-toolbox/src/synchronized.test.mts
Normal file
@@ -0,0 +1,142 @@
|
||||
import assert from 'node:assert'
|
||||
import { suite, test } from 'node:test'
|
||||
import { Synchronized } from './synchronized.mjs'
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-unused-vars
|
||||
async function* makeRangeGenerator(end = Infinity, progress = { yielded: 0 }, onYielded = (val: unknown) => {}) {
|
||||
for (let i = 0; i < end; i++) {
|
||||
await new Promise(resolve => setTimeout(resolve, 10))
|
||||
yield i
|
||||
onYielded(i)
|
||||
progress.yielded = i
|
||||
}
|
||||
return progress
|
||||
}
|
||||
|
||||
async function consume(
|
||||
iterable: AsyncGenerator,
|
||||
delay = 2500,
|
||||
// eslint-disable-next-line @typescript-eslint/no-unused-vars
|
||||
onConsumed = (val: unknown, iterable: AsyncGenerator) => Promise.resolve(false)
|
||||
) {
|
||||
for await (const val of iterable) {
|
||||
await new Promise(resolve => setTimeout(resolve, delay))
|
||||
if (await onConsumed(val, iterable)) {
|
||||
iterable.return(undefined)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
suite('success', () => {
|
||||
test('if works with multiple consumer', async () => {
|
||||
const progress = { yielded: 0 }
|
||||
const generator = makeRangeGenerator(3, progress)
|
||||
const forker = new Synchronized(generator)
|
||||
const first = forker.fork('first')
|
||||
const second = forker.fork('second')
|
||||
await Promise.all([consume(first, 50), consume(second, 500)])
|
||||
assert.strictEqual(progress.yielded, 2)
|
||||
})
|
||||
|
||||
test('if works with one consumer returning', async () => {
|
||||
const progress = { yielded: 0 }
|
||||
const generator = makeRangeGenerator(10, progress)
|
||||
const forker = new Synchronized(generator)
|
||||
const first = forker.fork('first')
|
||||
const second = forker.fork('second')
|
||||
await new Promise(resolve => setTimeout(resolve, 500))
|
||||
await Promise.all([
|
||||
consume(first, 1),
|
||||
consume(second, 1, async (val: unknown) => {
|
||||
return Promise.resolve(val === 2)
|
||||
}),
|
||||
])
|
||||
assert.strictEqual(progress.yielded, 9)
|
||||
})
|
||||
test('if works with all consumers returning', async () => {
|
||||
const progress = { yielded: 0 }
|
||||
const generator = makeRangeGenerator(10, progress)
|
||||
const forker = new Synchronized(generator)
|
||||
const first = forker.fork('first')
|
||||
const second = forker.fork('second')
|
||||
await new Promise(resolve => setTimeout(resolve, 500))
|
||||
await Promise.all([
|
||||
consume(first, 1, async (val: unknown) => {
|
||||
return Promise.resolve(val === 5)
|
||||
}),
|
||||
consume(second, 1, async (val: unknown) => {
|
||||
return Promise.resolve(val === 2)
|
||||
}),
|
||||
])
|
||||
assert.strictEqual(progress.yielded, 4)
|
||||
})
|
||||
})
|
||||
|
||||
suite('error', () => {
|
||||
test('it fails all if error on source', async () => {
|
||||
const progress = { yielded: 0 }
|
||||
const generator = makeRangeGenerator(5, progress, (val: unknown) => {
|
||||
if (val === 2) {
|
||||
throw new Error('source stop')
|
||||
}
|
||||
})
|
||||
const forker = new Synchronized(generator)
|
||||
const first = forker.fork('first')
|
||||
const second = forker.fork('second')
|
||||
await assert.rejects(Promise.all([consume(first, 1), consume(second, 1)]))
|
||||
|
||||
assert.strictEqual(progress.yielded, 1)
|
||||
})
|
||||
test('it flows until the end if one consumer throws', async () => {
|
||||
const progress = { yielded: 0 }
|
||||
const generator = makeRangeGenerator(10, progress)
|
||||
const forker = new Synchronized(generator)
|
||||
const first = forker.fork('first')
|
||||
const second = forker.fork('second')
|
||||
|
||||
const firstConsume = consume(first, 1)
|
||||
|
||||
const secondConsume = assert.rejects(() =>
|
||||
consume(second, 1, async (val: unknown, iterable) => {
|
||||
if (val === 1) {
|
||||
await iterable.throw(new Error('error on second'))
|
||||
}
|
||||
return false
|
||||
})
|
||||
)
|
||||
await Promise.allSettled([firstConsume, secondConsume])
|
||||
assert.strictEqual(progress.yielded, 9)
|
||||
})
|
||||
test('it throw if ALL consumer throws', async () => {
|
||||
const progress = { yielded: 0 }
|
||||
const generator = makeRangeGenerator(5, progress)
|
||||
const forker = new Synchronized(generator)
|
||||
const first = forker.fork('first')
|
||||
const second = forker.fork('second')
|
||||
|
||||
const firstConsume = consume(first, 1, async (val: unknown, iterable) => {
|
||||
if (val === 3) {
|
||||
await iterable.throw(new Error('error on first'))
|
||||
}
|
||||
return false
|
||||
})
|
||||
const secondConsume = consume(second, 1, async (val: unknown, iterable) => {
|
||||
if (val === 1) {
|
||||
await iterable.throw(new Error('error on second'))
|
||||
}
|
||||
return false
|
||||
})
|
||||
|
||||
await assert.rejects(Promise.all([firstConsume, secondConsume]))
|
||||
assert.strictEqual(progress.yielded, 2)
|
||||
})
|
||||
|
||||
test('It should not allow a new fork after the start', async () => {
|
||||
const progress = { yielded: 0 }
|
||||
const generator = makeRangeGenerator(5, progress)
|
||||
const forker = new Synchronized(generator)
|
||||
const first = forker.fork('first')
|
||||
await first.next()
|
||||
assert.throws(() => forker.fork('second'))
|
||||
})
|
||||
})
|
||||
66
@vates/generator-toolbox/src/throttle.mts
Normal file
66
@vates/generator-toolbox/src/throttle.mts
Normal file
@@ -0,0 +1,66 @@
|
||||
import assert from 'node:assert'
|
||||
|
||||
/**
|
||||
*
|
||||
* This class will throttle the production of a group of generators
|
||||
*
|
||||
* limits: depends on the event loop, so don't expect it to be very precise
|
||||
* works better with more small blocks
|
||||
* The source generator must yield object with a length property
|
||||
* Changing the speed will only be takin into account fot the next packets asked
|
||||
*/
|
||||
export class Throttle {
|
||||
#previousSlot = 0
|
||||
#bytesPerSecond: number | (() => number)
|
||||
get speed(): number {
|
||||
let speed: number
|
||||
if (typeof this.#bytesPerSecond === 'function') {
|
||||
speed = this.#bytesPerSecond()
|
||||
} else {
|
||||
speed = this.#bytesPerSecond
|
||||
}
|
||||
assert.ok(speed > 0, `speed must be greater than zero, ${speed} computed`)
|
||||
return speed
|
||||
}
|
||||
constructor(speed: number | (() => number)) {
|
||||
this.#bytesPerSecond = speed
|
||||
}
|
||||
|
||||
getNextSlot(length: number): { timeout?: ReturnType<typeof setTimeout>; promise?: Promise<unknown> } {
|
||||
assert.notStrictEqual(length, undefined, `throttled stream need to expose a length property }`)
|
||||
assert.ok(length > 0, `throttled stream must expose a positive length property , ${length} given }`)
|
||||
|
||||
const previous = this.#previousSlot
|
||||
const nextSlot = Math.round(previous + (length * 1000) / this.speed)
|
||||
if (nextSlot < Date.now()) {
|
||||
// we're above the limit, go now
|
||||
this.#previousSlot = Date.now()
|
||||
return {}
|
||||
}
|
||||
this.#previousSlot = nextSlot
|
||||
// wait till the next slot
|
||||
// it won't be extremely precise since the event loop is not
|
||||
let timeout: ReturnType<typeof setTimeout> | undefined = undefined
|
||||
const promise = new Promise(function (resolve) {
|
||||
timeout = setTimeout(resolve, nextSlot - Date.now())
|
||||
})
|
||||
return { promise, timeout }
|
||||
}
|
||||
|
||||
async *createThrottledGenerator(source: AsyncGenerator<{ length: number }>): AsyncGenerator<{ length: number }> {
|
||||
let timeout: ReturnType<typeof setTimeout> | undefined
|
||||
try {
|
||||
for await (const value of source) {
|
||||
const res = this.getNextSlot(value.length)
|
||||
timeout = res.timeout
|
||||
// wait for the time slot before yielding the data
|
||||
if (res.promise !== undefined) {
|
||||
await res.promise
|
||||
}
|
||||
yield value
|
||||
}
|
||||
} finally {
|
||||
clearTimeout(timeout)
|
||||
}
|
||||
}
|
||||
}
|
||||
63
@vates/generator-toolbox/src/throttle.test.mts
Normal file
63
@vates/generator-toolbox/src/throttle.test.mts
Normal file
@@ -0,0 +1,63 @@
|
||||
import { Throttle } from './throttle.mjs'
|
||||
import { suite, test } from 'node:test'
|
||||
import assert from 'node:assert'
|
||||
|
||||
async function* makeGenerator(chunkSize = 1024 * 1024, nbChunks = 100) {
|
||||
for (let i = 0; i < nbChunks; i++) {
|
||||
yield Buffer.allocUnsafe(chunkSize)
|
||||
}
|
||||
}
|
||||
|
||||
async function consumes(generator: AsyncGenerator) {
|
||||
// eslint-disable-next-line no-empty,@typescript-eslint/no-unused-vars
|
||||
for await (const data of generator) {
|
||||
}
|
||||
}
|
||||
|
||||
suite('it throttle one', () => {
|
||||
test('base test', async () => {
|
||||
// 10MB/s
|
||||
const throttler = new Throttle(10 * 1024 * 1024)
|
||||
const throttled = throttler.createThrottledGenerator(makeGenerator(1024 * 1024, 20))
|
||||
|
||||
const start = Date.now()
|
||||
await consumes(throttled)
|
||||
const end = Date.now()
|
||||
assert.strictEqual(Math.round((end - start) / 1000), 2)
|
||||
})
|
||||
test('variable speed test', async () => {
|
||||
// 10MB/s
|
||||
const throttler = new Throttle(() => 10 * 1024 * 1024)
|
||||
const throttled = throttler.createThrottledGenerator(makeGenerator(1024 * 1024, 20))
|
||||
|
||||
const start = Date.now()
|
||||
await consumes(throttled)
|
||||
const end = Date.now()
|
||||
assert.strictEqual(Math.round((end - start) / 1000), 2)
|
||||
})
|
||||
test('multiple generator test', async () => {
|
||||
// 10MB/s
|
||||
const throttler = new Throttle(10 * 1024 * 1024)
|
||||
const first = throttler.createThrottledGenerator(makeGenerator(1024 * 1024, 10))
|
||||
const second = throttler.createThrottledGenerator(makeGenerator(1024 * 1024, 10))
|
||||
const start = Date.now()
|
||||
|
||||
await Promise.all([consumes(first), consumes(second)])
|
||||
const end = Date.now()
|
||||
assert.strictEqual(Math.round((end - start) / 1000), 2)
|
||||
})
|
||||
test(`it doesn't accept negative value`, async () => {
|
||||
// 10MB/s
|
||||
let throttler = new Throttle(-10 * 1024 * 1024)
|
||||
let generator = throttler.createThrottledGenerator(makeGenerator(1024 * 1024, 10))
|
||||
await assert.rejects(() => consumes(generator))
|
||||
|
||||
let index = 5
|
||||
throttler = new Throttle(() => {
|
||||
index--
|
||||
return index > 0 ? 10 * 1024 * 1024 : -10 * 1024 * 1024
|
||||
})
|
||||
generator = throttler.createThrottledGenerator(makeGenerator(1024 * 1024, 10))
|
||||
await assert.rejects(() => consumes(generator))
|
||||
})
|
||||
})
|
||||
37
@vates/generator-toolbox/src/timeout.mts
Normal file
37
@vates/generator-toolbox/src/timeout.mts
Normal file
@@ -0,0 +1,37 @@
|
||||
import assert from 'node:assert'
|
||||
export class Timeout<T> implements AsyncGenerator {
|
||||
#source: AsyncGenerator<T>
|
||||
#timeout: number
|
||||
constructor(source: AsyncGenerator<T>, timeout: number) {
|
||||
assert.ok(timeout > 0, `Timeout value must be positive, ${timeout} received`)
|
||||
this.#source = source
|
||||
this.#timeout = timeout
|
||||
}
|
||||
async next(): Promise<IteratorResult<T>> {
|
||||
let timeout: ReturnType<typeof setTimeout>
|
||||
const promiseTimeout = new Promise((_, reject) => {
|
||||
timeout = setTimeout(() => {
|
||||
reject(new Error('Timeout reached '))
|
||||
}, this.#timeout)
|
||||
})
|
||||
const promiseNext = new Promise((resolve, reject) => {
|
||||
this.#source.next().then(res => {
|
||||
// ensure timetout won't fire later
|
||||
clearTimeout(timeout)
|
||||
resolve(res)
|
||||
}, reject)
|
||||
})
|
||||
// promiseTimeout will never resolve
|
||||
return Promise.race([promiseNext, promiseTimeout]) as Promise<IteratorResult<T>>
|
||||
}
|
||||
return(): Promise<IteratorResult<T>> {
|
||||
return this.#source.return(undefined)
|
||||
}
|
||||
throw(e: Error): Promise<IteratorResult<T>> {
|
||||
return this.#source.throw(e)
|
||||
}
|
||||
|
||||
[Symbol.asyncIterator](): AsyncGenerator<T> {
|
||||
return this
|
||||
}
|
||||
}
|
||||
67
@vates/generator-toolbox/src/timeout.test.mts
Normal file
67
@vates/generator-toolbox/src/timeout.test.mts
Normal file
@@ -0,0 +1,67 @@
|
||||
import { describe, it } from 'node:test'
|
||||
import assert from 'node:assert'
|
||||
import { Timeout } from './timeout.mjs'
|
||||
|
||||
describe('Timeout class', () => {
|
||||
it('should reject a timeout with a negative or zero value ', async () => {
|
||||
// Create a mock async generator that never resolves
|
||||
const mockGenerator = (async function* () {
|
||||
yield 1
|
||||
})()
|
||||
|
||||
assert.throws(() => new Timeout(mockGenerator, 0))
|
||||
assert.throws(() => new Timeout(mockGenerator, -10))
|
||||
})
|
||||
it('should resolve with the value from the source generator if it completes before the timeout', async () => {
|
||||
// Create a mock async generator that resolves immediately
|
||||
const mockGenerator = (async function* () {
|
||||
yield 1
|
||||
})()
|
||||
|
||||
const timeout = new Timeout(mockGenerator, 100) // 100ms timeout
|
||||
|
||||
const result = await timeout.next()
|
||||
assert.strictEqual(result.value, 1)
|
||||
assert.strictEqual(result.done, false)
|
||||
})
|
||||
|
||||
it('should reject with a timeout error if the source generator takes too long', async () => {
|
||||
// Create a mock async generator that never resolves
|
||||
// eslint-disable-next-line require-yield
|
||||
const mockGenerator = (async function* () {
|
||||
await new Promise(() => {}) // Never resolves
|
||||
})()
|
||||
|
||||
const timeout = new Timeout(mockGenerator, 10) // 10ms timeout
|
||||
|
||||
await assert.rejects(timeout.next(), 'Expected timeout error was not thrown')
|
||||
})
|
||||
it('should allow iteration using for-await-of', async () => {
|
||||
// Create a mock async generator that yields values
|
||||
const mockGenerator: AsyncGenerator<number> = (async function* () {
|
||||
yield 1
|
||||
yield 2
|
||||
yield 3
|
||||
})()
|
||||
|
||||
const timeout = new Timeout(mockGenerator, 100) // 100ms timeout
|
||||
|
||||
const results: Array<number> = []
|
||||
for await (const value of timeout) {
|
||||
results.push(value as number)
|
||||
}
|
||||
|
||||
assert.deepStrictEqual(results, [1, 2, 3])
|
||||
})
|
||||
|
||||
it('should handle the throw method correctly', async () => {
|
||||
// Create a mock async generator with a throw method
|
||||
const mockGenerator = (async function* () {
|
||||
yield 1
|
||||
})()
|
||||
|
||||
const timeout = new Timeout(mockGenerator, 100) // 100ms timeout
|
||||
|
||||
await assert.rejects(timeout.throw(new Error('Test error')), 'Expected error was not thrown')
|
||||
})
|
||||
})
|
||||
12
@vates/generator-toolbox/tsconfig.json
Normal file
12
@vates/generator-toolbox/tsconfig.json
Normal file
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"compilerOptions": {
|
||||
"target": "es2023",
|
||||
"module": "ESNext",
|
||||
"moduleResolution": "bundler",
|
||||
"strict": true,
|
||||
"declaration": true, // Generates `.d.ts` files for type safety
|
||||
"outDir": "dist" // Outputs compiled code to the `dist` directory
|
||||
},
|
||||
"include": ["src/**/*"],
|
||||
"exclude": ["node_modules", "dist"]
|
||||
}
|
||||
@@ -43,6 +43,7 @@
|
||||
|
||||
<!--packages-start-->
|
||||
|
||||
- @vates/generator-toolbox major
|
||||
- @vates/types major
|
||||
- @xen-orchestra/rest-api minor
|
||||
- @xen-orchestra/vmware-explorer patch
|
||||
|
||||
17
yarn.lock
17
yarn.lock
@@ -4536,6 +4536,13 @@
|
||||
dependencies:
|
||||
undici-types "~6.19.2"
|
||||
|
||||
"@types/node@^20.6":
|
||||
version "20.17.19"
|
||||
resolved "https://registry.yarnpkg.com/@types/node/-/node-20.17.19.tgz#0f2869555719bef266ca6e1827fcdca903c1a697"
|
||||
integrity sha512-LEwC7o1ifqg/6r2gn9Dns0f1rhK+fPFDoMiceTJ6kWmVk6bgXBI/9IOWfVan4WiAavK9pIVWdX0/e3J+eEUh5A==
|
||||
dependencies:
|
||||
undici-types "~6.19.2"
|
||||
|
||||
"@types/novnc__novnc@^1.5.0":
|
||||
version "1.5.0"
|
||||
resolved "https://registry.yarnpkg.com/@types/novnc__novnc/-/novnc__novnc-1.5.0.tgz#5f58687f0fa6591b75e91eb75237520da440dbb4"
|
||||
@@ -5420,7 +5427,7 @@ acorn-walk@^8.1.1:
|
||||
|
||||
acorn@5.X, acorn@^5.0.3:
|
||||
version "5.7.4"
|
||||
resolved "https://registry.yarnpkg.com/acorn/-/acorn-5.7.4.tgz#3e8d8a9947d0599a1796d10225d7432f4a4acf5e"
|
||||
resolved "https://registry.npmjs.org/acorn/-/acorn-5.7.4.tgz"
|
||||
integrity sha512-1D++VG7BhrtvQpNbBzovKNc1FLGGEE/oGe7b9xJm/RFHMBeUaUGpluV9RLjZa47YFdPcDAenEYuq9pQPcMdLJg==
|
||||
|
||||
acorn@^3.1.0:
|
||||
@@ -5430,12 +5437,12 @@ acorn@^3.1.0:
|
||||
|
||||
acorn@^4.0.4, acorn@~4.0.2:
|
||||
version "4.0.13"
|
||||
resolved "https://registry.yarnpkg.com/acorn/-/acorn-4.0.13.tgz#105495ae5361d697bd195c825192e1ad7f253787"
|
||||
resolved "https://registry.npmjs.org/acorn/-/acorn-4.0.13.tgz"
|
||||
integrity sha512-fu2ygVGuMmlzG8ZeRJ0bvR41nsAkxxhbyk8bZ1SS521Z7vmgJFTQQlfz/Mp/nJexGBz+v8sC9bM6+lNgskt4Ug==
|
||||
|
||||
acorn@^7.0.0, acorn@^7.1.1:
|
||||
version "7.4.1"
|
||||
resolved "https://registry.yarnpkg.com/acorn/-/acorn-7.4.1.tgz#feaed255973d2e77555b83dbc08851a6c63520fa"
|
||||
resolved "https://registry.npmjs.org/acorn/-/acorn-7.4.1.tgz"
|
||||
integrity sha512-nQyp0o1/mNdbTO1PO6kHkwSrmgZ0MT/jCCpNiwbUjGoRN4dlBhqJtoQuCnEOKzgTVwg0ZWiCoQy6SxMebQVh8A==
|
||||
|
||||
acorn@^8.11.0, acorn@^8.14.0, acorn@^8.4.1, acorn@^8.5.0, acorn@^8.8.2, acorn@^8.9.0:
|
||||
@@ -14134,7 +14141,7 @@ miller-rabin@^4.0.0:
|
||||
|
||||
mime-db@1.52.0:
|
||||
version "1.52.0"
|
||||
resolved "https://registry.yarnpkg.com/mime-db/-/mime-db-1.52.0.tgz#bbabcdc02859f4987301c856e3387ce5ec43bf70"
|
||||
resolved "https://registry.npmjs.org/mime-db/-/mime-db-1.52.0.tgz"
|
||||
integrity sha512-sPU4uV7dYlvtWJxwwxHD0PuihVNiE7TyAbQ5SWxDCB9mUYvOgroQOwYQQOKPJ8CIbE+1ETVlOoK1UC2nU3gYvg==
|
||||
|
||||
"mime-db@>= 1.43.0 < 2", mime-db@^1.52.0:
|
||||
@@ -19674,7 +19681,7 @@ typescript@5.2:
|
||||
resolved "https://registry.yarnpkg.com/typescript/-/typescript-5.2.2.tgz#5ebb5e5a5b75f085f22bc3f8460fba308310fa78"
|
||||
integrity sha512-mI4WrpHsbCIcwT9cF4FZvr80QUeKvsUsUvKDoR+X/7XHQH98xYD8YHZg7ANtz2GtZt/CBq2QJ0thkGJMHfqc1w==
|
||||
|
||||
typescript@~5.6.3:
|
||||
typescript@~5.6, typescript@~5.6.3:
|
||||
version "5.6.3"
|
||||
resolved "https://registry.yarnpkg.com/typescript/-/typescript-5.6.3.tgz#5f3449e31c9d94febb17de03cc081dd56d81db5b"
|
||||
integrity sha512-hjcS1mhfuyi4WW8IWtjP7brDrG2cuDZukyrYrSauoXGNgx0S7zceP07adYkJycEr56BOUTNPzbInooiN3fn1qw==
|
||||
|
||||
Reference in New Issue
Block a user