<commit_before>context("worker")

context = rzmq::init.context()
socket = rzmq::init.socket(context, "ZMQ_REP")
rzmq::bind.socket(socket, "tcp://*:55443")

start_worker = function(id="1", url="tcp://localhost:55443") {
	p = parallel::mcparallel(worker(id, url, 1024))
	msg = rzmq::receive.socket(socket)
	testthat::expect_equal(msg$id, "WORKER_UP")
    p
}

send_common = function() {
	rzmq::send.socket(socket, list(fun=function(x) x, const=list(), seed=1))
	msg = rzmq::receive.socket(socket)
	testthat::expect_equal(msg$id, "WORKER_READY")
}

shutdown_worker = function(p, id="1") {
	rzmq::send.socket(socket, list(id="WORKER_STOP"))
	msg = rzmq::receive.socket(socket)
    rzmq::send.socket(socket, "invalid") # already shut down
	testthat::expect_equal(msg$id, "WORKER_DONE")
	testthat::expect_equal(msg$worker_id, id)
	testthat::expect_is(msg$time, "proc_time")
	testthat::expect_is(msg$calls, "numeric")
	parallel::mccollect(p)
}

test_that("control flow", {
    p = start_worker()
    send_common()
    shutdown_worker(p)
})

test_that("common data redirect", {
    p = start_worker()

	rzmq::send.socket(socket, list(redirect="tcp://localhost:55443"))
	msg = rzmq::receive.socket(socket)
	testthat::expect_equal(msg$id, "WORKER_UP")

    send_common()
    shutdown_worker(p)
})
<commit_msg>retab<commit_after>context("worker")

context = rzmq::init.context()
socket = rzmq::init.socket(context, "ZMQ_REP")
rzmq::bind.socket(socket, "tcp://*:55443")

start_worker = function(id="1", url="tcp://localhost:55443") {
    p = parallel::mcparallel(worker(id, url, 1024))
    msg = rzmq::receive.socket(socket)
    testthat::expect_equal(msg$id, "WORKER_UP")
    p
}

send_common = function() {
    rzmq::send.socket(socket, list(fun=function(x) x, const=list(), seed=1))
    msg = rzmq::receive.socket(socket)
    testthat::expect_equal(msg$id, "WORKER_READY")
}

shutdown_worker = function(p, id="1") {
    rzmq::send.socket(socket, list(id="WORKER_STOP"))
    msg = rzmq::receive.socket(socket)
    rzmq::send.socket(socket, list()) # already shut down
    testthat::expect_equal(msg$id, "WORKER_DONE")
    testthat::expect_equal(msg$worker_id, id)
    testthat::expect_is(msg$time, "proc_time")
    testthat::expect_is(msg$calls, "numeric")
    parallel::mccollect(p)
}

test_that("control flow", {
    p = start_worker()
    send_common()
    shutdown_worker(p)
})

test_that("common data redirect", {
    p = start_worker()

    rzmq::send.socket(socket, list(redirect="tcp://localhost:55443"))
    msg = rzmq::receive.socket(socket)
    testthat::expect_equal(msg$id, "WORKER_UP")

    send_common()
    shutdown_worker(p)
})

{'fraction_non_alphanumeric': 0.1791271347248577, 'fraction_numerical': 0.01669829222011385, 'mean_word_length': 5.882506527415144, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '7459988', 'n_tokens_mistral': 1028, 'n_tokens_neox': 984, 'n_words': 142}