diff --git a/CHANGELOG.md b/CHANGELOG.md index 407b12e0..4a8ddab8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,8 @@ # Changelog +## [Unreleased] +- Add filtered live queries for parity with the JS SDK's `live(table).where(expr)`: `Surreal.selectLive(String table, String where)` appends a raw SurrealQL condition to the generated `LIVE SELECT * FROM type::table($tb)` statement (same trust model as `query()`), and `Surreal.selectLive(String table, String where, Map params)` additionally binds parameters — the safe way to inject condition values (e.g. `"age > $minAge"`; the binding name `tb` is reserved). Both keep the unfiltered variant's behavior: eager subscription-error surfacing (including an invalid WHERE clause), the live-query UUID available up front, and `kill()`/`close()` semantics [#194](https://github.com/surrealdb/surrealdb.java/pull/194). + ## [2.1.2] - 2026-06-24 - Add full geometry-type support for reading and writing all seven types — Point, LineString, Polygon, MultiPoint, MultiLineString, MultiPolygon, and GeometryCollection — via `Geometry` type-discrimination accessors (`getType()`, `isPolygon()`, …), readers returning `java.awt.geom.Point2D.Double` coordinates (x = longitude, y = latitude), and factory methods that serialize through `create`/`update` content and bound parameters. Also fixes an `UnsatisfiedLinkError` thrown when a `Geometry` was finalized (the native `deleteInstance` had no matching Rust symbol) [#183](https://github.com/surrealdb/surrealdb.java/pull/183). - Add `Surreal.kill(String)` / `Surreal.kill(java.util.UUID)` to terminate a live query by id, and `LiveStream.getQueryId()` to read the live-query UUID immediately, before the first notification; `selectLive` now starts the subscription through the public `LIVE SELECT` query path so the id is available up front. `kill()` stops notifications but does not close a local `LiveStream` — use `LiveStream.close()` to release a blocked `next()` [#184](https://github.com/surrealdb/surrealdb.java/pull/184). diff --git a/src/main/java/com/surrealdb/LiveStream.java b/src/main/java/com/surrealdb/LiveStream.java index 2b84c84d..f6184f8e 100644 --- a/src/main/java/com/surrealdb/LiveStream.java +++ b/src/main/java/com/surrealdb/LiveStream.java @@ -4,7 +4,8 @@ /** * Blocking iterator over live query notifications returned by - * {@link Surreal#selectLive(String)}. + * {@link Surreal#selectLive(String)} and its filtered + * {@link Surreal#selectLive(String, String, java.util.Map)} overloads. * *

