From cc3675fef9b88eb7c854979b74cbc6b5a5180e76 Mon Sep 17 00:00:00 2001 From: Michael Ilyin Date: Fri, 17 Jul 2026 11:56:01 +0200 Subject: [PATCH 1/8] Received preset encodings are value-only end to end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit zenoh-flat-jni#6 makes the delivered encoding handle CONDITIONAL on schema presence (binding-local encoding_if_schema behind prebindgen#84's field! leaf): a received preset re-sends through the id arm for free, so no per-message native handle — clone + Box + JVM wrapper + gc-Cleaner registration + Cleaner free — is ever materialized for it. Schema- carrying encodings keep the send-ready handle. Adapt the delivery plumbing: Sample.fromParts encH becomes nullable (Query's already was), the reply callback drops its non-null assertion, and EncodingHandleTest's preset round-trip flips to assert the received copy has NO handle — the new contract. CI pin bumped to the flat-jni#6 tip. Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 2 +- .../commonMain/kotlin/io/zenoh/FlatCallbacks.kt | 2 +- .../commonMain/kotlin/io/zenoh/sample/Sample.kt | 6 ++++-- .../kotlin/io/zenoh/EncodingHandleTest.kt | 16 ++++++++++------ 4 files changed, 16 insertions(+), 10 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c2122f30..ab0e3991 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: e8023170c0be3d8e5cd2cee17487870b27af6728 + ref: af690927669551791fda895e5af98ef7bc1220d6 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 94741393..c8581fbe 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt @@ -66,7 +66,7 @@ internal fun replyCallbackOf( if (isOk) { Reply.Success( replierId, - Sample.fromParts(keH!!, payloadH!!, encId!!, encSchema, encH!!, kindInt!!, ntp64, express!!, prioInt!!, ccInt!!, attachH, reliabilityInt!!, sourceZid, sourceEid!!, sourceSn!!) + Sample.fromParts(keH!!, payloadH!!, encId!!, encSchema, encH, kindInt!!, ntp64, express!!, prioInt!!, ccInt!!, attachH, reliabilityInt!!, sourceZid, sourceEid!!, sourceSn!!) ) } else { Reply.Error( 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 78edc600..e3cc70a4 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/sample/Sample.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/sample/Sample.kt @@ -59,7 +59,9 @@ data class Sample( * 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. The + * 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 * trailing `reliability` / `source*` leaves are part of the generated * decomposition but are not surfaced on the public [Sample] type. */ @@ -69,7 +71,7 @@ data class Sample( payloadH: io.zenoh.jni.bytes.ZBytes, encId: Int, encSchema: String?, - encH: io.zenoh.jni.bytes.Encoding, + encH: io.zenoh.jni.bytes.Encoding?, kindInt: Int, ntp64: Long?, express: Boolean, diff --git a/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingHandleTest.kt b/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingHandleTest.kt index d7685186..ac835c0e 100644 --- a/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingHandleTest.kt +++ b/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingHandleTest.kt @@ -33,9 +33,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 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. + * - 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`). */ class EncodingHandleTest { @@ -95,7 +97,7 @@ class EncodingHandleTest { } @Test - fun predefinedRoundTripStaysValueOnSendAndHandleOnReceive() { + fun predefinedRoundTripStaysValueOnlyEndToEnd() { val session = Zenoh.open(Config.loadDefault()) val keyExpr = KeyExpr.tryFrom("example/testing/encoding/preset") val received = mutableListOf() @@ -110,8 +112,10 @@ class EncodingHandleTest { assertNull(Encoding.TEXT_PLAIN.handle) assertEquals(1, received.size) assertEquals(Encoding.TEXT_PLAIN, received[0].encoding) - // …while the received copy arrived send-ready. - assertNotNull(received[0].encoding.handle) + // …and the received copy is value-only too: a schema-less encoding + // re-sends through the id arm for free, so no per-message native + // handle (clone + Box + wrapper + Cleaner) is ever materialized. + assertNull(received[0].encoding.handle) subscriber.close() session.close() From 86d0fe9ebf949661505a6304263ae10b748ab7bc Mon Sep 17 00:00:00 2001 From: Michael Ilyin Date: Fri, 17 Jul 2026 15:40:40 +0200 Subject: [PATCH 2/8] CI: bump zenoh-flat-jni pin to the #6 merge commit The previous pin was a branch commit deleted on merge; 1e040c9 is ZettaScaleLabs/zenoh-flat-jni#6 on main (conditional encoding handle, final fun!+sig! syntax). Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ab0e3991..d28833d5 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: af690927669551791fda895e5af98ef7bc1220d6 + ref: 1e040c96d0729c387d411a2356f361e9ea420877 path: zenoh-flat-jni - name: Check out zenoh-flat From 79b98f1bc84e2554e3c68ab7511e816ebff87a4a Mon Sep 17 00:00:00 2001 From: Michael Ilyin Date: Fri, 17 Jul 2026 17:28:43 +0200 Subject: [PATCH 3/8] KeyExpr: string-backed except declared MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consumes zenoh-flat-jni's string-delivering KeyExpr output. Ownership model (the Encoding model transplanted): a native handle exists ONLY behind Session.declareKeyExpr — the single case zenoh attaches a wire declaration (the only state beyond the string, paying off only on send through the declaring session). Everything else is a plain string value: - tryFrom validates natively once and releases the probe handle; autocanonize reads the canonized string and releases; - RECEIVED keyexprs (sample/query/reply) arrive as ONE eager string — no native allocation, nothing to free; the old per-message handle never carried a declaration (zenoh's RX path builds declaration-less) so it bought nothing on re-send while costing an unfreeable alloc and a second getStr crossing for every string-reading consumer; - sends pick the string or handle selector arm via KeyExpr.jniSel/jniStr/jniHandle (cloneHandle for consuming params); - algebra ops use the declared handle or a transient probe; - undeclare demotes to string-backed; close() is meaningful only for declared instances (doc matches upstream again). 113 jvmTest green incl. the new KeyExprHandleTest (constructed/declared/ received ownership + saved-keyexpr re-publish). CI pin bumped. Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 2 +- .../kotlin/io/zenoh/FlatCallbacks.kt | 14 +- .../src/commonMain/kotlin/io/zenoh/Session.kt | 33 +++-- .../kotlin/io/zenoh/keyexpr/KeyExpr.kt | 132 +++++++++++++----- .../kotlin/io/zenoh/liveliness/Liveliness.kt | 17 ++- .../commonMain/kotlin/io/zenoh/query/Query.kt | 11 +- .../kotlin/io/zenoh/sample/Sample.kt | 7 +- .../kotlin/io/zenoh/KeyExprHandleTest.kt | 92 ++++++++++++ 8 files changed, 236 insertions(+), 72 deletions(-) create mode 100644 zenoh-java/src/jvmTest/kotlin/io/zenoh/KeyExprHandleTest.kt diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d28833d5..b98958fa 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: 552555f462270dff552f72259165c1f62d7bf1fe 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..50509acf 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt @@ -39,34 +39,34 @@ 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, encH, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn -> + f(Sample.fromParts(keStr, payloadH, encId, encSchema, encH, 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, encH, 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, encH, 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, encH, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn, errPayloadH, errEncId, errEncSchema, errEncH -> 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, encH, kindInt!!, ntp64, express!!, prioInt!!, ccInt!!, attachH, reliabilityInt!!, sourceZid, sourceEid!!, sourceSn!!) ) } else { Reply.Error( 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/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..d1ca376c 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,7 +63,7 @@ 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?, @@ -70,7 +73,7 @@ class Query internal constructor( 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( @@ -99,7 +102,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 +141,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..02f99da0 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/sample/Sample.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/sample/Sample.kt @@ -57,7 +57,8 @@ data class Sample( * 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 + * crossing; the [KeyExpr] is string-backed (a received keyexpr never + * carries a wire declaration, so a native handle would buy nothing), 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 @@ -67,7 +68,7 @@ data class Sample( */ @Suppress("UNUSED_PARAMETER") fun fromParts( - keH: io.zenoh.jni.keyexpr.KeyExpr, + keStr: String, payloadH: io.zenoh.jni.bytes.ZBytes, encId: Int, encSchema: String?, @@ -83,7 +84,7 @@ data class Sample( sourceEid: Int, sourceSn: Long, ): Sample = Sample( - KeyExpr(keH), + KeyExpr(keStr), ZBytes.fromHandle(payloadH), Encoding(encId, encSchema, encH), io.zenoh.jni.sample.SampleKind.fromInt(kindInt).toPublic(), 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() + } +} From 4ddcddd1d9d75a311e99c4350a0639cea8e0aa27 Mon Sep 17 00:00:00 2001 From: Michael Ilyin Date: Fri, 17 Jul 2026 17:59:55 +0200 Subject: [PATCH 4/8] CI: bump flat-jni pin to the gc_managed-KeyExpr commit Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b98958fa..fe7201bb 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: 552555f462270dff552f72259165c1f62d7bf1fe + ref: 2d58775cd107b48923f661d470e22e55d7b05684 path: zenoh-flat-jni - name: Check out zenoh-flat From ec7521d3d8082a43ff728f6b9b2114ef87bb87d7 Mon Sep 17 00:00:00 2001 From: Michael Ilyin Date: Fri, 17 Jul 2026 18:25:20 +0200 Subject: [PATCH 5/8] examples: ping-pong exercises the receive-then-resend encoding path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ZPing puts carry a custom schema-carrying encoding and verify (after warmup) that the echoed sample brings it back; ZPong retransmits the RECEIVED payload AND encoding — a pong that drops metadata is not a faithful echo, and the round trip now measures the real receive-then-resend path the send-ready Encoding handle was built for. Co-Authored-By: Claude Fable 5 --- examples/src/main/java/io/zenoh/ZPing.java | 23 +++++++++++++++++++--- examples/src/main/java/io/zenoh/ZPong.java | 9 ++++++++- 2 files changed, 28 insertions(+), 4 deletions(-) diff --git a/examples/src/main/java/io/zenoh/ZPing.java b/examples/src/main/java/io/zenoh/ZPing.java index 34eefdd6..bb932c90 100644 --- a/examples/src/main/java/io/zenoh/ZPing.java +++ b/examples/src/main/java/io/zenoh/ZPing.java @@ -14,11 +14,13 @@ package io.zenoh; +import io.zenoh.bytes.Encoding; import io.zenoh.bytes.ZBytes; import io.zenoh.exceptions.ZError; import io.zenoh.keyexpr.KeyExpr; import io.zenoh.pubsub.Publisher; import io.zenoh.pubsub.PublisherOptions; +import io.zenoh.pubsub.PutOptions; import io.zenoh.qos.CongestionControl; import io.zenoh.sample.Sample; import picocli.CommandLine; @@ -63,18 +65,33 @@ public Integer call() throws Exception { } ZBytes payload = ZBytes.from(data); + // A custom, schema-carrying encoding rides every ping: the pong + // side retransmits the RECEIVED payload and encoding, so the + // round trip exercises the real receive-then-resend metadata path. + Encoding customEncoding = Encoding.from("application/custom;zping-schema"); + PutOptions putOptions = new PutOptions(); + putOptions.setEncoding(customEncoding); + // Warm-up System.out.println("Warming up for " + warmup + " seconds..."); long warmupEnd = System.currentTimeMillis() + (long) (warmup * 1000); + Sample echoed = null; while (System.currentTimeMillis() < warmupEnd) { - publisher.put(payload); - receiverQueue.take(); + publisher.put(payload, putOptions); + echoed = receiverQueue.take().orElse(null); + } + // Guard: the pong must echo the received encoding, not drop it — + // otherwise the round trip isn't measuring what it claims to. + if (echoed != null && !customEncoding.equals(echoed.getEncoding())) { + System.err.println("ZPong did not retransmit the received encoding (got " + + echoed.getEncoding() + ", expected " + customEncoding + ")"); + return 1; } List samples = new ArrayList<>(); for (int i = 0; i < n; i++) { long startTime = System.nanoTime(); - publisher.put(payload); + publisher.put(payload, putOptions); receiverQueue.take(); long elapsedTime = (System.nanoTime() - startTime) / 1000; // Convert to microseconds samples.add(elapsedTime); diff --git a/examples/src/main/java/io/zenoh/ZPong.java b/examples/src/main/java/io/zenoh/ZPong.java index c8119a94..42091c32 100644 --- a/examples/src/main/java/io/zenoh/ZPong.java +++ b/examples/src/main/java/io/zenoh/ZPong.java @@ -18,6 +18,7 @@ import io.zenoh.keyexpr.KeyExpr; import io.zenoh.pubsub.Publisher; import io.zenoh.pubsub.PublisherOptions; +import io.zenoh.pubsub.PutOptions; import io.zenoh.qos.CongestionControl; import picocli.CommandLine; @@ -53,7 +54,13 @@ public Integer call() throws Exception { session.declareSubscriber(keyExprPing, sample -> { try { - publisher.put(sample.getPayload()); + // Echo the RECEIVED data — payload AND encoding: a pong + // that drops metadata isn't a faithful echo, and the + // round trip is meant to exercise the real + // receive-then-resend path. + PutOptions options = new PutOptions(); + options.setEncoding(sample.getEncoding()); + publisher.put(sample.getPayload(), options); } catch (ZError e) { System.err.println("Error responding to ping: " + e.getMessage()); } From ed1381feb8aa7dad4133a9e92da9d4e426d327df Mon Sep 17 00:00:00 2001 From: Michael Ilyin Date: Fri, 17 Jul 2026 19:03:17 +0200 Subject: [PATCH 6/8] Encoding: received encodings are value-only (id, schema) Adapts to zenoh-flat-jni aecffac, which dropped the receive-side conditional handle leaf: the ping-pong A/B (ZPing/ZPong with a custom schema-carrying encoding, real payload+encoding retransmission) showed the per-receive handle lifecycle costs more than the schema re-decode it saves on resend (flat medians 68/68us handle vs 58/67us value-only at 8B/1KB, value-only never slower). Custom encodings still own a construction-born handle and send as a bare jlong; a saved received encoding re-sends through the (id, schema) value arm. Callback lambdas and Sample/Query.fromParts lose the encH leaf; EncodingHandleTest asserts value-only delivery; CI pin bumped. Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 2 +- .../kotlin/io/zenoh/FlatCallbacks.kt | 14 ++++----- .../kotlin/io/zenoh/bytes/Encoding.kt | 16 +++++----- .../commonMain/kotlin/io/zenoh/query/Query.kt | 3 +- .../kotlin/io/zenoh/sample/Sample.kt | 14 ++++----- .../kotlin/io/zenoh/EncodingHandleTest.kt | 31 +++++++++---------- 6 files changed, 38 insertions(+), 42 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fe7201bb..dcdc91c8 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: 2d58775cd107b48923f661d470e22e55d7b05684 + ref: aecffacefcde4f803d74829ad0ae59091a15eca0 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 50509acf..20dbae09 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt @@ -39,14 +39,14 @@ import io.zenoh.scouting.Hello internal fun sampleCallbackOf( f: (Sample) -> Unit ): io.zenoh.jni.sample.SampleCallback = - io.zenoh.jni.sample.SampleCallback { keStr, payloadH, encId, encSchema, encH, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn -> - f(Sample.fromParts(keStr, 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 { keStr, parameters, payloadH, encId, encSchema, encH, attachH, acceptsReplies, zq -> + 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 @@ -54,25 +54,25 @@ internal fun queryCallbackOf( // 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(keStr, 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, keStr, 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(keStr!!, 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/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/query/Query.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt index d1ca376c..283aacb4 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt @@ -68,7 +68,6 @@ class Query internal constructor( 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, @@ -80,7 +79,7 @@ class Query internal constructor( 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 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 02f99da0..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,16 +53,15 @@ 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] is string-backed (a received keyexpr never - * carries a wire declaration, so a native handle would buy nothing), 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 + * 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. */ @@ -72,7 +71,6 @@ data class Sample( payloadH: io.zenoh.jni.bytes.ZBytes, encId: Int, encSchema: String?, - encH: io.zenoh.jni.bytes.Encoding?, kindInt: Int, ntp64: Long?, express: Boolean, @@ -86,7 +84,7 @@ data class Sample( ): Sample = Sample( 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() From 3a08c27b15be3ad9374dacbf9ab70c7b5e20fa9d Mon Sep 17 00:00:00 2001 From: Michael Ilyin Date: Fri, 17 Jul 2026 19:07:45 +0200 Subject: [PATCH 7/8] ci: bump zenoh-flat-jni pin to e4a8ed7 Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dcdc91c8..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: aecffacefcde4f803d74829ad0ae59091a15eca0 + ref: e4a8ed7a354f6f44ab1e44f70771298dc82c9c67 path: zenoh-flat-jni - name: Check out zenoh-flat From 7700e82700b184d0d40a25700de954c29ea4a3b5 Mon Sep 17 00:00:00 2001 From: Michael Ilyin Date: Fri, 17 Jul 2026 19:25:14 +0200 Subject: [PATCH 8/8] examples: restore upstream ZPing/ZPong The custom-encoding echo instrumentation was measurement scaffolding for the Encoding handle A/B (verdict shipped: received encodings are value-only); the examples keep their upstream form. Co-Authored-By: Claude Fable 5 --- examples/src/main/java/io/zenoh/ZPing.java | 23 +++------------------- examples/src/main/java/io/zenoh/ZPong.java | 9 +-------- 2 files changed, 4 insertions(+), 28 deletions(-) diff --git a/examples/src/main/java/io/zenoh/ZPing.java b/examples/src/main/java/io/zenoh/ZPing.java index bb932c90..34eefdd6 100644 --- a/examples/src/main/java/io/zenoh/ZPing.java +++ b/examples/src/main/java/io/zenoh/ZPing.java @@ -14,13 +14,11 @@ package io.zenoh; -import io.zenoh.bytes.Encoding; import io.zenoh.bytes.ZBytes; import io.zenoh.exceptions.ZError; import io.zenoh.keyexpr.KeyExpr; import io.zenoh.pubsub.Publisher; import io.zenoh.pubsub.PublisherOptions; -import io.zenoh.pubsub.PutOptions; import io.zenoh.qos.CongestionControl; import io.zenoh.sample.Sample; import picocli.CommandLine; @@ -65,33 +63,18 @@ public Integer call() throws Exception { } ZBytes payload = ZBytes.from(data); - // A custom, schema-carrying encoding rides every ping: the pong - // side retransmits the RECEIVED payload and encoding, so the - // round trip exercises the real receive-then-resend metadata path. - Encoding customEncoding = Encoding.from("application/custom;zping-schema"); - PutOptions putOptions = new PutOptions(); - putOptions.setEncoding(customEncoding); - // Warm-up System.out.println("Warming up for " + warmup + " seconds..."); long warmupEnd = System.currentTimeMillis() + (long) (warmup * 1000); - Sample echoed = null; while (System.currentTimeMillis() < warmupEnd) { - publisher.put(payload, putOptions); - echoed = receiverQueue.take().orElse(null); - } - // Guard: the pong must echo the received encoding, not drop it — - // otherwise the round trip isn't measuring what it claims to. - if (echoed != null && !customEncoding.equals(echoed.getEncoding())) { - System.err.println("ZPong did not retransmit the received encoding (got " - + echoed.getEncoding() + ", expected " + customEncoding + ")"); - return 1; + publisher.put(payload); + receiverQueue.take(); } List samples = new ArrayList<>(); for (int i = 0; i < n; i++) { long startTime = System.nanoTime(); - publisher.put(payload, putOptions); + publisher.put(payload); receiverQueue.take(); long elapsedTime = (System.nanoTime() - startTime) / 1000; // Convert to microseconds samples.add(elapsedTime); diff --git a/examples/src/main/java/io/zenoh/ZPong.java b/examples/src/main/java/io/zenoh/ZPong.java index 42091c32..c8119a94 100644 --- a/examples/src/main/java/io/zenoh/ZPong.java +++ b/examples/src/main/java/io/zenoh/ZPong.java @@ -18,7 +18,6 @@ import io.zenoh.keyexpr.KeyExpr; import io.zenoh.pubsub.Publisher; import io.zenoh.pubsub.PublisherOptions; -import io.zenoh.pubsub.PutOptions; import io.zenoh.qos.CongestionControl; import picocli.CommandLine; @@ -54,13 +53,7 @@ public Integer call() throws Exception { session.declareSubscriber(keyExprPing, sample -> { try { - // Echo the RECEIVED data — payload AND encoding: a pong - // that drops metadata isn't a faithful echo, and the - // round trip is meant to exercise the real - // receive-then-resend path. - PutOptions options = new PutOptions(); - options.setEncoding(sample.getEncoding()); - publisher.put(sample.getPayload(), options); + publisher.put(sample.getPayload()); } catch (ZError e) { System.err.println("Error responding to ping: " + e.getMessage()); }