diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d28833d5..1524d036 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,7 +31,7 @@ jobs: uses: actions/checkout@v4 with: repository: ZettaScaleLabs/zenoh-flat-jni - ref: 1e040c96d0729c387d411a2356f361e9ea420877 + ref: e4a8ed7a354f6f44ab1e44f70771298dc82c9c67 path: zenoh-flat-jni - name: Check out zenoh-flat diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt index c8581fbe..20dbae09 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt @@ -39,40 +39,40 @@ import io.zenoh.scouting.Hello internal fun sampleCallbackOf( f: (Sample) -> Unit ): io.zenoh.jni.sample.SampleCallback = - io.zenoh.jni.sample.SampleCallback { keH, payloadH, encId, encSchema, encH, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn -> - f(Sample.fromParts(keH, payloadH, encId, encSchema, encH, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn)) + io.zenoh.jni.sample.SampleCallback { keStr, payloadH, encId, encSchema, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn -> + f(Sample.fromParts(keStr, payloadH, encId, encSchema, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn)) } internal fun queryCallbackOf( f: (Query) -> Unit ): io.zenoh.jni.query.QueryCallback = - io.zenoh.jni.query.QueryCallback { keH, parameters, payloadH, encId, encSchema, encH, attachH, acceptsReplies, zq -> - // The decomposed leaves — including the cloned `keH` key-expr handle and + io.zenoh.jni.query.QueryCallback { keStr, parameters, payloadH, encId, encSchema, attachH, acceptsReplies, zq -> + // The decomposed leaves — including the key-expr string and // the owned `zq` query handle — are folded into the SDK [Query]. Unlike // the decomposed read-only types (Sample/Hello), the query OWNS `zq` and // is NOT closed here: it may be retained beyond this callback (e.g. put // on a channel by a queue handler) and replied to later. The native query // is dropped when it is replied to (see [Query.reply]) or when [Query] is // closed — that drop is what finalizes the querier's get. - f(Query.fromParts(keH, parameters, payloadH, encId, encSchema, encH, attachH, acceptsReplies, zq)) + f(Query.fromParts(keStr, parameters, payloadH, encId, encSchema, attachH, acceptsReplies, zq)) } internal fun replyCallbackOf( f: (Reply) -> Unit ): io.zenoh.jni.query.ReplyCallback = - io.zenoh.jni.query.ReplyCallback { zid, eid, isOk, keH, payloadH, encId, encSchema, encH, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn, errPayloadH, errEncId, errEncSchema, errEncH -> + io.zenoh.jni.query.ReplyCallback { zid, eid, isOk, keStr, payloadH, encId, encSchema, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn, errPayloadH, errEncId, errEncSchema -> val replierId = zid?.let { EntityGlobalId(ZenohId(it), eid.toUInt()) } f( if (isOk) { Reply.Success( replierId, - Sample.fromParts(keH!!, payloadH!!, encId!!, encSchema, encH, kindInt!!, ntp64, express!!, prioInt!!, ccInt!!, attachH, reliabilityInt!!, sourceZid, sourceEid!!, sourceSn!!) + Sample.fromParts(keStr!!, payloadH!!, encId!!, encSchema, kindInt!!, ntp64, express!!, prioInt!!, ccInt!!, attachH, reliabilityInt!!, sourceZid, sourceEid!!, sourceSn!!) ) } else { Reply.Error( replierId, ZBytes.fromHandle(errPayloadH!!), - errEncId?.let { Encoding(it, errEncSchema, errEncH) } ?: Encoding.defaultEncoding() + errEncId?.let { Encoding(it, errEncSchema) } ?: Encoding.defaultEncoding() ) } ) diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt index 49b12215..693af037 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt @@ -31,6 +31,9 @@ import io.zenoh.handlers.BlockingQueueHandler import io.zenoh.handlers.Callback import io.zenoh.handlers.Handler import io.zenoh.keyexpr.KeyExpr +import io.zenoh.keyexpr.jniSel +import io.zenoh.keyexpr.jniStr +import io.zenoh.keyexpr.jniHandle import io.zenoh.liveliness.Liveliness import io.zenoh.pubsub.* import io.zenoh.qos.QoS @@ -388,7 +391,10 @@ class Session private constructor(private val config: Config) : AutoCloseable { fun declareKeyExpr(keyExpr: String): KeyExpr { val zSession = zSession ?: throw sessionClosedException val keyexpr = run { - KeyExpr(zSession.declareKeyexpr(keyExpr, throwZError)) + // The one place a KeyExpr carries a native handle: the wire + // declaration attached here makes sends through this session + // compact, so the handle is worth holding. + KeyExpr(keyExpr, zSession.declareKeyexpr(keyExpr, throwZError)) } strongDeclarations.add(keyexpr) return keyexpr @@ -406,13 +412,14 @@ class Session private constructor(private val config: Config) : AutoCloseable { @Throws(ZError::class) fun undeclare(keyExpr: KeyExpr) { val zSession = zSession ?: throw sessionClosedException - val handle = keyExpr.flat - if (handle.isClosed()) { + val handle = keyExpr.handle + if (handle == null || handle.isClosed()) { throw ZError("Attempting to undeclare a non declared key expression.") } run { zSession.undeclareKeyexpr(handle, throwZError) } + keyExpr.handle = null } /** @@ -591,7 +598,7 @@ class Session private constructor(private val config: Config) : AutoCloseable { // plain puts then cross no encoding data at all (see Publisher). val enc = options.encoding val zPublisher = zSession.declarePublisher( - keyExpr.cloneFlat(), + keyExpr.jniSel, keyExpr.jniStr, keyExpr.cloneHandle(), enc.jniSel, enc.jniId, enc.jniSchema, enc.jniHandle, options.congestionControl.jni, options.priority.jni, @@ -618,7 +625,7 @@ class Session private constructor(private val config: Config) : AutoCloseable { val zSession = zSession ?: throw sessionClosedException val subscriber = run { val zSubscriber = zSession.declareSubscriber( - keyExpr.cloneFlat(), + keyExpr.jniSel, keyExpr.jniStr, keyExpr.cloneHandle(), sampleCallbackOf { handler.handle(it) }, { handler.onClose() }, throwZError @@ -636,7 +643,7 @@ class Session private constructor(private val config: Config) : AutoCloseable { val zSession = zSession ?: throw sessionClosedException val subscriber = run { val zSubscriber = zSession.declareSubscriber( - keyExpr.cloneFlat(), + keyExpr.jniSel, keyExpr.jniStr, keyExpr.cloneHandle(), sampleCallbackOf { callback.run(it) }, { }, throwZError @@ -654,7 +661,7 @@ class Session private constructor(private val config: Config) : AutoCloseable { val zSession = zSession ?: throw sessionClosedException val queryable = run { val zQueryable = zSession.declareQueryable( - keyExpr.cloneFlat(), + keyExpr.jniSel, keyExpr.jniStr, keyExpr.cloneHandle(), options.complete, queryCallbackOf { handler.handle(it) }, { handler.onClose() }, @@ -673,7 +680,7 @@ class Session private constructor(private val config: Config) : AutoCloseable { val zSession = zSession ?: throw sessionClosedException val queryable = run { val zQueryable = zSession.declareQueryable( - keyExpr.cloneFlat(), + keyExpr.jniSel, keyExpr.jniStr, keyExpr.cloneHandle(), options.complete, queryCallbackOf { callback.run(it) }, { }, @@ -693,7 +700,7 @@ class Session private constructor(private val config: Config) : AutoCloseable { val zSession = zSession ?: throw sessionClosedException val querier = run { val zQuerier = zSession.declareQuerier( - keyExpr.cloneFlat(), + keyExpr.jniSel, keyExpr.jniStr, keyExpr.cloneHandle(), options.target.toFlat(), options.consolidationMode.toFlat(), options.congestionControl.jni, @@ -727,7 +734,7 @@ class Session private constructor(private val config: Config) : AutoCloseable { return run { val sel = selector.into() zSession.get( - sel.keyExpr.flat, + sel.keyExpr.jniSel, sel.keyExpr.jniStr, sel.keyExpr.jniHandle, sel.parameters?.toString(), options.timeout.toMillis(), options.target.toFlat(), @@ -757,7 +764,7 @@ class Session private constructor(private val config: Config) : AutoCloseable { run { val sel = selector.into() zSession.get( - sel.keyExpr.flat, + sel.keyExpr.jniSel, sel.keyExpr.jniStr, sel.keyExpr.jniHandle, sel.parameters?.toString(), options.timeout.toMillis(), options.target.toFlat(), @@ -782,7 +789,7 @@ class Session private constructor(private val config: Config) : AutoCloseable { run { val enc = putOptions.encoding zSession.put( - keyExpr.flat, + keyExpr.jniSel, keyExpr.jniStr, keyExpr.jniHandle, payload.into().bytes, enc.jniSel, enc.jniId, enc.jniSchema, enc.jniHandle, putOptions.congestionControl.jni, @@ -800,7 +807,7 @@ class Session private constructor(private val config: Config) : AutoCloseable { val zSession = zSession ?: return run { zSession.delete( - keyExpr.flat, + keyExpr.jniSel, keyExpr.jniStr, keyExpr.jniHandle, deleteOptions.congestionControl.jni, deleteOptions.priority.jni, deleteOptions.express, diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/Encoding.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/Encoding.kt index e22b4b1a..27cf33d8 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/Encoding.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/Encoding.kt @@ -38,13 +38,15 @@ import io.zenoh.jni.bytes.Encoding as JniEncoding * value-only: a send carries just their id inside the send call itself — no * native handle ever exists for them. An encoding that already owns a native * [handle] sends it as a bare `jlong` instead (the native side borrows and - * clones it, so the handle is reusable forever). Handles exist only where - * they are born without an extra crossing: custom (schema-carrying) encodings - * create one at construction, and received encodings (sample/query/reply) - * arrive with one in the same delivery crossing. Handle release is GC-managed - * (the JNI `Encoding` class is `gc_managed` — a shared Cleaner frees the - * native box once the handle is unreachable) — the value stays non-closeable - * either way. + * clones it, so the handle is reusable forever). Handles are born ONLY at + * construction of a custom (schema-carrying) encoding — construction is the + * one crossing. Received encodings (sample/query/reply) are always + * value-only `(id, schema)`: delivering a handle with each message + * measurably cost more than the schema re-decode it saved when one was + * re-sent, so a re-sent received encoding crosses through the `(id, schema)` + * value arm instead. Handle release is GC-managed (the JNI `Encoding` class + * is `gc_managed` — a shared Cleaner frees the native box once the handle is + * unreachable) — the value stays non-closeable either way. */ class Encoding internal constructor( internal val id: Int, diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/keyexpr/KeyExpr.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/keyexpr/KeyExpr.kt index cb581d3c..999da47d 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/keyexpr/KeyExpr.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/keyexpr/KeyExpr.kt @@ -51,36 +51,41 @@ import io.zenoh.query.Selector * * # Declaring a key expression from a session. * - * A [KeyExpr] acts as a container for the string representation of a key expression. Operations like `intersects`, - * `includes`, and `equals` are processed at the native layer using this string representation. For improved performance, - * consider initializing a [KeyExpr] through [Session.declareKeyExpr]. This method associates the [KeyExpr] with a native - * instance, thereby optimizing operation execution. However, it is crucial to manually invoke [close] on each [KeyExpr] - * instance before it is garbage collected to prevent memory leaks. - * - * As an alternative, employing a try-with-resources pattern using Kotlin's `use` block is recommended. This approach - * ensures that [close] is automatically called, safely managing the lifecycle of the [KeyExpr] instance. - * + * A [KeyExpr] is its validated string. A native instance exists ONLY behind a + * [Session.declareKeyExpr] result — the single case where zenoh attaches a wire + * declaration (a compact id replacing the string on the wire) worth carrying; + * such an instance should be [close]d (or `use`d) when no longer needed. + * Every other [KeyExpr] — constructed via [tryFrom]/[autocanonize] or received + * with a sample — is a plain value: nothing to close, no native resource. */ class KeyExpr internal constructor( - /** The owned native handle, passed (by reference) to raw `keyexpr_*` ops. */ - internal val flat: JniKeyExpr, - keyExprString: String? = null, + private val keyExprString: String, + /** + * The owned native handle — non-null ONLY for [Session.declareKeyExpr] + * results, whose wire declaration makes sends through the declaring + * session compact. Everything else is string-backed. + */ + internal var handle: JniKeyExpr? = null, ) : AutoCloseable, IntoSelector, SessionDeclaration { - private var keyExprStringLazy: String? = keyExprString + /** Clone the native handle before passing it to a consuming Rust API. */ + internal fun cloneHandle(): JniKeyExpr? = handle?.newClone(throwZError0) /** - * The string form, read LAZILY from the native handle on first use - * (toString/equals) — a received key expression is usually only matched - * or echoed by handle, so the common path never crosses for the string - * (forward-extraction rule: handle eager, string on demand). + * Run [body] with a native handle: the declared handle when present, + * else a transient one validated from the string (closed afterwards). + * Used by the native keyexpr algebra ops. */ - private val keyExprString: String - get() = keyExprStringLazy - ?: flat.getStr(throwZError0).also { keyExprStringLazy = it } - - /** Clone the native handle before passing it to a consuming Rust API. */ - internal fun cloneFlat(): JniKeyExpr = flat.newClone(throwZError0) + private inline fun withHandle(body: (JniKeyExpr) -> R): R { + val h = handle + if (h != null) return body(h) + val tmp = JniKeyExpr.newTryFrom(keyExprString, throwZError) + try { + return body(tmp) + } finally { + tmp.close() + } + } companion object { @@ -92,13 +97,20 @@ class KeyExpr internal constructor( * * You may use [autocanonize] instead if you are unsure if the expression you will use for construction will be canon. * + * The result is string-backed: the expression is validated natively + * once and the probe handle is released — an undeclared native + * keyexpr carries no state beyond its string. + * * @param keyExpr The intended key expression as a string. * @return The [KeyExpr] in case of success. * @throws ZError in the case of failure. */ @JvmStatic @Throws(ZError::class) - fun tryFrom(keyExpr: String): KeyExpr = KeyExpr(JniKeyExpr.newTryFrom(keyExpr, throwZError)) + fun tryFrom(keyExpr: String): KeyExpr { + JniKeyExpr.newTryFrom(keyExpr, throwZError).close() + return KeyExpr(keyExpr) + } /** * Autocanonize. @@ -112,7 +124,14 @@ class KeyExpr internal constructor( */ @JvmStatic @Throws(ZError::class) - fun autocanonize(keyExpr: String): KeyExpr = KeyExpr(JniKeyExpr.newAutocanonize(keyExpr, throwZError)) + fun autocanonize(keyExpr: String): KeyExpr { + val probe = JniKeyExpr.newAutocanonize(keyExpr, throwZError) + try { + return KeyExpr(probe.getStr(throwZError0)) + } finally { + probe.close() + } + } } /** @@ -121,8 +140,10 @@ class KeyExpr internal constructor( * Will return false as well if the key expression is not valid anymore. */ @Throws(ZError::class) - fun intersects(other: KeyExpr): Boolean = - this.flat.intersects(other.flat, throwZError0) + fun intersects(other: KeyExpr): Boolean = withHandle { h -> + other.handle?.let { h.intersects(it, throwZError0) } + ?: h.intersects(other.keyExprString, throwZError0) + } /** * Includes operation. This method returns `true` when all the keys defined by `other` also belong to the set @@ -130,8 +151,10 @@ class KeyExpr internal constructor( * Will return false as well if the key expression is not valid anymore. */ @Throws(ZError::class) - fun includes(other: KeyExpr): Boolean = - this.flat.includes(other.flat, throwZError0) + fun includes(other: KeyExpr): Boolean = withHandle { h -> + other.handle?.let { h.includes(it, throwZError0) } + ?: h.includes(other.keyExprString, throwZError0) + } /** * Returns the relation between 'this' and other from 'this''s point of view ([SetIntersectionLevel.INCLUDES] @@ -139,24 +162,43 @@ class KeyExpr internal constructor( * so you should favor these methods for most applications. */ @Throws(ZError::class) - fun relationTo(other: KeyExpr): SetIntersectionLevel = - SetIntersectionLevel.fromJni(this.flat.relationTo(other.flat, throwZError0)) + fun relationTo(other: KeyExpr): SetIntersectionLevel = withHandle { h -> + val raw = other.handle?.let { h.relationTo(it, throwZError0) } + ?: h.relationTo(other.keyExprString, throwZError0) + SetIntersectionLevel.fromJni(raw) + } /** * Joins both sides, inserting a / in between them. * This should be your preferred method when concatenating path segments. */ @Throws(ZError::class) - fun join(other: String): KeyExpr = - KeyExpr(JniKeyExpr.newJoin(this.flat, other, throwZError)) + fun join(other: String): KeyExpr { + // The companion factory validates and joins natively; the result is a + // fresh (declaration-less) handle — keep its canonical string only. + val probe = handle?.let { JniKeyExpr.newJoin(it, other, throwZError) } + ?: JniKeyExpr.newJoin(keyExprString, other, throwZError) + try { + return KeyExpr(probe.getStr(throwZError0)) + } finally { + probe.close() + } + } /** * Performs string concatenation and returns the result as a KeyExpr if possible. * You should probably prefer [join] as Zenoh may then take advantage of the hierarchical separation it inserts. */ @Throws(ZError::class) - fun concat(other: String): KeyExpr = - KeyExpr(JniKeyExpr.newConcat(this.flat, other, throwZError)) + fun concat(other: String): KeyExpr { + val probe = handle?.let { JniKeyExpr.newConcat(it, other, throwZError) } + ?: JniKeyExpr.newConcat(keyExprString, other, throwZError) + try { + return KeyExpr(probe.getStr(throwZError0)) + } finally { + probe.close() + } + } override fun toString(): String = keyExprString @@ -175,8 +217,8 @@ class KeyExpr internal constructor( * operations on it, but without the inner optimizations. */ override fun undeclare() { - // Frees the owned native key-expression handle. - flat.close() + handle?.close() + handle = null } override fun into(): Selector = Selector(this) @@ -194,3 +236,19 @@ class KeyExpr internal constructor( return keyExprString.hashCode() } } + +// The three slots of the generated key-expr selector block (`keyExprSel, +// keyExpr0, keyExpr1`), computed from one [KeyExpr] in a single expression per +// slot so every send call site stays one flat call: declared -> the bare +// handle (arm 1), string-backed -> the validated string rebuilt Rust-side +// inside the same call (arm 0). For CONSUMING params use [KeyExpr.cloneHandle] +// in the third slot instead — the handle arm takes the value by move. + +internal val KeyExpr.jniSel: Int + get() = if (handle != null) 1 else 0 + +internal val KeyExpr.jniStr: String? + get() = if (handle == null) toString() else null + +internal val KeyExpr.jniHandle: JniKeyExpr? + get() = handle diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/liveliness/Liveliness.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/liveliness/Liveliness.kt index 5bd332df..15205f20 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/liveliness/Liveliness.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/liveliness/Liveliness.kt @@ -23,6 +23,9 @@ import io.zenoh.handlers.BlockingQueueHandler import io.zenoh.handlers.Callback import io.zenoh.handlers.Handler import io.zenoh.keyexpr.KeyExpr +import io.zenoh.keyexpr.jniSel +import io.zenoh.keyexpr.jniStr +import io.zenoh.keyexpr.jniHandle import io.zenoh.pubsub.CallbackSubscriber import io.zenoh.pubsub.HandlerSubscriber import io.zenoh.pubsub.Subscriber @@ -51,7 +54,7 @@ class Liveliness internal constructor(private val session: Session) { @Throws(ZError::class) fun declareToken(keyExpr: KeyExpr): LivelinessToken { val zSession = session.zSession ?: throw Session.sessionClosedException - return LivelinessToken(zSession.livelinessDeclareToken(keyExpr.cloneFlat(), throwZError)) + return LivelinessToken(zSession.livelinessDeclareToken(keyExpr.jniSel, keyExpr.jniStr, keyExpr.cloneHandle(), throwZError)) } /** @@ -69,7 +72,7 @@ class Liveliness internal constructor(private val session: Session) { val zSession = session.zSession ?: throw Session.sessionClosedException val handler = BlockingQueueHandler(LinkedBlockingDeque()) zSession.livelinessGet( - keyExpr.flat, + keyExpr.jniSel, keyExpr.jniStr, keyExpr.jniHandle, timeout.toMillis(), replyCallbackOf { handler.handle(it) }, { handler.onClose() }, @@ -92,7 +95,7 @@ class Liveliness internal constructor(private val session: Session) { ) { val zSession = session.zSession ?: throw Session.sessionClosedException zSession.livelinessGet( - keyExpr.flat, + keyExpr.jniSel, keyExpr.jniStr, keyExpr.jniHandle, timeout.toMillis(), replyCallbackOf { callback.run(it) }, { }, @@ -115,7 +118,7 @@ class Liveliness internal constructor(private val session: Session) { ): R { val zSession = session.zSession ?: throw Session.sessionClosedException zSession.livelinessGet( - keyExpr.flat, + keyExpr.jniSel, keyExpr.jniStr, keyExpr.jniHandle, timeout.toMillis(), replyCallbackOf { handler.handle(it) }, { handler.onClose() }, @@ -139,7 +142,7 @@ class Liveliness internal constructor(private val session: Session) { val handler = BlockingQueueHandler(LinkedBlockingDeque()) val zSession = session.zSession ?: throw Session.sessionClosedException val zSubscriber = zSession.livelinessDeclareSubscriber( - keyExpr.cloneFlat(), + keyExpr.jniSel, keyExpr.jniStr, keyExpr.cloneHandle(), options.history, sampleCallbackOf { handler.handle(it) }, { handler.onClose() }, @@ -164,7 +167,7 @@ class Liveliness internal constructor(private val session: Session) { ): CallbackSubscriber { val zSession = session.zSession ?: throw Session.sessionClosedException val zSubscriber = zSession.livelinessDeclareSubscriber( - keyExpr.cloneFlat(), + keyExpr.jniSel, keyExpr.jniStr, keyExpr.cloneHandle(), options.history, sampleCallbackOf { callback.run(it) }, { }, @@ -190,7 +193,7 @@ class Liveliness internal constructor(private val session: Session) { ): HandlerSubscriber { val zSession = session.zSession ?: throw Session.sessionClosedException val zSubscriber = zSession.livelinessDeclareSubscriber( - keyExpr.cloneFlat(), + keyExpr.jniSel, keyExpr.jniStr, keyExpr.cloneHandle(), options.history, sampleCallbackOf { handler.handle(it) }, { handler.onClose() }, diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt index 8ddebb98..283aacb4 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt @@ -25,6 +25,9 @@ import io.zenoh.bytes.ZBytes import io.zenoh.exceptions.ZError import io.zenoh.exceptions.throwZError import io.zenoh.keyexpr.KeyExpr +import io.zenoh.keyexpr.jniSel +import io.zenoh.keyexpr.jniStr +import io.zenoh.keyexpr.jniHandle /** * Represents a Zenoh Query in Kotlin. @@ -60,24 +63,23 @@ class Query internal constructor( * keeps working after the callback returns). */ fun fromParts( - keH: io.zenoh.jni.keyexpr.KeyExpr, + keStr: String, parameters: String, payloadH: io.zenoh.jni.bytes.ZBytes?, encId: Int?, encSchema: String?, - encH: io.zenoh.jni.bytes.Encoding?, attachH: io.zenoh.jni.bytes.ZBytes?, acceptsRepliesInt: Int, zq: io.zenoh.jni.query.Query, ): Query { - val ke = KeyExpr(keH) + val ke = KeyExpr(keStr) val selector = if (parameters.isEmpty()) Selector(ke) else Selector(ke, Parameters.from(parameters)) return Query( ke, selector, payloadH?.let { ZBytes.fromHandle(it) }, - encId?.let { Encoding(it, encSchema, encH) }, + encId?.let { Encoding(it, encSchema) }, attachH?.let { ZBytes.fromHandle(it) }, io.zenoh.jni.query.ReplyKeyExpr.fromInt(acceptsRepliesInt).toPublic(), zq @@ -99,7 +101,7 @@ class Query internal constructor( val q = zQuery ?: throw ZError("Query is invalid") val enc = options.encoding q.replySuccess( - keyExpr.flat, + keyExpr.jniSel, keyExpr.jniStr, keyExpr.jniHandle, payload.into().bytes, enc.jniSel, enc.jniId, enc.jniSchema, enc.jniHandle, options.timeStamp?.ntpValue(), @@ -138,7 +140,7 @@ class Query internal constructor( fun replyDel(keyExpr: KeyExpr, options: ReplyDelOptions = ReplyDelOptions()) { val q = zQuery ?: throw ZError("Query is invalid") q.replyDelete( - keyExpr.flat, + keyExpr.jniSel, keyExpr.jniStr, keyExpr.jniHandle, options.timeStamp?.ntpValue(), options.attachment?.into()?.bytes, options.express, diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/sample/Sample.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/sample/Sample.kt index e3cc70a4..4438c3ec 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/sample/Sample.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/sample/Sample.kt @@ -53,25 +53,24 @@ data class Sample( internal companion object { /** - * Builds an SDK [Sample] from the 15 leaves of the ZSample canonical + * Builds an SDK [Sample] from the 14 leaves of the ZSample canonical * output — the lambda parameter list every decomposed sample delivery * uses (the `zReplySample` builder and the subscriber/liveliness * callbacks), in record order. The whole graph arrives in ONE JNI - * crossing; the [KeyExpr] retains the delivered handle leaf, and the - * [Encoding] retains its delivered handle (`encH`) so re-sending it - * crosses a bare `jlong` instead of rebuilding the native value — - * delivered ONLY for schema-carrying encodings (a preset re-sends - * through the id arm for free, so it arrives value-only). The + * crossing; the [KeyExpr] is string-backed (a received keyexpr never + * carries a wire declaration, so a native handle would buy nothing), + * and the [Encoding] is value-only `(id, schema?)` — a delivered + * native handle measurably cost more per receive than the + * schema-string re-decode it saved on the resent fraction. The * trailing `reliability` / `source*` leaves are part of the generated * decomposition but are not surfaced on the public [Sample] type. */ @Suppress("UNUSED_PARAMETER") fun fromParts( - keH: io.zenoh.jni.keyexpr.KeyExpr, + keStr: String, payloadH: io.zenoh.jni.bytes.ZBytes, encId: Int, encSchema: String?, - encH: io.zenoh.jni.bytes.Encoding?, kindInt: Int, ntp64: Long?, express: Boolean, @@ -83,9 +82,9 @@ data class Sample( sourceEid: Int, sourceSn: Long, ): Sample = Sample( - KeyExpr(keH), + KeyExpr(keStr), ZBytes.fromHandle(payloadH), - Encoding(encId, encSchema, encH), + Encoding(encId, encSchema), io.zenoh.jni.sample.SampleKind.fromInt(kindInt).toPublic(), ntp64?.let { TimeStamp(it) }, QoS( diff --git a/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingHandleTest.kt b/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingHandleTest.kt index ac835c0e..132b38a7 100644 --- a/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingHandleTest.kt +++ b/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingHandleTest.kt @@ -22,7 +22,6 @@ import io.zenoh.sample.Sample import org.junit.Assert.assertEquals import org.junit.Assert.assertNotNull import org.junit.Assert.assertNull -import org.junit.Assert.assertSame import org.junit.Test /** @@ -33,11 +32,11 @@ import org.junit.Test * the send call itself, no handle ever exists; * - custom (schema-carrying) encodings own a handle from construction — * construction is the one crossing, every send after is a bare jlong; - * - received SCHEMA-CARRYING encodings arrive WITH their handle in the same - * delivery crossing, so re-sending one (the save-and-republish scenario) - * never rebuilds the native value from its schema string; a received - * PRESET arrives value-only — its handle would buy nothing (the id arm is - * free), so the binding never materializes one (`encoding_if_schema`). + * - received encodings are ALWAYS value-only `(id, schema)`: a ping-pong A/B + * on the save-and-republish scenario showed the per-receive handle + * lifecycle (clone + Box + wrapper + Cleaner) costs more than the schema + * re-decode it saves on the resent fraction, so no handle is ever + * delivered — a re-sent received encoding crosses through the value arm. */ class EncodingHandleTest { @@ -58,14 +57,15 @@ class EncodingHandleTest { } @Test - fun receivedEncodingIsSendReadyAndResendsByHandle() { + fun receivedEncodingIsValueOnlyAndResendsThroughValueArm() { val custom = Encoding.from("application/custom;my-schema") val session = Zenoh.open(Config.loadDefault()) val keyExpr = KeyExpr.tryFrom("example/testing/encoding/handle") val received = mutableListOf() val subscriber = session.declareSubscriber(keyExpr) { received.add(it) } - // Send 1: user-created custom encoding — crosses by handle. + // Send 1: user-created custom encoding — crosses by its + // construction-born handle. val putOptions = PutOptions() putOptions.encoding = custom session.put(keyExpr, ZBytes.from("one"), putOptions) @@ -74,12 +74,12 @@ class EncodingHandleTest { assertEquals(1, received.size) val saved = received[0].encoding assertEquals(custom, saved) - // Delivered send-ready: the handle arrived with the sample. - assertNotNull(saved.handle) + // Value-only delivery: no per-message native handle materialized. + assertNull(saved.handle) - // Send 2: the user's scenario — re-publish with the SAVED encoding. - // The retained handle is what crosses; nothing is rebuilt. - val handleBefore = saved.handle + // Send 2: the save-and-republish scenario — the SAVED encoding + // re-sends through the (id, schema) value arm, rebuilt natively + // inside the same send crossing. val resendOptions = PutOptions() resendOptions.encoding = saved session.put(keyExpr, ZBytes.from("two"), resendOptions) @@ -87,10 +87,7 @@ class EncodingHandleTest { assertEquals(2, received.size) assertEquals(custom, received[1].encoding) - // The saved encoding still owns the same reusable handle (borrowed, - // never consumed, by the send). - assertSame(handleBefore, saved.handle) - assertNotNull(received[1].encoding.handle) + assertNull(received[1].encoding.handle) subscriber.close() session.close() diff --git a/zenoh-java/src/jvmTest/kotlin/io/zenoh/KeyExprHandleTest.kt b/zenoh-java/src/jvmTest/kotlin/io/zenoh/KeyExprHandleTest.kt new file mode 100644 index 00000000..afae4ab3 --- /dev/null +++ b/zenoh-java/src/jvmTest/kotlin/io/zenoh/KeyExprHandleTest.kt @@ -0,0 +1,92 @@ +// +// Copyright (c) 2026 ZettaScale Technology +// +// This program and the accompanying materials are made available under the +// terms of the Eclipse Public License 2.0 which is available at +// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0 +// which is available at https://www.apache.org/licenses/LICENSE-2.0. +// +// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0 +// +// Contributors: +// ZettaScale Zenoh Team, +// + +package io.zenoh + +import io.zenoh.bytes.ZBytes +import io.zenoh.keyexpr.KeyExpr +import io.zenoh.sample.Sample +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test + +/** + * The key-expression ownership model: a native handle exists ONLY behind + * `Session.declareKeyExpr` — the single case zenoh attaches a wire + * declaration worth carrying. Everything else — `tryFrom`, `autocanonize`, + * and every RECEIVED keyexpr (zenoh's RX path never attaches a declaration) + * — is a plain string-backed value: nothing to close, no native resource, + * no per-message allocation or leak. + */ +class KeyExprHandleTest { + + @Test + fun constructedKeyExprsAreStringBacked() { + val ke = KeyExpr.tryFrom("example/testing/keyexpr") + assertNull(ke.handle) + assertEquals("example/testing/keyexpr", ke.toString()) + // Canonization still happens natively; the result keeps the string. + val canon = KeyExpr.autocanonize("example/**/**") + assertNull(canon.handle) + assertEquals("example/**", canon.toString()) + // Algebra ops work on string-backed instances (transient native probe). + assertTrue(ke.intersects(KeyExpr.tryFrom("example/testing/*"))) + assertTrue(KeyExpr.tryFrom("example/**").includes(ke)) + assertEquals("example/testing/keyexpr/sub", ke.join("sub").toString()) + } + + @Test + fun declaredKeyExprOwnsAHandleUntilUndeclared() { + val session = Zenoh.open(Config.loadDefault()) + val declared = session.declareKeyExpr("example/testing/keyexpr/declared") + assertNotNull(declared.handle) + // Ops route through the declared handle. + assertTrue(declared.intersects(KeyExpr.tryFrom("example/testing/**"))) + // Undeclare demotes to string-backed; the string survives. + session.undeclare(declared) + assertNull(declared.handle) + assertEquals("example/testing/keyexpr/declared", declared.toString()) + session.close() + } + + @Test + fun receivedKeyExprIsStringBackedAndReusable() { + val session = Zenoh.open(Config.loadDefault()) + val keyExpr = KeyExpr.tryFrom("example/testing/keyexpr/received") + val received = mutableListOf() + val subscriber = session.declareSubscriber(keyExpr) { received.add(it) } + + session.put(keyExpr, ZBytes.from("one")) + Thread.sleep(500) + + assertEquals(1, received.size) + val saved = received[0].keyExpr + // Delivered as ONE eager string: no native allocation, nothing to + // free — a received keyexpr never carries a wire declaration, so a + // handle would buy nothing on re-send. + assertNull(saved.handle) + assertEquals(keyExpr, saved) + + // The user scenario: re-publish with the SAVED keyexpr. + session.put(saved, ZBytes.from("two")) + Thread.sleep(500) + assertEquals(2, received.size) + assertEquals(saved, received[1].keyExpr) + + subscriber.close() + session.close() + } +}