diff --git a/Cargo.lock b/Cargo.lock index ce0ad8d9..b936f310 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1507,11 +1507,17 @@ dependencies = [ name = "taskito-java" version = "0.16.4" dependencies = [ + "async-trait", + "crossbeam-channel", + "gethostname", "jni", + "log", "once_cell", "serde", "serde_json", "taskito-core", + "tokio", + "uuid", ] [[package]] diff --git a/crates/taskito-java/Cargo.toml b/crates/taskito-java/Cargo.toml index d65ba949..099f76b0 100644 --- a/crates/taskito-java/Cargo.toml +++ b/crates/taskito-java/Cargo.toml @@ -20,3 +20,9 @@ jni = "0.21" serde = { workspace = true } serde_json = { workspace = true } once_cell = "1" +tokio = { workspace = true } +async-trait = { workspace = true } +crossbeam-channel = { workspace = true } +uuid = { workspace = true } +log = { workspace = true } +gethostname = "1" diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs index fac7b047..f78ff855 100644 --- a/crates/taskito-java/src/convert.rs +++ b/crates/taskito-java/src/convert.rs @@ -150,6 +150,15 @@ impl<'a> From<&'a Job> for JobView<'a> { } } +/// Options accepted by `NativeQueue.runWorker`. +#[derive(Deserialize, Default)] +#[serde(rename_all = "camelCase", default)] +pub struct WorkerOptions { + pub queues: Option>, + pub channel_capacity: Option, + pub batch_size: Option, +} + /// Filter accepted by `NativeQueue.listJobs`. #[derive(Deserialize, Default)] #[serde(rename_all = "camelCase", default)] diff --git a/crates/taskito-java/src/dispatcher.rs b/crates/taskito-java/src/dispatcher.rs new file mode 100644 index 00000000..52a15546 --- /dev/null +++ b/crates/taskito-java/src/dispatcher.rs @@ -0,0 +1,207 @@ +//! `JavaDispatcher` — runs each job by calling back into Java. +//! +//! Implements the core [`WorkerDispatcher`] trait. The completion registry lives +//! here in Rust (the inverse of the Node shell's JS-promise bridge): for each +//! job a token + oneshot is registered, the job is submitted to Java via +//! `onJob`, and the awaiting task parks on the oneshot — no worker thread blocks. +//! Java runs the task (sync or async) and completes it through the native +//! `NativeWorker.completeJob/failJob/cancelJob` entry points (see `worker.rs`), +//! which resolve the oneshot. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use crossbeam_channel::Sender; +use jni::objects::{GlobalRef, JValue}; +use taskito_core::job::Job; +use taskito_core::scheduler::JobResult; +use taskito_core::worker::WorkerDispatcher; +use taskito_core::{Storage, StorageBackend}; +use tokio::sync::oneshot; + +use crate::jvm; + +/// The outcome Java reports for a submitted job. +pub enum TaskOutcome { + Success(Vec), + Failure(String), + Cancelled, +} + +/// Pending-job registry shared between the dispatcher and the Java completion +/// callbacks. Held alive by the worker handle so its pointer stays valid. +#[derive(Default)] +pub struct Registry { + pending: Mutex>>, + next_token: AtomicU64, +} + +impl Registry { + fn register(&self) -> (u64, oneshot::Receiver) { + let token = self.next_token.fetch_add(1, Ordering::Relaxed); + let (tx, rx) = oneshot::channel(); + self.pending.lock().unwrap().insert(token, tx); + (token, rx) + } + + fn forget(&self, token: u64) { + self.pending.lock().unwrap().remove(&token); + } + + /// Resolve a pending job. A no-op if the token already completed or timed out. + pub fn complete(&self, token: u64, outcome: TaskOutcome) { + if let Some(tx) = self.pending.lock().unwrap().remove(&token) { + let _ = tx.send(outcome); + } + } +} + +/// Executes jobs by dispatching them to a Java `WorkerBridge` callback object. +pub struct JavaDispatcher { + callbacks: GlobalRef, + registry: Arc, + storage: StorageBackend, +} + +impl JavaDispatcher { + pub fn new(callbacks: GlobalRef, registry: Arc, storage: StorageBackend) -> Self { + Self { + callbacks, + registry, + storage, + } + } +} + +#[async_trait::async_trait] +impl WorkerDispatcher for JavaDispatcher { + async fn run( + &self, + mut job_rx: tokio::sync::mpsc::Receiver, + result_tx: Sender, + ) { + while let Some(job) = job_rx.recv().await { + let callbacks = self.callbacks.clone(); + let registry = self.registry.clone(); + let storage = self.storage.clone(); + let result_tx = result_tx.clone(); + tokio::spawn(async move { + let result = run_one(&callbacks, ®istry, &storage, job).await; + let _ = result_tx.send(result); + }); + } + } + + fn shutdown(&self) {} +} + +/// Submit one job to Java, await its completion, and translate to a [`JobResult`]. +async fn run_one( + callbacks: &GlobalRef, + registry: &Registry, + storage: &StorageBackend, + job: Job, +) -> JobResult { + let started = Instant::now(); + let (token, rx) = registry.register(); + + if let Err(err) = submit_to_java(callbacks, token, &job) { + registry.forget(token); + return failure(job, err, started.elapsed().as_nanos() as i64, false); + } + + // `timeout_ms <= 0` means no limit. On timeout we drop the pending entry; a + // late completion then finds no sender and is harmlessly ignored. + let outcome = if job.timeout_ms > 0 { + match tokio::time::timeout(Duration::from_millis(job.timeout_ms as u64), rx).await { + Ok(result) => result, + Err(_) => { + registry.forget(token); + let wall = started.elapsed().as_nanos() as i64; + return failure(job, "task timed out".to_string(), wall, true); + } + } + } else { + rx.await + }; + + let wall = started.elapsed().as_nanos() as i64; + match outcome { + Ok(TaskOutcome::Success(result)) => JobResult::Success { + job_id: job.id, + result: Some(result), + task_name: job.task_name, + wall_time_ns: wall, + }, + Ok(TaskOutcome::Cancelled) => cancelled(job, wall), + Ok(TaskOutcome::Failure(error)) => { + // A failure on a cancel-requested job is a cancellation, not a fault. + if storage.is_cancel_requested(&job.id).unwrap_or(false) { + cancelled(job, wall) + } else { + failure(job, error, wall, false) + } + } + // The oneshot sender dropped without completing — the Java side died. + Err(_) => failure(job, "java task channel dropped".to_string(), wall, false), + } +} + +/// Invoke `WorkerBridge.onJob` on an attached thread. Local refs are freed when +/// the per-call attachment guard drops. +fn submit_to_java(callbacks: &GlobalRef, token: u64, job: &Job) -> Result<(), String> { + let vm = jvm::vm(); + let mut env = vm + .attach_current_thread() + .map_err(|e| format!("attach failed: {e}"))?; + let job_id = env.new_string(&job.id).map_err(|e| e.to_string())?; + let task_name = env.new_string(&job.task_name).map_err(|e| e.to_string())?; + let payload = env + .byte_array_from_slice(&job.payload) + .map_err(|e| e.to_string())?; + // An `Err(JavaException)` leaves the exception pending on the attached + // thread; clear it before returning so the thread isn't poisoned for the + // next call. + if let Err(e) = env.call_method( + callbacks, + "onJob", + "(JLjava/lang/String;Ljava/lang/String;[B)V", + &[ + JValue::Long(token as i64), + JValue::Object(&job_id), + JValue::Object(&task_name), + JValue::Object(&payload), + ], + ) { + let _ = env.exception_clear(); + return Err(e.to_string()); + } + if env.exception_check().unwrap_or(false) { + let _ = env.exception_clear(); + return Err("WorkerBridge.onJob threw".to_string()); + } + Ok(()) +} + +fn cancelled(job: Job, wall_time_ns: i64) -> JobResult { + JobResult::Cancelled { + job_id: job.id, + task_name: job.task_name, + wall_time_ns, + } +} + +fn failure(job: Job, error: String, wall_time_ns: i64, timed_out: bool) -> JobResult { + JobResult::Failure { + job_id: job.id, + error, + retry_count: job.retry_count, + max_retries: job.max_retries, + task_name: job.task_name, + wall_time_ns, + should_retry: true, + timed_out, + } +} diff --git a/crates/taskito-java/src/jvm.rs b/crates/taskito-java/src/jvm.rs index c5e3395a..346407bf 100644 --- a/crates/taskito-java/src/jvm.rs +++ b/crates/taskito-java/src/jvm.rs @@ -12,7 +12,6 @@ pub fn init(vm: JavaVM) { /// The cached VM (present after `JNI_OnLoad`). Used by the worker bridge to /// attach worker threads that invoke Java callbacks. -#[allow(dead_code)] pub fn vm() -> &'static JavaVM { JAVA_VM.get().expect("JNI_OnLoad has not run") } diff --git a/crates/taskito-java/src/lib.rs b/crates/taskito-java/src/lib.rs index 6fa1419a..daec4563 100644 --- a/crates/taskito-java/src/lib.rs +++ b/crates/taskito-java/src/lib.rs @@ -10,11 +10,13 @@ mod backend; mod convert; +mod dispatcher; mod error; mod ffi; mod handle; mod jvm; mod queue; +mod worker; use std::ffi::c_void; diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs new file mode 100644 index 00000000..0baa5bd3 --- /dev/null +++ b/crates/taskito-java/src/worker.rs @@ -0,0 +1,395 @@ +//! Worker wiring and the `NativeWorker` / `NativeQueue.runWorker` entry points. +//! +//! Mirrors the Node shell's `start_worker`: a scheduler loop feeds jobs to the +//! [`JavaDispatcher`], a result-drain loop applies outcomes and surfaces them to +//! Java for events/middleware, and a lifecycle loop registers + heartbeats the +//! worker. The completion registry (see [`crate::dispatcher`]) lets Java report +//! each job's outcome back. + +use std::sync::Arc; +use std::time::Duration; + +use jni::objects::{GlobalRef, JByteArray, JClass, JObject, JString, JValue}; +use jni::sys::jlong; +use jni::JNIEnv; +use taskito_core::scheduler::ResultOutcome; +use taskito_core::worker::WorkerDispatcher; +use taskito_core::{Scheduler, SchedulerConfig, Storage, StorageBackend}; +use tokio::sync::Notify; + +use crate::backend::QueueHandle; +use crate::convert::{parse_json, WorkerOptions}; +use crate::dispatcher::{JavaDispatcher, Registry, TaskOutcome}; +use crate::ffi::{guard, read_bytes, read_string}; +use crate::handle::{self, drop_handle, into_handle}; +use crate::jvm; + +const DEFAULT_QUEUE: &str = "default"; +const DEFAULT_CHANNEL_CAPACITY: usize = 128; +const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5); + +/// A running worker. The tokio runtime and the completion registry are held here +/// for the worker's lifetime; the registry's pointer (this handle) stays valid +/// until [`close`] runs. Fields drop in declaration order: the runtime stops +/// first (joining tasks), then the registry's last reference is released. +pub struct WorkerHandle { + _runtime: tokio::runtime::Runtime, + registry: Arc, + shutdown: Arc, + heartbeat_stop: Arc, +} + +/// Build and start a worker over `storage`, calling back into the Java +/// `WorkerBridge` (`callbacks`) for each job. +fn start_worker( + storage: StorageBackend, + namespace: Option, + options: WorkerOptions, + callbacks: GlobalRef, +) -> Result { + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .map_err(|e| crate::error::BindingError::new(format!("failed to start runtime: {e}")))?; + + let queues = options + .queues + .unwrap_or_else(|| vec![DEFAULT_QUEUE.to_string()]); + let capacity = options + .channel_capacity + .map(|c| (c as usize).max(1)) + .unwrap_or(DEFAULT_CHANNEL_CAPACITY); + let mut config = SchedulerConfig::default(); + if let Some(batch) = options.batch_size { + config.batch_size = batch.max(1) as usize; + } + + let registry = Arc::new(Registry::default()); + let dispatcher_storage = storage.clone(); + let lifecycle_storage = storage.clone(); + let queues_csv = queues.join(","); + let worker_id = format!("java-{}", uuid::Uuid::now_v7()); + + let scheduler = Arc::new(Scheduler::new(storage, queues, config, namespace)); + let shutdown = scheduler.shutdown_handle(); + let heartbeat_stop = Arc::new(Notify::new()); + + let (job_tx, job_rx) = tokio::sync::mpsc::channel(capacity); + // Unbounded so the dispatcher's `result_tx.send` (which runs inside a Tokio + // task) never blocks a runtime worker when the drain loop is momentarily + // slow. In-flight jobs are already bounded by the job channel + scheduler, + // so the result queue can't grow without bound. + let (result_tx, result_rx) = crossbeam_channel::unbounded(); + + // Scheduler loop: poll storage and dispatch ready jobs. + let scheduler_run = scheduler.clone(); + runtime.spawn(async move { + scheduler_run.run(job_tx).await; + }); + + // Dispatcher loop: submit each job to Java, report results. + let dispatcher = JavaDispatcher::new(callbacks.clone(), registry.clone(), dispatcher_storage); + runtime.spawn(async move { + dispatcher.run(job_rx, result_tx).await; + }); + + // Result-drain loop: apply outcomes and surface them to Java. crossbeam + // `recv` is blocking, so it runs on a blocking thread; it exits when the + // result sender has dropped (dispatcher done). + let drain_scheduler = scheduler; + let drain_callbacks = callbacks; + runtime.spawn_blocking(move || drain_results(result_rx, drain_scheduler, drain_callbacks)); + + // Lifecycle loop: register, heartbeat, unregister. + spawn_lifecycle( + &runtime, + lifecycle_storage, + worker_id, + queues_csv, + capacity, + heartbeat_stop.clone(), + ); + + Ok(WorkerHandle { + _runtime: runtime, + registry, + shutdown, + heartbeat_stop, + }) +} + +/// Drain results: apply each to storage and invoke `WorkerBridge.onOutcome`. +fn drain_results( + result_rx: crossbeam_channel::Receiver, + scheduler: Arc, + callbacks: GlobalRef, +) { + let vm = jvm::vm(); + let mut env = match vm.attach_current_thread() { + Ok(env) => env, + Err(e) => { + log::error!("[taskito-java] drain attach failed: {e}"); + return; + } + }; + while let Ok(result) = result_rx.recv() { + let outcome = match scheduler.handle_result(result) { + Ok(outcome) => outcome, + Err(e) => { + log::error!("[taskito-java] result handling error: {e}"); + continue; + } + }; + // Each outcome allocates several JNI locals on this long-lived attached + // env; scope them in a frame so a busy worker can't exhaust the + // local-reference table over the loop's lifetime. + let framed = env.with_local_frame(16, |env| { + if let Err(e) = call_on_outcome(env, &callbacks, &outcome) { + log::error!("[taskito-java] onOutcome failed: {e}"); + } + Ok::<(), jni::errors::Error>(()) + }); + if let Err(e) = framed { + log::error!("[taskito-java] drain local frame failed: {e}"); + } + } +} + +/// Invoke `WorkerBridge.onOutcome` for one finished job. +fn call_on_outcome( + env: &mut JNIEnv, + callbacks: &GlobalRef, + outcome: &ResultOutcome, +) -> Result<(), String> { + let (kind, job_id, task_name, error, retry_count, timed_out) = describe(outcome); + let kind_s = env.new_string(kind).map_err(|e| e.to_string())?; + let job_s = env.new_string(job_id).map_err(|e| e.to_string())?; + let task_s = env.new_string(task_name).map_err(|e| e.to_string())?; + let error_obj: JObject = match error { + Some(e) => env.new_string(e).map_err(|x| x.to_string())?.into(), + None => JObject::null(), + }; + // Clear a pending exception on the `Err(JavaException)` path before + // returning, so a thrown listener can't poison the reused env. + if let Err(e) = env.call_method( + callbacks, + "onOutcome", + "(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;IZ)V", + &[ + JValue::Object(&kind_s), + JValue::Object(&job_s), + JValue::Object(&task_s), + JValue::Object(&error_obj), + JValue::Int(retry_count), + JValue::Bool(u8::from(timed_out)), + ], + ) { + let _ = env.exception_clear(); + return Err(e.to_string()); + } + if env.exception_check().unwrap_or(false) { + let _ = env.exception_clear(); + return Err("WorkerBridge.onOutcome threw".to_string()); + } + Ok(()) +} + +/// Flatten a [`ResultOutcome`] into the fields `onOutcome` receives. `retry_count` +/// is -1 when not applicable. +fn describe(outcome: &ResultOutcome) -> (&str, &str, &str, Option<&str>, i32, bool) { + match outcome { + ResultOutcome::Success { job_id, task_name } => { + ("success", job_id, task_name, None, -1, false) + } + ResultOutcome::Retry { + job_id, + task_name, + error, + retry_count, + timed_out, + .. + } => ( + "retry", + job_id, + task_name, + Some(error), + *retry_count, + *timed_out, + ), + ResultOutcome::DeadLettered { + job_id, + task_name, + error, + timed_out, + .. + } => ("dead", job_id, task_name, Some(error), -1, *timed_out), + ResultOutcome::Cancelled { + job_id, task_name, .. + } => ("cancelled", job_id, task_name, None, -1, false), + } +} + +/// Register the worker, heartbeat every 5s until stopped, then unregister. +fn spawn_lifecycle( + runtime: &tokio::runtime::Runtime, + storage: StorageBackend, + worker_id: String, + queues_csv: String, + capacity: usize, + stop: Arc, +) { + let hostname = gethostname::gethostname().to_string_lossy().to_string(); + let pid = std::process::id() as i32; + if let Err(e) = storage.register_worker( + &worker_id, + &queues_csv, + None, + None, + None, + capacity as i32, + Some(&hostname), + Some(pid), + Some("java"), + ) { + log::warn!("[taskito-java] worker registration failed: {e}"); + } + runtime.spawn(async move { + loop { + tokio::select! { + _ = stop.notified() => break, + _ = tokio::time::sleep(HEARTBEAT_INTERVAL) => { + if let Err(e) = storage.heartbeat(&worker_id, None) { + log::warn!("[taskito-java] worker heartbeat failed: {e}"); + } + } + } + } + if let Err(e) = storage.unregister_worker(&worker_id) { + log::warn!("[taskito-java] worker unregister failed: {e}"); + } + }); +} + +/// Borrow a worker handle. +/// +/// # Safety +/// `handle` must be a live `WorkerHandle` pointer from [`start_worker`]. +unsafe fn borrow_worker<'a>(handle: jlong) -> &'a WorkerHandle { + handle::borrow::(handle) +} + +/// `long runWorker(long queueHandle, Object bridge, String optionsJson)` — start +/// a worker; returns its handle. `bridge` is a Java `WorkerBridge`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_runWorker<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + queue_handle: jlong, + bridge: JObject<'local>, + options_json: JString<'local>, +) -> jlong { + guard(&mut env, 0, |env| { + let queue = unsafe { handle::borrow::(queue_handle) }; + let raw = read_string(env, &options_json)?; + let options: WorkerOptions = parse_json(&raw, "worker options")?; + let callbacks = env + .new_global_ref(&bridge) + .map_err(|e| crate::error::BindingError::new(format!("global ref failed: {e}")))?; + let worker = start_worker( + queue.storage.clone(), + queue.namespace.clone(), + options, + callbacks, + )?; + Ok(into_handle(worker)) + }) +} + +/// `void completeJob(long workerHandle, long token, byte[] result)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorker_completeJob<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + token: jlong, + result: JByteArray<'local>, +) { + guard(&mut env, (), |env| { + let worker = unsafe { borrow_worker(handle) }; + let bytes = read_bytes(env, &result)?; + worker + .registry + .complete(token as u64, TaskOutcome::Success(bytes)); + Ok(()) + }) +} + +/// `void failJob(long workerHandle, long token, String error)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorker_failJob<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + token: jlong, + error: JString<'local>, +) { + guard(&mut env, (), |env| { + let worker = unsafe { borrow_worker(handle) }; + let message = read_string(env, &error)?; + worker + .registry + .complete(token as u64, TaskOutcome::Failure(message)); + Ok(()) + }) +} + +/// `void cancelJob(long workerHandle, long token)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorker_cancelJob( + mut env: JNIEnv, + _class: JClass, + handle: jlong, + token: jlong, +) { + guard(&mut env, (), |_env| { + let worker = unsafe { borrow_worker(handle) }; + worker + .registry + .complete(token as u64, TaskOutcome::Cancelled); + Ok(()) + }) +} + +/// `void stop(long workerHandle)` — stop the scheduler and heartbeat loops. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorker_stop( + mut env: JNIEnv, + _class: JClass, + handle: jlong, +) { + guard(&mut env, (), |_env| { + let worker = unsafe { borrow_worker(handle) }; + worker.shutdown.notify_one(); + worker.heartbeat_stop.notify_one(); + Ok(()) + }) +} + +/// `void close(long workerHandle)` — stop the runtime and reclaim the handle. +/// The Java side drains its handler executor before calling this, so no +/// `completeJob`/`failJob` can still borrow the handle being freed. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorker_close<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, +) { + // Dropping the handle stops the runtime (joining tasks); route through + // `guard` so a panic in that teardown can't unwind across the FFI boundary. + guard(&mut env, (), |_env| { + if handle != 0 { + unsafe { drop_handle::(handle) }; + } + Ok(()) + }) +} diff --git a/sdks/java/README.md b/sdks/java/README.md index caef584d..abb77919 100644 --- a/sdks/java/README.md +++ b/sdks/java/README.md @@ -3,12 +3,14 @@ A typed Java 11+ client over the Taskito Rust core, via a hand-written JNI shell (`crates/taskito-java`). -> Status: **early build-out** (Phase 0). Producer surface (enqueue / get / cancel -> / stats) is implemented and verified end-to-end against the native library. -> Worker execution, dashboard, webhooks, and the CLI follow in later phases. +> Status: **build-out** (Phases 0–2). Producer + inspection + admin + logs, and +> worker task execution, are implemented and verified end-to-end against the +> native library. Dashboard, webhooks, and the CLI follow in later phases. ## Usage +### Enqueue + ```java import org.byteveda.taskito.*; import java.util.Map; @@ -24,6 +26,30 @@ try (Queue queue = Taskito.builder().backend("sqlite").url("taskito.db").open()) } ``` +### Run a worker + +```java +import org.byteveda.taskito.events.EventName; + +Task add = Task.of("add", Map.class); + +Queue queue = Taskito.builder().backend("sqlite").url("taskito.db").open(); +Worker worker = queue.worker() + .handle(add, p -> ((Number) p.get("a")).intValue() + ((Number) p.get("b")).intValue()) + .concurrency(4) + .on(EventName.SUCCESS, e -> System.out.println("done: " + e.jobId)) + .start(); + +// Close on SIGTERM/Ctrl-C; awaitShutdown() then unblocks. (Don't put the worker +// in try-with-resources AND call awaitShutdown() inside it — the block can't +// exit to trigger close(), so it would deadlock.) +Runtime.getRuntime().addShutdownHook(new Thread(() -> { + worker.close(); + queue.close(); +})); +worker.awaitShutdown(); +``` + ## Structure ```text diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java index b3e64f81..72c54dee 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java @@ -127,8 +127,8 @@ public List metrics(String taskName, long sinceMs) { } @Override - public List listWorkers() { - return decodeList(backend.listWorkersJson(), Worker.class); + public List listWorkers() { + return decodeList(backend.listWorkersJson(), WorkerInfo.class); } // ── Admin ─────────────────────────────────────────────────────── @@ -210,6 +210,11 @@ public List getTaskLogs(String jobId) { return decodeList(backend.getTaskLogsJson(jobId), TaskLog.class); } + @Override + public Worker.Builder worker() { + return Worker.builder(backend, serializer); + } + @Override public void close() { backend.close(); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java index 60fae7f2..dbfab619 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java @@ -52,7 +52,7 @@ public interface Queue extends AutoCloseable { /** Per-execution metrics within the last {@code sinceMs}; null task = all. */ List metrics(String taskName, long sinceMs); - List listWorkers(); + List listWorkers(); // ── Admin ─────────────────────────────────────────────────────── @@ -89,6 +89,11 @@ public interface Queue extends AutoCloseable { List getTaskLogs(String jobId); + // ── Worker ────────────────────────────────────────────────────── + + /** Begin building a worker over this queue. */ + Worker.Builder worker(); + @Override void close(); } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/RegisteredTask.java b/sdks/java/src/main/java/org/byteveda/taskito/RegisteredTask.java new file mode 100644 index 00000000..721514c5 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/RegisteredTask.java @@ -0,0 +1,12 @@ +package org.byteveda.taskito; + +/** A worker-registered handler: payload type plus the function to run. */ +final class RegisteredTask { + final Class payloadType; + final TaskFunction handler; + + RegisteredTask(Class payloadType, TaskFunction handler) { + this.payloadType = payloadType; + this.handler = handler; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/TaskFunction.java b/sdks/java/src/main/java/org/byteveda/taskito/TaskFunction.java new file mode 100644 index 00000000..704bf3b4 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/TaskFunction.java @@ -0,0 +1,7 @@ +package org.byteveda.taskito; + +/** A task handler: receives a deserialized payload, returns a result (or null). */ +@FunctionalInterface +public interface TaskFunction { + R apply(T payload) throws Exception; +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/Worker.java index 6c01df5e..d4b52721 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Worker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Worker.java @@ -1,44 +1,172 @@ package org.byteveda.taskito; -import com.fasterxml.jackson.annotation.JsonCreator; -import com.fasterxml.jackson.annotation.JsonIgnoreProperties; -import com.fasterxml.jackson.annotation.JsonProperty; - -/** A registered worker (heartbeat + identity). Timestamps are Unix milliseconds. */ -@JsonIgnoreProperties(ignoreUnknown = true) -public final class Worker { - public final String workerId; - public final String queues; - public final String status; - public final long lastHeartbeat; - public final Long startedAt; - public final String hostname; - public final Integer pid; - public final String poolType; - public final int threads; - public final String tags; - - @JsonCreator - public Worker( - @JsonProperty("workerId") String workerId, - @JsonProperty("queues") String queues, - @JsonProperty("status") String status, - @JsonProperty("lastHeartbeat") long lastHeartbeat, - @JsonProperty("startedAt") Long startedAt, - @JsonProperty("hostname") String hostname, - @JsonProperty("pid") Integer pid, - @JsonProperty("poolType") String poolType, - @JsonProperty("threads") int threads, - @JsonProperty("tags") String tags) { - this.workerId = workerId; - this.queues = queues; - this.status = status; - this.lastHeartbeat = lastHeartbeat; - this.startedAt = startedAt; - this.hostname = hostname; - this.pid = pid; - this.poolType = poolType; - this.threads = threads; - this.tags = tags; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import org.byteveda.taskito.events.Emitter; +import org.byteveda.taskito.events.EventName; +import org.byteveda.taskito.events.OutcomeEvent; +import org.byteveda.taskito.serialization.Serializer; +import org.byteveda.taskito.spi.QueueBackend; +import org.byteveda.taskito.spi.WorkerControl; + +/** + * A running worker. Build one with {@link Queue#worker()}, register handlers, + * then {@link Builder#start()}. {@link #close()} stops it and drains in-flight + * jobs. + */ +public final class Worker implements AutoCloseable { + private static final long SHUTDOWN_TIMEOUT_SECONDS = 30; + + private final WorkerControl control; + private final ExecutorService executor; + private final CountDownLatch shutdown = new CountDownLatch(1); + private boolean closed; + + private Worker(WorkerControl control, ExecutorService executor) { + this.control = control; + this.executor = executor; + } + + public static Builder builder(QueueBackend backend, Serializer serializer) { + return new Builder(backend, serializer); + } + + /** Stop dispatching; in-flight jobs continue to drain. */ + public void stop() { + control.stop(); + } + + /** Block until {@link #close()} is called. */ + public void awaitShutdown() throws InterruptedException { + shutdown.await(); + } + + /** + * Stop dispatching, drain in-flight handler tasks, then free the native + * worker. Draining BEFORE {@code control.close()} is essential: a running + * handler may still call back into the native worker + * ({@code completeJob}/{@code failJob}), so the handle must outlive every + * handler task. Idempotent. + */ + @Override + public synchronized void close() { + if (closed) { + return; + } + closed = true; + control.stop(); // stop scheduling new work + executor.shutdown(); // stop accepting; let running handlers finish + try { + if (!executor.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + executor.shutdownNow(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + executor.shutdownNow(); + } + control.close(); // now safe to free the native handle — no handler can touch it + shutdown.countDown(); + } + + /** Registers handlers and worker options, then starts the worker. */ + public static final class Builder { + private static final ObjectMapper JSON = new ObjectMapper(); + + private final QueueBackend backend; + private final Serializer serializer; + private final Map handlers = new HashMap<>(); + private final Map>> listeners = + new EnumMap<>(EventName.class); + private List queues; + private int concurrency; + private Integer channelCapacity; + private Integer batchSize; + + Builder(QueueBackend backend, Serializer serializer) { + this.backend = backend; + this.serializer = serializer; + } + + public Builder handle(String taskName, Class payloadType, TaskFunction handler) { + handlers.put(taskName, new RegisteredTask(payloadType, cast(handler))); + return this; + } + + public Builder handle(Task task, TaskFunction handler) { + return handle(task.name(), task.payloadType(), handler); + } + + public Builder queues(String... queues) { + this.queues = Arrays.asList(queues); + return this; + } + + /** Fixed handler-thread count; 0 (default) uses a cached pool. */ + public Builder concurrency(int concurrency) { + this.concurrency = concurrency; + return this; + } + + public Builder channelCapacity(int channelCapacity) { + this.channelCapacity = channelCapacity; + return this; + } + + public Builder batchSize(int batchSize) { + this.batchSize = batchSize; + return this; + } + + public Builder on(EventName name, Consumer listener) { + listeners.computeIfAbsent(name, key -> new ArrayList<>()).add(listener); + return this; + } + + public Worker start() { + ExecutorService executor = concurrency > 0 + ? Executors.newFixedThreadPool(concurrency) + : Executors.newCachedThreadPool(); + Emitter emitter = new Emitter(); + listeners.forEach((name, bound) -> bound.forEach(listener -> emitter.on(name, listener))); + WorkerDispatchBridge bridge = + new WorkerDispatchBridge(handlers, serializer, executor, emitter); + WorkerControl control = backend.startWorker(bridge, encodeOptions()); + bridge.bind(control); + return new Worker(control, executor); + } + + private String encodeOptions() { + Map options = new LinkedHashMap<>(); + if (queues != null) { + options.put("queues", queues); + } + if (channelCapacity != null) { + options.put("channelCapacity", channelCapacity); + } + if (batchSize != null) { + options.put("batchSize", batchSize); + } + try { + return JSON.writeValueAsString(options); + } catch (Exception e) { + throw new TaskitoException("failed to encode worker options", e); + } + } + + @SuppressWarnings("unchecked") + private static TaskFunction cast(TaskFunction handler) { + return (TaskFunction) handler; + } } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.java new file mode 100644 index 00000000..00bc488e --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.java @@ -0,0 +1,71 @@ +package org.byteveda.taskito; + +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import org.byteveda.taskito.events.Emitter; +import org.byteveda.taskito.events.EventName; +import org.byteveda.taskito.events.OutcomeEvent; +import org.byteveda.taskito.serialization.Serializer; +import org.byteveda.taskito.spi.WorkerBridge; +import org.byteveda.taskito.spi.WorkerControl; + +/** + * Bridges native job dispatch to registered handlers. {@code onJob} hands work to + * an executor and completes it via the {@link WorkerControl}; {@code onOutcome} + * fans finished jobs out to event listeners. + */ +final class WorkerDispatchBridge implements WorkerBridge { + private final Map handlers; + private final Serializer serializer; + private final ExecutorService executor; + private final Emitter emitter; + // Resolved once startWorker returns; job tasks await it before completing. + private final CompletableFuture control = new CompletableFuture<>(); + + WorkerDispatchBridge( + Map handlers, + Serializer serializer, + ExecutorService executor, + Emitter emitter) { + this.handlers = handlers; + this.serializer = serializer; + this.executor = executor; + this.emitter = emitter; + } + + void bind(WorkerControl control) { + this.control.complete(control); + } + + @Override + public void onJob(long token, String jobId, String taskName, byte[] payload) { + executor.execute(() -> runJob(token, taskName, payload)); + } + + private void runJob(long token, String taskName, byte[] payload) { + WorkerControl bound = control.join(); + RegisteredTask task = handlers.get(taskName); + if (task == null) { + bound.failJob(token, "no handler registered for task '" + taskName + "'"); + return; + } + try { + Object argument = serializer.deserialize(payload, task.payloadType); + Object result = task.handler.apply(argument); + bound.completeJob(token, serializer.serialize(result)); + } catch (Throwable t) { + bound.failJob(token, describe(t)); + } + } + + @Override + public void onOutcome(String kind, String jobId, String taskName, String error, int retryCount, boolean timedOut) { + emitter.emit(new OutcomeEvent(EventName.fromKind(kind), jobId, taskName, error, retryCount, timedOut)); + } + + private static String describe(Throwable t) { + String message = t.getMessage(); + return message == null ? t.getClass().getSimpleName() : t.getClass().getSimpleName() + ": " + message; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/WorkerInfo.java b/sdks/java/src/main/java/org/byteveda/taskito/WorkerInfo.java new file mode 100644 index 00000000..c9795acf --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/WorkerInfo.java @@ -0,0 +1,44 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** A registered worker (heartbeat + identity). Timestamps are Unix milliseconds. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class WorkerInfo { + public final String workerId; + public final String queues; + public final String status; + public final long lastHeartbeat; + public final Long startedAt; + public final String hostname; + public final Integer pid; + public final String poolType; + public final int threads; + public final String tags; + + @JsonCreator + public WorkerInfo( + @JsonProperty("workerId") String workerId, + @JsonProperty("queues") String queues, + @JsonProperty("status") String status, + @JsonProperty("lastHeartbeat") long lastHeartbeat, + @JsonProperty("startedAt") Long startedAt, + @JsonProperty("hostname") String hostname, + @JsonProperty("pid") Integer pid, + @JsonProperty("poolType") String poolType, + @JsonProperty("threads") int threads, + @JsonProperty("tags") String tags) { + this.workerId = workerId; + this.queues = queues; + this.status = status; + this.lastHeartbeat = lastHeartbeat; + this.startedAt = startedAt; + this.hostname = hostname; + this.pid = pid; + this.poolType = poolType; + this.threads = threads; + this.tags = tags; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java b/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java new file mode 100644 index 00000000..8e40df08 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java @@ -0,0 +1,31 @@ +package org.byteveda.taskito.events; + +import java.util.EnumMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Consumer; + +/** Dispatches {@link OutcomeEvent}s to registered listeners. Thread-safe. */ +public final class Emitter { + private final Map>> listeners = new EnumMap<>(EventName.class); + + public void on(EventName name, Consumer listener) { + listeners.computeIfAbsent(name, key -> new CopyOnWriteArrayList<>()).add(listener); + } + + /** Deliver {@code event} to its listeners; a throwing listener never blocks the rest. */ + public void emit(OutcomeEvent event) { + List> bound = listeners.get(event.name); + if (bound == null) { + return; + } + for (Consumer listener : bound) { + try { + listener.accept(event); + } catch (RuntimeException ignored) { + // A listener fault must not break event dispatch. + } + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/events/EventName.java b/sdks/java/src/main/java/org/byteveda/taskito/events/EventName.java new file mode 100644 index 00000000..cbd90711 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/events/EventName.java @@ -0,0 +1,27 @@ +package org.byteveda.taskito.events; + +import org.byteveda.taskito.TaskitoException; + +/** Terminal outcome of a job, delivered to worker event listeners. */ +public enum EventName { + SUCCESS, + RETRY, + DEAD, + CANCELLED; + + /** Map a native outcome kind ("success"/"retry"/"dead"/"cancelled"). */ + public static EventName fromKind(String kind) { + switch (kind) { + case "success": + return SUCCESS; + case "retry": + return RETRY; + case "dead": + return DEAD; + case "cancelled": + return CANCELLED; + default: + throw new TaskitoException("unknown outcome kind: " + kind); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/events/OutcomeEvent.java b/sdks/java/src/main/java/org/byteveda/taskito/events/OutcomeEvent.java new file mode 100644 index 00000000..0c3b4a36 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/events/OutcomeEvent.java @@ -0,0 +1,21 @@ +package org.byteveda.taskito.events; + +/** A finished job's outcome. {@code error} is null on success/cancel; {@code retryCount} is -1 when N/A. */ +public final class OutcomeEvent { + public final EventName name; + public final String jobId; + public final String taskName; + public final String error; + public final int retryCount; + public final boolean timedOut; + + public OutcomeEvent( + EventName name, String jobId, String taskName, String error, int retryCount, boolean timedOut) { + this.name = name; + this.jobId = jobId; + this.taskName = taskName; + this.error = error; + this.retryCount = retryCount; + this.timedOut = timedOut; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/events/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/events/package-info.java new file mode 100644 index 00000000..31206064 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/events/package-info.java @@ -0,0 +1,2 @@ +/** Worker outcome events: EventName, OutcomeEvent, and the Emitter that dispatches them. */ +package org.byteveda.taskito.events; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java index d189e355..eda3f880 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java @@ -2,6 +2,8 @@ import java.util.Optional; import org.byteveda.taskito.spi.QueueBackend; +import org.byteveda.taskito.spi.WorkerBridge; +import org.byteveda.taskito.spi.WorkerControl; /** JNI-backed {@link QueueBackend} over a native queue handle. */ public final class JniQueueBackend implements QueueBackend { @@ -162,6 +164,11 @@ public String getTaskLogsJson(String jobId) { return NativeQueue.getTaskLogs(handle, jobId); } + @Override + public WorkerControl startWorker(WorkerBridge bridge, String optionsJson) { + return new JniWorkerControl(NativeQueue.runWorker(handle, bridge, optionsJson)); + } + /** Idempotent: the native handle is freed exactly once, so a second {@code close()} * (e.g. an explicit call plus try-with-resources) can't double-free. */ @Override diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java new file mode 100644 index 00000000..b0f602fc --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java @@ -0,0 +1,42 @@ +package org.byteveda.taskito.internal; + +import org.byteveda.taskito.spi.WorkerControl; + +/** JNI-backed {@link WorkerControl} over a native worker handle. */ +public final class JniWorkerControl implements WorkerControl { + private final long handle; + private volatile boolean closed; + + JniWorkerControl(long handle) { + this.handle = handle; + } + + @Override + public void completeJob(long token, byte[] result) { + NativeWorker.completeJob(handle, token, result); + } + + @Override + public void failJob(long token, String error) { + NativeWorker.failJob(handle, token, error); + } + + @Override + public void cancelJob(long token) { + NativeWorker.cancelJob(handle, token); + } + + @Override + public void stop() { + NativeWorker.stop(handle); + } + + /** Idempotent: frees the native worker handle exactly once. */ + @Override + public synchronized void close() { + if (!closed) { + closed = true; + NativeWorker.close(handle); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java index 1139afd1..49092e2c 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java @@ -84,4 +84,8 @@ public static native void writeTaskLog( long handle, String jobId, String taskName, String level, String message, String extraOrNull); public static native String getTaskLogs(long handle, String jobId); + + // ── Worker ────────────────────────────────────────────────────── + /** Start a worker; returns its handle. {@code bridge} is a {@code WorkerBridge}. */ + public static native long runWorker(long handle, Object bridge, String optionsJson); } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.java new file mode 100644 index 00000000..af8edd68 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.java @@ -0,0 +1,26 @@ +package org.byteveda.taskito.internal; + +/** + * JNI completion + control surface for a running worker. + * + *

