diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/Config.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/Config.kt index dcac0a62..c837c772 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/Config.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/Config.kt @@ -34,13 +34,14 @@ import java.nio.file.Path * * A default configuration can be loaded using [Config.loadDefault]. * - * The native configuration a Config wraps is consumed by [Zenoh.open] (or [Zenoh.scout]); a Config that is - * never used is released by the garbage-collection backstop. + * [Zenoh.open] (and [Zenoh.scout]) copy the native configuration rather than taking it, so a Config stays + * usable afterwards and its native memory is released by [close] — or, if never closed, by the + * garbage-collection backstop. * * Visit the [default configuration](https://github.com/eclipse-zenoh/zenoh/blob/main/DEFAULT_CONFIG.json5) for more * information on the Zenoh config parameters. */ -class Config internal constructor(internal val zConfig: JniConfig) { +class Config internal constructor(internal val zConfig: JniConfig) : AutoCloseable { companion object { @@ -145,4 +146,11 @@ class Config internal constructor(internal val zConfig: JniConfig) { @Throws(ZError::class) fun insertJson5(key: String, value: String) = zConfig.insertJson5(key, value, throwZError0, throwZError) + + /** + * Releases the native configuration. Idempotent. Any later use of this + * Config — [getJson], [insertJson5], or opening a session with it — fails + * with a [ZError] reporting a closed handle. + */ + override fun close() = zConfig.close() } diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/ZBytes.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/ZBytes.kt index 9d2927d4..7618a32b 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/ZBytes.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/bytes/ZBytes.kt @@ -14,6 +14,7 @@ package io.zenoh.bytes +import io.zenoh.exceptions.ZError import io.zenoh.exceptions.throwZError0 import io.zenoh.jni.bytes.ZBytes as JniZBytes @@ -44,13 +45,17 @@ import io.zenoh.jni.bytes.ZBytes as JniZBytes * garbage-collection backstop — payloads are the per-message hot path, * and registering a GC cleaner per message measured −23% throughput at * small payload sizes. In callback-based subscribers/queryables, access - * (or discard) payloads and attachments you care about; unread ones on + * (or [discard]) payloads and attachments you care about; unread ones on * dropped samples are the one place native memory can be retained. + * + * A payload you are not going to read can be released explicitly with + * [discard] / [close] (try-with-resources in Java, `use` in Kotlin) — see + * [discard] for what that means in each state. */ class ZBytes private constructor( initialBytes: ByteArray?, private var handle: JniZBytes?, -) : IntoZBytes { +) : IntoZBytes, AutoCloseable { /** * The materialized bytes, `null` until a handle-backed ZBytes is read. @@ -68,11 +73,13 @@ class ZBytes private constructor( * LAZILY on first access — one borrow-copy out of the native buffer via * `zZbytesToBytes` — then closes the native handle (forward-extraction * rule: the handle is delivered eagerly, the heavy bytes on demand). + * + * Throws [ZError] if this ZBytes was [discard]ed before being read. */ internal val bytes: ByteArray get() = eager ?: synchronized(this) { eager ?: run { - val h = handle!! + val h = handle ?: throw ZError("ZBytes was discarded before its bytes were read.") val b = h.toBytes(throwZError0) eager = b handle = null @@ -81,6 +88,26 @@ class ZBytes private constructor( } } + /** + * Releases the native buffer of an **unread** received payload without + * copying it out. Idempotent, and safe against a concurrent read: it takes + * the same monitor as the lazy materialization, so either the read wins + * (the bytes are materialized and stay readable) or the discard wins (any + * later read fails with [ZError]). + * + * On a ZBytes whose bytes are already available — one built by [from], or a + * received one already read — there is no native memory left to release and + * this does nothing: the bytes stay readable. + */ + @Synchronized + fun discard() { + handle?.close() + handle = null + } + + /** Equivalent to [discard]; lets a payload be used with `use` / try-with-resources. */ + override fun close() = discard() + companion object { /** diff --git a/zenoh-java/src/commonMain/kotlin/io/zenoh/session/SessionDeclaration.kt b/zenoh-java/src/commonMain/kotlin/io/zenoh/session/SessionDeclaration.kt index 2c3a9224..2c52c2b7 100644 --- a/zenoh-java/src/commonMain/kotlin/io/zenoh/session/SessionDeclaration.kt +++ b/zenoh-java/src/commonMain/kotlin/io/zenoh/session/SessionDeclaration.kt @@ -20,8 +20,15 @@ package io.zenoh.session * A session declaration is either a [io.zenoh.pubsub.Publisher], * a [io.zenoh.pubsub.Subscriber] or a [io.zenoh.query.Queryable] declared from a [io.zenoh.Session]. */ -interface SessionDeclaration { +interface SessionDeclaration : AutoCloseable { /** Undeclare a declaration. No further operations should be performed after calling this function. */ fun undeclare() + + /** + * Equivalent to [undeclare], so a declaration held through this interface can be used with + * try-with-resources (Java) or `use` (Kotlin). Every implementation in this SDK overrides it + * with the same behaviour. + */ + override fun close() = undeclare() } diff --git a/zenoh-java/src/jvmTest/java/io/zenoh/ConfigTest.java b/zenoh-java/src/jvmTest/java/io/zenoh/ConfigTest.java index 990c5452..87757c2c 100644 --- a/zenoh-java/src/jvmTest/java/io/zenoh/ConfigTest.java +++ b/zenoh-java/src/jvmTest/java/io/zenoh/ConfigTest.java @@ -349,4 +349,14 @@ public void insertIllFormattedJson5ShouldFailTest() throws ZError { assertTrue(retrievedEndpoints.contains("8.8.8.8")); } + + @Test + public void closeIsIdempotentAndInvalidatesTheConfigTest() { + Config config = Config.loadDefault(); + config.close(); + config.close(); + + assertThrows(ZError.class, () -> config.getJson("mode")); + assertThrows(ZError.class, () -> config.insertJson5("mode", "\"peer\"")); + } } diff --git a/zenoh-java/src/jvmTest/kotlin/io/zenoh/ZBytesLifecycleTest.kt b/zenoh-java/src/jvmTest/kotlin/io/zenoh/ZBytesLifecycleTest.kt new file mode 100644 index 00000000..e6835849 --- /dev/null +++ b/zenoh-java/src/jvmTest/kotlin/io/zenoh/ZBytesLifecycleTest.kt @@ -0,0 +1,120 @@ +// +// Copyright (c) 2026 ZettaScale Technology +// +// This program and the accompanying materials are made available under the +// terms of the Eclipse Public License 2.0 which is available at +// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0 +// which is available at https://www.apache.org/licenses/LICENSE-2.0. +// +// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0 +// +// Contributors: +// ZettaScale Zenoh Team, +// + +package io.zenoh + +import io.zenoh.bytes.ZBytes +import io.zenoh.exceptions.ZError +import io.zenoh.exceptions.throwZError0 +import io.zenoh.session.SessionDeclaration +import java.util.concurrent.CyclicBarrier +import io.zenoh.jni.bytes.ZBytes as JniZBytes +import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertThrows +import org.junit.Assert.assertTrue +import org.junit.Test + +/** + * The explicit lifecycle of a received payload: it owns a native buffer until + * its bytes are materialized (or discarded), and — unlike every other + * handle-owning class here — has no garbage-collection backstop, so + * [ZBytes.discard]/[ZBytes.close] is the only way to release one that is never + * read. A handle-backed instance is built here directly rather than through a + * session: what is under test is the wrapper's state machine, not delivery. + */ +class ZBytesLifecycleTest { + + private val payload = "the payload".encodeToByteArray() + + private fun received(): ZBytes = ZBytes.fromHandle(JniZBytes.newFromVec(payload, throwZError0)) + + @Test + fun discardBeforeMaterializationReleasesTheBufferAndInvalidatesTheBytes() { + val zbytes = received() + zbytes.discard() + + assertThrows(ZError::class.java) { zbytes.toBytes() } + } + + @Test + fun discardAfterMaterializationKeepsTheBytesReadable() { + val zbytes = received() + assertArrayEquals(payload, zbytes.toBytes()) + + zbytes.discard() + + assertArrayEquals(payload, zbytes.toBytes()) + } + + @Test + fun discardAndCloseAreIdempotent() { + val unread = received() + unread.close() + unread.close() + unread.discard() + + val read = received() + read.toBytes() + read.close() + read.discard() + } + + @Test + fun closingAValueZBytesIsANoOp() { + val zbytes = ZBytes.from(payload) + zbytes.close() + + assertArrayEquals(payload, zbytes.toBytes()) + } + + @Test + fun concurrentReadAndCloseHaveDeterministicOutcomes() { + repeat(500) { + val zbytes = received() + val barrier = CyclicBarrier(2) + var failure: Throwable? = null + + val reader = Thread { + barrier.await() + try { + assertArrayEquals(payload, zbytes.toBytes()) + } catch (e: ZError) { + // The discard won the race: a documented outcome. + } catch (e: Throwable) { + failure = e + } + } + val closer = Thread { + barrier.await() + try { + zbytes.close() + } catch (e: Throwable) { + failure = e + } + } + + reader.start() + closer.start() + reader.join() + closer.join() + + failure?.let { throw it } + } + } + + @Test + fun aSessionDeclarationIsAutoCloseable() { + assertTrue(AutoCloseable::class.java.isAssignableFrom(SessionDeclaration::class.java)) + } +}