context("worker usage")

test_that("connect to invalid endpoint errors", {
    w = methods::new(CMQWorker)
    expect_error(w$connect("tcp://localhost:12345", 0L))
    w$close()
})

test_that("recv without pending workers errors before timeout", {
    m = methods::new(CMQMaster)
    addr = m$listen("inproc://endpoint")
    expect_error(m$recv(-1L))
    m$close(500L)
})

test_that("recv timeout works", {
    m = methods::new(CMQMaster)
    addr = m$listen("inproc://endpoint")
    m$add_pending_workers(1L)
    expect_error(m$recv(0L))
    m$close(500L)
})

test_that("worker evaluation", {
    m = methods::new(CMQMaster)
    w = methods::new(CMQWorker, m$context())
    addr = m$listen("inproc://endpoint")
    m$add_pending_workers(1L)
    w$connect(addr, 500L)

    m$recv(500L)
    m$send(expression(5 * 2))
    status = w$process_one()
    result = m$recv(500L)

    expect_true(status)
    expect_equal(result, 10)

    w$close()
    m$close(500L)
})

test_that("export variable to worker", {
    m = methods::new(CMQMaster)
    w = methods::new(CMQWorker, m$context())
    addr = m$listen("inproc://endpoint")
    m$add_pending_workers(1L)
    w$connect(addr, 500L)

    m$add_env("x", 3)
    m$recv(500L)
    m$send(expression(5 + x))
    status = w$process_one()
    result = m$recv(500L)
    expect_true(status)
    expect_equal(result, 8)

    m$add_env("x", 5)
    m$send(expression(5 + x))
    status = w$process_one()
    result = m$recv(500L)
    expect_true(status)
    expect_equal(result, 10)

    w$close()
    m$close(500L)
})

test_that("load package on worker", {
    m = methods::new(CMQMaster)
    w = methods::new(CMQWorker, m$context())
    addr = m$listen("inproc://endpoint")
    m$add_pending_workers(1L)
    w$connect(addr, 500L)

    m$add_pkg("parallel")

    m$recv(500L)
    m$send(expression(splitIndices(1, 1)[[1]]))
    status = w$process_one()
    result = m$recv(500L)

    expect_true(status)
    expect_equal(result, 1)

    w$close()
    m$close(500L)
})

test_that("errors are sent back to master", {
    skip("this works interactively but evaluates the error on testthat")

    m = methods::new(CMQMaster)
    w = methods::new(CMQWorker, m$context())
    addr = m$listen("inproc://endpoint")
    m$add_pending_workers(1L)
    w$connect(addr, 500L)

    m$recv(500L)
    m$send(expression(stop("errmsg")))
    status = w$process_one()
    result = m$recv(500L)

    expect_true(status)
    expect_true(inherits(result, c("condition", "worker_error")))

    w$close()
    m$close(500L)
})

test_that("worker R API", {
    skip_on_os("windows")
    skip_if_not(has_connectivity("127.0.0.1")) # -> this or inproc w/ passing context

    m = methods::new(CMQMaster)
    addr = m$listen("tcp://127.0.0.1:*")
    m$add_pending_workers(1L)
#    addr = m$listen("inproc://endpoint") # mailbox.cpp assertion error

    p = parallel::mcparallel(worker(addr))
    expect_null(m$recv(5000L))
    m$send(expression(5 + 1))
    res = m$recv(500L)
    expect_equal(res[[1]], 6)

    m$send_shutdown()
    pc = parallel::mccollect(p, wait=TRUE, timeout=0.5)
    expect_equal(pc[[1]], NULL)
    m$close(500L)
})

test_that("communication with two workers", {
    skip_on_os("windows")
    skip_if_not(has_connectivity("127.0.0.1"))

    m = methods::new(CMQMaster)
    addr = m$listen("tcp://127.0.0.1:*")
    m$add_pending_workers(2L)
    w1 = parallel::mcparallel(worker(addr))
    w2 = parallel::mcparallel(worker(addr))

    expect_null(m$recv(5000L)) # worker 1 up
    m$send(expression({ Sys.sleep(0.5); 5 + 2 }))
    expect_null(m$recv(500L)) # worker 2 up
    m$send(expression({ Sys.sleep(0.5); 3 + 1 }))
    r1 = m$recv(1000L)
    m$send_shutdown()
    r2 = m$recv(1000L)
    m$send_shutdown()
    expect_equal(sort(c(r1, r2)), c(4,7))

    coll1 = parallel::mccollect(w1, wait=TRUE, timeout=0.5)
    expect_equal(names(coll1), as.character(w1$pid))
    coll2 = parallel::mccollect(w2, wait=TRUE, timeout=0.5)
    expect_equal(names(coll2), as.character(w2$pid))

    m$close(500L)
})