<commit_before>const {concat, EMPTY, ReplaySubject, isObservable, throwError, pipe} = require('rx')
const {catchError, filter, finalize, scan, switchMapTo, takeWhile} = require('rx/operators')
const {v4: uuid} = require('uuid')
const log = require('sepal/log').getLogger('rxjs')

const status$ = new ReplaySubject()

const push = (context, description) => {
    log.trace(`Finalize registered: ${description}`)
    status$.next({context, description, value: 1})
}

const pull = (context, description) => {
    log.trace(`Finalize completed: ${description}`)
    status$.next({context, description, value: -1})
}

const onFinalize = (context, callback, description) => {
    const complete = () => pull(context, description)
    push(context, description)

    return pipe(
        finalize(() => {
            try {
                const result = callback()
                if (isObservable(result)) {
                    result.subscribe({
                        error: error => {
                            log.error(`Finalize failed: ${description}`, error)
                            complete()
                        },
                        complete
                    })
                } else {
                    complete()
                }
            } catch (e) {
                complete()
                throw e
            }
        })
    )
}

// const finalize$ = defer(() => _finalizeSubject
const onFinalize$ = context =>
    status$.pipe(
        filter(({context: currentContext}) => context === currentContext),
        scan((acc, {value}) => acc + value, 0),
        takeWhile(acc => {
            if (acc > 0) {
                return true
            } else {
                log.debug('All finalize blocks completed')
                return false
            }
        }),
        switchMapTo(EMPTY)
    )

const finalize$ = (observable$, finalizeCallback, description) => {
    const context = uuid()
    return concat(
        observable$.pipe(
            onFinalize(context, finalizeCallback, description),
            catchError(e => concat(
                onFinalize$(context),
                throwError(e))
            )
        ),
        onFinalize$(context)
    )
}
<commit_msg>Replaced finalize$ global status with local.<commit_after>const {concat, EMPTY, ReplaySubject, isObservable, throwError, pipe} = require('rx')
const {catchError, finalize, scan, switchMapTo, takeWhile} = require('rx/operators')
const log = require('sepal/log').getLogger('rxjs')

const push = (status$, description) => {
    log.trace(`Finalize registered: ${description}`)
    status$.next({description, value: 1})
}

const pull = (status$, description) => {
    log.trace(`Finalize completed: ${description}`)
    status$.next({description, value: -1})
}

const onFinalize = (status$, callback, description) => {
    const complete = () => pull(status$, description)
    push(status$, description)

    return pipe(
        finalize(() => {
            try {
                const result = callback()
                if (isObservable(result)) {
                    result.subscribe({
                        error: error => {
                            log.error(`Finalize failed: ${description}`, error)
                            complete()
                        },
                        complete
                    })
                } else {
                    complete()
                }
            } catch (e) {
                complete()
                throw e
            }
        })
    )
}

// const finalize$ = defer(() => _finalizeSubject
const onFinalize$ = status$ =>
    status$.pipe(
        scan((acc, {value}) => acc + value, 0),
        takeWhile(acc => {
            if (acc > 0) {
                return true
            } else {
                log.debug('All finalize blocks completed')
                return false
            }
        }),
        switchMapTo(EMPTY)
    )

const finalize$ = (observable$, finalizeCallback, description) => {
    const status$ = new ReplaySubject()
    return concat(
        observable$.pipe(
            onFinalize(status$, finalizeCallback, description),
            catchError(e => concat(
                onFinalize$(status$),
                throwError(e))
            )
        ),
        onFinalize$(status$)
    )
}

{'fraction_non_alphanumeric': 0.12676382142030998, 'fraction_numerical': 0.002081887578070784, 'mean_word_length': 1.652760736196319, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 30, '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': '3498999', 'n_tokens_mistral': 1135, 'n_tokens_neox': 1093, 'n_words': 261}