The {@code handle} is the worker pointer from {@link NativeQueue#runWorker}. + * {@code completeJob}/{@code failJob}/{@code cancelJob} resolve an in-flight job + * identified by its {@code token} (delivered to {@code WorkerBridge.onJob}). + */ +public final class NativeWorker { + static { + NativeLoader.load(); + } + + private NativeWorker() {} + + public static native void completeJob(long handle, long token, byte[] result); + + public static native void failJob(long handle, long token, String error); + + public static native void cancelJob(long handle, long token); + + public static native void stop(long handle); + + public static native void close(long handle); +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java index f7d0b3e5..0f4ed0fa 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java @@ -82,6 +82,10 @@ public interface QueueBackend extends AutoCloseable { String getTaskLogsJson(String jobId); + // ── Worker ────────────────────────────────────────────────────── + /** Start a worker that dispatches jobs to {@code bridge}; returns its control. */ + WorkerControl startWorker(WorkerBridge bridge, String optionsJson); + @Override void close(); } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerBridge.java new file mode 100644 index 00000000..65bfdce7 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerBridge.java @@ -0,0 +1,15 @@ +package org.byteveda.taskito.spi; + +/** + * Callbacks the worker runtime invokes (on native threads). The SDK implements + * this to run tasks and surface outcomes. + * + *

{@code onJob} must return promptly — hand the work to an executor and + * complete it later via {@link WorkerControl}. {@code onOutcome} reports a + * finished job for events/middleware. + */ +public interface WorkerBridge { + void onJob(long token, String jobId, String taskName, byte[] payload); + + void onOutcome(String kind, String jobId, String taskName, String error, int retryCount, boolean timedOut); +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.java new file mode 100644 index 00000000..41ddb4f7 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.java @@ -0,0 +1,16 @@ +package org.byteveda.taskito.spi; + +/** Controls a running worker and completes its in-flight jobs. */ +public interface WorkerControl extends AutoCloseable { + void completeJob(long token, byte[] result); + + void failJob(long token, String error); + + void cancelJob(long token); + + /** Stop the scheduler and heartbeat loops; in-flight jobs drain. */ + void stop(); + + @Override + void close(); +} diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java new file mode 100644 index 00000000..c8680865 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java @@ -0,0 +1,50 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.byteveda.taskito.events.EventName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +class WorkerTest { + + @SuppressWarnings("unchecked") + private static Class> mapType() { + return (Class>) (Class) Map.class; + } + + @Test + @Timeout(30) + void runsTaskToCompletion(@TempDir Path dir) throws Exception { + Task> add = Task.of("add", mapType()); + try (Queue queue = + Taskito.builder().backend("sqlite").url(dir.resolve("t.db").toString()).open()) { + Map payload = new HashMap<>(); + payload.put("a", 2); + payload.put("b", 3); + String id = queue.enqueue(add, payload); + + CountDownLatch done = new CountDownLatch(1); + try (Worker worker = queue.worker() + .handle(add, (Map p) -> + ((Number) p.get("a")).intValue() + ((Number) p.get("b")).intValue()) + .on(EventName.SUCCESS, event -> done.countDown()) + .start()) { + assertTrue(done.await(20, TimeUnit.SECONDS), "task should complete"); + + Optional result = queue.getResult(id); + assertTrue(result.isPresent()); + assertEquals("5", new String(result.get(), StandardCharsets.UTF_8)); + } + } + } +}