<commit_before>const {Subject, ReplaySubject, defer, zip, concat, of} = require('rxjs')
const {delay, filter, finalize, first, map, mapTo, scan, tap} = require('rxjs/operators')
const {v4: uuid} = require('uuid')
const _ = require('lodash')
const {tag} = require('sepal/tag')
const log = require('sepal/log').getLogger('tokenLimiter')

const tokenLimiterTag = name => tag('TokenLimiter', name)

const TokenLimiter$ = ({name, rateWindowMs = 1000, maxRate, maxConcurrency}, onIdle) => {

    const requestId$ = new Subject()
    const rateToken$ = new Subject()
    const concurrencyToken$ = new Subject()
    const responseToken$ = new Subject()

    const msg = msg =>
        `${tokenLimiterTag(name)} ${msg}`

    const tokenId = ({requestId, rateToken, concurrencyToken}) =>
        `[${requestId}${rateToken ? `.R${rateToken}` : ''}${concurrencyToken ? `.C${concurrencyToken}` : ''}]`

    const initialToken$ = count =>
        of(...(_.range(1, count + 1)))

    const rateLimit$ = maxRate
        ? concat(initialToken$(maxRate), rateToken$)
        : requestId$.pipe(mapTo())

    const concurrencyLimit$ = maxConcurrency
        ? concat(initialToken$(maxConcurrency), concurrencyToken$)
        : requestId$.pipe(mapTo())

    const token$ = zip(
        requestId$,
        rateLimit$,
        concurrencyLimit$
    ).pipe(
        map(([requestId, rateToken, concurrencyToken]) =>
            ({requestId, rateToken, concurrencyToken})
        )
    )

    const serveToken = token =>
        responseToken$.next(token)

    const assertToken = token => {
        if (!token) {
            throw new Error('Token is required')
        }
    }

    const recycleRateToken = token => {
        assertToken(token)
        if (maxRate) {
            log.debug(msg(`recycling rate token ${token.rateToken}/${maxRate} from token ${tokenId(token)}`))
            rateToken$.next(token.rateToken)
        }
    }

    const recycleConcurrencytoken = (token, cancelled) => {
        assertToken(token)
        if (maxConcurrency) {
            log.debug(msg(`recycling ${cancelled ? 'cancelled ' : ''}concurrency token ${token.concurrencyToken}/${maxConcurrency} from token ${tokenId(token)}`))
            concurrencyToken$.next(token.concurrencyToken)
        }
    }

    token$.pipe(
        tap(token => serveToken(token)),
        delay(rateWindowMs)
    ).subscribe({
        next: token => recycleRateToken(token),
        error: error => log.fatal(msg('token stream failed:'), error),
        complete: () => log.fatal(msg('token stream completed'))
    })

    const count$ = new Subject()

    count$.pipe(
        scan((count, current) => count + current, 0)
    ).subscribe(
        count => {
            if (count === 0) {
                log.debug(() => msg('idle'))
                onIdle && onIdle()
            }
        }
    )

    const CurrentToken = () => {
        const current = {}
        count$.next(1)
        return {
            set(token) {
                current.token = token
            },
            recycleOrCancel() {
                if (current.token) {
                    count$.next(-1)
                    recycleConcurrencytoken(current.token, false)
                } else {
                    current.cancel = true
                }
            },
            recycleCancelled() {
                if (current.token && current.cancel) {
                    count$.next(-1)
                    recycleConcurrencytoken(current.token, true)
                }<commit_msg>Implemented TokenLimiter debounced idle detection.<commit_after>const {Subject, ReplaySubject, defer, zip, concat, of} = require('rxjs')
const {debounceTime, delay, filter, finalize, first, map, mapTo, scan, tap} = require('rxjs/operators')
const {v4: uuid} = require('uuid')
const _ = require('lodash')
const {tag} = require('sepal/tag')
const log = require('sepal/log').getLogger('tokenLimiter')

const tokenLimiterTag = name => tag('TokenLimiter', name)

const TokenLimiter$ = ({name, rateWindowMs = 1000, maxRate, maxConcurrency, idleMs = 5000}, onIdle) => {

    const requestId$ = new Subject()
    const rateToken$ = new Subject()
    const concurrencyToken$ = new Subject()
    const responseToken$ = new Subject()

    const msg = msg =>
        `${tokenLimiterTag(name)} ${msg}`

    const tokenId = ({requestId, rateToken, concurrencyToken}) =>
        `[${requestId}${rateToken ? `.R${rateToken}` : ''}${concurrencyToken ? `.C${concurrencyToken}` : ''}]`

    const initialToken$ = count =>
        of(...(_.range(1, count + 1)))

    const rateLimit$ = maxRate
        ? concat(initialToken$(maxRate), rateToken$)
        : requestId$.pipe(mapTo())

    const concurrencyLimit$ = maxConcurrency
        ? concat(initialToken$(maxConcurrency), concurrencyToken$)
        : requestId$.pipe(mapTo())

    const token$ = zip(
        requestId$,
        rateLimit$,
        concurrencyLimit$
    ).pipe(
        map(([requestId, rateToken, concurrencyToken]) =>
            ({requestId, rateToken, concurrencyToken})
        )
    )

    const serveToken = token =>
        responseToken$.next(token)

    const assertToken = token => {
        if (!token) {
            throw new Error('Token is required')
        }
    }

    const recycleRateToken = token => {
        assertToken(token)
        if (maxRate) {
            log.debug(msg(`recycling rate token ${token.rateToken}/${maxRate} from token ${tokenId(token)}`))
            rateToken$.next(token.rateToken)
        }
    }

    const recycleConcurrencytoken = (token, cancelled) => {
        assertToken(token)
        if (maxConcurrency) {
            log.debug(msg(`recycling ${cancelled ? 'cancelled ' : ''}concurrency token ${token.concurrencyToken}/${maxConcurrency} from token ${tokenId(token)}`))
            concurrencyToken$.next(token.concurrencyToken)
        }
    }

    token$.pipe(
        tap(token => serveToken(token)),
        delay(rateWindowMs)
    ).subscribe({
        next: token => recycleRateToken(token),
        error: error => log.fatal(msg('token stream failed:'), error),
        complete: () => log.fatal(msg('token stream completed'))
    })

    const count$ = new Subject()

    count$.pipe(
        scan((count, current) => count + current, 0),
        map(count => count === 0),
        debounceTime(idleMs)
    ).subscribe(
        idle => {
            if (idle) {
                log.debug(() => msg('idle'))
                onIdle && onIdle()
            }
        }
    )

    const CurrentToken = () => {
        const current = {}
        count$.next(1)
        return {
            set(token) {
                current.token = token
            },
            recycleOrCancel() {
                if (current.token) {
                    count$.next(-1)
                    recycleConcurrencytoken(current.token, false)
                } else {
                    current.cancel = true
                }
            },
            recycleCancelled() {
                if (current.token && current.cancel) {
                    count$.next(-1)
                    recycleConcurrencytoken(current.token, true)
                }
{'fraction_non_alphanumeric': 0.14048317274123726, 'fraction_numerical': 0.0039100684261974585, 'mean_word_length': 2.1578483245149913, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3618413', 'n_tokens_mistral': 2072, 'n_tokens_neox': 2002, 'n_words': 432}