Receive path goes value-only: KeyExpr string-backed except declared, Encoding always (id, schema)

.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

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()
                 )
             }
         )

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,

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,