Skip to content

Commit a6d64d4

Browse files
milyinmilyin
committed
native handle usage
1 parent b6b8cb2 commit a6d64d4

10 files changed

Lines changed: 88 additions & 82 deletions

File tree

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

Lines changed: 17 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -400,7 +400,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
400400
@Throws(ZError::class)
401401
fun declareKeyExpr(keyExpr: String): KeyExpr {
402402
return jniSession?.run {
403-
val ke = KeyExpr(keyExpr, io.zenoh.jni.NativeHandle(declareKeyExpr(keyExpr)))
403+
val ke = KeyExpr(keyExpr, declareKeyExpr(keyExpr))
404404
strongDeclarations.add(ke)
405405
ke
406406
} ?: throw sessionClosedException
@@ -420,12 +420,12 @@ class Session private constructor(private val config: Config) : AutoCloseable {
420420
val js = jniSession ?: throw sessionClosedException
421421
val nh = keyExpr.jniKeyExpr
422422
?: throw ZError("Attempting to undeclare a non declared key expression.")
423-
// `NativeHandle.consume` atomically takes the Long under the
424-
// write lock and nulls the field, so the generator-emitted
425-
// `undeclareKeyExpr` (Rust: `Arc::unwrap_or_clone(Arc::from_raw)`)
426-
// gets exactly-once handoff. By-value transfer IS the
427-
// destruction; no separate `keyExprDrop` is needed.
428-
nh.consume { handle -> js.undeclareKeyExpr(handle) }
423+
// `JNISession.undeclareKeyExpr` routes through the generator-
424+
// emitted wrapper, which does `keyExpr.consume { ... }` — the
425+
// NativeHandle's write lock provides exactly-once handoff and
426+
// by-value Rust transfer IS the destruction (no separate
427+
// `keyExprDrop` needed).
428+
js.undeclareKeyExpr(nh)
429429
}
430430

431431
/**
@@ -605,7 +605,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
605605
options.priority,
606606
options.encoding,
607607
declarePublisher(
608-
keyExpr.jniKeyExprHandle,
608+
keyExpr.jniKeyExpr,
609609
keyExpr.keyExpr,
610610
options.congestionControl.value,
611611
options.priority.value,
@@ -626,7 +626,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
626626
val subCallback = JNISampleCallback { sample ->
627627
handler.handle(sample.toPublic())
628628
}
629-
val subscriber = HandlerSubscriber(keyExpr, declareSubscriber(keyExpr.jniKeyExprHandle, keyExpr.keyExpr, subCallback, handler::onClose), handler.receiver())
629+
val subscriber = HandlerSubscriber(keyExpr, declareSubscriber(keyExpr.jniKeyExpr, keyExpr.keyExpr, subCallback, handler::onClose), handler.receiver())
630630
strongDeclarations.add(subscriber)
631631
subscriber
632632
} ?: throw (sessionClosedException)
@@ -640,7 +640,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
640640
val subCallback = JNISampleCallback { sample ->
641641
callback.run(sample.toPublic())
642642
}
643-
val subscriber = CallbackSubscriber(keyExpr, declareSubscriber(keyExpr.jniKeyExprHandle, keyExpr.keyExpr, subCallback, fun() {}))
643+
val subscriber = CallbackSubscriber(keyExpr, declareSubscriber(keyExpr.jniKeyExpr, keyExpr.keyExpr, subCallback, fun() {}))
644644
strongDeclarations.add(subscriber)
645645
subscriber
646646
} ?: throw (sessionClosedException)
@@ -668,7 +668,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
668668
)
669669
)
670670
}
671-
val queryable = HandlerQueryable(keyExpr, declareQueryable(keyExpr.jniKeyExprHandle, keyExpr.keyExpr, queryCallback, handler::onClose, options.complete), handler.receiver())
671+
val queryable = HandlerQueryable(keyExpr, declareQueryable(keyExpr.jniKeyExpr, keyExpr.keyExpr, queryCallback, handler::onClose, options.complete), handler.receiver())
672672
strongDeclarations.add(queryable)
673673
queryable
674674
} ?: throw (sessionClosedException)
@@ -696,7 +696,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
696696
)
697697
)
698698
}
699-
val queryable = CallbackQueryable(keyExpr, declareQueryable(keyExpr.jniKeyExprHandle, keyExpr.keyExpr, queryCallback, fun() {}, options.complete))
699+
val queryable = CallbackQueryable(keyExpr, declareQueryable(keyExpr.jniKeyExpr, keyExpr.keyExpr, queryCallback, fun() {}, options.complete))
700700
strongDeclarations.add(queryable)
701701
queryable
702702
} ?: throw (sessionClosedException)
@@ -712,7 +712,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
712712
keyExpr,
713713
QoS(congestionControl = options.congestionControl, priority = options.priority, express = options.express),
714714
declareQuerier(
715-
keyExpr.jniKeyExprHandle,
715+
keyExpr.jniKeyExpr,
716716
keyExpr.keyExpr,
717717
options.target.ordinal,
718718
options.consolidationMode.ordinal,
@@ -757,7 +757,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
757757
}
758758
val sel = selector.into()
759759
get(
760-
sel.keyExpr.jniKeyExprHandle,
760+
sel.keyExpr.jniKeyExpr,
761761
sel.keyExpr.keyExpr,
762762
sel.parameters?.toString(),
763763
getCallback,
@@ -806,7 +806,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
806806
}
807807
val sel = selector.into()
808808
get(
809-
sel.keyExpr.jniKeyExprHandle,
809+
sel.keyExpr.jniKeyExpr,
810810
sel.keyExpr.keyExpr,
811811
sel.parameters?.toString(),
812812
getCallback,
@@ -830,7 +830,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
830830
jniSession?.run {
831831
val encoding = (putOptions.encoding ?: Encoding.defaultEncoding()).toJni()
832832
put(
833-
keyExpr.jniKeyExprHandle,
833+
keyExpr.jniKeyExpr,
834834
keyExpr.keyExpr,
835835
payload.into().bytes,
836836
encoding,
@@ -847,7 +847,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
847847
internal fun resolveDelete(keyExpr: KeyExpr, deleteOptions: DeleteOptions) {
848848
jniSession?.run {
849849
delete(
850-
keyExpr.jniKeyExprHandle,
850+
keyExpr.jniKeyExpr,
851851
keyExpr.keyExpr,
852852
deleteOptions.congestionControl.value,
853853
deleteOptions.priority.value,

‎zenoh-java/src/commonMain/kotlin/io/zenoh/keyexpr/KeyExpr.kt‎

Lines changed: 14 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -71,23 +71,16 @@ class KeyExpr internal constructor(
7171
internal val keyExpr: String,
7272
/**
7373
* Native `Arc<KeyExpr<'static>>` handle when this expression was
74-
* session-declared; `null` otherwise. Wrapped in a [NativeHandle]
75-
* so the consume site (`Session.undeclare(keyExpr)`) gets atomic
76-
* exactly-once handoff via [NativeHandle.consume]; pure-borrow
77-
* read sites (`intersects`, `includes`, `declarePublisher`, …) go
78-
* through [jniKeyExprHandle] which is a non-locking snapshot.
74+
* session-declared; `null` for string-only key expressions. The
75+
* borrow-read JNI surface (`keyExprArg`, `JNIWrappers.*`) accepts
76+
* a [NativeHandle] directly so the inner `Long` never escapes
77+
* past the JNI boundary; the consume site
78+
* (`Session.undeclare(keyExpr)`) routes through
79+
* [NativeHandle.consume] for atomic exactly-once handoff.
7980
*/
8081
internal var jniKeyExpr: NativeHandle? = null,
8182
) : AutoCloseable, IntoSelector, SessionDeclaration {
8283

83-
/**
84-
* Snapshot of the current native handle Long, or `null` if the
85-
* key-expression is string-only or has been undeclared. Backed by
86-
* [NativeHandle.peek] (volatile read, no lock).
87-
*/
88-
internal val jniKeyExprHandle: Long?
89-
get() = jniKeyExpr?.peek()?.takeIf { it != 0L }
90-
9184
companion object {
9285

9386
/**
@@ -128,7 +121,7 @@ class KeyExpr internal constructor(
128121
*/
129122
@Throws(ZError::class)
130123
fun intersects(other: KeyExpr): Boolean =
131-
keyExprIntersects(jniKeyExprHandle, keyExpr, other.jniKeyExprHandle, other.keyExpr)
124+
keyExprIntersects(jniKeyExpr, keyExpr, other.jniKeyExpr, other.keyExpr)
132125

133126
/**
134127
* Includes operation. This method returns `true` when all the keys defined by `other` also belong to the set
@@ -137,7 +130,7 @@ class KeyExpr internal constructor(
137130
*/
138131
@Throws(ZError::class)
139132
fun includes(other: KeyExpr): Boolean =
140-
keyExprIncludes(jniKeyExprHandle, keyExpr, other.jniKeyExprHandle, other.keyExpr)
133+
keyExprIncludes(jniKeyExpr, keyExpr, other.jniKeyExpr, other.keyExpr)
141134

142135
/**
143136
* Returns the relation between 'this' and other from 'this''s point of view ([SetIntersectionLevel.INCLUDES]
@@ -147,7 +140,7 @@ class KeyExpr internal constructor(
147140
@Throws(ZError::class)
148141
fun relationTo(other: KeyExpr): SetIntersectionLevel =
149142
SetIntersectionLevel.fromInt(
150-
keyExprRelationTo(jniKeyExprHandle, keyExpr, other.jniKeyExprHandle, other.keyExpr)
143+
keyExprRelationTo(jniKeyExpr, keyExpr, other.jniKeyExpr, other.keyExpr)
151144
)
152145

153146
/**
@@ -156,8 +149,8 @@ class KeyExpr internal constructor(
156149
*/
157150
@Throws(ZError::class)
158151
fun join(other: String): KeyExpr {
159-
val r = keyExprJoin(jniKeyExprHandle, keyExpr, other)
160-
return KeyExpr(r.string, NativeHandle(r.handle))
152+
val r = keyExprJoin(jniKeyExpr, keyExpr, other)
153+
return KeyExpr(r.string, r.handle)
161154
}
162155

163156
/**
@@ -166,8 +159,8 @@ class KeyExpr internal constructor(
166159
*/
167160
@Throws(ZError::class)
168161
fun concat(other: String): KeyExpr {
169-
val r = keyExprConcat(jniKeyExprHandle, keyExpr, other)
170-
return KeyExpr(r.string, NativeHandle(r.handle))
162+
val r = keyExprConcat(jniKeyExpr, keyExpr, other)
163+
return KeyExpr(r.string, r.handle)
171164
}
172165

173166
override fun toString(): String = keyExpr
@@ -187,7 +180,7 @@ class KeyExpr internal constructor(
187180
* operations on it, but without the inner optimizations.
188181
*/
189182
override fun undeclare() {
190-
jniKeyExpr?.close { keyExprDrop(it) }
183+
jniKeyExpr?.let { keyExprDrop(it) }
191184
}
192185

193186
override fun into(): Selector = Selector(this)

‎zenoh-java/src/commonMain/kotlin/io/zenoh/liveliness/Liveliness.kt‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ class Liveliness internal constructor(private val session: Session) {
6060
@Throws(ZError::class)
6161
fun declareToken(keyExpr: KeyExpr): LivelinessToken {
6262
val jniSession = session.jniSession ?: throw Session.sessionClosedException
63-
return LivelinessToken(jniSession.declareLivelinessToken(keyExpr.jniKeyExprHandle, keyExpr.keyExpr))
63+
return LivelinessToken(jniSession.declareLivelinessToken(keyExpr.jniKeyExpr, keyExpr.keyExpr))
6464
}
6565

6666
/**
@@ -79,7 +79,7 @@ class Liveliness internal constructor(private val session: Session) {
7979
val handler = BlockingQueueHandler<Reply>(LinkedBlockingDeque())
8080
val getCallback = buildGetCallback(handler::handle)
8181
jniSession.livelinessGet(
82-
keyExpr.jniKeyExprHandle,
82+
keyExpr.jniKeyExpr,
8383
keyExpr.keyExpr,
8484
getCallback,
8585
timeout.toMillis(),
@@ -102,7 +102,7 @@ class Liveliness internal constructor(private val session: Session) {
102102
) {
103103
val jniSession = session.jniSession ?: throw Session.sessionClosedException
104104
jniSession.livelinessGet(
105-
keyExpr.jniKeyExprHandle,
105+
keyExpr.jniKeyExpr,
106106
keyExpr.keyExpr,
107107
buildGetCallback(callback),
108108
timeout.toMillis(),
@@ -125,7 +125,7 @@ class Liveliness internal constructor(private val session: Session) {
125125
): R {
126126
val jniSession = session.jniSession ?: throw Session.sessionClosedException
127127
jniSession.livelinessGet(
128-
keyExpr.jniKeyExprHandle,
128+
keyExpr.jniKeyExpr,
129129
keyExpr.keyExpr,
130130
buildGetCallback(handler::handle),
131131
timeout.toMillis(),
@@ -171,7 +171,7 @@ class Liveliness internal constructor(private val session: Session) {
171171
val handler = BlockingQueueHandler<Sample>(LinkedBlockingDeque())
172172
val jniSession = session.jniSession ?: throw Session.sessionClosedException
173173
val subCallback = buildSubscriberCallback(handler::handle)
174-
return HandlerSubscriber(keyExpr, jniSession.declareLivelinessSubscriber(keyExpr.jniKeyExprHandle, keyExpr.keyExpr, subCallback, options.history, handler::onClose), handler.receiver())
174+
return HandlerSubscriber(keyExpr, jniSession.declareLivelinessSubscriber(keyExpr.jniKeyExpr, keyExpr.keyExpr, subCallback, options.history, handler::onClose), handler.receiver())
175175
}
176176

177177
/**
@@ -190,7 +190,7 @@ class Liveliness internal constructor(private val session: Session) {
190190
): CallbackSubscriber {
191191
val jniSession = session.jniSession ?: throw Session.sessionClosedException
192192
val subCallback = buildSubscriberCallback(callback)
193-
return CallbackSubscriber(keyExpr, jniSession.declareLivelinessSubscriber(keyExpr.jniKeyExprHandle, keyExpr.keyExpr, subCallback, options.history, fun() {}))
193+
return CallbackSubscriber(keyExpr, jniSession.declareLivelinessSubscriber(keyExpr.jniKeyExpr, keyExpr.keyExpr, subCallback, options.history, fun() {}))
194194
}
195195

196196
/**
@@ -210,7 +210,7 @@ class Liveliness internal constructor(private val session: Session) {
210210
): HandlerSubscriber<R> {
211211
val jniSession = session.jniSession ?: throw Session.sessionClosedException
212212
val subCallback = buildSubscriberCallback(handler::handle)
213-
return HandlerSubscriber(keyExpr, jniSession.declareLivelinessSubscriber(keyExpr.jniKeyExprHandle, keyExpr.keyExpr, subCallback, options.history, handler::onClose), handler.receiver())
213+
return HandlerSubscriber(keyExpr, jniSession.declareLivelinessSubscriber(keyExpr.jniKeyExpr, keyExpr.keyExpr, subCallback, options.history, handler::onClose), handler.receiver())
214214
}
215215

216216
private fun buildSubscriberCallback(callback: Callback<Sample>): JNISampleCallback =

‎zenoh-java/src/commonMain/kotlin/io/zenoh/query/Querier.kt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -172,7 +172,7 @@ class Querier internal constructor(val keyExpr: KeyExpr, val qos: QoS, private v
172172
callback.run(reply)
173173
}
174174
jni.get(
175-
keyExpr.jniKeyExprHandle,
175+
keyExpr.jniKeyExpr,
176176
keyExpr.keyExpr,
177177
options.parameters?.toString(),
178178
getCallback,
@@ -206,7 +206,7 @@ class Querier internal constructor(val keyExpr: KeyExpr, val qos: QoS, private v
206206
handler.handle(reply)
207207
}
208208
jni.get(
209-
keyExpr.jniKeyExprHandle,
209+
keyExpr.jniKeyExpr,
210210
keyExpr.keyExpr,
211211
options.parameters?.toString(),
212212
getCallback,

‎zenoh-java/src/commonMain/kotlin/io/zenoh/query/Query.kt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ class Query internal constructor(
6868
val timestampEnabled = timestamp != null
6969
jniQuery?.apply {
7070
replySuccess(
71-
keyExpr.jniKeyExprHandle,
71+
keyExpr.jniKeyExpr,
7272
keyExpr.keyExpr,
7373
zbytes.bytes,
7474
(encoding ?: Encoding.defaultEncoding()).toJni(),
@@ -107,7 +107,7 @@ class Query internal constructor(
107107
val timestampEnabled = timestamp != null
108108
jniQuery?.apply {
109109
replyDelete(
110-
keyExpr.jniKeyExprHandle,
110+
keyExpr.jniKeyExpr,
111111
keyExpr.keyExpr,
112112
timestampEnabled,
113113
if (timestampEnabled) timestamp!!.ntpValue() else 0,

‎zenoh-jni-runtime/src/commonMain/kotlin/io/zenoh/jni/JNIKeyExpr.kt‎

Lines changed: 20 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -32,8 +32,15 @@ import io.zenoh.exceptions.ZError
3232
* Pick the declared handle if present, else the raw string. Returns
3333
* a JVM `Object` (boxed `java.lang.Long` or `java.lang.String`) which
3434
* the native dispatching converter resolves at runtime.
35+
*
36+
* Takes a [NativeHandle] (not a raw `Long`) so callers don't see the
37+
* inner pointer. The peek is unlocked — the borrow's Rust side runs
38+
* `Arc::increment_strong_count` and a concurrent close of this
39+
* keyExpr is the same race window the broader JNI layer already
40+
* tolerates for borrow-style calls.
3541
*/
36-
fun keyExprArg(handle: Long?, str: String): Any = handle ?: str
42+
fun keyExprArg(handle: NativeHandle?, str: String): Any =
43+
handle?.peek()?.takeIf { it != 0L } ?: str
3744

3845
@Throws(ZError::class)
3946
fun keyExprTryFrom(keyExpr: String): String = JNIWrappers.tryFrom(keyExpr)
@@ -42,38 +49,38 @@ fun keyExprTryFrom(keyExpr: String): String = JNIWrappers.tryFrom(keyExpr)
4249
fun keyExprAutocanonize(keyExpr: String): String = JNIWrappers.autocanonize(keyExpr)
4350

4451
@Throws(ZError::class)
45-
fun keyExprIntersects(a: Long?, aStr: String, b: Long?, bStr: String): Boolean =
52+
fun keyExprIntersects(a: NativeHandle?, aStr: String, b: NativeHandle?, bStr: String): Boolean =
4653
JNIWrappers.intersects(keyExprArg(a, aStr), keyExprArg(b, bStr))
4754

4855
@Throws(ZError::class)
49-
fun keyExprIncludes(a: Long?, aStr: String, b: Long?, bStr: String): Boolean =
56+
fun keyExprIncludes(a: NativeHandle?, aStr: String, b: NativeHandle?, bStr: String): Boolean =
5057
JNIWrappers.includes(keyExprArg(a, aStr), keyExprArg(b, bStr))
5158

5259
@Throws(ZError::class)
53-
fun keyExprRelationTo(a: Long?, aStr: String, b: Long?, bStr: String): Int =
60+
fun keyExprRelationTo(a: NativeHandle?, aStr: String, b: NativeHandle?, bStr: String): Int =
5461
JNIWrappers.relationTo(keyExprArg(a, aStr), keyExprArg(b, bStr))
5562

5663
/** Result of a join/concat: `(handle, canonicalString)`. */
57-
data class KeyExprResult(val handle: Long, val string: String)
64+
data class KeyExprResult(val handle: NativeHandle, val string: String)
5865

5966
@Throws(ZError::class)
60-
fun keyExprJoin(a: Long?, aStr: String, other: String): KeyExprResult {
61-
val handle = JNIWrappers.join(keyExprArg(a, aStr), other).peek()
67+
fun keyExprJoin(a: NativeHandle?, aStr: String, other: String): KeyExprResult {
68+
val handle = JNIWrappers.join(keyExprArg(a, aStr), other)
6269
// Match Rust's `KeyExpr::join` formatting: "{self}/{other}".
6370
return KeyExprResult(handle, "$aStr/$other")
6471
}
6572

6673
@Throws(ZError::class)
67-
fun keyExprConcat(a: Long?, aStr: String, other: String): KeyExprResult {
68-
val handle = JNIWrappers.concat(keyExprArg(a, aStr), other).peek()
74+
fun keyExprConcat(a: NativeHandle?, aStr: String, other: String): KeyExprResult {
75+
val handle = JNIWrappers.concat(keyExprArg(a, aStr), other)
6976
// Match Rust's `KeyExpr::concat` formatting: "{self}{other}".
7077
return KeyExprResult(handle, "$aStr$other")
7178
}
7279

7380
/**
74-
* Release the native `Arc<KeyExpr>` registration. No-op for `0L`
75-
* (string-only keyexprs never allocated an Arc).
81+
* Release the native `Arc<KeyExpr>` registration. No-op if the handle
82+
* has already been closed/consumed.
7683
*/
77-
fun keyExprDrop(handle: Long) {
78-
if (handle != 0L) JNINative.dropKeyExprViaJNI(handle)
84+
fun keyExprDrop(handle: NativeHandle) {
85+
handle.close { ptr -> JNINative.dropKeyExprViaJNI(ptr) }
7986
}

‎zenoh-jni-runtime/src/commonMain/kotlin/io/zenoh/jni/JNIQuerier.kt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ public class JNIQuerier(initialPtr: Long) : NativeHandle(initialPtr) {
2828

2929
@Throws(ZError::class)
3030
fun get(
31-
keyExprHandle: Long?,
31+
keyExprHandle: NativeHandle?,
3232
keyExprString: String,
3333
parameters: String?,
3434
callback: JNIGetCallback,

‎zenoh-jni-runtime/src/commonMain/kotlin/io/zenoh/jni/JNIQuery.kt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ public class JNIQuery(initialPtr: Long) : NativeHandle(initialPtr) {
2222

2323
@Throws(ZError::class)
2424
fun replySuccess(
25-
keyExprHandle: Long?,
25+
keyExprHandle: NativeHandle?,
2626
keyExprString: String,
2727
payload: ByteArray,
2828
encoding: JNIEncoding,
@@ -41,7 +41,7 @@ public class JNIQuery(initialPtr: Long) : NativeHandle(initialPtr) {
4141

4242
@Throws(ZError::class)
4343
fun replyDelete(
44-
keyExprHandle: Long?,
44+
keyExprHandle: NativeHandle?,
4545
keyExprString: String,
4646
timestampEnabled: Boolean,
4747
timestampNtp64: Long,

0 commit comments

Comments
 (0)