diff --git a/Cargo.lock b/Cargo.lock index 54411f13..ce0ad8d9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -120,6 +120,12 @@ dependencies = [ "shlex", ] +[[package]] +name = "cesu8" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d43a04d8753f35258c91f8ec639f792891f748a1edbd759cf1dcea3382ad83c" + [[package]] name = "cfg-if" version = "1.0.4" @@ -243,7 +249,7 @@ dependencies = [ "regex", "serde", "serde_json", - "thiserror", + "thiserror 2.0.18", ] [[package]] @@ -653,6 +659,50 @@ version = "1.0.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2" +[[package]] +name = "jni" +version = "0.21.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a87aa2bb7d2af34197c04845522473242e1aa17c12f4935d5856491a7fb8c97" +dependencies = [ + "cesu8", + "cfg-if", + "combine", + "jni-sys 0.3.1", + "log", + "thiserror 1.0.69", + "walkdir", + "windows-sys 0.45.0", +] + +[[package]] +name = "jni-sys" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41a652e1f9b6e0275df1f15b32661cf0d4b78d4d87ddec5e0c3c20f097433258" +dependencies = [ + "jni-sys 0.4.1", +] + +[[package]] +name = "jni-sys" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6377a88cb3910bee9b0fa88d4f42e1d2da8e79915598f65fb0c7ee14c878af2" +dependencies = [ + "jni-sys-macros", +] + +[[package]] +name = "jni-sys-macros" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" +dependencies = [ + "quote", + "syn", +] + [[package]] name = "js-sys" version = "0.3.88" @@ -1232,7 +1282,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a8a1f2315036ef6b1fbacd1972e8ee7688030b0a2121edfc2a6550febd41574d" dependencies = [ "hashbrown 0.16.1", - "thiserror", + "thiserror 2.0.18", ] [[package]] @@ -1260,6 +1310,15 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + [[package]] name = "scheduled-thread-pool" version = "0.2.7" @@ -1439,11 +1498,22 @@ dependencies = [ "serde", "serde_json", "tempfile", - "thiserror", + "thiserror 2.0.18", "tokio", "uuid", ] +[[package]] +name = "taskito-java" +version = "0.16.4" +dependencies = [ + "jni", + "once_cell", + "serde", + "serde_json", + "taskito-core", +] + [[package]] name = "taskito-mesh" version = "0.16.4" @@ -1525,13 +1595,33 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl 1.0.69", +] + [[package]] name = "thiserror" version = "2.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" dependencies = [ - "thiserror-impl", + "thiserror-impl 2.0.18", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn", ] [[package]] @@ -1673,6 +1763,16 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -1776,6 +1876,15 @@ dependencies = [ "semver", ] +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "windows-core" version = "0.62.2" @@ -1835,13 +1944,22 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.45.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75283be5efb2831d37ea142365f009c02ec203cd29a3ebecbc093d52315b66d0" +dependencies = [ + "windows-targets 0.42.2", +] + [[package]] name = "windows-sys" version = "0.60.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" dependencies = [ - "windows-targets", + "windows-targets 0.53.5", ] [[package]] @@ -1853,6 +1971,21 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e5180c00cd44c9b1c88adb3693291f1cd93605ded80c250a75d472756b4d071" +dependencies = [ + "windows_aarch64_gnullvm 0.42.2", + "windows_aarch64_msvc 0.42.2", + "windows_i686_gnu 0.42.2", + "windows_i686_msvc 0.42.2", + "windows_x86_64_gnu 0.42.2", + "windows_x86_64_gnullvm 0.42.2", + "windows_x86_64_msvc 0.42.2", +] + [[package]] name = "windows-targets" version = "0.53.5" @@ -1860,28 +1993,46 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" dependencies = [ "windows-link", - "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", + "windows_aarch64_gnullvm 0.53.1", + "windows_aarch64_msvc 0.53.1", + "windows_i686_gnu 0.53.1", "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_gnullvm", - "windows_x86_64_msvc", + "windows_i686_msvc 0.53.1", + "windows_x86_64_gnu 0.53.1", + "windows_x86_64_gnullvm 0.53.1", + "windows_x86_64_msvc 0.53.1", ] +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "597a5118570b68bc08d8d59125332c54f1ba9d9adeedeef5b99b02ba2b0698f8" + [[package]] name = "windows_aarch64_gnullvm" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" +[[package]] +name = "windows_aarch64_msvc" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e08e8864a60f06ef0d0ff4ba04124db8b0fb3be5776a5cd47641e942e58c4d43" + [[package]] name = "windows_aarch64_msvc" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" +[[package]] +name = "windows_i686_gnu" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c61d927d8da41da96a81f029489353e68739737d3beca43145c8afec9a31a84f" + [[package]] name = "windows_i686_gnu" version = "0.53.1" @@ -1894,24 +2045,48 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" +[[package]] +name = "windows_i686_msvc" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44d840b6ec649f480a41c8d80f9c65108b92d89345dd94027bfe06ac444d1060" + [[package]] name = "windows_i686_msvc" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" +[[package]] +name = "windows_x86_64_gnu" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8de912b8b8feb55c064867cf047dda097f92d51efad5b491dfb98f6bbb70cb36" + [[package]] name = "windows_x86_64_gnu" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26d41b46a36d453748aedef1486d5c7a85db22e56aff34643984ea85514e94a3" + [[package]] name = "windows_x86_64_gnullvm" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" +[[package]] +name = "windows_x86_64_msvc" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9aec5da331524158c6d1a4ac0ab1541149c0b9505fde06423b02f5ef0106b9f0" + [[package]] name = "windows_x86_64_msvc" version = "0.53.1" diff --git a/Cargo.toml b/Cargo.toml index 1f68d960..a32eeef1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["crates/taskito-core", "crates/taskito-python", "crates/taskito-node", "crates/taskito-workflows", "crates/taskito-mesh"] +members = ["crates/taskito-core", "crates/taskito-python", "crates/taskito-node", "crates/taskito-java", "crates/taskito-workflows", "crates/taskito-mesh"] resolver = "2" [workspace.dependencies] diff --git a/crates/taskito-java/Cargo.toml b/crates/taskito-java/Cargo.toml new file mode 100644 index 00000000..d65ba949 --- /dev/null +++ b/crates/taskito-java/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "taskito-java" +version = "0.16.4" +edition = "2021" +description = "Java (JNI) bindings for the Taskito task-queue core" +license = "MIT" +publish = false + +[lib] +crate-type = ["cdylib"] + +[features] +default = [] +postgres = ["taskito-core/postgres"] +redis = ["taskito-core/redis"] + +[dependencies] +taskito-core = { path = "../taskito-core" } +jni = "0.21" +serde = { workspace = true } +serde_json = { workspace = true } +once_cell = "1" diff --git a/crates/taskito-java/src/backend.rs b/crates/taskito-java/src/backend.rs new file mode 100644 index 00000000..69688b0a --- /dev/null +++ b/crates/taskito-java/src/backend.rs @@ -0,0 +1,70 @@ +//! Construct a [`QueueHandle`] from caller options (mirrors the Node shell). + +use taskito_core::{SqliteStorage, StorageBackend}; + +use crate::convert::OpenOptions; +use crate::error::BindingError; + +const DEFAULT_SQLITE_POOL: u32 = 8; +#[cfg(feature = "postgres")] +const DEFAULT_POSTGRES_POOL: u32 = 10; +#[cfg(feature = "postgres")] +const DEFAULT_POSTGRES_SCHEMA: &str = "taskito"; + +/// An open queue: the storage backend plus the default namespace applied to +/// enqueues that do not specify their own. +pub struct QueueHandle { + pub storage: StorageBackend, + pub namespace: Option, +} + +/// Error for a backend that is unknown or whose cargo feature is not compiled in. +fn unknown_backend(name: &str) -> BindingError { + BindingError::new(format!( + "backend '{name}' is not available (unknown, or this build omits its cargo feature)" + )) +} + +/// Reject an explicit zero pool size — r2d2 panics when `max_size == 0`, which +/// would take down the whole JVM. +fn resolve_pool_size(pool_size: Option, default: u32) -> Result { + match pool_size { + Some(0) => Err(BindingError::new("poolSize must be greater than 0")), + Some(n) => Ok(n), + None => Ok(default), + } +} + +/// Open the storage backend named by `options.backend` (default `"sqlite"`). +/// Returns an error if a requested backend was not compiled into this library. +pub fn open(options: OpenOptions) -> Result { + let storage = match options.backend.as_deref().unwrap_or("sqlite") { + "sqlite" => { + let pool = resolve_pool_size(options.pool_size, DEFAULT_SQLITE_POOL)?; + StorageBackend::Sqlite(SqliteStorage::with_pool_size(&options.dsn, pool)?) + } + #[cfg(feature = "postgres")] + "postgres" => { + let schema = options.schema.as_deref().unwrap_or(DEFAULT_POSTGRES_SCHEMA); + let pool = resolve_pool_size(options.pool_size, DEFAULT_POSTGRES_POOL)?; + StorageBackend::Postgres(taskito_core::PostgresStorage::with_schema_and_pool_size( + &options.dsn, + schema, + pool, + )?) + } + #[cfg(feature = "redis")] + "redis" => { + let storage = match options.prefix.as_deref() { + Some(prefix) => taskito_core::RedisStorage::with_prefix(&options.dsn, prefix), + None => taskito_core::RedisStorage::new(&options.dsn), + }?; + StorageBackend::Redis(storage) + } + other => return Err(unknown_backend(other)), + }; + Ok(QueueHandle { + storage, + namespace: options.namespace, + }) +} diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs new file mode 100644 index 00000000..fac7b047 --- /dev/null +++ b/crates/taskito-java/src/convert.rs @@ -0,0 +1,319 @@ +//! Serde views marshalled across the JNI boundary as JSON. +//! +//! Option and filter structs cross as JSON strings (decoded here); opaque job +//! payloads cross as raw `byte[]` and are never interpreted by the core. + +use serde::{Deserialize, Serialize}; +use taskito_core::job::{now_millis, Job, NewJob}; +use taskito_core::storage::models::{JobErrorRow, TaskLogRow, TaskMetricRow, WorkerRow}; +use taskito_core::storage::{DeadJob, QueueStats}; + +use crate::error::BindingError; + +const DEFAULT_QUEUE: &str = "default"; + +/// Options accepted by `NativeQueue.open`. +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct OpenOptions { + pub backend: Option, + pub dsn: String, + pub pool_size: Option, + /// Postgres schema; read only by the `postgres` backend. + #[cfg_attr(not(feature = "postgres"), allow(dead_code))] + pub schema: Option, + /// Redis key prefix; read only by the `redis` backend. + #[cfg_attr(not(feature = "redis"), allow(dead_code))] + pub prefix: Option, + pub namespace: Option, +} + +/// Per-enqueue options. Every field is optional; absent fields take core +/// defaults. +#[derive(Deserialize, Default)] +#[serde(rename_all = "camelCase", default)] +pub struct EnqueueOptions { + pub queue: Option, + pub priority: Option, + pub max_retries: Option, + pub timeout_ms: Option, + pub delay_ms: Option, + pub unique_key: Option, + pub metadata: Option, + pub namespace: Option, +} + +/// Parse a JSON argument, attributing any failure to the named field. +pub fn parse_json<'a, T: Deserialize<'a>>(raw: &'a str, what: &str) -> Result { + serde_json::from_str(raw).map_err(|e| BindingError::new(format!("invalid {what} JSON: {e}"))) +} + +/// Build a [`NewJob`] from an enqueue request, applying the queue's default +/// namespace when the request does not set one. +pub fn build_new_job( + task_name: String, + payload: Vec, + options: EnqueueOptions, + default_namespace: Option<&str>, +) -> NewJob { + let delay = options.delay_ms.unwrap_or(0).max(0); + NewJob { + queue: options.queue.unwrap_or_else(|| DEFAULT_QUEUE.to_string()), + task_name, + payload, + priority: options.priority.unwrap_or(0), + // Saturate rather than wrap/panic on an absurd delay. + scheduled_at: now_millis().saturating_add(delay), + max_retries: options.max_retries.unwrap_or(0), + timeout_ms: options.timeout_ms.unwrap_or(0), + unique_key: options.unique_key, + metadata: options.metadata, + notes: None, + depends_on: Vec::new(), + expires_at: None, + result_ttl_ms: None, + namespace: options + .namespace + .or_else(|| default_namespace.map(str::to_string)), + } +} + +/// Job-count snapshot returned by `NativeQueue.stats`. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct StatsView { + pub pending: i64, + pub running: i64, + pub completed: i64, + pub failed: i64, + pub dead: i64, + pub cancelled: i64, +} + +impl From for StatsView { + fn from(s: QueueStats) -> Self { + Self { + pending: s.pending, + running: s.running, + completed: s.completed, + failed: s.failed, + dead: s.dead, + cancelled: s.cancelled, + } + } +} + +/// Java-facing view of a job. Omits the opaque `payload`/`result` blobs; the +/// Java side deserializes those itself. `status` is the lowercase wire string +/// shared across SDKs. Timestamps are Unix milliseconds. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct JobView<'a> { + pub id: &'a str, + pub queue: &'a str, + pub task_name: &'a str, + pub status: &'static str, + pub priority: i32, + pub created_at: i64, + pub scheduled_at: i64, + pub started_at: Option, + pub completed_at: Option, + pub retry_count: i32, + pub max_retries: i32, + pub timeout_ms: i64, + pub progress: Option, + pub error: Option<&'a str>, + pub unique_key: Option<&'a str>, + pub namespace: Option<&'a str>, +} + +impl<'a> From<&'a Job> for JobView<'a> { + fn from(j: &'a Job) -> Self { + Self { + id: &j.id, + queue: &j.queue, + task_name: &j.task_name, + status: j.status.as_str(), + priority: j.priority, + created_at: j.created_at, + scheduled_at: j.scheduled_at, + started_at: j.started_at, + completed_at: j.completed_at, + retry_count: j.retry_count, + max_retries: j.max_retries, + timeout_ms: j.timeout_ms, + progress: j.progress, + error: j.error.as_deref(), + unique_key: j.unique_key.as_deref(), + namespace: j.namespace.as_deref(), + } + } +} + +/// Filter accepted by `NativeQueue.listJobs`. +#[derive(Deserialize, Default)] +#[serde(rename_all = "camelCase", default)] +pub struct JobFilter { + pub status: Option, + pub queue: Option, + pub task: Option, + pub limit: Option, + pub offset: Option, +} + +/// Map a lowercase status string to the core's `i32` status code. +pub fn status_code(status: &str) -> Option { + match status { + "pending" => Some(0), + "running" => Some(1), + "complete" | "completed" => Some(2), + "failed" => Some(3), + "dead" => Some(4), + "cancelled" => Some(5), + _ => None, + } +} + +/// Dead-letter entry view (omits the opaque payload). +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct DeadJobView<'a> { + pub id: &'a str, + pub original_job_id: &'a str, + pub queue: &'a str, + pub task_name: &'a str, + pub error: Option<&'a str>, + pub retry_count: i32, + pub failed_at: i64, + pub metadata: Option<&'a str>, + pub dlq_retry_count: i32, +} + +impl<'a> From<&'a DeadJob> for DeadJobView<'a> { + fn from(d: &'a DeadJob) -> Self { + Self { + id: &d.id, + original_job_id: &d.original_job_id, + queue: &d.queue, + task_name: &d.task_name, + error: d.error.as_deref(), + retry_count: d.retry_count, + failed_at: d.failed_at, + metadata: d.metadata.as_deref(), + dlq_retry_count: d.dlq_retry_count, + } + } +} + +/// One recorded error attempt for a job. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct JobErrorView<'a> { + pub id: &'a str, + pub job_id: &'a str, + pub attempt: i32, + pub error: &'a str, + pub failed_at: i64, +} + +impl<'a> From<&'a JobErrorRow> for JobErrorView<'a> { + fn from(e: &'a JobErrorRow) -> Self { + Self { + id: &e.id, + job_id: &e.job_id, + attempt: e.attempt, + error: &e.error, + failed_at: e.failed_at, + } + } +} + +/// A per-execution task metric. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct MetricView<'a> { + pub task_name: &'a str, + pub job_id: &'a str, + pub wall_time_ns: i64, + pub memory_bytes: i64, + pub succeeded: bool, + pub recorded_at: i64, +} + +impl<'a> From<&'a TaskMetricRow> for MetricView<'a> { + fn from(m: &'a TaskMetricRow) -> Self { + Self { + task_name: &m.task_name, + job_id: &m.job_id, + wall_time_ns: m.wall_time_ns, + memory_bytes: m.memory_bytes, + succeeded: m.succeeded, + recorded_at: m.recorded_at, + } + } +} + +/// A registered worker (heartbeat + identity). +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WorkerView<'a> { + pub worker_id: &'a str, + pub queues: &'a str, + pub status: &'a str, + pub last_heartbeat: i64, + pub started_at: Option, + pub hostname: Option<&'a str>, + pub pid: Option, + pub pool_type: Option<&'a str>, + pub threads: i32, + pub tags: Option<&'a str>, +} + +impl<'a> From<&'a WorkerRow> for WorkerView<'a> { + fn from(w: &'a WorkerRow) -> Self { + Self { + worker_id: &w.worker_id, + queues: &w.queues, + status: &w.status, + last_heartbeat: w.last_heartbeat, + started_at: w.started_at, + hostname: w.hostname.as_deref(), + pid: w.pid, + pool_type: w.pool_type.as_deref(), + threads: w.threads, + tags: w.tags.as_deref(), + } + } +} + +/// A task log line. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LogView<'a> { + pub id: &'a str, + pub job_id: &'a str, + pub task_name: &'a str, + pub level: &'a str, + pub message: &'a str, + pub extra: Option<&'a str>, + pub logged_at: i64, +} + +impl<'a> From<&'a TaskLogRow> for LogView<'a> { + fn from(l: &'a TaskLogRow) -> Self { + Self { + id: &l.id, + job_id: &l.job_id, + task_name: &l.task_name, + level: &l.level, + message: &l.message, + extra: l.extra.as_deref(), + logged_at: l.logged_at, + } + } +} + +/// Serialize a view to a JSON string for return across the boundary. +pub fn to_json(value: &T) -> Result { + serde_json::to_string(value).map_err(|e| BindingError::new(format!("serialize failed: {e}"))) +} diff --git a/crates/taskito-java/src/error.rs b/crates/taskito-java/src/error.rs new file mode 100644 index 00000000..930f203a --- /dev/null +++ b/crates/taskito-java/src/error.rs @@ -0,0 +1,36 @@ +//! Translate core errors into Java exceptions thrown across the JNI boundary. + +use jni::JNIEnv; +use taskito_core::QueueError; + +/// Fully-qualified JNI name (slashes, not dots) of the SDK's exception class. +pub const TASKITO_EXCEPTION: &str = "org/byteveda/taskito/TaskitoException"; + +/// A deferred Java exception. The binding builds one of these on failure and +/// throws it at the FFI boundary instead of unwinding a Rust panic across it. +pub struct BindingError { + class: &'static str, + message: String, +} + +impl BindingError { + /// A `TaskitoException` carrying `message`. + pub fn new(message: impl Into) -> Self { + Self { + class: TASKITO_EXCEPTION, + message: message.into(), + } + } + + /// Throw this as a pending Java exception. A failed throw means the JVM is + /// already in a bad state, so there is nothing further to do. + pub fn throw(&self, env: &mut JNIEnv) { + let _ = env.throw_new(self.class, &self.message); + } +} + +impl From for BindingError { + fn from(err: QueueError) -> Self { + Self::new(err.to_string()) + } +} diff --git a/crates/taskito-java/src/ffi.rs b/crates/taskito-java/src/ffi.rs new file mode 100644 index 00000000..c366defd --- /dev/null +++ b/crates/taskito-java/src/ffi.rs @@ -0,0 +1,114 @@ +//! JNI marshalling helpers and the panic-safety boundary. + +use std::panic::{catch_unwind, AssertUnwindSafe}; + +use jni::objects::{JByteArray, JObject, JObjectArray, JString}; +use jni::sys::{jbyteArray, jobjectArray, jstring}; +use jni::JNIEnv; + +use crate::error::BindingError; + +/// Run `body`, converting both `BindingError`s and Rust panics into thrown Java +/// exceptions and returning `fallback` in either case. +/// +/// A Rust panic must never unwind across the FFI boundary (undefined behaviour), +/// so every JNI entry point routes its work through this guard. +pub fn guard<'local, T>( + env: &mut JNIEnv<'local>, + fallback: T, + body: impl FnOnce(&mut JNIEnv<'local>) -> Result, +) -> T { + match catch_unwind(AssertUnwindSafe(|| body(env))) { + Ok(Ok(value)) => value, + Ok(Err(err)) => { + err.throw(env); + fallback + } + Err(_) => { + BindingError::new("taskito native panic").throw(env); + fallback + } + } +} + +/// Read a required Java `String` argument into an owned Rust `String`. +pub fn read_string(env: &mut JNIEnv, value: &JString) -> Result { + let java_str = env + .get_string(value) + .map_err(|e| BindingError::new(format!("invalid string argument: {e}")))?; + Ok(java_str.into()) +} + +/// Read an optional Java `String` argument; `null` becomes `None`. +pub fn read_optional_string( + env: &mut JNIEnv, + value: &JString, +) -> Result, BindingError> { + if value.is_null() { + Ok(None) + } else { + read_string(env, value).map(Some) + } +} + +/// Read a Java `byte[]` argument into an owned `Vec`. +pub fn read_bytes(env: &mut JNIEnv, value: &JByteArray) -> Result, BindingError> { + env.convert_byte_array(value) + .map_err(|e| BindingError::new(format!("invalid byte[] argument: {e}"))) +} + +/// Read a Java `byte[][]` argument into owned bytes per element. +pub fn read_bytes_array( + env: &mut JNIEnv, + value: &JObjectArray, +) -> Result>, BindingError> { + let len = env + .get_array_length(value) + .map_err(|e| BindingError::new(format!("invalid byte[][] argument: {e}")))?; + let mut out = Vec::with_capacity(len as usize); + for index in 0..len { + let element = env + .get_object_array_element(value, index) + .map_err(|e| BindingError::new(format!("invalid byte[][] element: {e}")))?; + // Auto-release the per-element local ref each iteration so a large array + // can't exhaust the JNI local-reference table before we return. + let element = env.auto_local(JByteArray::from(element)); + out.push(read_bytes(env, &element)?); + } + Ok(out) +} + +/// Build a Java `String` return value from an owned Rust `String`. +pub fn new_string(env: &mut JNIEnv, value: String) -> Result { + env.new_string(value) + .map(|s| s.into_raw()) + .map_err(|e| BindingError::new(format!("failed to allocate Java string: {e}"))) +} + +/// Build a Java `byte[]` return value from a slice. +pub fn new_bytes(env: &mut JNIEnv, value: &[u8]) -> Result { + env.byte_array_from_slice(value) + .map(|a| a.into_raw()) + .map_err(|e| BindingError::new(format!("failed to allocate byte[]: {e}"))) +} + +/// Build a Java `String[]` return value from owned strings. +pub fn new_string_array(env: &mut JNIEnv, values: &[String]) -> Result { + let class = env + .find_class("java/lang/String") + .map_err(|e| BindingError::new(format!("String class lookup failed: {e}")))?; + let array = env + .new_object_array(values.len() as i32, &class, JObject::null()) + .map_err(|e| BindingError::new(format!("failed to allocate String[]: {e}")))?; + for (index, value) in values.iter().enumerate() { + // Auto-release each element's local ref after storing it in the array, so + // a large String[] can't exhaust the JNI local-reference table. + let element = env.auto_local( + env.new_string(value) + .map_err(|e| BindingError::new(format!("failed to allocate Java string: {e}")))?, + ); + env.set_object_array_element(&array, index as i32, &element) + .map_err(|e| BindingError::new(format!("failed to set String[] element: {e}")))?; + } + Ok(array.into_raw()) +} diff --git a/crates/taskito-java/src/handle.rs b/crates/taskito-java/src/handle.rs new file mode 100644 index 00000000..7aedc6ce --- /dev/null +++ b/crates/taskito-java/src/handle.rs @@ -0,0 +1,29 @@ +//! Opaque Java `long` <-> heap-boxed Rust value. +//! +//! A handle is a `Box::into_raw` pointer as `jlong`; `close` reclaims it. Java +//! must not use a handle after `close`, nor `close` it concurrently with a call. + +use jni::sys::jlong; + +/// Box `value` on the heap and return its pointer as an opaque handle. +pub fn into_handle(value: T) -> jlong { + Box::into_raw(Box::new(value)) as jlong +} + +/// Borrow the value behind a handle. +/// +/// # Safety +/// `handle` must be a live pointer returned by [`into_handle`] for a `T` and not +/// yet passed to [`drop_handle`]. +pub unsafe fn borrow<'a, T>(handle: jlong) -> &'a T { + &*(handle as *const T) +} + +/// Reclaim and drop the value behind a handle. +/// +/// # Safety +/// `handle` must be a live pointer returned by [`into_handle`] for a `T`; it is +/// invalid afterward. +pub unsafe fn drop_handle(handle: jlong) { + drop(Box::from_raw(handle as *mut T)); +} diff --git a/crates/taskito-java/src/jvm.rs b/crates/taskito-java/src/jvm.rs new file mode 100644 index 00000000..c5e3395a --- /dev/null +++ b/crates/taskito-java/src/jvm.rs @@ -0,0 +1,18 @@ +//! Process-wide `JavaVM` handle, cached at `JNI_OnLoad`. + +use jni::JavaVM; +use once_cell::sync::OnceCell; + +static JAVA_VM: OnceCell = OnceCell::new(); + +/// Store the VM. Called once from `JNI_OnLoad`. +pub fn init(vm: JavaVM) { + let _ = JAVA_VM.set(vm); +} + +/// 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 new file mode 100644 index 00000000..6fa1419a --- /dev/null +++ b/crates/taskito-java/src/lib.rs @@ -0,0 +1,30 @@ +//! Java (JNI) bindings for the Taskito task-queue core. +//! +//! A thin binding shell — peer to the Python (`taskito-python`) and Node +//! (`taskito-node`) shells. All scheduling and storage logic lives in +//! `taskito-core`; this crate only marshals between the JVM and the core. +//! +//! Each JNI entry point lives in [`queue`] and is named +//! `Java_org_byteveda_taskito_internal_NativeQueue_` so the JVM links it +//! to the matching `native` method on the `NativeQueue` Java class. + +mod backend; +mod convert; +mod error; +mod ffi; +mod handle; +mod jvm; +mod queue; + +use std::ffi::c_void; + +use jni::sys::{jint, JNI_VERSION_1_8}; +use jni::JavaVM; + +/// Cache the `JavaVM` so worker threads can later attach to invoke Java +/// callbacks. The JVM calls this before any native method here. +#[no_mangle] +pub extern "system" fn JNI_OnLoad(vm: JavaVM, _reserved: *mut c_void) -> jint { + jvm::init(vm); + JNI_VERSION_1_8 +} diff --git a/crates/taskito-java/src/queue/admin.rs b/crates/taskito-java/src/queue/admin.rs new file mode 100644 index 00000000..91b6a479 --- /dev/null +++ b/crates/taskito-java/src/queue/admin.rs @@ -0,0 +1,194 @@ +//! Mutating administration entry points for `NativeQueue`. + +use jni::objects::{JClass, JString}; +use jni::sys::{jboolean, jlong, jstring, JNI_FALSE}; +use jni::JNIEnv; +use taskito_core::Storage; + +use super::borrow_queue; +use crate::convert::{to_json, DeadJobView}; +use crate::ffi::{guard, new_string, read_string}; + +/// `String listDead(long handle, long limit, long offset)` — dead-letter entries. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listDead<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + limit: jlong, + offset: jlong, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let dead = queue.storage.list_dead(limit.max(0), offset.max(0))?; + let views: Vec = dead.iter().map(DeadJobView::from).collect(); + new_string(env, to_json(&views)?) + }) +} + +/// `String retryDead(long handle, String deadId)` — re-enqueue; returns new id. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_retryDead<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + dead_id: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let id = read_string(env, &dead_id)?; + new_string(env, queue.storage.retry_dead(&id)?) + }) +} + +/// `boolean deleteDead(long handle, String deadId)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_deleteDead<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + dead_id: JString<'local>, +) -> jboolean { + guard(&mut env, JNI_FALSE, |env| { + let queue = unsafe { borrow_queue(handle) }; + let id = read_string(env, &dead_id)?; + Ok(super::to_jboolean(queue.storage.delete_dead(&id)?)) + }) +} + +/// `long purgeDead(long handle, long olderThanMs)` — returns rows removed. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_purgeDead<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + older_than_ms: jlong, +) -> jlong { + guard(&mut env, 0, |_env| { + let queue = unsafe { borrow_queue(handle) }; + Ok(queue.storage.purge_dead(older_than_ms)? as jlong) + }) +} + +/// `long purgeCompleted(long handle, long olderThanMs)` — returns rows removed. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_purgeCompleted<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + older_than_ms: jlong, +) -> jlong { + guard(&mut env, 0, |_env| { + let queue = unsafe { borrow_queue(handle) }; + Ok(queue.storage.purge_completed(older_than_ms)? as jlong) + }) +} + +/// `void pauseQueue(long handle, String queue)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_pauseQueue<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + queue_name: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let name = read_string(env, &queue_name)?; + queue.storage.pause_queue(&name)?; + Ok(()) + }) +} + +/// `void resumeQueue(long handle, String queue)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_resumeQueue<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + queue_name: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let name = read_string(env, &queue_name)?; + queue.storage.resume_queue(&name)?; + Ok(()) + }) +} + +/// `String listPausedQueues(long handle)` — a JSON array of queue names. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listPausedQueues<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + new_string(env, to_json(&queue.storage.list_paused_queues()?)?) + }) +} + +/// `String getSetting(long handle, String key)` — value, or `null` if unset. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_getSetting<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + key: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let key = read_string(env, &key)?; + match queue.storage.get_setting(&key)? { + Some(value) => new_string(env, value), + None => Ok(std::ptr::null_mut()), + } + }) +} + +/// `void setSetting(long handle, String key, String value)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_setSetting<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + key: JString<'local>, + value: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let key = read_string(env, &key)?; + let value = read_string(env, &value)?; + queue.storage.set_setting(&key, &value)?; + Ok(()) + }) +} + +/// `boolean deleteSetting(long handle, String key)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_deleteSetting<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + key: JString<'local>, +) -> jboolean { + guard(&mut env, JNI_FALSE, |env| { + let queue = unsafe { borrow_queue(handle) }; + let key = read_string(env, &key)?; + Ok(super::to_jboolean(queue.storage.delete_setting(&key)?)) + }) +} + +/// `String listSettings(long handle)` — a JSON map of key to value. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listSettings<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + new_string(env, to_json(&queue.storage.list_settings()?)?) + }) +} diff --git a/crates/taskito-java/src/queue/inspect.rs b/crates/taskito-java/src/queue/inspect.rs new file mode 100644 index 00000000..26523c67 --- /dev/null +++ b/crates/taskito-java/src/queue/inspect.rs @@ -0,0 +1,148 @@ +//! Read-only inspection entry points for `NativeQueue`. + +use std::collections::HashMap; + +use jni::objects::{JClass, JString}; +use jni::sys::{jlong, jstring}; +use jni::JNIEnv; +use taskito_core::Storage; + +use super::borrow_queue; +use crate::convert::{ + status_code, to_json, JobErrorView, JobFilter, JobView, MetricView, StatsView, WorkerView, +}; +use crate::error::BindingError; +use crate::ffi::{guard, new_string, read_optional_string, read_string}; + +const DEFAULT_LIMIT: i64 = 50; + +/// `String stats(long handle)` — job counts by status across all queues. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_stats<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + new_string(env, to_json(&StatsView::from(queue.storage.stats()?))?) + }) +} + +/// `String statsByQueue(long handle, String queue)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_statsByQueue<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + queue_name: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let name = read_string(env, &queue_name)?; + let stats = queue.storage.stats_by_queue(&name)?; + new_string(env, to_json(&StatsView::from(stats))?) + }) +} + +/// `String statsAllQueues(long handle)` — a JSON map of queue name to counts. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_statsAllQueues<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let all = queue.storage.stats_all_queues()?; + let mapped: HashMap = all + .into_iter() + .map(|(k, v)| (k, StatsView::from(v))) + .collect(); + new_string(env, to_json(&mapped)?) + }) +} + +/// `String listJobs(long handle, String filterJson)` — a JSON array of jobs. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listJobs<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + filter_json: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let raw = read_string(env, &filter_json)?; + let filter: JobFilter = crate::convert::parse_json(&raw, "job filter")?; + // Reject an unknown status so a typo fails loudly instead of matching all. + let status = match filter.status.as_deref() { + Some(s) => Some( + status_code(s).ok_or_else(|| BindingError::new(format!("unknown status '{s}'")))?, + ), + None => None, + }; + let limit = filter.limit.unwrap_or(DEFAULT_LIMIT).max(0); + let offset = filter.offset.unwrap_or(0).max(0); + let jobs = queue.storage.list_jobs( + status, + filter.queue.as_deref(), + filter.task.as_deref(), + limit, + offset, + queue.namespace.as_deref(), + )?; + let views: Vec = jobs.iter().map(JobView::from).collect(); + new_string(env, to_json(&views)?) + }) +} + +/// `String jobErrors(long handle, String jobId)` — one entry per failed attempt. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_jobErrors<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + job_id: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let id = read_string(env, &job_id)?; + let errors = queue.storage.get_job_errors(&id)?; + let views: Vec = errors.iter().map(JobErrorView::from).collect(); + new_string(env, to_json(&views)?) + }) +} + +/// `String metrics(long handle, String taskNameOrNull, long sinceMs)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_metrics<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + task_name: JString<'local>, + since_ms: jlong, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let task = read_optional_string(env, &task_name)?; + let metrics = queue.storage.get_metrics(task.as_deref(), since_ms)?; + let views: Vec = metrics.iter().map(MetricView::from).collect(); + new_string(env, to_json(&views)?) + }) +} + +/// `String listWorkers(long handle)` — registered workers. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listWorkers<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let workers = queue.storage.list_workers()?; + let views: Vec = workers.iter().map(WorkerView::from).collect(); + new_string(env, to_json(&views)?) + }) +} diff --git a/crates/taskito-java/src/queue/logs.rs b/crates/taskito-java/src/queue/logs.rs new file mode 100644 index 00000000..d6e418f4 --- /dev/null +++ b/crates/taskito-java/src/queue/logs.rs @@ -0,0 +1,53 @@ +//! Task-log entry points for `NativeQueue`. + +use jni::objects::{JClass, JString}; +use jni::sys::{jlong, jstring}; +use jni::JNIEnv; +use taskito_core::Storage; + +use super::borrow_queue; +use crate::convert::{to_json, LogView}; +use crate::ffi::{guard, new_string, read_optional_string, read_string}; + +/// `void writeTaskLog(long handle, String jobId, String taskName, String level, String message, String extraOrNull)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_writeTaskLog<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + job_id: JString<'local>, + task_name: JString<'local>, + level: JString<'local>, + message: JString<'local>, + extra: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let job_id = read_string(env, &job_id)?; + let task_name = read_string(env, &task_name)?; + let level = read_string(env, &level)?; + let message = read_string(env, &message)?; + let extra = read_optional_string(env, &extra)?; + queue + .storage + .write_task_log(&job_id, &task_name, &level, &message, extra.as_deref())?; + Ok(()) + }) +} + +/// `String getTaskLogs(long handle, String jobId)` — a JSON array of log lines. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_getTaskLogs<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + job_id: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let id = read_string(env, &job_id)?; + let logs = queue.storage.get_task_logs(&id)?; + let views: Vec = logs.iter().map(LogView::from).collect(); + new_string(env, to_json(&views)?) + }) +} diff --git a/crates/taskito-java/src/queue/mod.rs b/crates/taskito-java/src/queue/mod.rs new file mode 100644 index 00000000..146d3cab --- /dev/null +++ b/crates/taskito-java/src/queue/mod.rs @@ -0,0 +1,238 @@ +//! JNI entry points backing `org.byteveda.taskito.internal.NativeQueue`. +//! +//! Each function name encodes the Java package, class, and method the JVM links +//! it to. Every body runs inside [`guard`] so failures surface as Java +//! exceptions rather than unwinding across the FFI boundary. This module holds +//! the producer surface; inspection, admin, and log methods live in the sibling +//! submodules. + +mod admin; +mod inspect; +mod logs; + +use jni::objects::{JByteArray, JClass, JObjectArray, JString}; +use jni::sys::{jboolean, jbyteArray, jint, jlong, jobjectArray, jstring, JNI_FALSE, JNI_TRUE}; +use jni::JNIEnv; +use taskito_core::Storage; + +use crate::backend::{self, QueueHandle}; +use crate::convert::{build_new_job, parse_json, EnqueueOptions, OpenOptions}; +use crate::ffi::{ + guard, new_bytes, new_string, new_string_array, read_bytes, read_bytes_array, read_string, +}; +use crate::handle::{self, drop_handle, into_handle}; + +/// Borrow the queue behind a handle. +/// +/// # Safety +/// `handle` must be a live `QueueHandle` pointer (see [`crate::handle`]). +unsafe fn borrow_queue<'a>(handle: jlong) -> &'a QueueHandle { + handle::borrow::(handle) +} + +/// `long open(String optionsJson)` — open a backend, returning its handle. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_open<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + options_json: JString<'local>, +) -> jlong { + guard(&mut env, 0, |env| { + let raw = read_string(env, &options_json)?; + let options: OpenOptions = parse_json(&raw, "open options")?; + Ok(into_handle(backend::open(options)?)) + }) +} + +/// `void close(long handle)` — reclaim a handle. A zero handle is a no-op. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_close<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, +) { + // Routed through `guard` like every other entry point: a panic in a + // destructor must not unwind across the FFI boundary. + guard(&mut env, (), |_env| { + if handle != 0 { + unsafe { drop_handle::(handle) }; + } + Ok(()) + }) +} + +/// `String enqueue(long handle, String taskName, byte[] payload, String optionsJson)` +/// — enqueue a job and return its id. A set `uniqueKey` makes a duplicate +/// enqueue a no-op while the first job is pending/running (idempotency). +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_enqueue<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + task_name: JString<'local>, + payload: JByteArray<'local>, + options_json: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let task = read_string(env, &task_name)?; + let bytes = read_bytes(env, &payload)?; + let raw_opts = read_string(env, &options_json)?; + let options: EnqueueOptions = parse_json(&raw_opts, "enqueue options")?; + let unique = options.unique_key.is_some(); + let new_job = build_new_job(task, bytes, options, queue.namespace.as_deref()); + let job = if unique { + queue.storage.enqueue_unique(new_job) + } else { + queue.storage.enqueue(new_job) + }?; + new_string(env, job.id) + }) +} + +/// `String[] enqueueMany(long handle, String taskName, byte[][] payloads, String optionsJson)` +/// — enqueue a batch in one storage transaction. `optionsJson` is a JSON array +/// of per-job options, the same length as `payloads`. Returns the new job ids. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_enqueueMany<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + task_name: JString<'local>, + payloads: JObjectArray<'local>, + options_json: JString<'local>, +) -> jobjectArray { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let task = read_string(env, &task_name)?; + let payload_list = read_bytes_array(env, &payloads)?; + let raw_opts = read_string(env, &options_json)?; + let option_list: Vec = parse_json(&raw_opts, "enqueue batch options")?; + if option_list.len() != payload_list.len() { + return Err(crate::error::BindingError::new( + "payloads and options must have the same length", + )); + } + let namespace = queue.namespace.as_deref(); + let new_jobs = payload_list + .into_iter() + .zip(option_list) + .map(|(payload, options)| build_new_job(task.clone(), payload, options, namespace)) + .collect(); + let created = queue.storage.enqueue_batch(new_jobs)?; + let ids: Vec = created.into_iter().map(|job| job.id).collect(); + new_string_array(env, &ids) + }) +} + +/// `String getJob(long handle, String jobId)` — a JSON job view, or `null`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_getJob<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + job_id: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let id = read_string(env, &job_id)?; + match queue.storage.get_job(&id)? { + Some(job) => new_string( + env, + crate::convert::to_json(&crate::convert::JobView::from(&job))?, + ), + None => Ok(std::ptr::null_mut()), + } + }) +} + +/// `byte[] getResult(long handle, String jobId)` — the job's serialized result, +/// or `null` if absent or not yet complete. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_getResult<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + job_id: JString<'local>, +) -> jbyteArray { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let id = read_string(env, &job_id)?; + match queue.storage.get_job(&id)?.and_then(|job| job.result) { + Some(bytes) => new_bytes(env, &bytes), + None => Ok(std::ptr::null_mut()), + } + }) +} + +/// `boolean cancel(long handle, String jobId)` — cancel a pending job. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_cancel<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + job_id: JString<'local>, +) -> jboolean { + guard(&mut env, JNI_FALSE, |env| { + let queue = unsafe { borrow_queue(handle) }; + let id = read_string(env, &job_id)?; + Ok(to_jboolean(queue.storage.cancel_job(&id)?)) + }) +} + +/// `boolean requestCancel(long handle, String jobId)` — cooperatively cancel a +/// running job. Returns false if no such running job exists. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_requestCancel<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + job_id: JString<'local>, +) -> jboolean { + guard(&mut env, JNI_FALSE, |env| { + let queue = unsafe { borrow_queue(handle) }; + let id = read_string(env, &job_id)?; + Ok(to_jboolean(queue.storage.request_cancel(&id)?)) + }) +} + +/// `boolean isCancelRequested(long handle, String jobId)`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_isCancelRequested<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + job_id: JString<'local>, +) -> jboolean { + guard(&mut env, JNI_FALSE, |env| { + let queue = unsafe { borrow_queue(handle) }; + let id = read_string(env, &job_id)?; + Ok(to_jboolean(queue.storage.is_cancel_requested(&id)?)) + }) +} + +/// `void setProgress(long handle, String jobId, int progress)` — clamp to 0..100. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_setProgress<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + job_id: JString<'local>, + progress: jint, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let id = read_string(env, &job_id)?; + queue.storage.update_progress(&id, progress.clamp(0, 100))?; + Ok(()) + }) +} + +/// Map a Rust `bool` onto a JNI `jboolean`. +fn to_jboolean(value: bool) -> jboolean { + if value { + JNI_TRUE + } else { + JNI_FALSE + } +} diff --git a/sdks/java/.gitignore b/sdks/java/.gitignore new file mode 100644 index 00000000..67bcc2f7 --- /dev/null +++ b/sdks/java/.gitignore @@ -0,0 +1,2 @@ +.gradle/ +build/ diff --git a/sdks/java/README.md b/sdks/java/README.md new file mode 100644 index 00000000..caef584d --- /dev/null +++ b/sdks/java/README.md @@ -0,0 +1,71 @@ +# Taskito Java SDK + +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. + +## Usage + +```java +import org.byteveda.taskito.*; +import java.util.Map; + +Task sendEmail = Task.of("send_email", Map.class) + .withOptions(EnqueueOptions.builder().queue("emails").priority(5).build()); + +try (Queue queue = Taskito.builder().backend("sqlite").url("taskito.db").open()) { + String id = queue.enqueue(sendEmail, Map.of("to", "a@b.c")); + Job job = queue.getJob(id).orElseThrow(); // job.status == JobStatus.PENDING + QueueStats stats = queue.stats(); + queue.cancel(id); +} +``` + +## Structure + +```text +org.byteveda.taskito +├── Taskito entry point — Taskito.builder()...open() +├── Queue public interface +├── DefaultQueue package-private impl (not exported) +├── Task / Job / JobStatus / QueueStats / EnqueueOptions typed model +├── TaskitoException unchecked error type +├── serialization/ Serializer SPI + JsonSerializer (Jackson) default +├── spi/ QueueBackend — seam between API and the native layer +└── internal/ JNI bindings (NativeQueue, NativeLoader, JniQueueBackend) +``` + +The `spi.QueueBackend` seam keeps the public API independent of JNI: it can be +backed by the native library (default) or an in-memory fake in tests, and leaves +room for a future FFM/Panama backend without touching the API. + +## Build & checks + +```bash +./gradlew build # cargoBuild → stage native → compile → test → jar +./gradlew test # JUnit 5 +./gradlew spotlessApply # format (palantir-java-format) +./gradlew spotlessCheck # verify formatting +./gradlew checkstyleMain # static analysis +./gradlew check # test + spotlessCheck + checkstyle +``` + +The Rust shell is built and checked with the workspace tooling: + +```bash +cargo build -p taskito-java --release --features postgres,redis +cargo fmt -p taskito-java -- --check +cargo clippy -p taskito-java -- -D warnings +``` + +### Native library resolution + +At runtime the platform binary is extracted from the JAR and loaded. For local +development against a freshly built library, point the loader at it directly: + +```bash +-Dtaskito.native.lib=/abs/path/to/target/release/libtaskito_java.so +``` diff --git a/sdks/java/build.gradle.kts b/sdks/java/build.gradle.kts new file mode 100644 index 00000000..a2a17239 --- /dev/null +++ b/sdks/java/build.gradle.kts @@ -0,0 +1,95 @@ +plugins { + `java-library` + checkstyle + id("com.diffplug.spotless") version "6.25.0" +} + +group = "org.byteveda" +version = "0.16.4" + +java { + toolchain { + languageVersion.set(JavaLanguageVersion.of(11)) + } + withSourcesJar() + withJavadocJar() +} + +repositories { + mavenCentral() +} + +// --- Code integrity: formatting + static analysis ------------------------- + +spotless { + java { + target("src/**/*.java") + palantirJavaFormat("2.50.0") // modern 4-space formatter; `spotlessApply` to fix + removeUnusedImports() + trimTrailingWhitespace() + endWithNewline() + } +} + +checkstyle { + toolVersion = "10.21.4" + configFile = file("config/checkstyle/checkstyle.xml") + isIgnoreFailures = false +} +// Native staging copies binaries under build/resources; never lint those. +tasks.withType().configureEach { + source = fileTree("src") { include("**/*.java") } +} + +dependencies { + api("com.fasterxml.jackson.core:jackson-databind:2.17.2") + + testImplementation(platform("org.junit:junit-bom:5.10.3")) + testImplementation("org.junit.jupiter:junit-jupiter") + testRuntimeOnly("org.junit.platform:junit-platform-launcher") +} + +// --- Native (Rust cdylib) ------------------------------------------------- + +val crateDir = layout.projectDirectory.dir("../../crates/taskito-java") +val cargoTargetDir = layout.projectDirectory.dir("../../target") +val nativeStaging = layout.buildDirectory.dir("native") + +// Build the native library for the local platform. +val cargoBuild by tasks.registering(Exec::class) { + workingDir = crateDir.asFile + commandLine("cargo", "build", "--release", "--features", "postgres,redis") +} + +// Stage the built library under its platform-classifier resource path. +val copyNative by tasks.registering(Copy::class) { + dependsOn(cargoBuild) + from(cargoTargetDir.dir("release")) { + include("libtaskito_java.so", "libtaskito_java.dylib", "taskito_java.dll") + } + into(nativeStaging.map { it.dir("org/byteveda/taskito/native/${platformClassifier()}") }) +} + +sourceSets["main"].resources.srcDir(nativeStaging) +tasks.named("processResources") { dependsOn(copyNative) } + +tasks.test { + useJUnitPlatform() +} + +/** Resource classifier for the local platform, e.g. "linux-x86_64". */ +fun platformClassifier(): String { + val os = System.getProperty("os.name").lowercase() + val arch = System.getProperty("os.arch").lowercase() + val osDir = when { + os.contains("win") -> "windows" + os.contains("mac") || os.contains("darwin") -> "osx" + else -> "linux" + } + val archDir = when (arch) { + "amd64", "x86_64" -> "x86_64" + "aarch64", "arm64" -> "aarch64" + else -> arch + } + return "$osDir-$archDir" +} diff --git a/sdks/java/config/checkstyle/checkstyle.xml b/sdks/java/config/checkstyle/checkstyle.xml new file mode 100644 index 00000000..47dc3584 --- /dev/null +++ b/sdks/java/config/checkstyle/checkstyle.xml @@ -0,0 +1,39 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/sdks/java/gradle/wrapper/gradle-wrapper.jar b/sdks/java/gradle/wrapper/gradle-wrapper.jar new file mode 100644 index 00000000..b498d244 Binary files /dev/null and b/sdks/java/gradle/wrapper/gradle-wrapper.jar differ diff --git a/sdks/java/gradle/wrapper/gradle-wrapper.properties b/sdks/java/gradle/wrapper/gradle-wrapper.properties new file mode 100644 index 00000000..00cde9d2 --- /dev/null +++ b/sdks/java/gradle/wrapper/gradle-wrapper.properties @@ -0,0 +1,6 @@ +distributionBase=GRADLE_USER_HOME +distributionPath=wrapper/dists +distributionUrl=https\://services.gradle.org/distributions/gradle-8.12-bin.zip +distributionSha256Sum=7a00d51fb93147819aab76024feece20b6b84e420694101f276be952e08bef03 +zipStoreBase=GRADLE_USER_HOME +zipStorePath=wrapper/dists diff --git a/sdks/java/gradlew b/sdks/java/gradlew new file mode 100755 index 00000000..17a91706 --- /dev/null +++ b/sdks/java/gradlew @@ -0,0 +1,176 @@ +#!/usr/bin/env sh + +############################################################################## +## +## Gradle start up script for UN*X +## +############################################################################## + +# Attempt to set APP_HOME +# Resolve links: $0 may be a link +PRG="$0" +# Need this for relative symlinks. +while [ -h "$PRG" ] ; do + ls=`ls -ld "$PRG"` + link=`expr "$ls" : '.*-> \(.*\)$'` + if expr "$link" : '/.*' > /dev/null; then + PRG="$link" + else + PRG=`dirname "$PRG"`"/$link" + fi +done +SAVED="`pwd`" +cd "`dirname \"$PRG\"`/" >/dev/null +APP_HOME="`pwd -P`" +cd "$SAVED" >/dev/null + +APP_NAME="Gradle" +APP_BASE_NAME=`basename "$0"` + +# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. +DEFAULT_JVM_OPTS="" + +# Use the maximum available, or set MAX_FD != -1 to use that value. +MAX_FD="maximum" + +warn () { + echo "$*" +} + +die () { + echo + echo "$*" + echo + exit 1 +} + +# OS specific support (must be 'true' or 'false'). +cygwin=false +msys=false +darwin=false +nonstop=false +case "`uname`" in + CYGWIN* ) + cygwin=true + ;; + Darwin* ) + darwin=true + ;; + MINGW* ) + msys=true + ;; + NONSTOP* ) + nonstop=true + ;; +esac + +CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar + +# Determine the Java command to use to start the JVM. +if [ -n "$JAVA_HOME" ] ; then + if [ -x "$JAVA_HOME/jre/sh/java" ] ; then + # IBM's JDK on AIX uses strange locations for the executables + JAVACMD="$JAVA_HOME/jre/sh/java" + else + JAVACMD="$JAVA_HOME/bin/java" + fi + if [ ! -x "$JAVACMD" ] ; then + die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME + +Please set the JAVA_HOME variable in your environment to match the +location of your Java installation." + fi +else + JAVACMD="java" + which java >/dev/null 2>&1 || die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. + +Please set the JAVA_HOME variable in your environment to match the +location of your Java installation." +fi + +# Increase the maximum file descriptors if we can. +if [ "$cygwin" = "false" -a "$darwin" = "false" -a "$nonstop" = "false" ] ; then + MAX_FD_LIMIT=`ulimit -H -n` + if [ $? -eq 0 ] ; then + if [ "$MAX_FD" = "maximum" -o "$MAX_FD" = "max" ] ; then + MAX_FD="$MAX_FD_LIMIT" + fi + ulimit -n $MAX_FD + if [ $? -ne 0 ] ; then + warn "Could not set maximum file descriptor limit: $MAX_FD" + fi + else + warn "Could not query maximum file descriptor limit: $MAX_FD_LIMIT" + fi +fi + +# For Darwin, add options to specify how the application appears in the dock +if $darwin; then + GRADLE_OPTS="$GRADLE_OPTS \"-Xdock:name=$APP_NAME\" \"-Xdock:icon=$APP_HOME/media/gradle.icns\"" +fi + +# For Cygwin, switch paths to Windows format before running java +if $cygwin ; then + APP_HOME=`cygpath --path --mixed "$APP_HOME"` + CLASSPATH=`cygpath --path --mixed "$CLASSPATH"` + JAVACMD=`cygpath --unix "$JAVACMD"` + + # We build the pattern for arguments to be converted via cygpath + ROOTDIRSRAW=`find -L / -maxdepth 1 -mindepth 1 -type d 2>/dev/null` + SEP="" + for dir in $ROOTDIRSRAW ; do + ROOTDIRS="$ROOTDIRS$SEP$dir" + SEP="|" + done + OURCYGPATTERN="(^($ROOTDIRS))" + # Add a user-defined pattern to the cygpath arguments + if [ "$GRADLE_CYGPATTERN" != "" ] ; then + OURCYGPATTERN="$OURCYGPATTERN|($GRADLE_CYGPATTERN)" + fi + # Now convert the arguments - kludge to limit ourselves to /bin/sh + i=0 + for arg in "$@" ; do + CHECK=`echo "$arg"|egrep -c "$OURCYGPATTERN" -` + CHECK2=`echo "$arg"|egrep -c "^-"` ### Determine if an option + + if [ $CHECK -ne 0 ] && [ $CHECK2 -eq 0 ] ; then ### Added a condition + eval `echo args$i`=`cygpath --path --ignore --mixed "$arg"` + else + eval `echo args$i`="\"$arg\"" + fi + i=$((i+1)) + done + case $i in + (0) set -- ;; + (1) set -- "$args0" ;; + (2) set -- "$args0" "$args1" ;; + (3) set -- "$args0" "$args1" "$args2" ;; + (4) set -- "$args0" "$args1" "$args2" "$args3" ;; + (5) set -- "$args0" "$args1" "$args2" "$args3" "$args4" ;; + (6) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" ;; + (7) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" ;; + (8) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" ;; + (9) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" "$args8" ;; + esac +fi + +# Escape application args +save () { + for i do printf %s\\n "$i" | sed "s/'/'\\\\''/g;1s/^/'/;\$s/\$/' \\\\/" ; done + echo " " +} +APP_ARGS=$(save "$@") + +# Collect all arguments for the java command, following the shell quoting and substitution rules +if $JAVACMD --add-opens java.base/java.lang=ALL-UNNAMED -version ; then + DEFAULT_JVM_OPTS="--add-opens java.base/java.lang=ALL-UNNAMED $DEFAULT_JVM_OPTS" +fi + +eval set -- $DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS "\"-Dorg.gradle.appname=$APP_BASE_NAME\"" -classpath "\"$CLASSPATH\"" org.gradle.wrapper.GradleWrapperMain "$APP_ARGS" + +# by default we should be in the correct project dir, but when run from Finder on Mac, the cwd is wrong +if [ "$(uname)" = "Darwin" ] && [ "$HOME" = "$PWD" ]; then + cd "$(dirname "$0")" +fi + +exec "$JAVACMD" "$@" diff --git a/sdks/java/gradlew.bat b/sdks/java/gradlew.bat new file mode 100644 index 00000000..e95643d6 --- /dev/null +++ b/sdks/java/gradlew.bat @@ -0,0 +1,84 @@ +@if "%DEBUG%" == "" @echo off +@rem ########################################################################## +@rem +@rem Gradle startup script for Windows +@rem +@rem ########################################################################## + +@rem Set local scope for the variables with windows NT shell +if "%OS%"=="Windows_NT" setlocal + +set DIRNAME=%~dp0 +if "%DIRNAME%" == "" set DIRNAME=. +set APP_BASE_NAME=%~n0 +set APP_HOME=%DIRNAME% + +@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. +set DEFAULT_JVM_OPTS= + +@rem Find java.exe +if defined JAVA_HOME goto findJavaFromJavaHome + +set JAVA_EXE=java.exe +%JAVA_EXE% -version >NUL 2>&1 +if "%ERRORLEVEL%" == "0" goto init + +echo. +echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. +echo. +echo Please set the JAVA_HOME variable in your environment to match the +echo location of your Java installation. + +goto fail + +:findJavaFromJavaHome +set JAVA_HOME=%JAVA_HOME:"=% +set JAVA_EXE=%JAVA_HOME%/bin/java.exe + +if exist "%JAVA_EXE%" goto init + +echo. +echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME% +echo. +echo Please set the JAVA_HOME variable in your environment to match the +echo location of your Java installation. + +goto fail + +:init +@rem Get command-line arguments, handling Windows variants + +if not "%OS%" == "Windows_NT" goto win9xME_args + +:win9xME_args +@rem Slurp the command line arguments. +set CMD_LINE_ARGS= +set _SKIP=2 + +:win9xME_args_slurp +if "x%~1" == "x" goto execute + +set CMD_LINE_ARGS=%* + +:execute +@rem Setup the command line + +set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar + +@rem Execute Gradle +"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %CMD_LINE_ARGS% + +:end +@rem End local scope for the variables with windows NT shell +if "%ERRORLEVEL%"=="0" goto mainEnd + +:fail +rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of +rem the _cmd.exe /c_ return code! +if not "" == "%GRADLE_EXIT_CONSOLE%" exit 1 +exit /b 1 + +:mainEnd +if "%OS%"=="Windows_NT" endlocal + +:omega diff --git a/sdks/java/settings.gradle.kts b/sdks/java/settings.gradle.kts new file mode 100644 index 00000000..ca2d61ba --- /dev/null +++ b/sdks/java/settings.gradle.kts @@ -0,0 +1 @@ +rootProject.name = "taskito" diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DeadJob.java b/sdks/java/src/main/java/org/byteveda/taskito/DeadJob.java new file mode 100644 index 00000000..ffe43fa1 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/DeadJob.java @@ -0,0 +1,41 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Immutable view of a dead-letter entry. Timestamps are Unix milliseconds. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class DeadJob { + public final String id; + public final String originalJobId; + public final String queue; + public final String taskName; + public final String error; + public final int retryCount; + public final long failedAt; + public final String metadata; + public final int dlqRetryCount; + + @JsonCreator + public DeadJob( + @JsonProperty("id") String id, + @JsonProperty("originalJobId") String originalJobId, + @JsonProperty("queue") String queue, + @JsonProperty("taskName") String taskName, + @JsonProperty("error") String error, + @JsonProperty("retryCount") int retryCount, + @JsonProperty("failedAt") long failedAt, + @JsonProperty("metadata") String metadata, + @JsonProperty("dlqRetryCount") int dlqRetryCount) { + this.id = id; + this.originalJobId = originalJobId; + this.queue = queue; + this.taskName = taskName; + this.error = error; + this.retryCount = retryCount; + this.failedAt = failedAt; + this.metadata = metadata; + this.dlqRetryCount = dlqRetryCount; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java new file mode 100644 index 00000000..b3e64f81 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java @@ -0,0 +1,253 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.databind.JavaType; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.byteveda.taskito.serialization.Serializer; +import org.byteveda.taskito.spi.QueueBackend; + +/** + * Default {@link Queue}: maps the typed public API onto a {@link QueueBackend}, + * serializing payloads with the configured {@link Serializer} and decoding + * native JSON views with a private mapper. + */ +final class DefaultQueue implements Queue { + private static final ObjectMapper VIEWS = new ObjectMapper(); + + private final QueueBackend backend; + private final Serializer serializer; + + DefaultQueue(QueueBackend backend, Serializer serializer) { + this.backend = backend; + this.serializer = serializer; + } + + // ── Producer ──────────────────────────────────────────────────── + + @Override + public String enqueue(Task task, T payload) { + return enqueue(task, payload, task.options()); + } + + @Override + public String enqueue(Task task, T payload, EnqueueOptions options) { + return backend.enqueue(task.name(), serializer.serialize(payload), encode(options)); + } + + @Override + public String enqueue(String taskName, Object payload) { + return backend.enqueue(taskName, serializer.serialize(payload), encode(EnqueueOptions.none())); + } + + @Override + public List enqueueMany(Task task, List payloads) { + return enqueueMany(task, payloads, task.options()); + } + + @Override + public List enqueueMany(Task task, List payloads, EnqueueOptions options) { + byte[][] bytes = new byte[payloads.size()][]; + List perJob = new ArrayList<>(payloads.size()); + for (int i = 0; i < payloads.size(); i++) { + bytes[i] = serializer.serialize(payloads.get(i)); + perJob.add(options); + } + return Arrays.asList(backend.enqueueMany(task.name(), bytes, encode(perJob))); + } + + @Override + public Optional getJob(String jobId) { + return backend.getJobJson(jobId).map(json -> decode(json, Job.class)); + } + + @Override + public Optional getResult(String jobId) { + return backend.getResult(jobId); + } + + @Override + public Optional getResult(String jobId, Class type) { + return backend.getResult(jobId).map(bytes -> serializer.deserialize(bytes, type)); + } + + @Override + public boolean cancel(String jobId) { + return backend.cancel(jobId); + } + + @Override + public boolean requestCancel(String jobId) { + return backend.requestCancel(jobId); + } + + @Override + public boolean isCancelRequested(String jobId) { + return backend.isCancelRequested(jobId); + } + + @Override + public void setProgress(String jobId, int progress) { + backend.setProgress(jobId, progress); + } + + // ── Inspection ────────────────────────────────────────────────── + + @Override + public QueueStats stats() { + return decode(backend.statsJson(), QueueStats.class); + } + + @Override + public QueueStats statsByQueue(String queue) { + return decode(backend.statsByQueueJson(queue), QueueStats.class); + } + + @Override + public Map statsAllQueues() { + return decodeMap(backend.statsAllQueuesJson(), QueueStats.class); + } + + @Override + public List listJobs(JobFilter filter) { + return decodeList(backend.listJobsJson(encode(filter)), Job.class); + } + + @Override + public List jobErrors(String jobId) { + return decodeList(backend.jobErrorsJson(jobId), JobError.class); + } + + @Override + public List metrics(String taskName, long sinceMs) { + return decodeList(backend.metricsJson(taskName, sinceMs), TaskMetric.class); + } + + @Override + public List listWorkers() { + return decodeList(backend.listWorkersJson(), Worker.class); + } + + // ── Admin ─────────────────────────────────────────────────────── + + @Override + public List listDead(long limit, long offset) { + return decodeList(backend.listDeadJson(limit, offset), DeadJob.class); + } + + @Override + public String retryDead(String deadId) { + return backend.retryDead(deadId); + } + + @Override + public boolean deleteDead(String deadId) { + return backend.deleteDead(deadId); + } + + @Override + public long purgeDead(long olderThanMs) { + return backend.purgeDead(olderThanMs); + } + + @Override + public long purgeCompleted(long olderThanMs) { + return backend.purgeCompleted(olderThanMs); + } + + @Override + public void pauseQueue(String queue) { + backend.pauseQueue(queue); + } + + @Override + public void resumeQueue(String queue) { + backend.resumeQueue(queue); + } + + @Override + public List listPausedQueues() { + return decodeList(backend.listPausedQueuesJson(), String.class); + } + + @Override + public Optional getSetting(String key) { + return backend.getSetting(key); + } + + @Override + public void setSetting(String key, String value) { + backend.setSetting(key, value); + } + + @Override + public boolean deleteSetting(String key) { + return backend.deleteSetting(key); + } + + @Override + public Map listSettings() { + return decodeMap(backend.listSettingsJson(), String.class); + } + + // ── Logs ──────────────────────────────────────────────────────── + + @Override + public void writeTaskLog(String jobId, String taskName, String level, String message) { + backend.writeTaskLog(jobId, taskName, level, message, null); + } + + @Override + public void writeTaskLog(String jobId, String taskName, String level, String message, String extra) { + backend.writeTaskLog(jobId, taskName, level, message, extra); + } + + @Override + public List getTaskLogs(String jobId) { + return decodeList(backend.getTaskLogsJson(jobId), TaskLog.class); + } + + @Override + public void close() { + backend.close(); + } + + // ── JSON helpers ──────────────────────────────────────────────── + + private static String encode(Object value) { + try { + return VIEWS.writeValueAsString(value); + } catch (Exception e) { + throw new TaskitoException("failed to encode request", e); + } + } + + private static R decode(String json, Class type) { + try { + return VIEWS.readValue(json, type); + } catch (Exception e) { + throw new TaskitoException("failed to decode native response", e); + } + } + + private static List decodeList(String json, Class element) { + JavaType type = VIEWS.getTypeFactory().constructCollectionType(List.class, element); + try { + return VIEWS.readValue(json, type); + } catch (Exception e) { + throw new TaskitoException("failed to decode native response", e); + } + } + + private static Map decodeMap(String json, Class value) { + JavaType type = VIEWS.getTypeFactory().constructMapType(Map.class, String.class, value); + try { + return VIEWS.readValue(json, type); + } catch (Exception e) { + throw new TaskitoException("failed to decode native response", e); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/EnqueueOptions.java b/sdks/java/src/main/java/org/byteveda/taskito/EnqueueOptions.java new file mode 100644 index 00000000..96667661 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/EnqueueOptions.java @@ -0,0 +1,100 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Immutable per-enqueue options. Unset fields take core defaults. */ +@JsonInclude(JsonInclude.Include.NON_NULL) +public final class EnqueueOptions { + @JsonProperty("queue") private final String queue; + @JsonProperty("priority") private final Integer priority; + @JsonProperty("maxRetries") private final Integer maxRetries; + @JsonProperty("timeoutMs") private final Long timeoutMs; + @JsonProperty("delayMs") private final Long delayMs; + @JsonProperty("uniqueKey") private final String uniqueKey; + @JsonProperty("metadata") private final String metadata; + @JsonProperty("namespace") private final String namespace; + + private EnqueueOptions(Builder b) { + this.queue = b.queue; + this.priority = b.priority; + this.maxRetries = b.maxRetries; + this.timeoutMs = b.timeoutMs; + this.delayMs = b.delayMs; + this.uniqueKey = b.uniqueKey; + this.metadata = b.metadata; + this.namespace = b.namespace; + } + + public static EnqueueOptions none() { + return builder().build(); + } + + public static Builder builder() { + return new Builder(); + } + + public static final class Builder { + private String queue; + private Integer priority; + private Integer maxRetries; + private Long timeoutMs; + private Long delayMs; + private String uniqueKey; + private String metadata; + private String namespace; + + public Builder queue(String queue) { + this.queue = queue; + return this; + } + + public Builder priority(int priority) { + this.priority = priority; + return this; + } + + public Builder maxRetries(int maxRetries) { + if (maxRetries < 0) { + throw new IllegalArgumentException("maxRetries must be >= 0"); + } + this.maxRetries = maxRetries; + return this; + } + + public Builder timeoutMs(long timeoutMs) { + if (timeoutMs < 0) { + throw new IllegalArgumentException("timeoutMs must be >= 0"); + } + this.timeoutMs = timeoutMs; + return this; + } + + public Builder delayMs(long delayMs) { + if (delayMs < 0) { + throw new IllegalArgumentException("delayMs must be >= 0"); + } + this.delayMs = delayMs; + return this; + } + + public Builder uniqueKey(String uniqueKey) { + this.uniqueKey = uniqueKey; + return this; + } + + public Builder metadata(String metadata) { + this.metadata = metadata; + return this; + } + + public Builder namespace(String namespace) { + this.namespace = namespace; + return this; + } + + public EnqueueOptions build() { + return new EnqueueOptions(this); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Job.java b/sdks/java/src/main/java/org/byteveda/taskito/Job.java new file mode 100644 index 00000000..1d331556 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/Job.java @@ -0,0 +1,62 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Immutable view of a job. Timestamps are Unix milliseconds. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class Job { + public final String id; + public final String queue; + public final String taskName; + public final JobStatus status; + public final int priority; + public final long createdAt; + public final long scheduledAt; + public final Long startedAt; + public final Long completedAt; + public final int retryCount; + public final int maxRetries; + public final long timeoutMs; + public final Integer progress; + public final String error; + public final String uniqueKey; + public final String namespace; + + @JsonCreator + public Job( + @JsonProperty("id") String id, + @JsonProperty("queue") String queue, + @JsonProperty("taskName") String taskName, + @JsonProperty("status") JobStatus status, + @JsonProperty("priority") int priority, + @JsonProperty("createdAt") long createdAt, + @JsonProperty("scheduledAt") long scheduledAt, + @JsonProperty("startedAt") Long startedAt, + @JsonProperty("completedAt") Long completedAt, + @JsonProperty("retryCount") int retryCount, + @JsonProperty("maxRetries") int maxRetries, + @JsonProperty("timeoutMs") long timeoutMs, + @JsonProperty("progress") Integer progress, + @JsonProperty("error") String error, + @JsonProperty("uniqueKey") String uniqueKey, + @JsonProperty("namespace") String namespace) { + this.id = id; + this.queue = queue; + this.taskName = taskName; + this.status = status; + this.priority = priority; + this.createdAt = createdAt; + this.scheduledAt = scheduledAt; + this.startedAt = startedAt; + this.completedAt = completedAt; + this.retryCount = retryCount; + this.maxRetries = maxRetries; + this.timeoutMs = timeoutMs; + this.progress = progress; + this.error = error; + this.uniqueKey = uniqueKey; + this.namespace = namespace; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/JobError.java b/sdks/java/src/main/java/org/byteveda/taskito/JobError.java new file mode 100644 index 00000000..5f81dad5 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/JobError.java @@ -0,0 +1,29 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** One recorded error attempt for a job. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class JobError { + public final String id; + public final String jobId; + public final int attempt; + public final String error; + public final long failedAt; + + @JsonCreator + public JobError( + @JsonProperty("id") String id, + @JsonProperty("jobId") String jobId, + @JsonProperty("attempt") int attempt, + @JsonProperty("error") String error, + @JsonProperty("failedAt") long failedAt) { + this.id = id; + this.jobId = jobId; + this.attempt = attempt; + this.error = error; + this.failedAt = failedAt; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/JobFilter.java b/sdks/java/src/main/java/org/byteveda/taskito/JobFilter.java new file mode 100644 index 00000000..2c6f3183 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/JobFilter.java @@ -0,0 +1,68 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Immutable filter for {@link Queue#listJobs(JobFilter)}. Unset fields are ignored. */ +@JsonInclude(JsonInclude.Include.NON_NULL) +public final class JobFilter { + @JsonProperty("status") private final String status; + @JsonProperty("queue") private final String queue; + @JsonProperty("task") private final String task; + @JsonProperty("limit") private final Integer limit; + @JsonProperty("offset") private final Integer offset; + + private JobFilter(Builder b) { + this.status = b.status; + this.queue = b.queue; + this.task = b.task; + this.limit = b.limit; + this.offset = b.offset; + } + + public static JobFilter all() { + return builder().build(); + } + + public static Builder builder() { + return new Builder(); + } + + public static final class Builder { + private String status; + private String queue; + private String task; + private Integer limit; + private Integer offset; + + /** Lowercase wire status: pending/running/complete/failed/dead/cancelled. */ + public Builder status(JobStatus status) { + this.status = status.wire(); + return this; + } + + public Builder queue(String queue) { + this.queue = queue; + return this; + } + + public Builder task(String task) { + this.task = task; + return this; + } + + public Builder limit(int limit) { + this.limit = limit; + return this; + } + + public Builder offset(int offset) { + this.offset = offset; + return this; + } + + public JobFilter build() { + return new JobFilter(this); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/JobStatus.java b/sdks/java/src/main/java/org/byteveda/taskito/JobStatus.java new file mode 100644 index 00000000..85195385 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/JobStatus.java @@ -0,0 +1,32 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; +import java.util.Locale; + +/** Lifecycle state of a job. Wire form is the lowercase name, shared across SDKs. */ +public enum JobStatus { + PENDING, + RUNNING, + COMPLETE, + FAILED, + DEAD, + CANCELLED; + + @JsonValue + public String wire() { + return name().toLowerCase(Locale.ROOT); + } + + @JsonCreator + public static JobStatus fromWire(String wire) { + if (wire == null) { + throw new TaskitoException("job status is null"); + } + try { + return valueOf(wire.toUpperCase(Locale.ROOT)); + } catch (IllegalArgumentException e) { + throw new TaskitoException("unknown job status: " + wire, e); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java new file mode 100644 index 00000000..60fae7f2 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java @@ -0,0 +1,94 @@ +package org.byteveda.taskito; + +import java.util.List; +import java.util.Map; +import java.util.Optional; + +/** A Taskito queue. Obtain one from {@link Taskito#builder()}. */ +public interface Queue extends AutoCloseable { + // ── Producer ──────────────────────────────────────────────────── + + /** Enqueue a typed payload using the task's default options; returns the job id. */ + String enqueue(Task task, T payload); + + String enqueue(Task task, T payload, EnqueueOptions options); + + /** Enqueue by task name with an arbitrary payload and default options. */ + String enqueue(String taskName, Object payload); + + /** Enqueue a batch in one storage call; returns ids in input order. */ + List enqueueMany(Task task, List payloads); + + List enqueueMany(Task task, List payloads, EnqueueOptions options); + + Optional getJob(String jobId); + + /** The job's raw serialized result, if complete. */ + Optional getResult(String jobId); + + /** The job's result deserialized to {@code type}, if complete. */ + Optional getResult(String jobId, Class type); + + boolean cancel(String jobId); + + boolean requestCancel(String jobId); + + boolean isCancelRequested(String jobId); + + void setProgress(String jobId, int progress); + + // ── Inspection ────────────────────────────────────────────────── + + QueueStats stats(); + + QueueStats statsByQueue(String queue); + + Map statsAllQueues(); + + List listJobs(JobFilter filter); + + List jobErrors(String jobId); + + /** Per-execution metrics within the last {@code sinceMs}; null task = all. */ + List metrics(String taskName, long sinceMs); + + List listWorkers(); + + // ── Admin ─────────────────────────────────────────────────────── + + List listDead(long limit, long offset); + + /** Re-enqueue a dead-letter entry; returns the new job id. */ + String retryDead(String deadId); + + boolean deleteDead(String deadId); + + long purgeDead(long olderThanMs); + + long purgeCompleted(long olderThanMs); + + void pauseQueue(String queue); + + void resumeQueue(String queue); + + List listPausedQueues(); + + Optional getSetting(String key); + + void setSetting(String key, String value); + + boolean deleteSetting(String key); + + Map listSettings(); + + // ── Logs ──────────────────────────────────────────────────────── + + void writeTaskLog(String jobId, String taskName, String level, String message); + + void writeTaskLog(String jobId, String taskName, String level, String message, String extra); + + List getTaskLogs(String jobId); + + @Override + void close(); +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/QueueStats.java b/sdks/java/src/main/java/org/byteveda/taskito/QueueStats.java new file mode 100644 index 00000000..d177a626 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/QueueStats.java @@ -0,0 +1,32 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Job counts by status across all queues. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class QueueStats { + public final long pending; + public final long running; + public final long completed; + public final long failed; + public final long dead; + public final long cancelled; + + @JsonCreator + public QueueStats( + @JsonProperty("pending") long pending, + @JsonProperty("running") long running, + @JsonProperty("completed") long completed, + @JsonProperty("failed") long failed, + @JsonProperty("dead") long dead, + @JsonProperty("cancelled") long cancelled) { + this.pending = pending; + this.running = running; + this.completed = completed; + this.failed = failed; + this.dead = dead; + this.cancelled = cancelled; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Task.java b/sdks/java/src/main/java/org/byteveda/taskito/Task.java new file mode 100644 index 00000000..7aec0ef5 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/Task.java @@ -0,0 +1,40 @@ +package org.byteveda.taskito; + +import java.util.Objects; + +/** Typed task descriptor: a name, its payload type, and default enqueue options. */ +public final class Task { + private final String name; + private final Class payloadType; + private final EnqueueOptions options; + + private Task(String name, Class payloadType, EnqueueOptions options) { + this.name = Objects.requireNonNull(name, "task name must not be null"); + if (name.trim().isEmpty()) { + throw new IllegalArgumentException("task name must not be blank"); + } + this.payloadType = Objects.requireNonNull(payloadType, "payloadType must not be null"); + this.options = Objects.requireNonNull(options, "options must not be null"); + } + + public static Task of(String name, Class payloadType) { + return new Task<>(name, payloadType, EnqueueOptions.none()); + } + + /** A copy of this task with the given default options. */ + public Task withOptions(EnqueueOptions options) { + return new Task<>(name, payloadType, options); + } + + public String name() { + return name; + } + + public Class payloadType() { + return payloadType; + } + + public EnqueueOptions options() { + return options; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/TaskLog.java b/sdks/java/src/main/java/org/byteveda/taskito/TaskLog.java new file mode 100644 index 00000000..09f7c4f6 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/TaskLog.java @@ -0,0 +1,35 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** A task log line. Timestamps are Unix milliseconds. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class TaskLog { + public final String id; + public final String jobId; + public final String taskName; + public final String level; + public final String message; + public final String extra; + public final long loggedAt; + + @JsonCreator + public TaskLog( + @JsonProperty("id") String id, + @JsonProperty("jobId") String jobId, + @JsonProperty("taskName") String taskName, + @JsonProperty("level") String level, + @JsonProperty("message") String message, + @JsonProperty("extra") String extra, + @JsonProperty("loggedAt") long loggedAt) { + this.id = id; + this.jobId = jobId; + this.taskName = taskName; + this.level = level; + this.message = message; + this.extra = extra; + this.loggedAt = loggedAt; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/TaskMetric.java b/sdks/java/src/main/java/org/byteveda/taskito/TaskMetric.java new file mode 100644 index 00000000..037639d9 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/TaskMetric.java @@ -0,0 +1,32 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** A per-execution task metric. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class TaskMetric { + public final String taskName; + public final String jobId; + public final long wallTimeNs; + public final long memoryBytes; + public final boolean succeeded; + public final long recordedAt; + + @JsonCreator + public TaskMetric( + @JsonProperty("taskName") String taskName, + @JsonProperty("jobId") String jobId, + @JsonProperty("wallTimeNs") long wallTimeNs, + @JsonProperty("memoryBytes") long memoryBytes, + @JsonProperty("succeeded") boolean succeeded, + @JsonProperty("recordedAt") long recordedAt) { + this.taskName = taskName; + this.jobId = jobId; + this.wallTimeNs = wallTimeNs; + this.memoryBytes = memoryBytes; + this.succeeded = succeeded; + this.recordedAt = recordedAt; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java new file mode 100644 index 00000000..fd7f88b2 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java @@ -0,0 +1,83 @@ +package org.byteveda.taskito; + +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.LinkedHashMap; +import java.util.Map; +import org.byteveda.taskito.internal.JniQueueBackend; +import org.byteveda.taskito.serialization.JsonSerializer; +import org.byteveda.taskito.serialization.Serializer; +import org.byteveda.taskito.spi.QueueBackend; + +/** Entry point: open a {@link Queue} over a storage backend. */ +public final class Taskito { + private Taskito() {} + + public static Builder builder() { + return new Builder(); + } + + /** Configures and opens a {@link Queue}. */ + public static final class Builder { + private static final ObjectMapper JSON = new ObjectMapper(); + + private final Map options = new LinkedHashMap<>(); + private Serializer serializer = new JsonSerializer(); + + public Builder backend(String backend) { + options.put("backend", backend); + return this; + } + + /** Connection string: a file path for SQLite, a URL for Postgres/Redis. */ + public Builder url(String dsn) { + options.put("dsn", dsn); + return this; + } + + public Builder poolSize(int poolSize) { + options.put("poolSize", poolSize); + return this; + } + + public Builder schema(String schema) { + options.put("schema", schema); + return this; + } + + public Builder prefix(String prefix) { + options.put("prefix", prefix); + return this; + } + + public Builder namespace(String namespace) { + options.put("namespace", namespace); + return this; + } + + public Builder serializer(Serializer serializer) { + this.serializer = serializer; + return this; + } + + /** Open over an explicit backend, e.g. an in-memory fake in tests. */ + public Queue open(QueueBackend backend) { + return new DefaultQueue(backend, serializer); + } + + /** Open the native backend described by the configured options. */ + public Queue open() { + if (!options.containsKey("dsn")) { + throw new TaskitoException("url (dsn) is required"); + } + return new DefaultQueue(JniQueueBackend.open(encodeOptions()), serializer); + } + + private String encodeOptions() { + try { + return JSON.writeValueAsString(options); + } catch (Exception e) { + throw new TaskitoException("failed to encode open options", e); + } + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/TaskitoException.java b/sdks/java/src/main/java/org/byteveda/taskito/TaskitoException.java new file mode 100644 index 00000000..b8f9ccd5 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/TaskitoException.java @@ -0,0 +1,12 @@ +package org.byteveda.taskito; + +/** Unchecked exception for Taskito SDK and native errors. */ +public class TaskitoException extends RuntimeException { + public TaskitoException(String message) { + super(message); + } + + public TaskitoException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/Worker.java new file mode 100644 index 00000000..6c01df5e --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/Worker.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 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; + } +} 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 new file mode 100644 index 00000000..d189e355 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java @@ -0,0 +1,174 @@ +package org.byteveda.taskito.internal; + +import java.util.Optional; +import org.byteveda.taskito.spi.QueueBackend; + +/** JNI-backed {@link QueueBackend} over a native queue handle. */ +public final class JniQueueBackend implements QueueBackend { + private final long handle; + private volatile boolean closed; + + private JniQueueBackend(long handle) { + this.handle = handle; + } + + /** Open a native backend from its JSON options. */ + public static JniQueueBackend open(String optionsJson) { + return new JniQueueBackend(NativeQueue.open(optionsJson)); + } + + @Override + public String enqueue(String taskName, byte[] payload, String optionsJson) { + return NativeQueue.enqueue(handle, taskName, payload, optionsJson); + } + + @Override + public String[] enqueueMany(String taskName, byte[][] payloads, String optionsJson) { + return NativeQueue.enqueueMany(handle, taskName, payloads, optionsJson); + } + + @Override + public Optional getJobJson(String jobId) { + return Optional.ofNullable(NativeQueue.getJob(handle, jobId)); + } + + @Override + public Optional getResult(String jobId) { + return Optional.ofNullable(NativeQueue.getResult(handle, jobId)); + } + + @Override + public boolean cancel(String jobId) { + return NativeQueue.cancel(handle, jobId); + } + + @Override + public boolean requestCancel(String jobId) { + return NativeQueue.requestCancel(handle, jobId); + } + + @Override + public boolean isCancelRequested(String jobId) { + return NativeQueue.isCancelRequested(handle, jobId); + } + + @Override + public void setProgress(String jobId, int progress) { + NativeQueue.setProgress(handle, jobId, progress); + } + + @Override + public String statsJson() { + return NativeQueue.stats(handle); + } + + @Override + public String statsByQueueJson(String queue) { + return NativeQueue.statsByQueue(handle, queue); + } + + @Override + public String statsAllQueuesJson() { + return NativeQueue.statsAllQueues(handle); + } + + @Override + public String listJobsJson(String filterJson) { + return NativeQueue.listJobs(handle, filterJson); + } + + @Override + public String jobErrorsJson(String jobId) { + return NativeQueue.jobErrors(handle, jobId); + } + + @Override + public String metricsJson(String taskNameOrNull, long sinceMs) { + return NativeQueue.metrics(handle, taskNameOrNull, sinceMs); + } + + @Override + public String listWorkersJson() { + return NativeQueue.listWorkers(handle); + } + + @Override + public String listDeadJson(long limit, long offset) { + return NativeQueue.listDead(handle, limit, offset); + } + + @Override + public String retryDead(String deadId) { + return NativeQueue.retryDead(handle, deadId); + } + + @Override + public boolean deleteDead(String deadId) { + return NativeQueue.deleteDead(handle, deadId); + } + + @Override + public long purgeDead(long olderThanMs) { + return NativeQueue.purgeDead(handle, olderThanMs); + } + + @Override + public long purgeCompleted(long olderThanMs) { + return NativeQueue.purgeCompleted(handle, olderThanMs); + } + + @Override + public void pauseQueue(String queue) { + NativeQueue.pauseQueue(handle, queue); + } + + @Override + public void resumeQueue(String queue) { + NativeQueue.resumeQueue(handle, queue); + } + + @Override + public String listPausedQueuesJson() { + return NativeQueue.listPausedQueues(handle); + } + + @Override + public Optional getSetting(String key) { + return Optional.ofNullable(NativeQueue.getSetting(handle, key)); + } + + @Override + public void setSetting(String key, String value) { + NativeQueue.setSetting(handle, key, value); + } + + @Override + public boolean deleteSetting(String key) { + return NativeQueue.deleteSetting(handle, key); + } + + @Override + public String listSettingsJson() { + return NativeQueue.listSettings(handle); + } + + @Override + public void writeTaskLog(String jobId, String taskName, String level, String message, String extraOrNull) { + NativeQueue.writeTaskLog(handle, jobId, taskName, level, message, extraOrNull); + } + + @Override + public String getTaskLogsJson(String jobId) { + return NativeQueue.getTaskLogs(handle, jobId); + } + + /** 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 + public synchronized void close() { + if (!closed) { + closed = true; + NativeQueue.close(handle); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeLoader.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeLoader.java new file mode 100644 index 00000000..45a4c8db --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeLoader.java @@ -0,0 +1,75 @@ +package org.byteveda.taskito.internal; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Locale; + +/** + * Loads the native Taskito library. + * + *

Order: an explicit {@code -Dtaskito.native.lib=/path} override, otherwise + * the platform binary bundled in the JAR is extracted to a temp file and loaded. + */ +public final class NativeLoader { + private static final String LIB = "taskito_java"; + private static boolean loaded; + + private NativeLoader() {} + + /** Load the library once per process. */ + public static synchronized void load() { + if (loaded) { + return; + } + String override = System.getProperty("taskito.native.lib"); + if (override != null) { + System.load(override); + } else { + System.load(extractBundled()); + } + loaded = true; + } + + /** Extract the bundled platform binary to a temp file; return its path. */ + private static String extractBundled() { + String resource = "/org/byteveda/taskito/native/" + platformDir() + "/" + System.mapLibraryName(LIB); + InputStream in = NativeLoader.class.getResourceAsStream(resource); + if (in == null) { + throw new UnsatisfiedLinkError( + "no bundled native library for platform '" + platformDir() + "' (" + resource + + "); set -Dtaskito.native.lib=/path/to/library"); + } + try { + Path dir = Files.createTempDirectory("taskito-native"); + dir.toFile().deleteOnExit(); + Path target = dir.resolve(System.mapLibraryName(LIB)); + try (InputStream src = in; OutputStream out = Files.newOutputStream(target)) { + byte[] buf = new byte[8192]; + int n; + while ((n = src.read(buf)) != -1) { + out.write(buf, 0, n); + } + } + target.toFile().deleteOnExit(); + return target.toAbsolutePath().toString(); + } catch (IOException e) { + throw new UnsatisfiedLinkError("failed to extract native library: " + e.getMessage()); + } + } + + /** Resource classifier directory, e.g. {@code linux-x86_64}. */ + static String platformDir() { + String os = System.getProperty("os.name", "").toLowerCase(Locale.ROOT); + String arch = System.getProperty("os.arch", "").toLowerCase(Locale.ROOT); + String osDir = os.contains("win") ? "windows" + : (os.contains("mac") || os.contains("darwin")) ? "osx" + : "linux"; + String archDir = (arch.equals("amd64") || arch.equals("x86_64")) ? "x86_64" + : (arch.equals("aarch64") || arch.equals("arm64")) ? "aarch64" + : arch; + return osDir + "-" + archDir; + } +} 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 new file mode 100644 index 00000000..1139afd1 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java @@ -0,0 +1,87 @@ +package org.byteveda.taskito.internal; + +/** + * JNI surface over the Taskito core. + * + *

Opaque job payloads cross as {@code byte[]}; options, filters, and views + * cross as JSON strings. Methods throw {@link org.byteveda.taskito.TaskitoException} + * on native failure. The {@code handle} is an opaque pointer from {@link #open}. + */ +public final class NativeQueue { + static { + NativeLoader.load(); + } + + private NativeQueue() {} + + // ── Lifecycle ─────────────────────────────────────────────────── + public static native long open(String optionsJson); + + public static native void close(long handle); + + // ── Producer ──────────────────────────────────────────────────── + public static native String enqueue(long handle, String taskName, byte[] payload, String optionsJson); + + public static native String[] enqueueMany(long handle, String taskName, byte[][] payloads, String optionsJson); + + /** Returns a JSON job view, or {@code null} if absent. */ + public static native String getJob(long handle, String jobId); + + /** Returns the job's serialized result, or {@code null} if absent/incomplete. */ + public static native byte[] getResult(long handle, String jobId); + + public static native boolean cancel(long handle, String jobId); + + public static native boolean requestCancel(long handle, String jobId); + + public static native boolean isCancelRequested(long handle, String jobId); + + public static native void setProgress(long handle, String jobId, int progress); + + // ── Inspection ────────────────────────────────────────────────── + public static native String stats(long handle); + + public static native String statsByQueue(long handle, String queue); + + public static native String statsAllQueues(long handle); + + public static native String listJobs(long handle, String filterJson); + + public static native String jobErrors(long handle, String jobId); + + public static native String metrics(long handle, String taskNameOrNull, long sinceMs); + + public static native String listWorkers(long handle); + + // ── Admin ─────────────────────────────────────────────────────── + public static native String listDead(long handle, long limit, long offset); + + public static native String retryDead(long handle, String deadId); + + public static native boolean deleteDead(long handle, String deadId); + + public static native long purgeDead(long handle, long olderThanMs); + + public static native long purgeCompleted(long handle, long olderThanMs); + + public static native void pauseQueue(long handle, String queue); + + public static native void resumeQueue(long handle, String queue); + + public static native String listPausedQueues(long handle); + + /** Returns the value, or {@code null} if unset. */ + public static native String getSetting(long handle, String key); + + public static native void setSetting(long handle, String key, String value); + + public static native boolean deleteSetting(long handle, String key); + + public static native String listSettings(long handle); + + // ── Logs ──────────────────────────────────────────────────────── + 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); +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/package-info.java new file mode 100644 index 00000000..2c079164 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/package-info.java @@ -0,0 +1,2 @@ +/** Implementation details: JNI bindings, native-library loading, and the default backend. Not public API. */ +package org.byteveda.taskito.internal; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/package-info.java new file mode 100644 index 00000000..67e0d8b5 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/package-info.java @@ -0,0 +1,11 @@ +/** + * Taskito Java SDK — a typed client over the Rust task-queue core. + * + *

Open a {@link org.byteveda.taskito.Queue} via + * {@link org.byteveda.taskito.Taskito#builder()}. Enqueue typed payloads with a + * {@link org.byteveda.taskito.Task}; payloads serialize through a + * {@link org.byteveda.taskito.serialization.Serializer}. The implementation + * talks to the core through the {@link org.byteveda.taskito.spi.QueueBackend} + * seam, whose default implementation lives in {@code internal}. + */ +package org.byteveda.taskito; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/serialization/JsonSerializer.java b/sdks/java/src/main/java/org/byteveda/taskito/serialization/JsonSerializer.java new file mode 100644 index 00000000..396edbd0 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/serialization/JsonSerializer.java @@ -0,0 +1,35 @@ +package org.byteveda.taskito.serialization; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.byteveda.taskito.TaskitoException; + +/** Default {@link Serializer}: JSON via Jackson. */ +public final class JsonSerializer implements Serializer { + private final ObjectMapper mapper; + + public JsonSerializer() { + this(new ObjectMapper()); + } + + public JsonSerializer(ObjectMapper mapper) { + this.mapper = mapper; + } + + @Override + public byte[] serialize(Object value) { + try { + return mapper.writeValueAsBytes(value); + } catch (Exception e) { + throw new TaskitoException("failed to serialize payload", e); + } + } + + @Override + public T deserialize(byte[] bytes, Class type) { + try { + return mapper.readValue(bytes, type); + } catch (Exception e) { + throw new TaskitoException("failed to deserialize payload", e); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/serialization/Serializer.java b/sdks/java/src/main/java/org/byteveda/taskito/serialization/Serializer.java new file mode 100644 index 00000000..feedc94c --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/serialization/Serializer.java @@ -0,0 +1,8 @@ +package org.byteveda.taskito.serialization; + +/** Converts task payloads and results to and from the opaque bytes the core stores. */ +public interface Serializer { + byte[] serialize(Object value); + + T deserialize(byte[] bytes, Class type); +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/serialization/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/serialization/package-info.java new file mode 100644 index 00000000..030feb5d --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/serialization/package-info.java @@ -0,0 +1,2 @@ +/** Payload codec: the {@link org.byteveda.taskito.serialization.Serializer} SPI and its JSON default. */ +package org.byteveda.taskito.serialization; 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 new file mode 100644 index 00000000..f7d0b3e5 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java @@ -0,0 +1,87 @@ +package org.byteveda.taskito.spi; + +import java.util.Optional; + +/** + * Low-level queue operations a backend provides, in native-shaped terms (opaque + * {@code byte[]} payloads, JSON strings for options and views). + * + *

This is the seam between the public API and its implementation. The default + * implementation is JNI-backed; alternatives (an FFM backend, or an in-memory + * fake for tests) can be supplied without touching the public API. Methods that + * return a JSON collection never return {@code null}; nullable scalars are + * returned as {@link Optional}. + */ +public interface QueueBackend extends AutoCloseable { + // ── Producer ──────────────────────────────────────────────────── + + /** Enqueue one job; {@code optionsJson} is a single {@code EnqueueOptions} object. */ + String enqueue(String taskName, byte[] payload, String optionsJson); + + /** + * Enqueue a batch. Unlike {@link #enqueue}, {@code optionsJson} is a JSON + * array of per-job {@code EnqueueOptions}, the same length as + * {@code payloads}. Returns the job ids in input order. + */ + String[] enqueueMany(String taskName, byte[][] payloads, String optionsJson); + + Optional getJobJson(String jobId); + + Optional getResult(String jobId); + + boolean cancel(String jobId); + + boolean requestCancel(String jobId); + + boolean isCancelRequested(String jobId); + + void setProgress(String jobId, int progress); + + // ── Inspection ────────────────────────────────────────────────── + String statsJson(); + + String statsByQueueJson(String queue); + + String statsAllQueuesJson(); + + String listJobsJson(String filterJson); + + String jobErrorsJson(String jobId); + + String metricsJson(String taskNameOrNull, long sinceMs); + + String listWorkersJson(); + + // ── Admin ─────────────────────────────────────────────────────── + String listDeadJson(long limit, long offset); + + String retryDead(String deadId); + + boolean deleteDead(String deadId); + + long purgeDead(long olderThanMs); + + long purgeCompleted(long olderThanMs); + + void pauseQueue(String queue); + + void resumeQueue(String queue); + + String listPausedQueuesJson(); + + Optional getSetting(String key); + + void setSetting(String key, String value); + + boolean deleteSetting(String key); + + String listSettingsJson(); + + // ── Logs ──────────────────────────────────────────────────────── + void writeTaskLog(String jobId, String taskName, String level, String message, String extraOrNull); + + String getTaskLogsJson(String jobId); + + @Override + void close(); +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/package-info.java new file mode 100644 index 00000000..5412d5f7 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/package-info.java @@ -0,0 +1,2 @@ +/** Backend seam: {@link org.byteveda.taskito.spi.QueueBackend} decouples the public API from the JNI layer. */ +package org.byteveda.taskito.spi; diff --git a/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java b/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java new file mode 100644 index 00000000..c04418dd --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java @@ -0,0 +1,108 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class QueueTest { + + private static final Task> SEND_EMAIL = Task.of("send_email", mapType()); + + @SuppressWarnings("unchecked") + private static Class> mapType() { + return (Class>) (Class) Map.class; + } + + private Queue open(Path dir) { + return Taskito.builder().backend("sqlite").url(dir.resolve("t.db").toString()).open(); + } + + @Test + void enqueueGetCancelStats(@TempDir Path dir) { + try (Queue queue = open(dir)) { + String id = queue.enqueue( + SEND_EMAIL, + Collections.singletonMap("to", "a@b.c"), + EnqueueOptions.builder().priority(5).queue("emails").build()); + assertFalse(id.isEmpty()); + + Optional job = queue.getJob(id); + assertTrue(job.isPresent()); + assertEquals("send_email", job.get().taskName); + assertEquals(JobStatus.PENDING, job.get().status); + assertEquals(5, job.get().priority); + assertEquals("emails", job.get().queue); + + assertEquals(1, queue.stats().pending); + + assertTrue(queue.cancel(id)); + assertEquals(0, queue.stats().pending); + } + } + + @Test + void getMissingJobIsEmpty(@TempDir Path dir) { + try (Queue queue = open(dir)) { + assertTrue(queue.getJob("nope").isEmpty()); + } + } + + @Test + void enqueueManyAndFilteredList(@TempDir Path dir) { + try (Queue queue = open(dir)) { + List ids = queue.enqueueMany( + SEND_EMAIL.withOptions(EnqueueOptions.builder().queue("emails").build()), + Arrays.asList( + Collections.singletonMap("to", "a"), + Collections.singletonMap("to", "b"), + Collections.singletonMap("to", "c"))); + assertEquals(3, ids.size()); + assertEquals(3, queue.statsByQueue("emails").pending); + + List jobs = queue.listJobs( + JobFilter.builder().queue("emails").status(JobStatus.PENDING).build()); + assertEquals(3, jobs.size()); + } + } + + @Test + void settingsRoundTrip(@TempDir Path dir) { + try (Queue queue = open(dir)) { + queue.setSetting("theme", "dark"); + assertEquals("dark", queue.getSetting("theme").orElse("")); + assertEquals("dark", queue.listSettings().get("theme")); + assertTrue(queue.deleteSetting("theme")); + assertTrue(queue.getSetting("theme").isEmpty()); + } + } + + @Test + void pauseAndResumeQueue(@TempDir Path dir) { + try (Queue queue = open(dir)) { + queue.pauseQueue("emails"); + assertTrue(queue.listPausedQueues().contains("emails")); + queue.resumeQueue("emails"); + assertFalse(queue.listPausedQueues().contains("emails")); + } + } + + @Test + void taskLogsRoundTrip(@TempDir Path dir) { + try (Queue queue = open(dir)) { + String id = queue.enqueue("send_email", Collections.singletonMap("to", "a")); + queue.writeTaskLog(id, "send_email", "info", "hello"); + List logs = queue.getTaskLogs(id); + assertEquals(1, logs.size()); + assertEquals("hello", logs.get(0).message); + } + } +}