module executor

import time

fn test_run_drains_accepted_jobs_then_wait_returns() {
	mut ex := new(queue_size: 4)!
	ran := chan int{cap: 3}
	result := chan string{cap: 1}

	for i in 0 .. 3 {
		ex.try_post(fn [ran, i] () ! {
			ran <- i
		})!
	}

	runner := spawn fn [mut ex, result] () {
		ex.run() or {
			result <- err.msg()
			return
		}
		result <- 'ok'
	}()

	assert wait_for_int(ran, 'first run job missing') == 0
	assert wait_for_int(ran, 'second run job missing') == 1
	assert wait_for_int(ran, 'third run job missing') == 2
	ex.stop()
	assert wait_for_string(result, 'run did not return after stop') == 'ok'
	ex.wait()!
	runner.wait()
}

fn test_stop_from_owner_callback_does_not_deadlock() {
	mut ex := new(queue_size: 2)!
	ran := chan bool{cap: 1}
	result := chan string{cap: 1}

	ex.try_post(fn [mut ex, ran] () ! {
		ex.stop()
		ran <- true
	})!

	runner := spawn fn [mut ex, result] () {
		ex.run() or {
			result <- err.msg()
			return
		}
		result <- 'ok'
	}()

	assert wait_for_bool(ran, 'owner callback did not run')
	assert wait_for_string(result, 'run did not return after owner stop') == 'ok'
	ex.wait()!
	runner.wait()
}

fn test_wait_from_owner_callback_after_stop_returns_error_without_consuming_wait() {
	mut ex := new(queue_size: 2)!
	wait_result := chan string{cap: 1}
	run_result := chan string{cap: 1}

	ex.try_post(fn [mut ex, wait_result] () ! {
		ex.stop()
		ex.wait() or {
			wait_result <- err.msg()
			return
		}
		wait_result <- 'ok'
	})!

	runner := spawn fn [mut ex, run_result] () {
		ex.run() or {
			run_result <- err.msg()
			return
		}
		run_result <- 'ok'
	}()

	assert wait_for_string(wait_result, 'owner wait did not return') == 'executor: wait cannot be called from executor owner thread'
	assert wait_for_string(run_result, 'run did not return after owner stop') == 'ok'
	ex.wait()!
	runner.wait()
}

fn test_wait_before_terminal_pump_returns_error() {
	mut ex := new(queue_size: 1)!
	ex.wait() or {
		assert err.msg() == 'executor: wait requires a running pump or stopped executor'
		ex.stop()
		return
	}
	assert false, 'wait returned successfully before a terminal owner pump'
}

fn test_wait_before_runner_run_does_not_consume_wait() {
	mut ex := new(queue_size: 2)!
	start_run := chan bool{cap: 1}
	started := chan bool{cap: 1}
	result := chan string{cap: 1}

	ex.try_post(fn [started] () ! {
		started <- true
	})!

	runner := spawn fn [mut ex, start_run, result] () {
		_ := <-start_run
		ex.run() or {
			result <- err.msg()
			return
		}
		result <- 'ok'
	}()

	mut saw_precondition := false
	ex.wait() or {
		assert err.msg() == 'executor: wait requires a running pump or stopped executor'
		saw_precondition = true
	}
	assert saw_precondition

	start_run <- true
	assert wait_for_bool(started, 'runner did not start owner pump')
	ex.stop()
	assert wait_for_string(result, 'run did not return after stop') == 'ok'
	ex.wait()!
	mut saw_second_wait := false
	ex.wait() or {
		assert err.msg() == 'executor: wait was already called'
		saw_second_wait = true
	}
	assert saw_second_wait
	runner.wait()
}

fn test_run_after_stop_closes_admission_and_second_run_is_stopped() {
	mut ex := new(queue_size: 1)!
	ex.stop()

	mut rejected_after_stop := false
	ex.try_post(fn () ! {}) or {
		assert err.msg() == 'executor: executor is closed'
		rejected_after_stop = true
	}
	assert rejected_after_stop

	ex.run() or {
		assert err.msg() == 'executor: executor is stopped'
		return
	}
	assert false, 'run after stop returned successfully'
}

fn test_second_wait_returns_error() {
	mut ex := new(queue_size: 1)!
	started := chan bool{cap: 1}
	release := chan bool{cap: 1}
	result := chan string{cap: 1}

	ex.try_post(fn [started, release] () ! {
		started <- true
		_ := <-release
	})!

	runner := spawn fn [mut ex, result] () {
		ex.run() or {
			result <- err.msg()
			return
		}
		result <- 'ok'
	}()

	assert wait_for_bool(started, 'run did not start before wait test')
	release <- true
	ex.stop()
	assert wait_for_string(result, 'run did not stop') == 'ok'
	ex.wait()!
	ex.wait() or {
		assert err.msg() == 'executor: wait was already called'
		runner.wait()
		return
	}
	assert false, 'second wait returned successfully'
}

fn test_second_concurrent_run_returns_error() {
	mut ex := new(queue_size: 2)!
	started := chan bool{cap: 1}
	release := chan bool{cap: 1}
	first_result := chan string{cap: 1}
	second_result := chan string{cap: 1}

	ex.try_post(fn [started, release] () ! {
		started <- true
		_ := <-release
	})!

	first := spawn fn [mut ex, first_result] () {
		ex.run() or {
			first_result <- err.msg()
			return
		}
		first_result <- 'ok'
	}()
	assert wait_for_bool(started, 'first run did not start owner job')

	second := spawn fn [mut ex, second_result] () {
		ex.run() or {
			second_result <- err.msg()
			return
		}
		second_result <- 'ok'
	}()

	msg := wait_for_string(second_result, 'second run did not return')
	assert msg == 'executor: owner pump is already running'

	release <- true
	ex.stop()
	assert wait_for_string(first_result, 'first run did not stop') == 'ok'
	ex.wait()!
	first.wait()
	second.wait()
}

fn wait_for_bool(signal chan bool, message string) bool {
	select {
		value := <-signal {
			return value
		}
		1 * time.second {
			assert false, message
		}
	}
	return false
}

fn wait_for_int(signal chan int, message string) int {
	select {
		value := <-signal {
			return value
		}
		1 * time.second {
			assert false, message
		}
	}
	return 0
}

fn wait_for_string(signal chan string, message string) string {
	select {
		value := <-signal {
			return value
		}
		1 * time.second {
			assert false, message
		}
	}
	return ''
}