From b73579e0547993acdae7acf5b49678be89af0574 Mon Sep 17 00:00:00 2001 From: Michael Ilyin Date: Thu, 16 Jul 2026 14:33:38 +0200 Subject: [PATCH 1/3] Make Encoding and ZenohId.toString pure JVM values MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An encoding IS its (id, schema) pair — Zenoh's own representation; the textual form is derived from a fixed table. With the id<->name table already JVM-side (the generated ENCODING_* constants) and the companion zenoh-flat-jni change crossing Encoding by value in both directions, the SDK Encoding becomes a plain immutable value: no native handle, no native calls, no lazy caches, no locking, nothing to close. This also removes a native memory leak: received Sample/Query/ReplyError previously retained a cloned Encoding handle with no close path. - Encoding.kt: immutable (id, schema); from()/toString()/withSchema() implement Zenoh's exact conversion rules (incl. the custom-encoding name-preserving withSchema and the render-only CUSTOM<->"" table row); equality on (id, schema), matching Zenoh core. - ZenohId.toString: the little-endian lowercase-hex rule in pure Kotlin. - Callback/fromParts leaves follow the new (encId, encSchema) decomposition; outbound sites pass .id/.schema directly. - Correspondence tests (EncodingCorrespondenceTest, ZenohIdCorrespondenceTest) verify the pure implementations against the native ones across the whole predefined id range, parse/render edge shapes, and random ids — the contract for any JVM-side reimplementation of zenoh-flat API. One test caught and fixed a real divergence (custom-encoding withSchema). Pairs with ZettaScaleLabs/zenoh-flat-jni#4 (CI pin bumped). jvmTest: 105 tests, 0 failures. --- .github/workflows/ci.yml | 2 +- ZENOH_FLAT_TRANSITION.md | 3 +- .../kotlin/io/zenoh/FlatCallbacks.kt | 14 +- .../src/commonMain/kotlin/io/zenoh/Session.kt | 12 +- .../kotlin/io/zenoh/bytes/Encoding.kt | 472 +++++++----------- .../kotlin/io/zenoh/config/ZenohId.kt | 17 +- .../kotlin/io/zenoh/pubsub/Publisher.kt | 4 +- .../kotlin/io/zenoh/query/Querier.kt | 8 +- .../commonMain/kotlin/io/zenoh/query/Query.kt | 10 +- .../kotlin/io/zenoh/sample/Sample.kt | 4 +- .../io/zenoh/EncodingCorrespondenceTest.kt | 115 +++++ .../io/zenoh/ZenohIdCorrespondenceTest.kt | 58 +++ 12 files changed, 398 insertions(+), 321 deletions(-) create mode 100644 zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingCorrespondenceTest.kt create mode 100644 zenoh-java/src/jvmTest/kotlin/io/zenoh/ZenohIdCorrespondenceTest.kt diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 28a2fab3..98fc0a1b 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: c82656b2f41093d04921ead50516219ae74c7e96 + ref: abbc4832e8658109fdf8f9b1face95a9f21a0b15 path: zenoh-flat-jni - name: Check out zenoh-flat diff --git a/ZENOH_FLAT_TRANSITION.md b/ZENOH_FLAT_TRANSITION.md index e03f18d5..af9d7639 100644 --- a/ZENOH_FLAT_TRANSITION.md +++ b/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 | +| `encoding-pure-value` | `Encoding` and `ZenohId.toString` become pure JVM values (no JNI crossings, no retained handles — fixes a per-message native leak), verified by correspondence tests against the native implementation; pairs with [zenoh-flat-jni#4](https://github.com/ZettaScaleLabs/zenoh-flat-jni/pull/4) | 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) diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/FlatCallbacks.kt index 63d401e3..4e35a7cf 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 { 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 @@ -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(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() ) } ) diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt index 10473468..8f91a17f 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt @@ -735,8 +735,8 @@ class Session private constructor(private val config: Config) : AutoCloseable { options.qos.express, options.payload?.into()?.bytes, true, - enc.idForWire(), - enc.schemaForWire(), + enc.id, + enc.schema, options.attachment?.into()?.bytes, replyCallbackOf { handler.handle(it) }, { handler.onClose() }, @@ -768,8 +768,8 @@ class Session private constructor(private val config: Config) : AutoCloseable { options.qos.express, options.payload?.into()?.bytes, true, - enc.idForWire(), - enc.schemaForWire(), + enc.id, + enc.schema, options.attachment?.into()?.bytes, replyCallbackOf { callback.run(it) }, { }, @@ -787,8 +787,8 @@ class Session private constructor(private val config: Config) : AutoCloseable { keyExpr.flat, payload.into().bytes, true, - enc.idForWire(), - enc.schemaForWire(), + enc.id, + enc.schema, putOptions.congestionControl.jni, putOptions.priority.jni, putOptions.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 affd91ed..841f10a0 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/Encoding.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/Encoding.kt @@ -14,114 +14,7 @@ package io.zenoh.bytes -import io.zenoh.exceptions.throwZError0 -import io.zenoh.jni.bytes.ENCODING_APPLICATION_CBOR -import io.zenoh.jni.bytes.ENCODING_APPLICATION_CBOR_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_CDR -import io.zenoh.jni.bytes.ENCODING_APPLICATION_CDR_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_COAP_PAYLOAD -import io.zenoh.jni.bytes.ENCODING_APPLICATION_COAP_PAYLOAD_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JAVA_SERIALIZED_OBJECT -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JAVA_SERIALIZED_OBJECT_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JSON -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JSONPATH -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JSONPATH_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JSON_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JSON_PATCH_JSON -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JSON_PATCH_JSON_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JSON_SEQ -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JSON_SEQ_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JWT -import io.zenoh.jni.bytes.ENCODING_APPLICATION_JWT_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_MP4 -import io.zenoh.jni.bytes.ENCODING_APPLICATION_MP4_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_OCTET_STREAM -import io.zenoh.jni.bytes.ENCODING_APPLICATION_OCTET_STREAM_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_OPENMETRICS_TEXT -import io.zenoh.jni.bytes.ENCODING_APPLICATION_OPENMETRICS_TEXT_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_PROTOBUF -import io.zenoh.jni.bytes.ENCODING_APPLICATION_PROTOBUF_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_PYTHON_SERIALIZED_OBJECT -import io.zenoh.jni.bytes.ENCODING_APPLICATION_PYTHON_SERIALIZED_OBJECT_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_SOAP_XML -import io.zenoh.jni.bytes.ENCODING_APPLICATION_SOAP_XML_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_SQL -import io.zenoh.jni.bytes.ENCODING_APPLICATION_SQL_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_XML -import io.zenoh.jni.bytes.ENCODING_APPLICATION_XML_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_X_WWW_FORM_URLENCODED -import io.zenoh.jni.bytes.ENCODING_APPLICATION_X_WWW_FORM_URLENCODED_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_YAML -import io.zenoh.jni.bytes.ENCODING_APPLICATION_YAML_ID -import io.zenoh.jni.bytes.ENCODING_APPLICATION_YANG -import io.zenoh.jni.bytes.ENCODING_APPLICATION_YANG_ID -import io.zenoh.jni.bytes.ENCODING_AUDIO_AAC -import io.zenoh.jni.bytes.ENCODING_AUDIO_AAC_ID -import io.zenoh.jni.bytes.ENCODING_AUDIO_FLAC -import io.zenoh.jni.bytes.ENCODING_AUDIO_FLAC_ID -import io.zenoh.jni.bytes.ENCODING_AUDIO_MP4 -import io.zenoh.jni.bytes.ENCODING_AUDIO_MP4_ID -import io.zenoh.jni.bytes.ENCODING_AUDIO_OGG -import io.zenoh.jni.bytes.ENCODING_AUDIO_OGG_ID -import io.zenoh.jni.bytes.ENCODING_AUDIO_VORBIS -import io.zenoh.jni.bytes.ENCODING_AUDIO_VORBIS_ID -import io.zenoh.jni.bytes.ENCODING_IMAGE_BMP -import io.zenoh.jni.bytes.ENCODING_IMAGE_BMP_ID -import io.zenoh.jni.bytes.ENCODING_IMAGE_GIF -import io.zenoh.jni.bytes.ENCODING_IMAGE_GIF_ID -import io.zenoh.jni.bytes.ENCODING_IMAGE_JPEG -import io.zenoh.jni.bytes.ENCODING_IMAGE_JPEG_ID -import io.zenoh.jni.bytes.ENCODING_IMAGE_PNG -import io.zenoh.jni.bytes.ENCODING_IMAGE_PNG_ID -import io.zenoh.jni.bytes.ENCODING_IMAGE_WEBP -import io.zenoh.jni.bytes.ENCODING_IMAGE_WEBP_ID -import io.zenoh.jni.bytes.ENCODING_TEXT_CSS -import io.zenoh.jni.bytes.ENCODING_TEXT_CSS_ID -import io.zenoh.jni.bytes.ENCODING_TEXT_CSV -import io.zenoh.jni.bytes.ENCODING_TEXT_CSV_ID -import io.zenoh.jni.bytes.ENCODING_TEXT_HTML -import io.zenoh.jni.bytes.ENCODING_TEXT_HTML_ID -import io.zenoh.jni.bytes.ENCODING_TEXT_JAVASCRIPT -import io.zenoh.jni.bytes.ENCODING_TEXT_JAVASCRIPT_ID -import io.zenoh.jni.bytes.ENCODING_TEXT_JSON -import io.zenoh.jni.bytes.ENCODING_TEXT_JSON5 -import io.zenoh.jni.bytes.ENCODING_TEXT_JSON5_ID -import io.zenoh.jni.bytes.ENCODING_TEXT_JSON_ID -import io.zenoh.jni.bytes.ENCODING_TEXT_MARKDOWN -import io.zenoh.jni.bytes.ENCODING_TEXT_MARKDOWN_ID -import io.zenoh.jni.bytes.ENCODING_TEXT_PLAIN -import io.zenoh.jni.bytes.ENCODING_TEXT_PLAIN_ID -import io.zenoh.jni.bytes.ENCODING_TEXT_XML -import io.zenoh.jni.bytes.ENCODING_TEXT_XML_ID -import io.zenoh.jni.bytes.ENCODING_TEXT_YAML -import io.zenoh.jni.bytes.ENCODING_TEXT_YAML_ID -import io.zenoh.jni.bytes.ENCODING_VIDEO_H261 -import io.zenoh.jni.bytes.ENCODING_VIDEO_H261_ID -import io.zenoh.jni.bytes.ENCODING_VIDEO_H263 -import io.zenoh.jni.bytes.ENCODING_VIDEO_H263_ID -import io.zenoh.jni.bytes.ENCODING_VIDEO_H264 -import io.zenoh.jni.bytes.ENCODING_VIDEO_H264_ID -import io.zenoh.jni.bytes.ENCODING_VIDEO_H265 -import io.zenoh.jni.bytes.ENCODING_VIDEO_H265_ID -import io.zenoh.jni.bytes.ENCODING_VIDEO_H266 -import io.zenoh.jni.bytes.ENCODING_VIDEO_H266_ID -import io.zenoh.jni.bytes.ENCODING_VIDEO_MP4 -import io.zenoh.jni.bytes.ENCODING_VIDEO_MP4_ID -import io.zenoh.jni.bytes.ENCODING_VIDEO_OGG -import io.zenoh.jni.bytes.ENCODING_VIDEO_OGG_ID -import io.zenoh.jni.bytes.ENCODING_VIDEO_RAW -import io.zenoh.jni.bytes.ENCODING_VIDEO_RAW_ID -import io.zenoh.jni.bytes.ENCODING_VIDEO_VP8 -import io.zenoh.jni.bytes.ENCODING_VIDEO_VP8_ID -import io.zenoh.jni.bytes.ENCODING_VIDEO_VP9 -import io.zenoh.jni.bytes.ENCODING_VIDEO_VP9_ID -import io.zenoh.jni.bytes.ENCODING_ZENOH_BYTES -import io.zenoh.jni.bytes.ENCODING_ZENOH_BYTES_ID -import io.zenoh.jni.bytes.ENCODING_ZENOH_SERIALIZED -import io.zenoh.jni.bytes.ENCODING_ZENOH_SERIALIZED_ID -import io.zenoh.jni.bytes.ENCODING_ZENOH_STRING -import io.zenoh.jni.bytes.ENCODING_ZENOH_STRING_ID -import io.zenoh.jni.bytes.Encoding as JniEncoding +import io.zenoh.jni.bytes.* /** * Default encoding values used by Zenoh. @@ -133,157 +26,149 @@ import io.zenoh.jni.bytes.Encoding as JniEncoding * * A set of associated constants are provided to cover the most common encodings for user convenience. * - * Internally the encoding is kept as its canonical textual form ([repr], e.g. - * `"text/plain"` or `"text/plain;utf-8"`); a native `ZEncoding` handle is built - * on demand for each native crossing (the raw `z_*` API takes the encoding by - * reference, so the transient handle is closed by the caller after the call). + * An encoding **is** its decomposed pair `(id, schema)` — exactly Zenoh's own + * representation; the textual form (e.g. `"text/plain;utf-8"`) is derived from + * a fixed id↔name table. Both the table (the generated `ENCODING_*` constants, + * regenerated from Zenoh) and the conversion rules live JVM-side, so this class + * is a plain immutable value: no native handle, no native calls, nothing to + * close. Its correspondence to the native implementation is verified by + * `EncodingCorrespondenceTest`. */ -class Encoding private constructor( - private var reprLazy: String?, - internal val id: Int?, - private val handle: JniEncoding?, +class Encoding internal constructor( + internal val id: Int, + internal val schema: String? = null, ) { - internal constructor(repr: String) : this(repr, null, null) - - private var schemaLazy: String? = null - private var schemaKnown: Boolean = false - private var idCached: Int = id ?: 0 - private var idKnown: Boolean = id != null - - /** - * Ensure the lossless decomposed form `(id, schema)` is cached. Zenoh's - * encoding IS `(id, schema)`, so this loses nothing. A handle-backed - * (received) Encoding already knows its [id]; its schema is read LAZILY - * through the handle on first need. A predefined constant is born fully - * decomposed (repr + id from the generated consts, schema known-absent) and - * never enters this path. A repr-primary Encoding built by [from] derives - * BOTH once from a transient handle built off [repr], then frees it — - * caching pure JVM values and retaining no native handle. The cache is - * reused across every native crossing, so a reused - * encoding (the normal case — a publisher publishes one data type) never - * re-parses its string per call. - */ - private fun ensureDecomposed() { - if (idKnown && schemaKnown) return - synchronized(this) { - if (idKnown && schemaKnown) return - if (handle != null) { - if (!schemaKnown) { - schemaLazy = handle.getSchema(throwZError0) - schemaKnown = true - } - } else { - val h = JniEncoding.newFromString(repr, throwZError0) - try { - if (!idKnown) { - idCached = h.getId(throwZError0) - idKnown = true - } - if (!schemaKnown) { - schemaLazy = h.getSchema(throwZError0) - schemaKnown = true - } - } finally { - h.close() - } - } - } - } - - /** - * Encoding id for the OUTBOUND native crossing. Cached once (see - * [ensureDecomposed]) and reused across every put/get/reply — the wire - * carries this cheap primitive instead of a freshly parsed string. - */ - internal fun idForWire(): Int { - ensureDecomposed() - return idCached - } - - /** Optional schema for the OUTBOUND native crossing. Cached once. */ - internal fun schemaForWire(): String? { - ensureDecomposed() - return schemaLazy - } - - /** - * Canonical display string. A handle-backed (received) Encoding - * materializes it LAZILY on first use (toString/equals) via the native - * accessor — received encodings are usually only forwarded or compared by - * id, so the common path never builds the string (forward-extraction rule). - */ - internal val repr: String - get() = reprLazy ?: synchronized(this) { - reprLazy - ?: handle!!.toStr(throwZError0) - .also { reprLazy = it } - } - companion object { /** - * A predefined constant: its canonical string and wire id come from - * the generated `ENCODING_*` consts (single source of truth in Rust), - * and a predefined encoding never carries a schema — the decomposed - * `(id, schema)` form is fully known up front, so these constants - * never touch native for decomposition (see [ensureDecomposed]). + * Mirror of Zenoh's `CUSTOM_ENCODING_ID`: a custom (non-predefined) + * encoding carries its whole textual form in the schema slot. */ - private fun predefined(repr: String, id: Int): Encoding = - Encoding(repr, id, null).apply { schemaKnown = true } + internal const val CUSTOM_ENCODING_ID: Int = 0xFFFF - @JvmField val ZENOH_BYTES = predefined(ENCODING_ZENOH_BYTES, ENCODING_ZENOH_BYTES_ID) - @JvmField val ZENOH_STRING = predefined(ENCODING_ZENOH_STRING, ENCODING_ZENOH_STRING_ID) - @JvmField val ZENOH_SERIALIZED = predefined(ENCODING_ZENOH_SERIALIZED, ENCODING_ZENOH_SERIALIZED_ID) - @JvmField val APPLICATION_OCTET_STREAM = predefined(ENCODING_APPLICATION_OCTET_STREAM, ENCODING_APPLICATION_OCTET_STREAM_ID) - @JvmField val TEXT_PLAIN = predefined(ENCODING_TEXT_PLAIN, ENCODING_TEXT_PLAIN_ID) - @JvmField val APPLICATION_JSON = predefined(ENCODING_APPLICATION_JSON, ENCODING_APPLICATION_JSON_ID) - @JvmField val TEXT_JSON = predefined(ENCODING_TEXT_JSON, ENCODING_TEXT_JSON_ID) - @JvmField val APPLICATION_CDR = predefined(ENCODING_APPLICATION_CDR, ENCODING_APPLICATION_CDR_ID) - @JvmField val APPLICATION_CBOR = predefined(ENCODING_APPLICATION_CBOR, ENCODING_APPLICATION_CBOR_ID) - @JvmField val APPLICATION_YAML = predefined(ENCODING_APPLICATION_YAML, ENCODING_APPLICATION_YAML_ID) - @JvmField val TEXT_YAML = predefined(ENCODING_TEXT_YAML, ENCODING_TEXT_YAML_ID) - @JvmField val TEXT_JSON5 = predefined(ENCODING_TEXT_JSON5, ENCODING_TEXT_JSON5_ID) - @JvmField val APPLICATION_PYTHON_SERIALIZED_OBJECT = predefined(ENCODING_APPLICATION_PYTHON_SERIALIZED_OBJECT, ENCODING_APPLICATION_PYTHON_SERIALIZED_OBJECT_ID) - @JvmField val APPLICATION_PROTOBUF = predefined(ENCODING_APPLICATION_PROTOBUF, ENCODING_APPLICATION_PROTOBUF_ID) - @JvmField val APPLICATION_JAVA_SERIALIZED_OBJECT = predefined(ENCODING_APPLICATION_JAVA_SERIALIZED_OBJECT, ENCODING_APPLICATION_JAVA_SERIALIZED_OBJECT_ID) - @JvmField val APPLICATION_OPENMETRICS_TEXT = predefined(ENCODING_APPLICATION_OPENMETRICS_TEXT, ENCODING_APPLICATION_OPENMETRICS_TEXT_ID) - @JvmField val IMAGE_PNG = predefined(ENCODING_IMAGE_PNG, ENCODING_IMAGE_PNG_ID) - @JvmField val IMAGE_JPEG = predefined(ENCODING_IMAGE_JPEG, ENCODING_IMAGE_JPEG_ID) - @JvmField val IMAGE_GIF = predefined(ENCODING_IMAGE_GIF, ENCODING_IMAGE_GIF_ID) - @JvmField val IMAGE_BMP = predefined(ENCODING_IMAGE_BMP, ENCODING_IMAGE_BMP_ID) - @JvmField val IMAGE_WEBP = predefined(ENCODING_IMAGE_WEBP, ENCODING_IMAGE_WEBP_ID) - @JvmField val APPLICATION_XML = predefined(ENCODING_APPLICATION_XML, ENCODING_APPLICATION_XML_ID) - @JvmField val APPLICATION_X_WWW_FORM_URLENCODED = predefined(ENCODING_APPLICATION_X_WWW_FORM_URLENCODED, ENCODING_APPLICATION_X_WWW_FORM_URLENCODED_ID) - @JvmField val TEXT_HTML = predefined(ENCODING_TEXT_HTML, ENCODING_TEXT_HTML_ID) - @JvmField val TEXT_XML = predefined(ENCODING_TEXT_XML, ENCODING_TEXT_XML_ID) - @JvmField val TEXT_CSS = predefined(ENCODING_TEXT_CSS, ENCODING_TEXT_CSS_ID) - @JvmField val TEXT_JAVASCRIPT = predefined(ENCODING_TEXT_JAVASCRIPT, ENCODING_TEXT_JAVASCRIPT_ID) - @JvmField val TEXT_MARKDOWN = predefined(ENCODING_TEXT_MARKDOWN, ENCODING_TEXT_MARKDOWN_ID) - @JvmField val TEXT_CSV = predefined(ENCODING_TEXT_CSV, ENCODING_TEXT_CSV_ID) - @JvmField val APPLICATION_SQL = predefined(ENCODING_APPLICATION_SQL, ENCODING_APPLICATION_SQL_ID) - @JvmField val APPLICATION_COAP_PAYLOAD = predefined(ENCODING_APPLICATION_COAP_PAYLOAD, ENCODING_APPLICATION_COAP_PAYLOAD_ID) - @JvmField val APPLICATION_JSON_PATCH_JSON = predefined(ENCODING_APPLICATION_JSON_PATCH_JSON, ENCODING_APPLICATION_JSON_PATCH_JSON_ID) - @JvmField val APPLICATION_JSON_SEQ = predefined(ENCODING_APPLICATION_JSON_SEQ, ENCODING_APPLICATION_JSON_SEQ_ID) - @JvmField val APPLICATION_JSONPATH = predefined(ENCODING_APPLICATION_JSONPATH, ENCODING_APPLICATION_JSONPATH_ID) - @JvmField val APPLICATION_JWT = predefined(ENCODING_APPLICATION_JWT, ENCODING_APPLICATION_JWT_ID) - @JvmField val APPLICATION_MP4 = predefined(ENCODING_APPLICATION_MP4, ENCODING_APPLICATION_MP4_ID) - @JvmField val APPLICATION_SOAP_XML = predefined(ENCODING_APPLICATION_SOAP_XML, ENCODING_APPLICATION_SOAP_XML_ID) - @JvmField val APPLICATION_YANG = predefined(ENCODING_APPLICATION_YANG, ENCODING_APPLICATION_YANG_ID) - @JvmField val AUDIO_AAC = predefined(ENCODING_AUDIO_AAC, ENCODING_AUDIO_AAC_ID) - @JvmField val AUDIO_FLAC = predefined(ENCODING_AUDIO_FLAC, ENCODING_AUDIO_FLAC_ID) - @JvmField val AUDIO_MP4 = predefined(ENCODING_AUDIO_MP4, ENCODING_AUDIO_MP4_ID) - @JvmField val AUDIO_OGG = predefined(ENCODING_AUDIO_OGG, ENCODING_AUDIO_OGG_ID) - @JvmField val AUDIO_VORBIS = predefined(ENCODING_AUDIO_VORBIS, ENCODING_AUDIO_VORBIS_ID) - @JvmField val VIDEO_H261 = predefined(ENCODING_VIDEO_H261, ENCODING_VIDEO_H261_ID) - @JvmField val VIDEO_H263 = predefined(ENCODING_VIDEO_H263, ENCODING_VIDEO_H263_ID) - @JvmField val VIDEO_H264 = predefined(ENCODING_VIDEO_H264, ENCODING_VIDEO_H264_ID) - @JvmField val VIDEO_H265 = predefined(ENCODING_VIDEO_H265, ENCODING_VIDEO_H265_ID) - @JvmField val VIDEO_H266 = predefined(ENCODING_VIDEO_H266, ENCODING_VIDEO_H266_ID) - @JvmField val VIDEO_MP4 = predefined(ENCODING_VIDEO_MP4, ENCODING_VIDEO_MP4_ID) - @JvmField val VIDEO_OGG = predefined(ENCODING_VIDEO_OGG, ENCODING_VIDEO_OGG_ID) - @JvmField val VIDEO_RAW = predefined(ENCODING_VIDEO_RAW, ENCODING_VIDEO_RAW_ID) - @JvmField val VIDEO_VP8 = predefined(ENCODING_VIDEO_VP8, ENCODING_VIDEO_VP8_ID) - @JvmField val VIDEO_VP9 = predefined(ENCODING_VIDEO_VP9, ENCODING_VIDEO_VP9_ID) + /** + * The id↔name table, built from the generated constants (single source + * of truth in Rust) plus the custom-id row Zenoh keys with the empty + * name. + */ + private val idToName: Map = mapOf( + ENCODING_ZENOH_BYTES_ID to ENCODING_ZENOH_BYTES, + ENCODING_ZENOH_STRING_ID to ENCODING_ZENOH_STRING, + ENCODING_ZENOH_SERIALIZED_ID to ENCODING_ZENOH_SERIALIZED, + ENCODING_APPLICATION_OCTET_STREAM_ID to ENCODING_APPLICATION_OCTET_STREAM, + ENCODING_TEXT_PLAIN_ID to ENCODING_TEXT_PLAIN, + ENCODING_APPLICATION_JSON_ID to ENCODING_APPLICATION_JSON, + ENCODING_TEXT_JSON_ID to ENCODING_TEXT_JSON, + ENCODING_APPLICATION_CDR_ID to ENCODING_APPLICATION_CDR, + ENCODING_APPLICATION_CBOR_ID to ENCODING_APPLICATION_CBOR, + ENCODING_APPLICATION_YAML_ID to ENCODING_APPLICATION_YAML, + ENCODING_TEXT_YAML_ID to ENCODING_TEXT_YAML, + ENCODING_TEXT_JSON5_ID to ENCODING_TEXT_JSON5, + ENCODING_APPLICATION_PYTHON_SERIALIZED_OBJECT_ID to ENCODING_APPLICATION_PYTHON_SERIALIZED_OBJECT, + ENCODING_APPLICATION_PROTOBUF_ID to ENCODING_APPLICATION_PROTOBUF, + ENCODING_APPLICATION_JAVA_SERIALIZED_OBJECT_ID to ENCODING_APPLICATION_JAVA_SERIALIZED_OBJECT, + ENCODING_APPLICATION_OPENMETRICS_TEXT_ID to ENCODING_APPLICATION_OPENMETRICS_TEXT, + ENCODING_IMAGE_PNG_ID to ENCODING_IMAGE_PNG, + ENCODING_IMAGE_JPEG_ID to ENCODING_IMAGE_JPEG, + ENCODING_IMAGE_GIF_ID to ENCODING_IMAGE_GIF, + ENCODING_IMAGE_BMP_ID to ENCODING_IMAGE_BMP, + ENCODING_IMAGE_WEBP_ID to ENCODING_IMAGE_WEBP, + ENCODING_APPLICATION_XML_ID to ENCODING_APPLICATION_XML, + ENCODING_APPLICATION_X_WWW_FORM_URLENCODED_ID to ENCODING_APPLICATION_X_WWW_FORM_URLENCODED, + ENCODING_TEXT_HTML_ID to ENCODING_TEXT_HTML, + ENCODING_TEXT_XML_ID to ENCODING_TEXT_XML, + ENCODING_TEXT_CSS_ID to ENCODING_TEXT_CSS, + ENCODING_TEXT_JAVASCRIPT_ID to ENCODING_TEXT_JAVASCRIPT, + ENCODING_TEXT_MARKDOWN_ID to ENCODING_TEXT_MARKDOWN, + ENCODING_TEXT_CSV_ID to ENCODING_TEXT_CSV, + ENCODING_APPLICATION_SQL_ID to ENCODING_APPLICATION_SQL, + ENCODING_APPLICATION_COAP_PAYLOAD_ID to ENCODING_APPLICATION_COAP_PAYLOAD, + ENCODING_APPLICATION_JSON_PATCH_JSON_ID to ENCODING_APPLICATION_JSON_PATCH_JSON, + ENCODING_APPLICATION_JSON_SEQ_ID to ENCODING_APPLICATION_JSON_SEQ, + ENCODING_APPLICATION_JSONPATH_ID to ENCODING_APPLICATION_JSONPATH, + ENCODING_APPLICATION_JWT_ID to ENCODING_APPLICATION_JWT, + ENCODING_APPLICATION_MP4_ID to ENCODING_APPLICATION_MP4, + ENCODING_APPLICATION_SOAP_XML_ID to ENCODING_APPLICATION_SOAP_XML, + ENCODING_APPLICATION_YANG_ID to ENCODING_APPLICATION_YANG, + ENCODING_AUDIO_AAC_ID to ENCODING_AUDIO_AAC, + ENCODING_AUDIO_FLAC_ID to ENCODING_AUDIO_FLAC, + ENCODING_AUDIO_MP4_ID to ENCODING_AUDIO_MP4, + ENCODING_AUDIO_OGG_ID to ENCODING_AUDIO_OGG, + ENCODING_AUDIO_VORBIS_ID to ENCODING_AUDIO_VORBIS, + ENCODING_VIDEO_H261_ID to ENCODING_VIDEO_H261, + ENCODING_VIDEO_H263_ID to ENCODING_VIDEO_H263, + ENCODING_VIDEO_H264_ID to ENCODING_VIDEO_H264, + ENCODING_VIDEO_H265_ID to ENCODING_VIDEO_H265, + ENCODING_VIDEO_H266_ID to ENCODING_VIDEO_H266, + ENCODING_VIDEO_MP4_ID to ENCODING_VIDEO_MP4, + ENCODING_VIDEO_OGG_ID to ENCODING_VIDEO_OGG, + ENCODING_VIDEO_RAW_ID to ENCODING_VIDEO_RAW, + ENCODING_VIDEO_VP8_ID to ENCODING_VIDEO_VP8, + ENCODING_VIDEO_VP9_ID to ENCODING_VIDEO_VP9, + CUSTOM_ENCODING_ID to "", + ) + + // Parse-side reverse table. Zenoh's `STR_TO_ID` has NO empty-string + // row (the `CUSTOM ↔ ""` mapping is render-side only), so a leading + // `;` input falls through to the custom arm with the whole string as + // schema — exclude the custom row here to match. + private val nameToId: Map = idToName.entries + .filter { (_, name) -> name.isNotEmpty() } + .associate { (id, name) -> name to id } + + @JvmField val ZENOH_BYTES = Encoding(ENCODING_ZENOH_BYTES_ID) + @JvmField val ZENOH_STRING = Encoding(ENCODING_ZENOH_STRING_ID) + @JvmField val ZENOH_SERIALIZED = Encoding(ENCODING_ZENOH_SERIALIZED_ID) + @JvmField val APPLICATION_OCTET_STREAM = Encoding(ENCODING_APPLICATION_OCTET_STREAM_ID) + @JvmField val TEXT_PLAIN = Encoding(ENCODING_TEXT_PLAIN_ID) + @JvmField val APPLICATION_JSON = Encoding(ENCODING_APPLICATION_JSON_ID) + @JvmField val TEXT_JSON = Encoding(ENCODING_TEXT_JSON_ID) + @JvmField val APPLICATION_CDR = Encoding(ENCODING_APPLICATION_CDR_ID) + @JvmField val APPLICATION_CBOR = Encoding(ENCODING_APPLICATION_CBOR_ID) + @JvmField val APPLICATION_YAML = Encoding(ENCODING_APPLICATION_YAML_ID) + @JvmField val TEXT_YAML = Encoding(ENCODING_TEXT_YAML_ID) + @JvmField val TEXT_JSON5 = Encoding(ENCODING_TEXT_JSON5_ID) + @JvmField val APPLICATION_PYTHON_SERIALIZED_OBJECT = Encoding(ENCODING_APPLICATION_PYTHON_SERIALIZED_OBJECT_ID) + @JvmField val APPLICATION_PROTOBUF = Encoding(ENCODING_APPLICATION_PROTOBUF_ID) + @JvmField val APPLICATION_JAVA_SERIALIZED_OBJECT = Encoding(ENCODING_APPLICATION_JAVA_SERIALIZED_OBJECT_ID) + @JvmField val APPLICATION_OPENMETRICS_TEXT = Encoding(ENCODING_APPLICATION_OPENMETRICS_TEXT_ID) + @JvmField val IMAGE_PNG = Encoding(ENCODING_IMAGE_PNG_ID) + @JvmField val IMAGE_JPEG = Encoding(ENCODING_IMAGE_JPEG_ID) + @JvmField val IMAGE_GIF = Encoding(ENCODING_IMAGE_GIF_ID) + @JvmField val IMAGE_BMP = Encoding(ENCODING_IMAGE_BMP_ID) + @JvmField val IMAGE_WEBP = Encoding(ENCODING_IMAGE_WEBP_ID) + @JvmField val APPLICATION_XML = Encoding(ENCODING_APPLICATION_XML_ID) + @JvmField val APPLICATION_X_WWW_FORM_URLENCODED = Encoding(ENCODING_APPLICATION_X_WWW_FORM_URLENCODED_ID) + @JvmField val TEXT_HTML = Encoding(ENCODING_TEXT_HTML_ID) + @JvmField val TEXT_XML = Encoding(ENCODING_TEXT_XML_ID) + @JvmField val TEXT_CSS = Encoding(ENCODING_TEXT_CSS_ID) + @JvmField val TEXT_JAVASCRIPT = Encoding(ENCODING_TEXT_JAVASCRIPT_ID) + @JvmField val TEXT_MARKDOWN = Encoding(ENCODING_TEXT_MARKDOWN_ID) + @JvmField val TEXT_CSV = Encoding(ENCODING_TEXT_CSV_ID) + @JvmField val APPLICATION_SQL = Encoding(ENCODING_APPLICATION_SQL_ID) + @JvmField val APPLICATION_COAP_PAYLOAD = Encoding(ENCODING_APPLICATION_COAP_PAYLOAD_ID) + @JvmField val APPLICATION_JSON_PATCH_JSON = Encoding(ENCODING_APPLICATION_JSON_PATCH_JSON_ID) + @JvmField val APPLICATION_JSON_SEQ = Encoding(ENCODING_APPLICATION_JSON_SEQ_ID) + @JvmField val APPLICATION_JSONPATH = Encoding(ENCODING_APPLICATION_JSONPATH_ID) + @JvmField val APPLICATION_JWT = Encoding(ENCODING_APPLICATION_JWT_ID) + @JvmField val APPLICATION_MP4 = Encoding(ENCODING_APPLICATION_MP4_ID) + @JvmField val APPLICATION_SOAP_XML = Encoding(ENCODING_APPLICATION_SOAP_XML_ID) + @JvmField val APPLICATION_YANG = Encoding(ENCODING_APPLICATION_YANG_ID) + @JvmField val AUDIO_AAC = Encoding(ENCODING_AUDIO_AAC_ID) + @JvmField val AUDIO_FLAC = Encoding(ENCODING_AUDIO_FLAC_ID) + @JvmField val AUDIO_MP4 = Encoding(ENCODING_AUDIO_MP4_ID) + @JvmField val AUDIO_OGG = Encoding(ENCODING_AUDIO_OGG_ID) + @JvmField val AUDIO_VORBIS = Encoding(ENCODING_AUDIO_VORBIS_ID) + @JvmField val VIDEO_H261 = Encoding(ENCODING_VIDEO_H261_ID) + @JvmField val VIDEO_H263 = Encoding(ENCODING_VIDEO_H263_ID) + @JvmField val VIDEO_H264 = Encoding(ENCODING_VIDEO_H264_ID) + @JvmField val VIDEO_H265 = Encoding(ENCODING_VIDEO_H265_ID) + @JvmField val VIDEO_H266 = Encoding(ENCODING_VIDEO_H266_ID) + @JvmField val VIDEO_MP4 = Encoding(ENCODING_VIDEO_MP4_ID) + @JvmField val VIDEO_OGG = Encoding(ENCODING_VIDEO_OGG_ID) + @JvmField val VIDEO_RAW = Encoding(ENCODING_VIDEO_RAW_ID) + @JvmField val VIDEO_VP8 = Encoding(ENCODING_VIDEO_VP8_ID) + @JvmField val VIDEO_VP9 = Encoding(ENCODING_VIDEO_VP9_ID) /** The default [Encoding] is [ZENOH_BYTES]. */ @JvmStatic fun defaultEncoding() = ZENOH_BYTES @@ -292,62 +177,67 @@ class Encoding private constructor( * Parse a textual encoding (e.g. `"text/plain"`, `"text/plain;utf-8"`, * `"my_encoding"`). Well-known names resolve to their canonical id; * everything else is preserved as a custom encoding. + * + * Implements Zenoh's parse rule: everything before the first `;` is + * looked up as a known name — a match yields `(id, rest-if-nonempty)`, + * a miss yields the custom id with the whole string as schema. */ - @JvmStatic fun from(s: String): Encoding = Encoding(s) - - /** - * Decodes a native `ZEncoding` handle into a value [Encoding] and frees - * the handle. Used when an accessor / callback hands back an encoding. - */ - internal fun fromHandle(handle: JniEncoding): Encoding { - try { - return Encoding(handle.toStr(throwZError0)) - } finally { - handle.close() + @JvmStatic fun from(s: String): Encoding { + if (s.isEmpty()) return ZENOH_BYTES + val sep = s.indexOf(';') + val name = if (sep >= 0) s.substring(0, sep) else s + val rest = if (sep >= 0) s.substring(sep + 1) else "" + val id = nameToId[name] + return if (id != null) { + Encoding(id, rest.takeIf { it.isNotEmpty() }) + } else { + Encoding(CUSTOM_ENCODING_ID, s) } } - - /** Wrap the decomposed `(handle, id)` leaves. Schema and the - * canonical string stay lazy through the handle. */ - internal fun fromParts(encH: JniEncoding, id: Int): Encoding = - Encoding(null, id, encH) } - /** - * Builds a fresh native `ZEncoding` handle from [repr]. The raw `z_*` - * encoding parameters take it **by reference** (not consumed), so the - * caller MUST close the returned handle after the native call. - */ - internal fun toZEncoding(): JniEncoding = - if (handle != null) { - handle.newClone(throwZError0) - } else { - JniEncoding.newFromString(repr, throwZError0) - } - /** * Set a schema to this encoding. Zenoh does not define what a schema is and its semantics is left to the implementer. * E.g. a common schema for `text/plain` encoding is `utf-8`. + * + * For a custom encoding the schema slot holds `"name[;schema]"`; per + * Zenoh's rule the name part is preserved and only the schema part is + * replaced. */ fun withSchema(schema: String): Encoding { - // `withSchema` takes the base encoding flattened to `(id, schema)`; this - // Encoding already exposes that decomposition lazily. - val withSchema = JniEncoding.newWithSchema(idForWire(), schemaForWire(), schema, throwZError0) - try { - return Encoding(withSchema.toStr(throwZError0)) - } finally { - withSchema.close() - } + if (id != CUSTOM_ENCODING_ID) return Encoding(id, schema) + val name = this.schema?.substringBefore(';') ?: "" + return Encoding( + id, + when { + name.isEmpty() -> schema + schema.isEmpty() -> name + else -> "$name;$schema" + }, + ) } - override fun toString(): String = repr + /** + * Canonical textual form, per Zenoh's rendering rule: the known name, + * `"name;schema"`, the bare schema for a custom encoding, or + * `"unknown(id)"` for an unrecognized id. + */ + override fun toString(): String { + val name = idToName[id] + return when { + name == null -> if (schema == null) "unknown($id)" else "unknown($id);$schema" + schema == null -> name + name.isEmpty() -> schema + else -> "$name;$schema" + } + } override fun equals(other: Any?): Boolean { if (this === other) return true if (javaClass != other?.javaClass) return false other as Encoding - return repr == other.repr + return id == other.id && schema == other.schema } - override fun hashCode(): Int = repr.hashCode() + override fun hashCode(): Int = 31 * id + (schema?.hashCode() ?: 0) } diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/config/ZenohId.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/config/ZenohId.kt index ff2b59f9..2bd980a7 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/config/ZenohId.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/config/ZenohId.kt @@ -14,7 +14,6 @@ package io.zenoh.config -import io.zenoh.exceptions.throwZError0 import io.zenoh.jni.config.ZenohId as JniZenohId import kotlin.math.absoluteValue @@ -23,8 +22,22 @@ import kotlin.math.absoluteValue */ data class ZenohId internal constructor(internal val inner: JniZenohId) { + /** + * The standard string form: the id bytes read as a little-endian integer, + * rendered as lowercase hex without leading zeros (`"0"` for the zero id) — + * Zenoh's own rule (uhlc `ID`'s `Debug`), implemented in pure Kotlin over + * the value-class bytes; correspondence with the native formatter is + * verified by `ZenohIdCorrespondenceTest`. + */ override fun toString(): String { - return inner.toStr(throwZError0) + val hex = "0123456789abcdef" + val sb = StringBuilder(inner.bytes.size * 2) + for (i in inner.bytes.indices.reversed()) { + val b = inner.bytes[i].toInt() and 0xFF + sb.append(hex[b ushr 4]).append(hex[b and 0x0F]) + } + val s = sb.trimStart('0').toString() + return s.ifEmpty { "0" } } override fun equals(other: Any?): Boolean { diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt index 523b1cae..fd410eb8 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt @@ -134,8 +134,8 @@ class Publisher internal constructor( p.put( payload.into().bytes, true, - encoding.idForWire(), - encoding.schemaForWire(), + encoding.id, + encoding.schema, attachment?.into()?.bytes, throwZError, ) diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt index 37945776..3acf207d 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt @@ -150,8 +150,8 @@ class Querier internal constructor(val keyExpr: KeyExpr, val qos: QoS, private v options.parameters?.toString(), options.payload?.into()?.bytes, true, - enc.idForWire(), - enc.schemaForWire(), + enc.id, + enc.schema, options.attachment?.into()?.bytes, replyCallbackOf { callback.run(it) }, { }, @@ -166,8 +166,8 @@ class Querier internal constructor(val keyExpr: KeyExpr, val qos: QoS, private v options.parameters?.toString(), options.payload?.into()?.bytes, true, - enc.idForWire(), - enc.schemaForWire(), + enc.id, + enc.schema, options.attachment?.into()?.bytes, replyCallbackOf { 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 e523bf8c..633c011c 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt @@ -59,8 +59,8 @@ class Query internal constructor( keH: io.zenoh.jni.keyexpr.KeyExpr, parameters: String, payloadH: io.zenoh.jni.bytes.ZBytes?, - encH: io.zenoh.jni.bytes.Encoding?, encId: Int?, + encSchema: String?, attachH: io.zenoh.jni.bytes.ZBytes?, acceptsRepliesInt: Int, zq: io.zenoh.jni.query.Query, @@ -72,7 +72,7 @@ class Query internal constructor( ke, selector, payloadH?.let { ZBytes.fromHandle(it) }, - encH?.let { Encoding.fromParts(it, encId!!) }, + encId?.let { Encoding(it, encSchema) }, attachH?.let { ZBytes.fromHandle(it) }, io.zenoh.jni.query.ReplyKeyExpr.fromInt(acceptsRepliesInt).toPublic(), zq @@ -96,8 +96,8 @@ class Query internal constructor( keyExpr.flat, payload.into().bytes, true, - options.encoding.idForWire(), - options.encoding.schemaForWire(), + options.encoding.id, + options.encoding.schema, options.timeStamp?.ntpValue(), options.attachment?.into()?.bytes, options.express, @@ -154,7 +154,7 @@ class Query internal constructor( @Throws(ZError::class) fun replyErr(message: IntoZBytes, options: ReplyErrOptions = ReplyErrOptions()) { val q = zQuery ?: throw ZError("Query is invalid") - q.replyError(message.into().bytes, true, options.encoding.idForWire(), options.encoding.schemaForWire(), throwZError) + q.replyError(message.into().bytes, true, options.encoding.id, options.encoding.schema, throwZError) q.close() zQuery = null } 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 86476eda..d7489659 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/sample/Sample.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/sample/Sample.kt @@ -65,8 +65,8 @@ data class Sample( fun fromParts( keH: io.zenoh.jni.keyexpr.KeyExpr, payloadH: io.zenoh.jni.bytes.ZBytes, - encH: io.zenoh.jni.bytes.Encoding, encId: Int, + encSchema: String?, kindInt: Int, ntp64: Long?, express: Boolean, @@ -80,7 +80,7 @@ data class Sample( ): Sample = Sample( KeyExpr(keH), ZBytes.fromHandle(payloadH), - Encoding.fromParts(encH, encId), + 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/EncodingCorrespondenceTest.kt b/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingCorrespondenceTest.kt new file mode 100644 index 00000000..668b22dd --- /dev/null +++ b/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingCorrespondenceTest.kt @@ -0,0 +1,115 @@ +// +// Copyright (c) 2023 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.Encoding +import io.zenoh.exceptions.throwZError0 +import io.zenoh.jni.bytes.Encoding as JniEncoding +import org.junit.Assert.assertEquals +import org.junit.Test + +/** + * Correspondence tests for the pure-JVM [Encoding] implementation. + * + * The SDK implements the encoding string ↔ `(id, schema)` conversion in pure + * Kotlin (no JNI crossing), on the contract that any JVM-side reimplementation + * of zenoh-flat API must be verified against the native implementation. These + * tests drive both implementations — the pure one and the native oracle (the + * generated `Encoding` handle methods) — over the whole predefined id range + * plus the edge shapes of the parse/render rules, asserting equal results. + */ +class EncodingCorrespondenceTest { + + /** Native oracle: the canonical string of `(id, schema)`. */ + private fun nativeRender(id: Int, schema: String?): String { + val h = JniEncoding.newFromId(id, schema, throwZError0) + try { + return h.toStr(throwZError0) + } finally { + h.close() + } + } + + /** Native oracle: parse a string into `(id, schema, canonical string)`. */ + private fun nativeParse(s: String): Triple { + val h = JniEncoding.newFromString(s, throwZError0) + try { + return Triple(h.getId(throwZError0), h.getSchema(throwZError0), h.toStr(throwZError0)) + } finally { + h.close() + } + } + + @Test + fun renderMatchesNativeAcrossIdRange() { + // The whole predefined range, a gap past it (unknown ids), and the + // custom id — with and without a schema. + val ids = (0..64) + listOf(1000, 0xFFFE, 0xFFFF) + for (id in ids) { + for (schema in listOf(null, "utf-8")) { + assertEquals( + "render mismatch for (id=$id, schema=$schema)", + nativeRender(id, schema), + Encoding(id, schema).toString(), + ) + } + } + } + + @Test + fun parseMatchesNativeOnNamesAndEdges() { + // Every predefined canonical name (obtained FROM the native table so the + // corpus can't drift), plus the parse-rule edge shapes. + val names = (0..52).map { nativeRender(it, null) } + val edges = listOf( + "", + "text/plain", + "text/plain;utf-8", + "text/plain;", + "my_custom_encoding", + "custom;with;semicolons", + ";leading_separator", + "unknown_name;schema", + "zenoh/bytes;s", + ) + for (s in names + names.map { "$it;schema" } + edges) { + val (nid, nschema, nstr) = nativeParse(s) + val pure = Encoding.from(s) + assertEquals("id mismatch for \"$s\"", nid, pure.id) + assertEquals("schema mismatch for \"$s\"", nschema, pure.schema) + assertEquals("render mismatch for \"$s\"", nstr, pure.toString()) + } + } + + @Test + fun withSchemaMatchesNative() { + val bases = listOf( + Encoding.TEXT_PLAIN, + Encoding.ZENOH_BYTES, + Encoding.from("my_custom_encoding"), + Encoding.from("text/plain;old-schema"), + ) + for (base in bases) { + val pure = base.withSchema("new-schema") + val w = JniEncoding.newWithSchema(base.id, base.schema, "new-schema", throwZError0) + val nativeStr = try { + w.toStr(throwZError0) + } finally { + w.close() + } + assertEquals("withSchema mismatch for $base", nativeStr, pure.toString()) + } + } +} diff --git a/zenoh-java/src/jvmTest/kotlin/io/zenoh/ZenohIdCorrespondenceTest.kt b/zenoh-java/src/jvmTest/kotlin/io/zenoh/ZenohIdCorrespondenceTest.kt new file mode 100644 index 00000000..f531c240 --- /dev/null +++ b/zenoh-java/src/jvmTest/kotlin/io/zenoh/ZenohIdCorrespondenceTest.kt @@ -0,0 +1,58 @@ +// +// Copyright (c) 2023 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.config.ZenohId +import io.zenoh.exceptions.throwZError0 +import io.zenoh.jni.config.ZenohId as JniZenohId +import kotlin.random.Random +import org.junit.Assert.assertEquals +import org.junit.Test + +/** + * Correspondence test for the pure-JVM [ZenohId.toString] — the id bytes read + * as a little-endian integer, rendered as lowercase hex without leading zeros + * (`"0"` for the zero id). Verified against the native formatter + * (`zenoh_id_to_string`) over edge patterns and a deterministic random corpus. + */ +class ZenohIdCorrespondenceTest { + + private fun assertCorresponds(bytes: ByteArray) { + val jni = JniZenohId(bytes) + assertEquals( + "zid render mismatch for ${bytes.joinToString(",")}", + jni.toStr(throwZError0), + ZenohId(jni).toString(), + ) + } + + @Test + fun rendersLikeNative() { + val edges = listOf( + ByteArray(16), // zero id + ByteArray(16).also { it[0] = 1 }, // smallest nonzero (LE low byte) + ByteArray(16).also { it[15] = 1 }, // highest byte only + ByteArray(16).also { it[0] = 0x0F }, // sub-0x10 low byte + ByteArray(16) { 0xFF.toByte() }, // all ones + ByteArray(16) { i -> i.toByte() }, // ascending with leading zero byte + ) + edges.forEach(::assertCorresponds) + + val rng = Random(20260716) + repeat(32) { + assertCorresponds(rng.nextBytes(16)) + } + } +} From 4eecd3a1f635ee0e582d9d8ac6cec32cab34e155 Mon Sep 17 00:00:00 2001 From: Michael Ilyin Date: Thu, 16 Jul 2026 15:33:07 +0200 Subject: [PATCH 2/3] Publisher default encoding at declare + PinnedEncoding per-put override MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two encoding fast paths over the dual-arm expansion (prebindgen#80, zenoh-flat#3, zenoh-flat-jni#4): - The publisher's default encoding is set NATIVELY once at declare time (PublisherOptions.encoding -> declarePublisher's encoding arm); plain put(payload) passes absent (sel -1), so no encoding data crosses per call. Previously the JVM tier re-sent (id, schema) on every put. - PinnedEncoding (Encoding.pinned()) preallocates the native form; passing it in PutOptions/ReplyOptions/GetOptions crosses only a borrowed handle (native clone = Arc bump) — no schema-string traffic in hot publish loops. close() releases the handle (falls back to the plain (id, schema) arm); a finalizer backstops leaks. A PinnedEncoding equals its plain counterpart. - All outbound sites route through one Encoding?.forWire() helper emitting the selector tuple (-1 absent / 0 value / 1 pinned handle). Tests: PinnedEncodingTest (publisher-default applies natively end-to-end; pinned override reused across puts; post-close fallback), correspondence tests updated to the dual-arm newWithSchema. jvmTest: 107 tests, 0 failures. CI pins: zenoh-flat-jni cbbca40, zenoh-flat d792547. --- .github/workflows/ci.yml | 4 +- ZENOH_FLAT_TRANSITION.md | 2 +- .../src/commonMain/kotlin/io/zenoh/Session.kt | 23 +++-- .../kotlin/io/zenoh/bytes/Encoding.kt | 78 ++++++++++++++++- .../kotlin/io/zenoh/pubsub/Publisher.kt | 19 ++-- .../kotlin/io/zenoh/query/Querier.kt | 11 ++- .../commonMain/kotlin/io/zenoh/query/Query.kt | 12 ++- .../io/zenoh/EncodingCorrespondenceTest.kt | 4 +- .../kotlin/io/zenoh/PinnedEncodingTest.kt | 87 +++++++++++++++++++ 9 files changed, 213 insertions(+), 27 deletions(-) create mode 100644 zenoh-java/src/jvmTest/kotlin/io/zenoh/PinnedEncodingTest.kt diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 98fc0a1b..0e3587f2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,14 +31,14 @@ jobs: uses: actions/checkout@v4 with: repository: ZettaScaleLabs/zenoh-flat-jni - ref: abbc4832e8658109fdf8f9b1face95a9f21a0b15 + ref: cbbca40e2f5f0d84d94ba9d05e64766c807750fd path: zenoh-flat-jni - name: Check out zenoh-flat uses: actions/checkout@v4 with: repository: ZettaScaleLabs/zenoh-flat - ref: 7b151d7808583ed15884fada2fd40d5db387c982 + ref: d792547220f20e0f0f957fbaa6e135e8d5114a82 path: zenoh-flat - name: Use prebindgen from GitHub main diff --git a/ZENOH_FLAT_TRANSITION.md b/ZENOH_FLAT_TRANSITION.md index af9d7639..4ff857d2 100644 --- a/ZENOH_FLAT_TRANSITION.md +++ b/ZENOH_FLAT_TRANSITION.md @@ -26,7 +26,7 @@ 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 | merged | -| `encoding-pure-value` | `Encoding` and `ZenohId.toString` become pure JVM values (no JNI crossings, no retained handles — fixes a per-message native leak), verified by correspondence tests against the native implementation; pairs with [zenoh-flat-jni#4](https://github.com/ZettaScaleLabs/zenoh-flat-jni/pull/4) | in progress | +| `encoding-pure-value` | `Encoding`/`ZenohId.toString` become pure JVM values (correspondence-tested, fixes a per-message native leak); publisher **default encoding set natively at declare** (plain puts cross no encoding data); `PinnedEncoding` preallocates the native form for per-put overrides (handle-only crossing). 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), [prebindgen#80](https://github.com/milyin/prebindgen/pull/80) | 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) diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt index 8f91a17f..4bbafed7 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt @@ -18,6 +18,7 @@ import io.zenoh.annotations.Unstable import io.zenoh.bytes.Encoding import io.zenoh.bytes.IntoZBytes import io.zenoh.bytes.ZBytes +import io.zenoh.bytes.forWire import io.zenoh.bytes.into import io.zenoh.config.ZenohId import io.zenoh.exceptions.ZError @@ -587,8 +588,15 @@ 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.forWire() val zPublisher = zSession.declarePublisher( keyExpr.cloneFlat(), + enc.sel, + enc.id, + enc.schema, + enc.handle, options.congestionControl.jni, options.priority.jni, options.express, @@ -722,7 +730,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.forWire() zSession.get( sel.keyExpr.flat, sel.parameters?.toString(), @@ -734,9 +742,10 @@ class Session private constructor(private val config: Config) : AutoCloseable { options.qos.priority.jni, options.qos.express, options.payload?.into()?.bytes, - true, + enc.sel, enc.id, enc.schema, + enc.handle, options.attachment?.into()?.bytes, replyCallbackOf { handler.handle(it) }, { handler.onClose() }, @@ -755,7 +764,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.forWire() zSession.get( sel.keyExpr.flat, sel.parameters?.toString(), @@ -767,9 +776,10 @@ class Session private constructor(private val config: Config) : AutoCloseable { options.qos.priority.jni, options.qos.express, options.payload?.into()?.bytes, - true, + enc.sel, enc.id, enc.schema, + enc.handle, options.attachment?.into()?.bytes, replyCallbackOf { callback.run(it) }, { }, @@ -782,13 +792,14 @@ 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.forWire() zSession.put( keyExpr.flat, payload.into().bytes, - true, + enc.sel, enc.id, enc.schema, + enc.handle, putOptions.congestionControl.jni, putOptions.priority.jni, putOptions.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 841f10a0..5a47a8b3 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/Encoding.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/Encoding.kt @@ -33,8 +33,12 @@ import io.zenoh.jni.bytes.* * is a plain immutable value: no native handle, no native calls, nothing to * close. Its correspondence to the native implementation is verified by * `EncodingCorrespondenceTest`. + * + * For a hot publish loop with a custom schema-bearing encoding, [pinned] + * preallocates the native form once so each call crosses only a handle — + * see [PinnedEncoding]. */ -class Encoding internal constructor( +open class Encoding internal constructor( internal val id: Int, internal val schema: String? = null, ) { @@ -196,6 +200,18 @@ class Encoding internal constructor( } } + /** + * Preallocate the native form of this encoding. The returned + * [PinnedEncoding] IS this encoding (usable anywhere an [Encoding] is — + * e.g. `PutOptions.encoding` or `PublisherOptions.encoding`), but native + * calls pass its prebuilt handle instead of re-crossing the `(id, schema)` + * pair — for a custom schema-bearing encoding in a hot publish loop this + * removes the per-call string traffic. The native side only borrows the + * handle (a cheap clone per call), so the pinned encoding is reusable + * until [PinnedEncoding.close]s. + */ + fun pinned(): PinnedEncoding = PinnedEncoding(id, schema) + /** * Set a schema to this encoding. Zenoh does not define what a schema is and its semantics is left to the implementer. * E.g. a common schema for `text/plain` encoding is `utf-8`. @@ -234,10 +250,66 @@ class Encoding internal constructor( override fun equals(other: Any?): Boolean { if (this === other) return true - if (javaClass != other?.javaClass) return false - other as Encoding + // `is`, not javaClass: a [PinnedEncoding] equals its plain counterpart + // (pinning is a transport optimization, not part of the value). + if (other !is Encoding) return false return id == other.id && schema == other.schema } override fun hashCode(): Int = 31 * id + (schema?.hashCode() ?: 0) } + +/** + * The selector-tuple wire form of an optional encoding, matching the generated + * externs' dual-arm expansion: `sel == -1` = absent, arm `0` = the decomposed + * `(id, schema)` value, arm `1` = a pinned native handle (borrowed by the + * call). Computed once per call so a concurrent [PinnedEncoding.close] cannot + * desynchronize the tuple. + */ +internal class EncodingWire( + val sel: Int, + val id: Int?, + val schema: String?, + val handle: io.zenoh.jni.bytes.Encoding?, +) + +/** See [EncodingWire]. A closed [PinnedEncoding] falls back to arm 0. */ +internal fun Encoding?.forWire(): EncodingWire { + if (this == null) return EncodingWire(-1, null, null, null) + val h = (this as? PinnedEncoding)?.handleOrNull() + return if (h != null) { + EncodingWire(1, null, null, h) + } else { + EncodingWire(0, id, schema, null) + } +} + +/** + * An [Encoding] with its native form preallocated (see [Encoding.pinned]). + * Native calls pass the prebuilt handle (borrowed — cloned natively per call, + * an `Arc` bump) instead of re-crossing the `(id, schema)` pair, removing the + * per-call schema-string traffic in hot publish loops. + * + * Owns one native handle: [close] releases it deterministically (after which + * native calls fall back to the plain `(id, schema)` crossing); a finalizer + * backstops leaks. + */ +class PinnedEncoding internal constructor(id: Int, schema: String?) : + Encoding(id, schema), AutoCloseable { + + internal val handle: io.zenoh.jni.bytes.Encoding = + io.zenoh.jni.bytes.Encoding.newFromId(id, schema, io.zenoh.exceptions.throwZError0) + + /** Whether the pinned native form is still available. */ + internal fun handleOrNull(): io.zenoh.jni.bytes.Encoding? = + handle.takeIf { !it.isClosed() } + + override fun close() { + handle.close() + } + + @Suppress("removal") + protected fun finalize() { + close() + } +} diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt index fd410eb8..0d7e80c8 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt @@ -17,6 +17,7 @@ package io.zenoh.pubsub import io.zenoh.* import io.zenoh.bytes.Encoding import io.zenoh.bytes.IntoZBytes +import io.zenoh.bytes.forWire import io.zenoh.bytes.ZBytes import io.zenoh.exceptions.ZError import io.zenoh.exceptions.throwZError @@ -80,13 +81,15 @@ class Publisher internal constructor( /** Performs a PUT operation on the specified [keyExpr] with the specified [payload]. */ @Throws(ZError::class) fun put(payload: IntoZBytes) { - performPut(payload, encoding, null) + // No per-call encoding: the publisher's default encoding — set + // NATIVELY at declare time — applies, so no encoding data crosses. + performPut(payload, null, null) } /** Performs a PUT operation on the specified [keyExpr] with the specified [payload]. */ @Throws(ZError::class) fun put(payload: IntoZBytes, options: PutOptions) { - performPut(payload, options.encoding ?: this.encoding, options.attachment) + performPut(payload, options.encoding, options.attachment) } /** Performs a PUT operation on the specified [keyExpr] with the specified [payload]. */ @@ -129,13 +132,17 @@ class Publisher internal constructor( } @Throws(ZError::class) - private fun performPut(payload: IntoZBytes, encoding: Encoding, attachment: IntoZBytes?) { + private fun performPut(payload: IntoZBytes, encoding: Encoding?, attachment: IntoZBytes?) { val p = zPublisher ?: throw publisherNotValid + // `null` encoding = absent (`sel -1`): the publisher's native default + // applies. A [io.zenoh.bytes.PinnedEncoding] crosses as its handle. + val enc = encoding.forWire() p.put( payload.into().bytes, - true, - encoding.id, - encoding.schema, + enc.sel, + enc.id, + enc.schema, + enc.handle, attachment?.into()?.bytes, throwZError, ) diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt index 3acf207d..ece2f61c 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt @@ -17,6 +17,7 @@ package io.zenoh.query import io.zenoh.annotations.Unstable import io.zenoh.replyCallbackOf import io.zenoh.bytes.Encoding +import io.zenoh.bytes.forWire import io.zenoh.bytes.IntoZBytes import io.zenoh.bytes.ZBytes import io.zenoh.exceptions.ZError @@ -145,13 +146,14 @@ class Querier internal constructor(val keyExpr: KeyExpr, val qos: QoS, private v private fun resolveGetWithCallback(callback: Callback, options: GetOptions) { val q = zQuerier ?: throw ZError("Querier is not valid.") - val enc = options.encoding ?: Encoding.defaultEncoding() + val enc = options.encoding.forWire() q.get( options.parameters?.toString(), options.payload?.into()?.bytes, - true, + enc.sel, enc.id, enc.schema, + enc.handle, options.attachment?.into()?.bytes, replyCallbackOf { callback.run(it) }, { }, @@ -161,13 +163,14 @@ class Querier internal constructor(val keyExpr: KeyExpr, val qos: QoS, private v private fun resolveGetWithHandler(handler: Handler, options: GetOptions): R { val q = zQuerier ?: throw ZError("Querier is not valid.") - val enc = options.encoding ?: Encoding.defaultEncoding() + val enc = options.encoding.forWire() q.get( options.parameters?.toString(), options.payload?.into()?.bytes, - true, + enc.sel, enc.id, enc.schema, + enc.handle, options.attachment?.into()?.bytes, replyCallbackOf { 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 633c011c..528ee59a 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt @@ -16,6 +16,7 @@ package io.zenoh.query import io.zenoh.ZenohType import io.zenoh.bytes.Encoding +import io.zenoh.bytes.forWire import io.zenoh.bytes.IntoZBytes import io.zenoh.bytes.ZBytes import io.zenoh.exceptions.ZError @@ -92,12 +93,14 @@ class Query internal constructor( @JvmOverloads fun reply(keyExpr: KeyExpr, payload: IntoZBytes, options: ReplyOptions = ReplyOptions()) { val q = zQuery ?: throw ZError("Query is invalid") + val enc = options.encoding.forWire() q.replySuccess( keyExpr.flat, payload.into().bytes, - true, - options.encoding.id, - options.encoding.schema, + enc.sel, + enc.id, + enc.schema, + enc.handle, options.timeStamp?.ntpValue(), options.attachment?.into()?.bytes, options.express, @@ -154,7 +157,8 @@ class Query internal constructor( @Throws(ZError::class) fun replyErr(message: IntoZBytes, options: ReplyErrOptions = ReplyErrOptions()) { val q = zQuery ?: throw ZError("Query is invalid") - q.replyError(message.into().bytes, true, options.encoding.id, options.encoding.schema, throwZError) + val enc = options.encoding.forWire() + q.replyError(message.into().bytes, enc.sel, enc.id, enc.schema, enc.handle, throwZError) q.close() zQuery = null } diff --git a/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingCorrespondenceTest.kt b/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingCorrespondenceTest.kt index 668b22dd..149cdb87 100644 --- a/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingCorrespondenceTest.kt +++ b/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingCorrespondenceTest.kt @@ -103,7 +103,9 @@ class EncodingCorrespondenceTest { ) for (base in bases) { val pure = base.withSchema("new-schema") - val w = JniEncoding.newWithSchema(base.id, base.schema, "new-schema", throwZError0) + // The `e` param crosses via the dual-arm expansion: arm 0 = the + // decomposed (id, schema) value. + val w = JniEncoding.newWithSchema(0, base.id, base.schema, null, "new-schema", throwZError0) val nativeStr = try { w.toStr(throwZError0) } finally { diff --git a/zenoh-java/src/jvmTest/kotlin/io/zenoh/PinnedEncodingTest.kt b/zenoh-java/src/jvmTest/kotlin/io/zenoh/PinnedEncodingTest.kt new file mode 100644 index 00000000..1db5ae32 --- /dev/null +++ b/zenoh-java/src/jvmTest/kotlin/io/zenoh/PinnedEncodingTest.kt @@ -0,0 +1,87 @@ +// +// Copyright (c) 2023 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.Encoding +import io.zenoh.bytes.ZBytes +import io.zenoh.keyexpr.KeyExpr +import io.zenoh.pubsub.PublisherOptions +import io.zenoh.pubsub.PutOptions +import io.zenoh.sample.Sample +import org.junit.Assert.assertEquals +import org.junit.Test + +/** + * End-to-end tests of the two encoding fast paths: + * * the publisher's **default encoding**, set natively once at declare time + * (plain `put`s cross no encoding data at all), and + * * a **pinned** encoding ([Encoding.pinned]) passed per call as a + * preallocated native handle (no schema-string crossing per put). + */ +class PinnedEncodingTest { + + private val custom = Encoding.from("application/custom;my-schema") + + @Test + fun publisherDefaultEncodingAppliesNatively() { + val session = Zenoh.open(Config.loadDefault()) + val keyExpr = KeyExpr.tryFrom("example/testing/pinned/default") + val received = mutableListOf() + val subscriber = session.declareSubscriber(keyExpr) { received.add(it) } + + val options = PublisherOptions() + options.encoding = custom + val publisher = session.declarePublisher(keyExpr, options) + // Plain put: NO encoding crosses — the native publisher default applies. + publisher.put(ZBytes.from("hello")) + Thread.sleep(500) + + publisher.close() + subscriber.close() + session.close() + assertEquals(1, received.size) + assertEquals(custom, received[0].encoding) + } + + @Test + fun pinnedEncodingOverridesPerPut() { + val session = Zenoh.open(Config.loadDefault()) + val keyExpr = KeyExpr.tryFrom("example/testing/pinned/override") + val received = mutableListOf() + val subscriber = session.declareSubscriber(keyExpr) { received.add(it) } + val publisher = session.declarePublisher(keyExpr) + + val pinned = custom.pinned() + val options = PutOptions() + options.encoding = pinned + // The pinned handle is BORROWED per call — reusable across the loop. + repeat(3) { publisher.put(ZBytes.from("hello #$it"), options) } + Thread.sleep(500) + assertEquals(3, received.size) + received.forEach { assertEquals(custom, it.encoding) } + + // After close() the pinned form falls back to the plain (id, schema) + // crossing — same value, still correct. + pinned.close() + publisher.put(ZBytes.from("after close"), options) + Thread.sleep(500) + + publisher.close() + subscriber.close() + session.close() + assertEquals(4, received.size) + assertEquals(custom, received[3].encoding) + } +} From e6c44f77f9278abcc292cd878b8653be08ef3c91 Mon Sep 17 00:00:00 2001 From: Michael Ilyin Date: Thu, 16 Jul 2026 16:06:46 +0200 Subject: [PATCH 3/3] Move encoding/zid logic to the shared bindings tier; drop pinning MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit zenoh-flat-jni is the shared tier reused by both zenoh-java and zenoh-kotlin, so the pure-JVM conversion logic moves there (EncodingCodec, ZenohId.zidString — next to the generated constants they are built from). io.zenoh.bytes.Encoding shrinks to a thin facade: the established constants and API delegating to the shared codec, zero logic. Drop PinnedEncoding and the selector-tuple wire helper: with the publisher's default encoding set natively at declare time, the hot publish loop crosses no encoding data at all, and per-call overrides crossing (id, schema) cost only a short string decode — not worth a resource class in the public API. Encoding params revert to the plain (present, id, schema) crossing. jvmTest: 106 tests, 0 failures (correspondence suites now exercise the shared codec; PublisherEncodingTest verifies the native declare-time default + per-put override end-to-end). CI pins: zenoh-flat-jni 9bc155b, zenoh-flat main (6d22091, #3 merged). --- .github/workflows/ci.yml | 4 +- ZENOH_FLAT_TRANSITION.md | 2 +- .../src/commonMain/kotlin/io/zenoh/Session.kt | 33 ++- .../kotlin/io/zenoh/bytes/Encoding.kt | 215 ++---------------- .../kotlin/io/zenoh/config/ZenohId.kt | 20 +- .../kotlin/io/zenoh/pubsub/Publisher.kt | 13 +- .../kotlin/io/zenoh/query/Querier.kt | 19 +- .../commonMain/kotlin/io/zenoh/query/Query.kt | 14 +- .../io/zenoh/EncodingCorrespondenceTest.kt | 5 +- .../kotlin/io/zenoh/PinnedEncodingTest.kt | 87 ------- .../kotlin/io/zenoh/PublisherEncodingTest.kt | 62 +++++ 11 files changed, 120 insertions(+), 354 deletions(-) delete mode 100644 zenoh-java/src/jvmTest/kotlin/io/zenoh/PinnedEncodingTest.kt create mode 100644 zenoh-java/src/jvmTest/kotlin/io/zenoh/PublisherEncodingTest.kt diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0e3587f2..ef67473a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,14 +31,14 @@ jobs: uses: actions/checkout@v4 with: repository: ZettaScaleLabs/zenoh-flat-jni - ref: cbbca40e2f5f0d84d94ba9d05e64766c807750fd + ref: 9bc155b3272c8c905bf95614cb7eb7e6d656c72f path: zenoh-flat-jni - name: Check out zenoh-flat uses: actions/checkout@v4 with: repository: ZettaScaleLabs/zenoh-flat - ref: d792547220f20e0f0f957fbaa6e135e8d5114a82 + ref: 6d22091d75912be1c14330b0f78fa1d4bf258ea3 path: zenoh-flat - name: Use prebindgen from GitHub main diff --git a/ZENOH_FLAT_TRANSITION.md b/ZENOH_FLAT_TRANSITION.md index 4ff857d2..492d969a 100644 --- a/ZENOH_FLAT_TRANSITION.md +++ b/ZENOH_FLAT_TRANSITION.md @@ -26,7 +26,7 @@ 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 | merged | -| `encoding-pure-value` | `Encoding`/`ZenohId.toString` become pure JVM values (correspondence-tested, fixes a per-message native leak); publisher **default encoding set natively at declare** (plain puts cross no encoding data); `PinnedEncoding` preallocates the native form for per-put overrides (handle-only crossing). 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), [prebindgen#80](https://github.com/milyin/prebindgen/pull/80) | in progress | +| [#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) diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt index 4bbafed7..846dec59 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/Session.kt @@ -18,7 +18,6 @@ import io.zenoh.annotations.Unstable import io.zenoh.bytes.Encoding import io.zenoh.bytes.IntoZBytes import io.zenoh.bytes.ZBytes -import io.zenoh.bytes.forWire import io.zenoh.bytes.into import io.zenoh.config.ZenohId import io.zenoh.exceptions.ZError @@ -590,13 +589,12 @@ class Session private constructor(private val config: Config) : AutoCloseable { 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.forWire() + val enc = options.encoding val zPublisher = zSession.declarePublisher( keyExpr.cloneFlat(), - enc.sel, + true, enc.id, enc.schema, - enc.handle, options.congestionControl.jni, options.priority.jni, options.express, @@ -730,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.forWire() + val enc = options.encoding zSession.get( sel.keyExpr.flat, sel.parameters?.toString(), @@ -742,10 +740,9 @@ class Session private constructor(private val config: Config) : AutoCloseable { options.qos.priority.jni, options.qos.express, options.payload?.into()?.bytes, - enc.sel, - enc.id, - enc.schema, - enc.handle, + enc != null, + enc?.id ?: 0, + enc?.schema, options.attachment?.into()?.bytes, replyCallbackOf { handler.handle(it) }, { handler.onClose() }, @@ -764,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.forWire() + val enc = options.encoding zSession.get( sel.keyExpr.flat, sel.parameters?.toString(), @@ -776,10 +773,9 @@ class Session private constructor(private val config: Config) : AutoCloseable { options.qos.priority.jni, options.qos.express, options.payload?.into()?.bytes, - enc.sel, - enc.id, - enc.schema, - enc.handle, + enc != null, + enc?.id ?: 0, + enc?.schema, options.attachment?.into()?.bytes, replyCallbackOf { callback.run(it) }, { }, @@ -792,14 +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.forWire() + val enc = putOptions.encoding zSession.put( keyExpr.flat, payload.into().bytes, - enc.sel, - enc.id, - enc.schema, - enc.handle, + enc != null, + enc?.id ?: 0, + enc?.schema, putOptions.congestionControl.jni, putOptions.priority.jni, putOptions.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 5a47a8b3..dfa0e19c 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/Encoding.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/Encoding.kt @@ -26,100 +26,19 @@ import io.zenoh.jni.bytes.* * * A set of associated constants are provided to cover the most common encodings for user convenience. * - * An encoding **is** its decomposed pair `(id, schema)` — exactly Zenoh's own - * representation; the textual form (e.g. `"text/plain;utf-8"`) is derived from - * a fixed id↔name table. Both the table (the generated `ENCODING_*` constants, - * regenerated from Zenoh) and the conversion rules live JVM-side, so this class - * is a plain immutable value: no native handle, no native calls, nothing to - * close. Its correspondence to the native implementation is verified by - * `EncodingCorrespondenceTest`. - * - * For a hot publish loop with a custom schema-bearing encoding, [pinned] - * preallocates the native form once so each call crosses only a handle — - * see [PinnedEncoding]. + * An encoding **is** its decomposed pair `(id, schema)` — Zenoh's own + * representation. This class is a plain immutable value; all conversion logic + * (the id↔name table and Zenoh's parse/render rules) lives in the shared + * bindings tier ([EncodingCodec]), verified against the native implementation + * by `EncodingCorrespondenceTest`. No native handle, no native calls, nothing + * to close. */ -open class Encoding internal constructor( +class Encoding internal constructor( internal val id: Int, internal val schema: String? = null, ) { companion object { - /** - * Mirror of Zenoh's `CUSTOM_ENCODING_ID`: a custom (non-predefined) - * encoding carries its whole textual form in the schema slot. - */ - internal const val CUSTOM_ENCODING_ID: Int = 0xFFFF - - /** - * The id↔name table, built from the generated constants (single source - * of truth in Rust) plus the custom-id row Zenoh keys with the empty - * name. - */ - private val idToName: Map = mapOf( - ENCODING_ZENOH_BYTES_ID to ENCODING_ZENOH_BYTES, - ENCODING_ZENOH_STRING_ID to ENCODING_ZENOH_STRING, - ENCODING_ZENOH_SERIALIZED_ID to ENCODING_ZENOH_SERIALIZED, - ENCODING_APPLICATION_OCTET_STREAM_ID to ENCODING_APPLICATION_OCTET_STREAM, - ENCODING_TEXT_PLAIN_ID to ENCODING_TEXT_PLAIN, - ENCODING_APPLICATION_JSON_ID to ENCODING_APPLICATION_JSON, - ENCODING_TEXT_JSON_ID to ENCODING_TEXT_JSON, - ENCODING_APPLICATION_CDR_ID to ENCODING_APPLICATION_CDR, - ENCODING_APPLICATION_CBOR_ID to ENCODING_APPLICATION_CBOR, - ENCODING_APPLICATION_YAML_ID to ENCODING_APPLICATION_YAML, - ENCODING_TEXT_YAML_ID to ENCODING_TEXT_YAML, - ENCODING_TEXT_JSON5_ID to ENCODING_TEXT_JSON5, - ENCODING_APPLICATION_PYTHON_SERIALIZED_OBJECT_ID to ENCODING_APPLICATION_PYTHON_SERIALIZED_OBJECT, - ENCODING_APPLICATION_PROTOBUF_ID to ENCODING_APPLICATION_PROTOBUF, - ENCODING_APPLICATION_JAVA_SERIALIZED_OBJECT_ID to ENCODING_APPLICATION_JAVA_SERIALIZED_OBJECT, - ENCODING_APPLICATION_OPENMETRICS_TEXT_ID to ENCODING_APPLICATION_OPENMETRICS_TEXT, - ENCODING_IMAGE_PNG_ID to ENCODING_IMAGE_PNG, - ENCODING_IMAGE_JPEG_ID to ENCODING_IMAGE_JPEG, - ENCODING_IMAGE_GIF_ID to ENCODING_IMAGE_GIF, - ENCODING_IMAGE_BMP_ID to ENCODING_IMAGE_BMP, - ENCODING_IMAGE_WEBP_ID to ENCODING_IMAGE_WEBP, - ENCODING_APPLICATION_XML_ID to ENCODING_APPLICATION_XML, - ENCODING_APPLICATION_X_WWW_FORM_URLENCODED_ID to ENCODING_APPLICATION_X_WWW_FORM_URLENCODED, - ENCODING_TEXT_HTML_ID to ENCODING_TEXT_HTML, - ENCODING_TEXT_XML_ID to ENCODING_TEXT_XML, - ENCODING_TEXT_CSS_ID to ENCODING_TEXT_CSS, - ENCODING_TEXT_JAVASCRIPT_ID to ENCODING_TEXT_JAVASCRIPT, - ENCODING_TEXT_MARKDOWN_ID to ENCODING_TEXT_MARKDOWN, - ENCODING_TEXT_CSV_ID to ENCODING_TEXT_CSV, - ENCODING_APPLICATION_SQL_ID to ENCODING_APPLICATION_SQL, - ENCODING_APPLICATION_COAP_PAYLOAD_ID to ENCODING_APPLICATION_COAP_PAYLOAD, - ENCODING_APPLICATION_JSON_PATCH_JSON_ID to ENCODING_APPLICATION_JSON_PATCH_JSON, - ENCODING_APPLICATION_JSON_SEQ_ID to ENCODING_APPLICATION_JSON_SEQ, - ENCODING_APPLICATION_JSONPATH_ID to ENCODING_APPLICATION_JSONPATH, - ENCODING_APPLICATION_JWT_ID to ENCODING_APPLICATION_JWT, - ENCODING_APPLICATION_MP4_ID to ENCODING_APPLICATION_MP4, - ENCODING_APPLICATION_SOAP_XML_ID to ENCODING_APPLICATION_SOAP_XML, - ENCODING_APPLICATION_YANG_ID to ENCODING_APPLICATION_YANG, - ENCODING_AUDIO_AAC_ID to ENCODING_AUDIO_AAC, - ENCODING_AUDIO_FLAC_ID to ENCODING_AUDIO_FLAC, - ENCODING_AUDIO_MP4_ID to ENCODING_AUDIO_MP4, - ENCODING_AUDIO_OGG_ID to ENCODING_AUDIO_OGG, - ENCODING_AUDIO_VORBIS_ID to ENCODING_AUDIO_VORBIS, - ENCODING_VIDEO_H261_ID to ENCODING_VIDEO_H261, - ENCODING_VIDEO_H263_ID to ENCODING_VIDEO_H263, - ENCODING_VIDEO_H264_ID to ENCODING_VIDEO_H264, - ENCODING_VIDEO_H265_ID to ENCODING_VIDEO_H265, - ENCODING_VIDEO_H266_ID to ENCODING_VIDEO_H266, - ENCODING_VIDEO_MP4_ID to ENCODING_VIDEO_MP4, - ENCODING_VIDEO_OGG_ID to ENCODING_VIDEO_OGG, - ENCODING_VIDEO_RAW_ID to ENCODING_VIDEO_RAW, - ENCODING_VIDEO_VP8_ID to ENCODING_VIDEO_VP8, - ENCODING_VIDEO_VP9_ID to ENCODING_VIDEO_VP9, - CUSTOM_ENCODING_ID to "", - ) - - // Parse-side reverse table. Zenoh's `STR_TO_ID` has NO empty-string - // row (the `CUSTOM ↔ ""` mapping is render-side only), so a leading - // `;` input falls through to the custom arm with the whole string as - // schema — exclude the custom row here to match. - private val nameToId: Map = idToName.entries - .filter { (_, name) -> name.isNotEmpty() } - .associate { (id, name) -> name to id } - @JvmField val ZENOH_BYTES = Encoding(ENCODING_ZENOH_BYTES_ID) @JvmField val ZENOH_STRING = Encoding(ENCODING_ZENOH_STRING_ID) @JvmField val ZENOH_SERIALIZED = Encoding(ENCODING_ZENOH_SERIALIZED_ID) @@ -181,135 +100,31 @@ open class Encoding internal constructor( * Parse a textual encoding (e.g. `"text/plain"`, `"text/plain;utf-8"`, * `"my_encoding"`). Well-known names resolve to their canonical id; * everything else is preserved as a custom encoding. - * - * Implements Zenoh's parse rule: everything before the first `;` is - * looked up as a known name — a match yields `(id, rest-if-nonempty)`, - * a miss yields the custom id with the whole string as schema. */ @JvmStatic fun from(s: String): Encoding { - if (s.isEmpty()) return ZENOH_BYTES - val sep = s.indexOf(';') - val name = if (sep >= 0) s.substring(0, sep) else s - val rest = if (sep >= 0) s.substring(sep + 1) else "" - val id = nameToId[name] - return if (id != null) { - Encoding(id, rest.takeIf { it.isNotEmpty() }) - } else { - Encoding(CUSTOM_ENCODING_ID, s) - } + val (id, schema) = EncodingCodec.parse(s) + return Encoding(id, schema) } } - /** - * Preallocate the native form of this encoding. The returned - * [PinnedEncoding] IS this encoding (usable anywhere an [Encoding] is — - * e.g. `PutOptions.encoding` or `PublisherOptions.encoding`), but native - * calls pass its prebuilt handle instead of re-crossing the `(id, schema)` - * pair — for a custom schema-bearing encoding in a hot publish loop this - * removes the per-call string traffic. The native side only borrows the - * handle (a cheap clone per call), so the pinned encoding is reusable - * until [PinnedEncoding.close]s. - */ - fun pinned(): PinnedEncoding = PinnedEncoding(id, schema) - /** * Set a schema to this encoding. Zenoh does not define what a schema is and its semantics is left to the implementer. * E.g. a common schema for `text/plain` encoding is `utf-8`. - * - * For a custom encoding the schema slot holds `"name[;schema]"`; per - * Zenoh's rule the name part is preserved and only the schema part is - * replaced. */ fun withSchema(schema: String): Encoding { - if (id != CUSTOM_ENCODING_ID) return Encoding(id, schema) - val name = this.schema?.substringBefore(';') ?: "" - return Encoding( - id, - when { - name.isEmpty() -> schema - schema.isEmpty() -> name - else -> "$name;$schema" - }, - ) + val (nid, nschema) = EncodingCodec.withSchema(id, this.schema, schema) + return Encoding(nid, nschema) } - /** - * Canonical textual form, per Zenoh's rendering rule: the known name, - * `"name;schema"`, the bare schema for a custom encoding, or - * `"unknown(id)"` for an unrecognized id. - */ - override fun toString(): String { - val name = idToName[id] - return when { - name == null -> if (schema == null) "unknown($id)" else "unknown($id);$schema" - schema == null -> name - name.isEmpty() -> schema - else -> "$name;$schema" - } - } + /** Canonical textual form. */ + override fun toString(): String = EncodingCodec.render(id, schema) override fun equals(other: Any?): Boolean { if (this === other) return true - // `is`, not javaClass: a [PinnedEncoding] equals its plain counterpart - // (pinning is a transport optimization, not part of the value). - if (other !is Encoding) return false + if (javaClass != other?.javaClass) return false + other as Encoding return id == other.id && schema == other.schema } override fun hashCode(): Int = 31 * id + (schema?.hashCode() ?: 0) } - -/** - * The selector-tuple wire form of an optional encoding, matching the generated - * externs' dual-arm expansion: `sel == -1` = absent, arm `0` = the decomposed - * `(id, schema)` value, arm `1` = a pinned native handle (borrowed by the - * call). Computed once per call so a concurrent [PinnedEncoding.close] cannot - * desynchronize the tuple. - */ -internal class EncodingWire( - val sel: Int, - val id: Int?, - val schema: String?, - val handle: io.zenoh.jni.bytes.Encoding?, -) - -/** See [EncodingWire]. A closed [PinnedEncoding] falls back to arm 0. */ -internal fun Encoding?.forWire(): EncodingWire { - if (this == null) return EncodingWire(-1, null, null, null) - val h = (this as? PinnedEncoding)?.handleOrNull() - return if (h != null) { - EncodingWire(1, null, null, h) - } else { - EncodingWire(0, id, schema, null) - } -} - -/** - * An [Encoding] with its native form preallocated (see [Encoding.pinned]). - * Native calls pass the prebuilt handle (borrowed — cloned natively per call, - * an `Arc` bump) instead of re-crossing the `(id, schema)` pair, removing the - * per-call schema-string traffic in hot publish loops. - * - * Owns one native handle: [close] releases it deterministically (after which - * native calls fall back to the plain `(id, schema)` crossing); a finalizer - * backstops leaks. - */ -class PinnedEncoding internal constructor(id: Int, schema: String?) : - Encoding(id, schema), AutoCloseable { - - internal val handle: io.zenoh.jni.bytes.Encoding = - io.zenoh.jni.bytes.Encoding.newFromId(id, schema, io.zenoh.exceptions.throwZError0) - - /** Whether the pinned native form is still available. */ - internal fun handleOrNull(): io.zenoh.jni.bytes.Encoding? = - handle.takeIf { !it.isClosed() } - - override fun close() { - handle.close() - } - - @Suppress("removal") - protected fun finalize() { - close() - } -} diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/config/ZenohId.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/config/ZenohId.kt index 2bd980a7..5f3c34ae 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/config/ZenohId.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/config/ZenohId.kt @@ -15,6 +15,7 @@ package io.zenoh.config import io.zenoh.jni.config.ZenohId as JniZenohId +import io.zenoh.jni.config.zidString import kotlin.math.absoluteValue /** @@ -23,22 +24,11 @@ import kotlin.math.absoluteValue data class ZenohId internal constructor(internal val inner: JniZenohId) { /** - * The standard string form: the id bytes read as a little-endian integer, - * rendered as lowercase hex without leading zeros (`"0"` for the zero id) — - * Zenoh's own rule (uhlc `ID`'s `Debug`), implemented in pure Kotlin over - * the value-class bytes; correspondence with the native formatter is - * verified by `ZenohIdCorrespondenceTest`. + * The standard string form, computed in pure Kotlin by the shared + * bindings tier ([zidString]); correspondence with the native formatter + * is verified by `ZenohIdCorrespondenceTest`. */ - override fun toString(): String { - val hex = "0123456789abcdef" - val sb = StringBuilder(inner.bytes.size * 2) - for (i in inner.bytes.indices.reversed()) { - val b = inner.bytes[i].toInt() and 0xFF - sb.append(hex[b ushr 4]).append(hex[b and 0x0F]) - } - val s = sb.trimStart('0').toString() - return s.ifEmpty { "0" } - } + override fun toString(): String = inner.zidString() override fun equals(other: Any?): Boolean { if (this === other) return true diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt index 0d7e80c8..039dc1ac 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/pubsub/Publisher.kt @@ -17,7 +17,6 @@ package io.zenoh.pubsub import io.zenoh.* import io.zenoh.bytes.Encoding import io.zenoh.bytes.IntoZBytes -import io.zenoh.bytes.forWire import io.zenoh.bytes.ZBytes import io.zenoh.exceptions.ZError import io.zenoh.exceptions.throwZError @@ -134,15 +133,13 @@ class Publisher internal constructor( @Throws(ZError::class) private fun performPut(payload: IntoZBytes, encoding: Encoding?, attachment: IntoZBytes?) { val p = zPublisher ?: throw publisherNotValid - // `null` encoding = absent (`sel -1`): the publisher's native default - // applies. A [io.zenoh.bytes.PinnedEncoding] crosses as its handle. - val enc = encoding.forWire() + // `null` encoding = absent: the publisher's default encoding — set + // NATIVELY at declare time — applies, and no encoding data crosses. p.put( payload.into().bytes, - enc.sel, - enc.id, - enc.schema, - enc.handle, + encoding != null, + encoding?.id ?: 0, + encoding?.schema, attachment?.into()?.bytes, throwZError, ) diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt index ece2f61c..65fde3ca 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt @@ -17,7 +17,6 @@ package io.zenoh.query import io.zenoh.annotations.Unstable import io.zenoh.replyCallbackOf import io.zenoh.bytes.Encoding -import io.zenoh.bytes.forWire import io.zenoh.bytes.IntoZBytes import io.zenoh.bytes.ZBytes import io.zenoh.exceptions.ZError @@ -146,14 +145,13 @@ class Querier internal constructor(val keyExpr: KeyExpr, val qos: QoS, private v private fun resolveGetWithCallback(callback: Callback, options: GetOptions) { val q = zQuerier ?: throw ZError("Querier is not valid.") - val enc = options.encoding.forWire() + val enc = options.encoding q.get( options.parameters?.toString(), options.payload?.into()?.bytes, - enc.sel, - enc.id, - enc.schema, - enc.handle, + enc != null, + enc?.id ?: 0, + enc?.schema, options.attachment?.into()?.bytes, replyCallbackOf { callback.run(it) }, { }, @@ -163,14 +161,13 @@ class Querier internal constructor(val keyExpr: KeyExpr, val qos: QoS, private v private fun resolveGetWithHandler(handler: Handler, options: GetOptions): R { val q = zQuerier ?: throw ZError("Querier is not valid.") - val enc = options.encoding.forWire() + val enc = options.encoding q.get( options.parameters?.toString(), options.payload?.into()?.bytes, - enc.sel, - enc.id, - enc.schema, - enc.handle, + enc != null, + enc?.id ?: 0, + enc?.schema, options.attachment?.into()?.bytes, replyCallbackOf { 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 528ee59a..895540bb 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt @@ -16,7 +16,6 @@ package io.zenoh.query import io.zenoh.ZenohType import io.zenoh.bytes.Encoding -import io.zenoh.bytes.forWire import io.zenoh.bytes.IntoZBytes import io.zenoh.bytes.ZBytes import io.zenoh.exceptions.ZError @@ -93,14 +92,13 @@ class Query internal constructor( @JvmOverloads fun reply(keyExpr: KeyExpr, payload: IntoZBytes, options: ReplyOptions = ReplyOptions()) { val q = zQuery ?: throw ZError("Query is invalid") - val enc = options.encoding.forWire() + val enc = options.encoding q.replySuccess( keyExpr.flat, payload.into().bytes, - enc.sel, - enc.id, - enc.schema, - enc.handle, + enc != null, + enc?.id ?: 0, + enc?.schema, options.timeStamp?.ntpValue(), options.attachment?.into()?.bytes, options.express, @@ -157,8 +155,8 @@ class Query internal constructor( @Throws(ZError::class) fun replyErr(message: IntoZBytes, options: ReplyErrOptions = ReplyErrOptions()) { val q = zQuery ?: throw ZError("Query is invalid") - val enc = options.encoding.forWire() - q.replyError(message.into().bytes, enc.sel, enc.id, enc.schema, enc.handle, throwZError) + val enc = options.encoding + q.replyError(message.into().bytes, enc != null, enc?.id ?: 0, enc?.schema, throwZError) q.close() zQuery = null } diff --git a/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingCorrespondenceTest.kt b/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingCorrespondenceTest.kt index 149cdb87..8c3468db 100644 --- a/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingCorrespondenceTest.kt +++ b/zenoh-java/src/jvmTest/kotlin/io/zenoh/EncodingCorrespondenceTest.kt @@ -103,9 +103,8 @@ class EncodingCorrespondenceTest { ) for (base in bases) { val pure = base.withSchema("new-schema") - // The `e` param crosses via the dual-arm expansion: arm 0 = the - // decomposed (id, schema) value. - val w = JniEncoding.newWithSchema(0, base.id, base.schema, null, "new-schema", throwZError0) + // The `e` param crosses as its decomposed (id, schema) value. + val w = JniEncoding.newWithSchema(base.id, base.schema, "new-schema", throwZError0) val nativeStr = try { w.toStr(throwZError0) } finally { diff --git a/zenoh-java/src/jvmTest/kotlin/io/zenoh/PinnedEncodingTest.kt b/zenoh-java/src/jvmTest/kotlin/io/zenoh/PinnedEncodingTest.kt deleted file mode 100644 index 1db5ae32..00000000 --- a/zenoh-java/src/jvmTest/kotlin/io/zenoh/PinnedEncodingTest.kt +++ /dev/null @@ -1,87 +0,0 @@ -// -// Copyright (c) 2023 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.Encoding -import io.zenoh.bytes.ZBytes -import io.zenoh.keyexpr.KeyExpr -import io.zenoh.pubsub.PublisherOptions -import io.zenoh.pubsub.PutOptions -import io.zenoh.sample.Sample -import org.junit.Assert.assertEquals -import org.junit.Test - -/** - * End-to-end tests of the two encoding fast paths: - * * the publisher's **default encoding**, set natively once at declare time - * (plain `put`s cross no encoding data at all), and - * * a **pinned** encoding ([Encoding.pinned]) passed per call as a - * preallocated native handle (no schema-string crossing per put). - */ -class PinnedEncodingTest { - - private val custom = Encoding.from("application/custom;my-schema") - - @Test - fun publisherDefaultEncodingAppliesNatively() { - val session = Zenoh.open(Config.loadDefault()) - val keyExpr = KeyExpr.tryFrom("example/testing/pinned/default") - val received = mutableListOf() - val subscriber = session.declareSubscriber(keyExpr) { received.add(it) } - - val options = PublisherOptions() - options.encoding = custom - val publisher = session.declarePublisher(keyExpr, options) - // Plain put: NO encoding crosses — the native publisher default applies. - publisher.put(ZBytes.from("hello")) - Thread.sleep(500) - - publisher.close() - subscriber.close() - session.close() - assertEquals(1, received.size) - assertEquals(custom, received[0].encoding) - } - - @Test - fun pinnedEncodingOverridesPerPut() { - val session = Zenoh.open(Config.loadDefault()) - val keyExpr = KeyExpr.tryFrom("example/testing/pinned/override") - val received = mutableListOf() - val subscriber = session.declareSubscriber(keyExpr) { received.add(it) } - val publisher = session.declarePublisher(keyExpr) - - val pinned = custom.pinned() - val options = PutOptions() - options.encoding = pinned - // The pinned handle is BORROWED per call — reusable across the loop. - repeat(3) { publisher.put(ZBytes.from("hello #$it"), options) } - Thread.sleep(500) - assertEquals(3, received.size) - received.forEach { assertEquals(custom, it.encoding) } - - // After close() the pinned form falls back to the plain (id, schema) - // crossing — same value, still correct. - pinned.close() - publisher.put(ZBytes.from("after close"), options) - Thread.sleep(500) - - publisher.close() - subscriber.close() - session.close() - assertEquals(4, received.size) - assertEquals(custom, received[3].encoding) - } -} diff --git a/zenoh-java/src/jvmTest/kotlin/io/zenoh/PublisherEncodingTest.kt b/zenoh-java/src/jvmTest/kotlin/io/zenoh/PublisherEncodingTest.kt new file mode 100644 index 00000000..b53daf69 --- /dev/null +++ b/zenoh-java/src/jvmTest/kotlin/io/zenoh/PublisherEncodingTest.kt @@ -0,0 +1,62 @@ +// +// Copyright (c) 2023 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.Encoding +import io.zenoh.bytes.ZBytes +import io.zenoh.keyexpr.KeyExpr +import io.zenoh.pubsub.PublisherOptions +import io.zenoh.pubsub.PutOptions +import io.zenoh.sample.Sample +import org.junit.Assert.assertEquals +import org.junit.Test + +/** + * End-to-end test of the publisher's **default encoding**: set natively once + * at declare time ([PublisherOptions.encoding]), applied by zenoh itself to + * every plain `put` — which crosses **no encoding data at all** — and + * overridden per call via [PutOptions.encoding]. + */ +class PublisherEncodingTest { + + private val custom = Encoding.from("application/custom;my-schema") + + @Test + fun publisherDefaultAppliesNativelyAndPerPutOverrides() { + val session = Zenoh.open(Config.loadDefault()) + val keyExpr = KeyExpr.tryFrom("example/testing/publisher/encoding") + val received = mutableListOf() + val subscriber = session.declareSubscriber(keyExpr) { received.add(it) } + + val options = PublisherOptions() + options.encoding = custom + val publisher = session.declarePublisher(keyExpr, options) + + // Plain put: NO encoding crosses — the native publisher default applies. + publisher.put(ZBytes.from("default")) + // Per-put override still works. + val putOptions = PutOptions() + putOptions.encoding = Encoding.TEXT_PLAIN + publisher.put(ZBytes.from("override"), putOptions) + Thread.sleep(500) + + publisher.close() + subscriber.close() + session.close() + assertEquals(2, received.size) + assertEquals(custom, received[0].encoding) + assertEquals(Encoding.TEXT_PLAIN, received[1].encoding) + } +}