Encoding as a pure JVM value + native publisher default + pinned per-put override

.github/workflows/ci.yml

@@ -31,14 +31,14 @@ jobs:
         uses: actions/checkout@v4
         with:
           repository: ZettaScaleLabs/zenoh-flat-jni
-          ref: c82656b2f41093d04921ead50516219ae74c7e96
+          ref: 9bc155b3272c8c905bf95614cb7eb7e6d656c72f
           path: zenoh-flat-jni
 
       - name: Check out zenoh-flat
         uses: actions/checkout@v4
         with:
           repository: ZettaScaleLabs/zenoh-flat
-          ref: 7b151d7808583ed15884fada2fd40d5db387c982
+          ref: 6d22091d75912be1c14330b0f78fa1d4bf258ea3
           path: zenoh-flat
 
       - name: Use prebindgen from GitHub main

ZENOH_FLAT_TRANSITION.md

@@ -25,7 +25,8 @@ zenoh (Rust)
 
 | PR | Scope | Status |
 | --- | --- | --- |
-| [#481](https://github.com/eclipse-zenoh/zenoh-java/pull/481) | Use receiver-style zenoh-flat-jni bindings (generated Session/Query/Publisher methods, split key-expr overloads, de-prefixed callback names); +61–83% subscriber throughput | CI green |
+| [#481](https://github.com/eclipse-zenoh/zenoh-java/pull/481) | Use receiver-style zenoh-flat-jni bindings (generated Session/Query/Publisher methods, split key-expr overloads, de-prefixed callback names); +61–83% subscriber throughput | merged |
+| [#484](https://github.com/eclipse-zenoh/zenoh-java/pull/484) | `Encoding`/`ZenohId.toString` become pure JVM values (fixes a per-message native leak); the conversion logic lives in the **shared tier** (`EncodingCodec`/`zidString` in zenoh-flat-jni, correspondence-tested against native, reusable by zenoh-kotlin) with only a thin facade here; publisher **default encoding set natively at declare** (plain puts cross no encoding data). Pairs with [zenoh-flat-jni#4](https://github.com/ZettaScaleLabs/zenoh-flat-jni/pull/4), [zenoh-flat#3](https://github.com/ZettaScaleLabs/zenoh-flat/pull/3) (merged), [prebindgen#80](https://github.com/milyin/prebindgen/pull/80) (merged) | in progress |
 
 Companion PRs in the upstream repos are coordinated per constituent PR (e.g.
 [ZettaScaleLabs/zenoh-flat-jni#3](https://github.com/ZettaScaleLabs/zenoh-flat-jni/pull/3)

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, encH, encId, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn ->
-        f(Sample.fromParts(keH, payloadH, encH, encId, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn))
+    io.zenoh.jni.sample.SampleCallback { keH, payloadH, encId, encSchema, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn ->
+        f(Sample.fromParts(keH, 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, encH, encId, attachH, acceptsReplies, zq ->
+    io.zenoh.jni.query.QueryCallback { keH, parameters, payloadH, encId, encSchema, attachH, acceptsReplies, zq ->
         // The decomposed leaves — including the cloned `keH` key-expr handle 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, encH, encId, attachH, acceptsReplies, zq))
+        f(Query.fromParts(keH, 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, encH, encId, kindInt, ntp64, express, prioInt, ccInt, attachH, reliabilityInt, sourceZid, sourceEid, sourceSn, errPayloadH, errEncH, errEncId ->
+    io.zenoh.jni.query.ReplyCallback { zid, eid, isOk, keH, 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!!, encH!!, encId!!, kindInt!!, ntp64, express!!, prioInt!!, ccInt!!, attachH, reliabilityInt!!, sourceZid, sourceEid!!, sourceSn!!)
+                    Sample.fromParts(keH!!, payloadH!!, encId!!, encSchema, kindInt!!, ntp64, express!!, prioInt!!, ccInt!!, attachH, reliabilityInt!!, sourceZid, sourceEid!!, sourceSn!!)
                 )
             } else {
                 Reply.Error(
                     replierId,
                     ZBytes.fromHandle(errPayloadH!!),
-                    errEncH?.let { Encoding.fromParts(it, errEncId!!) } ?: Encoding.defaultEncoding()
+                    errEncId?.let { Encoding(it, errEncSchema) } ?: Encoding.defaultEncoding()
                 )
             }
         )

zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt

@@ -587,8 +587,14 @@ class Session private constructor(private val config: Config) : AutoCloseable {
     internal fun resolvePublisher(keyExpr: KeyExpr, options: PublisherOptions): Publisher {
         val zSession = zSession ?: throw sessionClosedException
         val publisher = run {
+            // The publisher's default encoding is set NATIVELY here, once —
+            // plain puts then cross no encoding data at all (see Publisher).
+            val enc = options.encoding
             val zPublisher = zSession.declarePublisher(
                 keyExpr.cloneFlat(),
+                true,
+                enc.id,
+                enc.schema,
                 options.congestionControl.jni,
                 options.priority.jni,
                 options.express,
@@ -722,7 +728,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
         val zSession = zSession ?: throw sessionClosedException
         return run {
             val sel = selector.into()
-            val enc = options.encoding ?: Encoding.defaultEncoding()
+            val enc = options.encoding
             zSession.get(
                 sel.keyExpr.flat,
                 sel.parameters?.toString(),
@@ -734,9 +740,9 @@ class Session private constructor(private val config: Config) : AutoCloseable {
                 options.qos.priority.jni,
                 options.qos.express,
                 options.payload?.into()?.bytes,
-                true,
-                enc.idForWire(),
-                enc.schemaForWire(),
+                enc != null,
+                enc?.id ?: 0,
+                enc?.schema,
                 options.attachment?.into()?.bytes,
                 replyCallbackOf { handler.handle(it) },
                 { handler.onClose() },
@@ -755,7 +761,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
         val zSession = zSession ?: throw sessionClosedException
         run {
             val sel = selector.into()
-            val enc = options.encoding ?: Encoding.defaultEncoding()
+            val enc = options.encoding
             zSession.get(
                 sel.keyExpr.flat,
                 sel.parameters?.toString(),
@@ -767,9 +773,9 @@ class Session private constructor(private val config: Config) : AutoCloseable {
                 options.qos.priority.jni,
                 options.qos.express,
                 options.payload?.into()?.bytes,
-                true,
-                enc.idForWire(),
-                enc.schemaForWire(),
+                enc != null,
+                enc?.id ?: 0,
+                enc?.schema,
                 options.attachment?.into()?.bytes,
                 replyCallbackOf { callback.run(it) },
                 { },
@@ -782,13 +788,13 @@ class Session private constructor(private val config: Config) : AutoCloseable {
     internal fun resolvePut(keyExpr: KeyExpr, payload: IntoZBytes, putOptions: PutOptions) {
         val zSession = zSession ?: return
         run {
-            val enc = putOptions.encoding ?: Encoding.defaultEncoding()
+            val enc = putOptions.encoding
             zSession.put(
                 keyExpr.flat,
                 payload.into().bytes,
-                true,
-                enc.idForWire(),
-                enc.schemaForWire(),
+                enc != null,
+                enc?.id ?: 0,
+                enc?.schema,
                 putOptions.congestionControl.jni,
                 putOptions.priority.jni,
                 putOptions.express,