* Typical usage: diff --git a/src/main/java/com/surrealdb/Surreal.java b/src/main/java/com/surrealdb/Surreal.java index 187f26f5..33920a25 100644 --- a/src/main/java/com/surrealdb/Surreal.java +++ b/src/main/java/com/surrealdb/Surreal.java @@ -149,6 +149,9 @@ private static native long upsertRecordIdRangeValue(long ptr, String table, long private static native LiveStream selectLive(long ptr, String table); + private static native LiveStream selectLiveWhere(long ptr, String table, String where, String[] paramsKey, + long[] valuePtrs); + private static native void kill(long ptr, String queryId); @Override @@ -289,6 +292,98 @@ public LiveStream selectLive(String table) { return selectLive(getPtr(), table); } + /** + * Starts a filtered live query on the given table and returns a blocking stream + * of notifications (CREATE, UPDATE, DELETE) for the records that match the + * given {@code WHERE} condition. This mirrors the JS SDK's + * {@code live(table).where(expr)}. + * + *

+ * The {@code where} argument is a raw SurrealQL expression appended verbatim to + * the generated {@code LIVE SELECT * FROM type::table($tb) WHERE ...} statement + * — the same trust model as {@link #query(String)}. Never concatenate untrusted + * values into it; use {@link #selectLive(String, String, Map)} to bind values + * safely. + * + *

+ * Like {@link #selectLive(String)}, this method blocks until the live query + * subscription is fully established on the server, and subscription errors + * (e.g. the table does not exist, or the {@code where} expression is invalid) + * are thrown immediately as a {@link SurrealException} rather than being + * deferred to the first {@link LiveStream#next()} call. + * + * @param table + * table name to watch (must already exist) + * @param where + * raw SurrealQL condition (without the {@code WHERE} keyword), e.g. + * {@code "age > 18"} + * @return a LiveStream delivering only matching notifications; the caller must + * call {@link LiveStream#close()} when done + * @throws SurrealException + * if live queries are not supported, the table does not exist, the + * condition is invalid, or the subscription fails + * @see #selectLive(String, String, Map) + */ + public LiveStream selectLive(String table, String where) { + return selectLive(table, where, java.util.Collections.emptyMap()); + } + + /** + * Starts a filtered live query on the given table with parameter bindings and + * returns a blocking stream of notifications (CREATE, UPDATE, DELETE) for the + * records that match the given {@code WHERE} condition. This mirrors the JS + * SDK's {@code live(table).where(expr)}. + * + *

+ * The {@code where} argument is a raw SurrealQL expression appended verbatim to + * the generated {@code LIVE SELECT * FROM type::table($tb) WHERE ...} statement + * — the same trust model as {@link #query(String, Map)}. Values should be + * referenced as {@code $name} placeholders and supplied through {@code params} + * (bound exactly like {@link #query(String, Map)}), which is the safe way to + * inject values: + * + * 

{@code
+	 * Map params = new HashMap<>();
+	 * params.put("minAge", 18);
+	 * try (LiveStream stream = surreal.selectLive("person", "age > $minAge", params)) {
+	 * 	...
+	 * }
+	 * }
+ * + *

+ * The parameter name {@code tb} is reserved (it binds the table name) and is + * rejected. Like {@link #selectLive(String)}, subscription errors (e.g. the + * table does not exist, or the {@code where} expression is invalid) are thrown + * immediately as a {@link SurrealException} rather than being deferred to the + * first {@link LiveStream#next()} call. + * + * @param table + * table name to watch (must already exist) + * @param where + * raw SurrealQL condition (without the {@code WHERE} keyword), + * referencing {@code params} as {@code $name} placeholders + * @param params + * parameter values to bind to the condition + * @return a LiveStream delivering only matching notifications; the caller must + * call {@link LiveStream#close()} when done + * @throws SurrealException + * if the condition is null or empty, a reserved parameter name is + * used, live queries are not supported, the table does not exist, + * the condition is invalid, or the subscription fails + */ + public LiveStream selectLive(String table, String where, Map params) { + if (where == null || where.trim().isEmpty()) { + throw new SurrealException("The where condition must not be null or empty"); + } + final Map valueMuts = ValueBuilder.convertParams(params); + final String[] keys = valueMuts.keySet().toArray(new String[0]); + final long[] values = new long[keys.length]; + for (int i = 0; i < keys.length; i++) { + values[i] = valueMuts.get(keys[i]).getPtr(); + } + return selectLiveWhere(getPtr(), table, where, keys, values); + } + /** * Stops the live query with the given id. After the server processes the kill, * the query delivers no further notifications. diff --git a/src/main/rust/surreal.rs b/src/main/rust/surreal.rs index 6cbceb9e..4b479042 100644 --- a/src/main/rust/surreal.rs +++ b/src/main/rust/surreal.rs @@ -515,7 +515,81 @@ pub extern "system" fn Java_com_surrealdb_Surreal_run<'local>( /// JNI implementation of `Surreal.selectLive(long ptr, String table)`. /// -/// Starts a live query on `table` and returns a fully-constructed Java +/// Starts an unfiltered live query on `table`. See [`start_live_select`] for +/// the shared architecture (eager error surfacing, up-front UUID, notification +/// thread). +#[no_mangle] +pub extern "system" fn Java_com_surrealdb_Surreal_selectLive<'local>( + mut env: EnvUnowned<'local>, + _class: JClass<'local>, + ptr: jlong, + table: JString<'local>, +) -> jobject { + with_env_body!(env, env, { + let surreal = get_surreal_ref!(env, ptr, null_mut); + let table = get_rust_string!(env, &table, null_mut); + + // Run the LIVE SELECT synchronously on the calling thread. type::table($tb) + // binds the table name safely (no SQL injection). + let mut params = BTreeMap::new(); + params.insert("tb".to_string(), Value::String(table)); + match start_live_select(env, surreal, "LIVE SELECT * FROM type::table($tb)", params) { + Ok(obj) => obj, + Err(e) => e.exception(env, null_mut), + } + }) +} + +/// JNI implementation of `Surreal.selectLiveWhere(long ptr, String table, +/// String where, String[] paramsKey, long[] valuePtrs)`, backing the filtered +/// `Surreal.selectLive(table, where[, params])` overloads (parity with the JS +/// SDK's `live(table).where(expr)`). +/// +/// The `where` clause is raw SurrealQL appended verbatim to the generated +/// statement — the same trust model as `query()` — so values must be injected +/// via the `params` bindings (built exactly like `queryWithBindings`). The +/// table name itself remains safely bound through `type::table($tb)`; the +/// binding name `tb` is therefore reserved. An invalid `where` expression +/// fails the statement synchronously and is surfaced as an eager exception, +/// like every other subscription error in [`start_live_select`]. +#[no_mangle] +pub extern "system" fn Java_com_surrealdb_Surreal_selectLiveWhere<'local>( + mut env: EnvUnowned<'local>, + _class: JClass<'local>, + ptr: jlong, + table: JString<'local>, + where_clause: JString<'local>, + params_keys: JObjectArray<'local, JString<'local>>, + params_values: JLongArray<'local>, +) -> jobject { + with_env_body!(env, env, { + let surreal = get_surreal_ref!(env, ptr, null_mut); + let table = get_rust_string!(env, &table, null_mut); + let where_clause = get_rust_string!(env, &where_clause, null_mut); + let mut params = build_params_map!(env, params_keys, params_values, null_mut); + if params.contains_key("tb") { + return SurrealError::SurrealDBJni( + "The parameter name 'tb' is reserved by selectLive (it binds the table name); use a different binding name".to_string(), + ) + .exception(env, null_mut); + } + params.insert("tb".to_string(), Value::String(table)); + // Deliberate contract (also documented on the Java `selectLive` overloads): + // the `where` predicate is raw SurrealQL trusted exactly like `query(String)` + // input — it is concatenated, not bound. Values must go through the `params` + // bindings above; only the table name is bound (`$tb`). Passing the clause as + // a positional `format!` argument keeps it inert to `format!` itself. + let sql = format!("LIVE SELECT * FROM type::table($tb) WHERE {}", where_clause); + match start_live_select(env, surreal, &sql, params) { + Ok(obj) => obj, + Err(e) => e.exception(env, null_mut), + } + }) +} + +/// Shared implementation behind the `selectLive` JNI entry points. +/// +/// Runs the given `LIVE SELECT` statement and returns a fully-constructed Java /// `LiveStream` object carrying both the native `LiveStreamChannel` handle and /// the live-query UUID. /// @@ -537,110 +611,97 @@ pub extern "system" fn Java_com_surrealdb_Surreal_run<'local>( /// /// Unlike the `.select(table).live()` builder (whose query id is private), the /// raw `LIVE SELECT` runs synchronously on the calling thread, so subscription -/// errors (e.g. the table does not exist) are surfaced eagerly via `take(0)` -/// rather than deferred to `next()`. The statement returns a `Value::Uuid`, -/// which we read up front so it is available from `LiveStream.getQueryId()` -/// before any notification arrives. A dedicated OS thread then drives the -/// notification stream on the shared tokio runtime, forwarding notifications -/// through an unbounded `async_channel` that the Java side reads via -/// `nextNative`. `take(0)` and `stream(0)` read independent maps, so reading -/// the UUID does not consume the stream. -#[no_mangle] -pub extern "system" fn Java_com_surrealdb_Surreal_selectLive<'local>( - mut env: EnvUnowned<'local>, - _class: JClass<'local>, - ptr: jlong, - table: JString<'local>, -) -> jobject { - with_env_body!(env, env, { - let surreal = get_surreal_ref!(env, ptr, null_mut); - let table = get_rust_string!(env, &table, null_mut); - - // Run the LIVE SELECT synchronously on the calling thread. type::table($tb) - // binds the table name safely (no SQL injection). - let mut params = BTreeMap::new(); - params.insert("tb".to_string(), Value::String(table)); - let res = - surrealdb_query::(surreal, "LIVE SELECT * FROM type::table($tb)", Some(params)); - let mut res = check_query_result!(env, res, null_mut); - - // The statement result at index 0 is the live-query UUID. Reading it via - // take(0) also surfaces subscription errors (e.g. the table does not exist). - let uuid_str = match res.take::(0) { - Ok(Value::Uuid(uuid)) => uuid.to_string(), - // Some servers/protocols return the live-query id wrapped in a one-element - // array; unwrap it, mirroring the SDK's `.select(table).live()` builder. - Ok(Value::Array(mut arr)) if arr.len() == 1 => match arr.pop() { - Some(Value::Uuid(uuid)) => uuid.to_string(), - other => { - return SurrealError::SurrealDBJni(format!( - "LIVE SELECT did not return a UUID: {other:?}" - )) - .exception(env, null_mut); - } - }, - Ok(other) => { - return SurrealError::SurrealDBJni(format!( - "LIVE SELECT did not return a UUID: {}", - other.to_sql() - )) - .exception(env, null_mut); +/// errors (e.g. the table does not exist, or an invalid WHERE clause) are +/// surfaced eagerly via `take(0)` rather than deferred to `next()`. The +/// statement returns a `Value::Uuid`, which we read up front so it is +/// available from `LiveStream.getQueryId()` before any notification arrives. +/// A dedicated OS thread then drives the notification stream on the shared +/// tokio runtime, forwarding notifications through an unbounded +/// `async_channel` that the Java side reads via `nextNative`. `take(0)` and +/// `stream(0)` read independent maps, so reading the UUID does not consume the +/// stream. +fn start_live_select( + env: &mut Env, + surreal: &Surreal, + sql: &str, + params: BTreeMap, +) -> StdResult { + let mut res = + surrealdb_query::(surreal, sql, Some(params)).map_err(SurrealError::SurrealDB)?; + + // The statement result at index 0 is the live-query UUID. Reading it via + // take(0) also surfaces subscription errors (e.g. the table does not exist). + let uuid_str = match res.take::(0) { + Ok(Value::Uuid(uuid)) => uuid.to_string(), + // Some servers/protocols return the live-query id wrapped in a one-element + // array; unwrap it, mirroring the SDK's `.select(table).live()` builder. + Ok(Value::Array(mut arr)) if arr.len() == 1 => match arr.pop() { + Some(Value::Uuid(uuid)) => uuid.to_string(), + other => { + return Err(SurrealError::SurrealDBJni(format!( + "LIVE SELECT did not return a UUID: {other:?}" + ))); } - Err(e) => return SurrealError::SurrealDB(e).exception(env, null_mut), - }; - - // The notification stream lives in a separate map from the results, so the - // take(0) above does not disturb it. - let qstream = match res.stream::(0) { - Ok(s) => s, - Err(e) => return SurrealError::SurrealDB(e).exception(env, null_mut), - }; + }, + Ok(other) => { + return Err(SurrealError::SurrealDBJni(format!( + "LIVE SELECT did not return a UUID: {}", + other.to_sql() + ))); + } + Err(e) => return Err(SurrealError::SurrealDB(e)), + }; - // Notification channel: background thread produces, nextNative consumes. - let (tx, rx) = async_channel::unbounded(); - // Shutdown channel: dropping shutdown_tx signals the background thread to exit. - let (shutdown_tx, shutdown_rx) = async_channel::bounded::<()>(1); - - let tx_thread = tx.clone(); - let join_handle = std::thread::spawn(move || { - TOKIO_RUNTIME.block_on(async move { - let mut qstream = qstream; - loop { - tokio::select! { - _ = shutdown_rx.recv() => break, - item = qstream.next() => match item { - Some(i) => { - let _ = tx_thread.send(i).await; - } - None => break, - }, - } + // The notification stream lives in a separate map from the results, so the + // take(0) above does not disturb it. + let qstream = res.stream::(0).map_err(SurrealError::SurrealDB)?; + + // Notification channel: background thread produces, nextNative consumes. + let (tx, rx) = async_channel::unbounded(); + // Shutdown channel: dropping shutdown_tx signals the background thread to exit. + let (shutdown_tx, shutdown_rx) = async_channel::bounded::<()>(1); + + let tx_thread = tx.clone(); + let join_handle = std::thread::spawn(move || { + TOKIO_RUNTIME.block_on(async move { + let mut qstream = qstream; + loop { + tokio::select! { + _ = shutdown_rx.recv() => break, + item = qstream.next() => match item { + Some(i) => { + let _ = tx_thread.send(i).await; + } + None => break, + }, } - }); + } }); - - let recv_mutex = std::sync::Arc::new(parking_lot::Mutex::new(())); - let handle = JniTypes::new_live_stream(( - recv_mutex, - parking_lot::Mutex::new(Some(join_handle)), - parking_lot::Mutex::new(Some(shutdown_tx)), - parking_lot::Mutex::new(Some(rx)), - )); - - // Construct and return a LiveStream(handle, queryId), mirroring how live.rs - // builds a LiveNotification. - let uuid_raw = new_string!(env, uuid_str, null_mut); - let uuid_jstr = unsafe { JObject::from_raw(env, uuid_raw) }; - let class = match env.find_class(jni_str!("com/surrealdb/LiveStream")) { - Ok(c) => c, - Err(e) => return SurrealError::from(e).exception(env, null_mut), - }; - let args = [JValue::Long(handle), JValue::Object(&uuid_jstr)]; - match env.new_object(class, jni_sig!("(JLjava/lang/String;)V"), &args) { - Ok(obj) => obj.into_raw(), - Err(e) => SurrealError::from(e).exception(env, null_mut), - } - }) + }); + + let recv_mutex = std::sync::Arc::new(parking_lot::Mutex::new(())); + let handle = JniTypes::new_live_stream(( + recv_mutex, + parking_lot::Mutex::new(Some(join_handle)), + parking_lot::Mutex::new(Some(shutdown_tx)), + parking_lot::Mutex::new(Some(rx)), + )); + + // Construct and return a LiveStream(handle, queryId), mirroring how live.rs + // builds a LiveNotification. + let uuid_raw = env + .new_string(uuid_str) + .map_err(SurrealError::from)? + .into_raw(); + let uuid_jstr = unsafe { JObject::from_raw(env, uuid_raw) }; + let class = env + .find_class(jni_str!("com/surrealdb/LiveStream")) + .map_err(SurrealError::from)?; + let args = [JValue::Long(handle), JValue::Object(&uuid_jstr)]; + let obj = env + .new_object(class, jni_sig!("(JLjava/lang/String;)V"), &args) + .map_err(SurrealError::from)?; + Ok(obj.into_raw()) } /// JNI implementation of `Surreal.kill(long ptr, String queryId)`. diff --git a/src/test/java/com/surrealdb/LiveSelectWhereTests.java b/src/test/java/com/surrealdb/LiveSelectWhereTests.java new file mode 100644 index 00000000..48cfbbe0 --- /dev/null +++ b/src/test/java/com/surrealdb/LiveSelectWhereTests.java @@ -0,0 +1,355 @@ +package com.surrealdb; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; +import org.junit.jupiter.api.Test; + +import com.surrealdb.pojos.Person; + +/** + * Tests for filtered live queries: {@link Surreal#selectLive(String, String)} + * and {@link Surreal#selectLive(String, String, Map)} (parity with the JS SDK's + * {@code live(table).where(expr)}). Follows the timeout-safe patterns of + * {@link LiveQueryTests}: {@code next()} is always called on a daemon consumer + * thread that exits when the stream closes, so a blocking call can never hang + * the test. + */ +public class LiveSelectWhereTests { + + /** + * Starts a daemon consumer thread that drains {@code stream} into + * {@code events} (as "ACTION id" strings) until the stream closes, and waits + * until it is running. + */ + private static Thread startConsumer(LiveStream stream, List events, AtomicReference error) + throws Exception { + CountDownLatch consuming = new CountDownLatch(1); + Thread consumer = new Thread(() -> { + try { + consuming.countDown(); + Optional n; + while ((n = stream.next()).isPresent()) { + LiveNotification notification = n.get(); + String id = notification.getValue().getObject().get("id").toString(); + events.add(notification.getAction().toUpperCase() + " " + id); + } + } catch (Throwable t) { + error.set(t); + } + }); + consumer.setDaemon(true); + consumer.start(); + assertTrue(consuming.await(2, TimeUnit.SECONDS), "Consumer thread did not start in time"); + Thread.sleep(500); + return consumer; + } + + /** + * Only records matching the parameterized WHERE condition produce CREATE + * notifications; non-matching records are filtered server-side. + */ + @Test + void selectLiveWhere_onlyMatchingCreatesNotify() throws Exception { + List events = Collections.synchronizedList(new ArrayList<>()); + AtomicReference error = new AtomicReference<>(); + + try (Surreal surreal = new Surreal()) { + surreal.connect("memory").useNs("test").useDb("test"); + surreal.query("DEFINE TABLE person SCHEMALESS"); + + Map params = new HashMap<>(); + params.put("minCategory", 1); + try (LiveStream stream = surreal.selectLive("person", "category > $minCategory", params)) { + startConsumer(stream, events, error); + + // tobie has category 1 (does not match "category > 1"), jaime has + // category 2 (matches). Create the non-matching record first: had it + // produced a notification, it would arrive before jaime's. + surreal.create(new RecordId("person", 1), Helpers.tobie); + surreal.create(new RecordId("person", 2), Helpers.jaime); + Thread.sleep(800); + + assertEquals(Collections.singletonList("CREATE person:2"), new ArrayList<>(events), + "Only the matching CREATE should be delivered"); + } + if (error.get() != null) { + fail("next() threw an exception: " + error.get()); + } + } + } + + /** + * UPDATEs that move a record across the filter boundary. Observed engine + * semantics (embedded memory engine): the condition is evaluated against the + * record's NEW value, so an update moving a record INTO the filter delivers an + * UPDATE notification, while an update moving a record OUT of the filter + * delivers nothing (no DELETE-style notification). + */ + @Test + void selectLiveWhere_updateAcrossFilterBoundary() throws Exception { + List events = Collections.synchronizedList(new ArrayList<>()); + AtomicReference error = new AtomicReference<>(); + + try (Surreal surreal = new Surreal()) { + surreal.connect("memory").useNs("test").useDb("test"); + // Records exist before the live query starts: no CREATE notifications. + surreal.create(new RecordId("person", 1), Helpers.tobie); // category 1: outside filter + surreal.create(new RecordId("person", 2), Helpers.jaime); // category 2: inside filter + + Map params = new HashMap<>(); + params.put("minCategory", 1); + try (LiveStream stream = surreal.selectLive("person", "category > $minCategory", params)) { + startConsumer(stream, events, error); + + // Moves person:1 INTO the filter (1 -> 5): notifies with the new value. + surreal.query("UPDATE person:1 SET category = 5"); + Thread.sleep(800); + assertEquals(Collections.singletonList("UPDATE person:1"), new ArrayList<>(events), + "An update moving a record into the filter should notify"); + + // Moves person:2 OUT of the filter (2 -> 0): no notification. + surreal.query("UPDATE person:2 SET category = 0"); + Thread.sleep(800); + assertEquals(Collections.singletonList("UPDATE person:1"), new ArrayList<>(events), + "An update moving a record out of the filter should not notify"); + } + if (error.get() != null) { + fail("next() threw an exception: " + error.get()); + } + } + } + + /** + * The two-argument overload (no bindings) works with a literal condition. + */ + @Test + void selectLiveWhere_literalCondition() throws Exception { + List events = Collections.synchronizedList(new ArrayList<>()); + AtomicReference error = new AtomicReference<>(); + + try (Surreal surreal = new Surreal()) { + surreal.connect("memory").useNs("test").useDb("test"); + surreal.query("DEFINE TABLE person SCHEMALESS"); + + try (LiveStream stream = surreal.selectLive("person", "category > 1")) { + startConsumer(stream, events, error); + + surreal.create(new RecordId("person", 1), Helpers.tobie); // category 1: filtered out + surreal.create(new RecordId("person", 2), Helpers.jaime); // category 2: matches + Thread.sleep(800); + + assertEquals(Collections.singletonList("CREATE person:2"), new ArrayList<>(events)); + } + if (error.get() != null) { + fail("next() threw an exception: " + error.get()); + } + } + } + + /** + * The live query UUID is available up front, exactly like the unfiltered + * variant. + */ + @Test + void selectLiveWhere_exposesQueryIdUpFront() { + try (Surreal surreal = new Surreal()) { + surreal.connect("memory").useNs("test").useDb("test"); + surreal.query("DEFINE TABLE person SCHEMALESS"); + try (LiveStream stream = surreal.selectLive("person", "category > 1")) { + String queryId = stream.getQueryId(); + assertNotNull(queryId, "Live query id should be available before any notification"); + assertEquals(queryId, java.util.UUID.fromString(queryId).toString()); + } + } + } + + /** + * {@link Surreal#kill(String)} stops a filtered live query: notifications + * before the kill arrive, none after — same contract as the unfiltered variant + * ({@code LiveQueryTests.killLiveQuery_byQueryId}). + */ + @Test + void selectLiveWhere_killStopsNotifications() throws Exception { + List events = Collections.synchronizedList(new ArrayList<>()); + AtomicReference error = new AtomicReference<>(); + + try (Surreal surreal = new Surreal()) { + surreal.connect("memory").useNs("test").useDb("test"); + surreal.query("DEFINE TABLE person SCHEMALESS"); + + try (LiveStream stream = surreal.selectLive("person", "category > 1")) { + String queryId = stream.getQueryId(); + assertNotNull(queryId); + startConsumer(stream, events, error); + + // A matching create before the kill is delivered. + surreal.create(new RecordId("person", 2), Helpers.jaime); + Thread.sleep(500); + assertEquals(1, events.size(), "Notification before kill should be delivered"); + + // After the kill, a further matching create must NOT be delivered. + surreal.kill(queryId); + Thread.sleep(300); + surreal.create(new RecordId("person", 3), Helpers.emmanuel); + Thread.sleep(800); + assertEquals(1, events.size(), "No notifications should be delivered after kill"); + } + if (error.get() != null) { + fail("next() threw an exception: " + error.get()); + } + } + } + + /** + * close() unblocks a blocked next(), exactly like the unfiltered variant. + */ + @Test + void selectLiveWhere_closeUnblocksBlockedNext() throws Exception { + AtomicReference> result = new AtomicReference<>(); + AtomicReference error = new AtomicReference<>(); + CountDownLatch consuming = new CountDownLatch(1); + + try (Surreal surreal = new Surreal()) { + surreal.connect("memory").useNs("test").useDb("test"); + surreal.query("DEFINE TABLE person SCHEMALESS"); + + LiveStream stream = surreal.selectLive("person", "category > 1"); + Thread consumer = new Thread(() -> { + try { + consuming.countDown(); + result.set(stream.next()); + } catch (Throwable t) { + error.set(t); + } + }); + consumer.setDaemon(true); + consumer.start(); + + assertTrue(consuming.await(2, TimeUnit.SECONDS)); + Thread.sleep(500); + + stream.close(); + + consumer.join(5000); + assertFalse(consumer.isAlive(), "Consumer thread still alive after close() — next() never returned"); + if (error.get() != null) { + fail("next() threw instead of returning empty: " + error.get()); + } + assertNotNull(result.get(), "next() should have returned after close()"); + assertFalse(result.get().isPresent(), "next() should return empty after stream is closed"); + } + } + + /** + * An invalid WHERE clause surfaces an eager error, matching how selectLive + * surfaces subscription errors today (e.g. a non-existent table). + */ + @Test + void selectLiveWhere_invalidConditionThrowsImmediately() { + try (Surreal surreal = new Surreal()) { + surreal.connect("memory").useNs("test").useDb("test"); + surreal.query("DEFINE TABLE person SCHEMALESS"); + assertThrows(SurrealException.class, () -> surreal.selectLive("person", "][ not a condition")); + } + } + + /** + * A null or blank WHERE clause is rejected eagerly with a clear message; use + * {@link Surreal#selectLive(String)} for an unfiltered live query. + */ + @Test + void selectLiveWhere_nullOrEmptyConditionThrows() { + try (Surreal surreal = new Surreal()) { + surreal.connect("memory").useNs("test").useDb("test"); + surreal.query("DEFINE TABLE person SCHEMALESS"); + assertThrows(SurrealException.class, () -> surreal.selectLive("person", null)); + assertThrows(SurrealException.class, () -> surreal.selectLive("person", " ")); + } + } + + /** + * The binding name {@code tb} is reserved for the table parameter and must be + * rejected rather than silently overridden. + */ + @Test + void selectLiveWhere_reservedParamNameThrows() { + try (Surreal surreal = new Surreal()) { + surreal.connect("memory").useNs("test").useDb("test"); + surreal.query("DEFINE TABLE person SCHEMALESS"); + Map params = new HashMap<>(); + params.put("tb", "other_table"); + assertThrows(SurrealException.class, () -> surreal.selectLive("person", "category > $tb", params)); + } + } + + /** + * A filtered live query on a non-existent table throws immediately, like the + * unfiltered variant. + */ + @Test + void selectLiveWhere_onNonExistentTableThrowsImmediately() { + try (Surreal surreal = new Surreal()) { + surreal.connect("memory").useNs("test").useDb("test"); + assertThrows(SurrealException.class, () -> surreal.selectLive("no_such_table", "category > 1")); + } + } + + /** + * The typed Person mapping still works through a filtered stream. + */ + @Test + void selectLiveWhere_notificationCarriesRecordContent() throws Exception { + AtomicReference received = new AtomicReference<>(); + AtomicReference error = new AtomicReference<>(); + CountDownLatch consuming = new CountDownLatch(1); + + try (Surreal surreal = new Surreal()) { + surreal.connect("memory").useNs("test").useDb("test"); + surreal.query("DEFINE TABLE person SCHEMALESS"); + + Map params = new HashMap<>(); + params.put("name", "Jaime"); + try (LiveStream stream = surreal.selectLive("person", "name = $name", params)) { + Thread consumer = new Thread(() -> { + try { + consuming.countDown(); + stream.next().ifPresent(received::set); + } catch (Throwable t) { + error.set(t); + } + }); + consumer.setDaemon(true); + consumer.start(); + + assertTrue(consuming.await(2, TimeUnit.SECONDS)); + Thread.sleep(500); + + surreal.create(new RecordId("person", 2), Helpers.jaime); + + consumer.join(5000); + assertFalse(consumer.isAlive(), "Consumer thread still blocked — next() never returned"); + } + if (error.get() != null) { + fail("next() threw an exception: " + error.get()); + } + assertNotNull(received.get(), "No notification received after matching CREATE"); + assertEquals("CREATE", received.get().getAction().toUpperCase()); + Person person = received.get().getValue().get(Person.class); + assertEquals(Helpers.jaime.name, person.name); + } + } +}