<commit_before>const {Subject, of} = require('rx')
const {finalize, first, map, filter, catchError} = require('rx/operators')
const {Worker, MessageChannel} = require('worker_threads')
const path = require('path')
const _ = require('lodash')
const Transport = require('./transport')
const service = require('sepal/service')

const WORKER_PATH = path.join(__dirname, 'worker.js')

const bootstrapWorker$ = ({workerId, jobName, logConfig}) => {
    const worker$ = new Subject()
    const worker = new Worker(WORKER_PATH)
    const {port1: localPort, port2: remotePort} = new MessageChannel()
    worker.on('message', message => {
        message.ready && worker$.next({worker, port: localPort})
    })
    worker.postMessage({workerId, jobName, logConfig, port: remotePort}, [remotePort])
    return worker$.pipe(
        first()
    )
}

const setupWorker = ({workerId, jobName, jobPath, worker, port}) => {
    const disposables = []
    const transport = Transport({port, jobName, workerId})

    transport.onChannel(
        ({conversationId: serviceId, in$: response$, out$: request$}) => {
            if (service) {
                service.start(serviceId, request$, response$)
            }
        }
    )
 
    const submit$ = ({requestId, initArgs, args, args$, cmd$}) => {
        const {in$: request$, out$: response$} = transport.createChannel('job')

        const start = () =>
            request$.next({start: {requestId, jobPath, initArgs, args}})

        const stop = () =>
            request$.complete()

        args$ && args$.subscribe(
            value => request$.next({next: {requestId, value}})
            // [TODO] handle error
        )

        cmd$ && cmd$.subscribe(
            cmd => request$.next({next: {requestId, cmd}})
        )

        start()
<commit_msg>Minor fix.<commit_after>const {Subject, of} = require('rx')
const {finalize, first, map, filter, catchError} = require('rx/operators')
const {Worker, MessageChannel} = require('worker_threads')
const path = require('path')
const _ = require('lodash')
const Transport = require('./transport')
const service = require('sepal/service')

const WORKER_PATH = path.join(__dirname, 'worker.js')

const bootstrapWorker$ = ({workerId, jobName, logConfig}) => {
    const worker$ = new Subject()
    const worker = new Worker(WORKER_PATH)
    const {port1: localPort, port2: remotePort} = new MessageChannel()
    worker.on('message', message => {
        message.ready && worker$.next({worker, port: localPort})
    })
    worker.postMessage({workerId, jobName, logConfig, port: remotePort}, [remotePort])
    return worker$.pipe(
        first()
    )
}

const setupWorker = ({workerId, jobName, jobPath, worker, port}) => {
    const disposables = []
    const transport = Transport({port, jobName, workerId})

    transport.onChannel(
        ({conversationId: serviceId, in$: response$, out$: request$}) => {
            if (serviceId) {
                service.start(serviceId, request$, response$)
            }
        }
    )
 
    const submit$ = ({requestId, initArgs, args, args$, cmd$}) => {
        const {in$: request$, out$: response$} = transport.createChannel('job')

        const start = () =>
            request$.next({start: {requestId, jobPath, initArgs, args}})

        const stop = () =>
            request$.complete()

        args$ && args$.subscribe(
            value => request$.next({next: {requestId, value}})
            // [TODO] handle error
        )

        cmd$ && cmd$.subscribe(
            cmd => request$.next({next: {requestId, cmd}})
        )

        start()

{'fraction_non_alphanumeric': 0.15643234231731037, 'fraction_numerical': 0.0011114198388441233, 'mean_word_length': 2.891891891891892, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, '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': '2119307', 'n_tokens_mistral': 1028, 'n_tokens_neox': 1003, 'n_words': 261}