Skip to content

Commit a1a720d

Browse files
milyinmilyin
committed
wrappers directly in Session and. Liveliness
1 parent 248f645 commit a1a720d

4 files changed

Lines changed: 229 additions & 453 deletions

File tree

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

Lines changed: 188 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -15,17 +15,21 @@
1515
package io.zenoh
1616

1717
import io.zenoh.annotations.Unstable
18+
import io.zenoh.bytes.Encoding
1819
import io.zenoh.bytes.IntoZBytes
1920
import io.zenoh.bytes.ZBytes
21+
import io.zenoh.bytes.into
2022
import io.zenoh.config.ZenohId
2123
import io.zenoh.exceptions.ZError
24+
import io.zenoh.exceptions.wrapJNIExceptionAsZError
2225
import io.zenoh.handlers.BlockingQueueHandler
2326
import io.zenoh.handlers.Callback
2427
import io.zenoh.handlers.Handler
25-
import io.zenoh.jni.JNISession
28+
import io.zenoh.jni.session.ZSession
2629
import io.zenoh.keyexpr.KeyExpr
2730
import io.zenoh.liveliness.Liveliness
2831
import io.zenoh.pubsub.*
32+
import io.zenoh.qos.QoS
2933
import io.zenoh.query.*
3034
import io.zenoh.query.Query
3135
import io.zenoh.query.Queryable
@@ -53,7 +57,7 @@ import java.util.concurrent.LinkedBlockingDeque
5357
*/
5458
class Session private constructor(private val config: Config) : AutoCloseable {
5559

56-
internal var jniSession: JNISession? = null
60+
internal var zSession: ZSession? = null
5761

5862
// Subscribers and Queryables that keep running despite losing references to them.
5963
private var strongDeclarations = mutableListOf<SessionDeclaration>()
@@ -98,8 +102,8 @@ class Session private constructor(private val config: Config) : AutoCloseable {
98102
true
99103
}
100104

101-
jniSession?.close()
102-
jniSession = null
105+
zSession?.close()
106+
zSession = null
103107
}
104108

105109
protected fun finalize() {
@@ -382,11 +386,12 @@ class Session private constructor(private val config: Config) : AutoCloseable {
382386
*/
383387
@Throws(ZError::class)
384388
fun declareKeyExpr(keyExpr: String): KeyExpr {
385-
return jniSession?.run {
386-
val keyexpr = declareKeyExpr(keyExpr)
387-
strongDeclarations.add(keyexpr)
388-
keyexpr
389-
} ?: throw sessionClosedException
389+
val zSession = zSession ?: throw sessionClosedException
390+
val keyexpr = wrapJNIExceptionAsZError {
391+
KeyExpr(io.zenoh.jni.keyexpr.KeyExpr(keyExpr, zSession.zSessionDeclareKeyexpr(keyExpr)))
392+
}
393+
strongDeclarations.add(keyexpr)
394+
return keyexpr
390395
}
391396

392397
/**
@@ -400,9 +405,14 @@ class Session private constructor(private val config: Config) : AutoCloseable {
400405
*/
401406
@Throws(ZError::class)
402407
fun undeclare(keyExpr: KeyExpr) {
403-
return jniSession?.run {
404-
undeclareKeyExpr(keyExpr)
405-
} ?: throw (sessionClosedException)
408+
val zSession = zSession ?: throw sessionClosedException
409+
val handle = keyExpr.flat.keyExprNative
410+
if (handle == null || handle.isClosed()) {
411+
throw ZError("Attempting to undeclare a non declared key expression.")
412+
}
413+
wrapJNIExceptionAsZError {
414+
zSession.zSessionUndeclareKeyexpr(handle)
415+
}
406416
}
407417

408418
/**
@@ -556,7 +566,7 @@ class Session private constructor(private val config: Config) : AutoCloseable {
556566

557567
/** Returns if session is open or has been closed. */
558568
fun isClosed(): Boolean {
559-
return jniSession == null
569+
return zSession == null
560570
}
561571

562572
/**
@@ -575,66 +585,126 @@ class Session private constructor(private val config: Config) : AutoCloseable {
575585

576586
@Throws(ZError::class)
577587
internal fun resolvePublisher(keyExpr: KeyExpr, options: PublisherOptions): Publisher {
578-
return jniSession?.run {
579-
val publisher = declarePublisher(keyExpr, options)
580-
weakDeclarations.add(WeakReference(publisher))
581-
publisher
582-
} ?: throw (sessionClosedException)
588+
val zSession = zSession ?: throw sessionClosedException
589+
val publisher = wrapJNIExceptionAsZError {
590+
val zPublisher = zSession.zSessionDeclarePublisher(
591+
keyExpr.flat,
592+
options.congestionControl.jni,
593+
options.priority.jni,
594+
options.express,
595+
options.reliability.jni
596+
)
597+
Publisher(
598+
keyExpr,
599+
options.congestionControl,
600+
options.priority,
601+
options.encoding,
602+
zPublisher,
603+
)
604+
}
605+
weakDeclarations.add(WeakReference(publisher))
606+
return publisher
583607
}
584608

585609
@Throws(ZError::class)
586610
internal fun <R> resolveSubscriberWithHandler(
587611
keyExpr: KeyExpr, handler: Handler<Sample, R>
588612
): HandlerSubscriber<R> {
589-
return jniSession?.run {
590-
val subscriber = declareSubscriberWithHandler(keyExpr, handler)
591-
strongDeclarations.add(subscriber)
592-
subscriber
593-
} ?: throw (sessionClosedException)
613+
val zSession = zSession ?: throw sessionClosedException
614+
val subscriber = wrapJNIExceptionAsZError {
615+
val zSubscriber = zSession.sessionDeclareSubscriber(
616+
keyExpr.flat,
617+
io.zenoh.jni.callbacks.SampleCallback { flat -> handler.handle(Sample.from(flat)) },
618+
io.zenoh.jni.callbacks.Callback { handler.onClose() }
619+
)
620+
HandlerSubscriber(keyExpr, zSubscriber, handler.receiver())
621+
}
622+
strongDeclarations.add(subscriber)
623+
return subscriber
594624
}
595625

596626
@Throws(ZError::class)
597627
internal fun resolveSubscriberWithCallback(
598628
keyExpr: KeyExpr, callback: Callback<Sample>
599629
): CallbackSubscriber {
600-
return jniSession?.run {
601-
val subscriber = declareSubscriberWithCallback(keyExpr, callback)
602-
strongDeclarations.add(subscriber)
603-
subscriber
604-
} ?: throw (sessionClosedException)
630+
val zSession = zSession ?: throw sessionClosedException
631+
val subscriber = wrapJNIExceptionAsZError {
632+
val zSubscriber = zSession.sessionDeclareSubscriber(
633+
keyExpr.flat,
634+
io.zenoh.jni.callbacks.SampleCallback { flat -> callback.run(Sample.from(flat)) },
635+
io.zenoh.jni.callbacks.Callback { }
636+
)
637+
CallbackSubscriber(keyExpr, zSubscriber)
638+
}
639+
strongDeclarations.add(subscriber)
640+
return subscriber
605641
}
606642

607643
@Throws(ZError::class)
608644
internal fun <R> resolveQueryableWithHandler(
609645
keyExpr: KeyExpr, handler: Handler<Query, R>, options: QueryableOptions
610646
): HandlerQueryable<R> {
611-
return jniSession?.run {
612-
val queryable = declareQueryableWithHandler(keyExpr, handler, options)
613-
strongDeclarations.add(queryable)
614-
queryable
615-
} ?: throw (sessionClosedException)
647+
val zSession = zSession ?: throw sessionClosedException
648+
val queryable = wrapJNIExceptionAsZError {
649+
val zQueryable = zSession.sessionDeclareQueryable(
650+
keyExpr.flat,
651+
options.complete,
652+
io.zenoh.jni.callbacks.QueryCallback { flat -> handler.handle(Query.from(flat)) },
653+
io.zenoh.jni.callbacks.Callback { handler.onClose() }
654+
)
655+
HandlerQueryable(keyExpr, zQueryable, handler.receiver())
656+
}
657+
strongDeclarations.add(queryable)
658+
return queryable
616659
}
617660

618661
@Throws(ZError::class)
619662
internal fun resolveQueryableWithCallback(
620663
keyExpr: KeyExpr, callback: Callback<Query>, options: QueryableOptions
621664
): CallbackQueryable {
622-
return jniSession?.run {
623-
val queryable = declareQueryableWithCallback(keyExpr, callback, options)
624-
strongDeclarations.add(queryable)
625-
queryable
626-
} ?: throw (sessionClosedException)
665+
val zSession = zSession ?: throw sessionClosedException
666+
val queryable = wrapJNIExceptionAsZError {
667+
val zQueryable = zSession.sessionDeclareQueryable(
668+
keyExpr.flat,
669+
options.complete,
670+
io.zenoh.jni.callbacks.QueryCallback { flat -> callback.run(Query.from(flat)) },
671+
io.zenoh.jni.callbacks.Callback { }
672+
)
673+
CallbackQueryable(keyExpr, zQueryable)
674+
}
675+
strongDeclarations.add(queryable)
676+
return queryable
627677
}
628678

679+
@OptIn(Unstable::class)
629680
private fun resolveQuerier(
630681
keyExpr: KeyExpr,
631682
options: QuerierOptions
632683
): Querier {
633-
return jniSession?.run {
634-
val querier = declareQuerier(keyExpr, options)
635-
weakDeclarations.add(WeakReference(querier))
636-
querier
637-
} ?: throw sessionClosedException
684+
val zSession = zSession ?: throw sessionClosedException
685+
val querier = wrapJNIExceptionAsZError {
686+
val zQuerier = zSession.zSessionDeclareQuerier(
687+
keyExpr.flat,
688+
options.target.toFlat(),
689+
options.consolidationMode.toFlat(),
690+
options.congestionControl.jni,
691+
options.priority.jni,
692+
options.express,
693+
options.timeout.toMillis(),
694+
options.acceptReplies.toFlat()
695+
)
696+
Querier(
697+
keyExpr,
698+
QoS(
699+
congestionControl = options.congestionControl,
700+
priority = options.priority,
701+
express = options.express
702+
),
703+
zQuerier
704+
)
705+
}
706+
weakDeclarations.add(WeakReference(querier))
707+
return querier
638708
}
639709

640710
@Throws(ZError::class)
@@ -643,11 +713,27 @@ class Session private constructor(private val config: Config) : AutoCloseable {
643713
handler: Handler<Reply, R>,
644714
options: GetOptions
645715
): R {
646-
return jniSession?.performGetWithHandler(
647-
selector,
648-
handler,
649-
options
650-
) ?: throw sessionClosedException
716+
val zSession = zSession ?: throw sessionClosedException
717+
return wrapJNIExceptionAsZError {
718+
val sel = selector.into()
719+
zSession.sessionGet(
720+
sel.keyExpr.flat,
721+
sel.parameters?.toString(),
722+
options.timeout.toMillis(),
723+
options.target.toFlat(),
724+
options.consolidation.toFlat(),
725+
options.acceptReplies.toFlat(),
726+
options.qos.congestionControl.jni,
727+
options.qos.priority.jni,
728+
options.qos.express,
729+
options.payload?.into()?.inner,
730+
(options.encoding ?: Encoding.defaultEncoding()).toFlat(),
731+
options.attachment?.into()?.inner,
732+
io.zenoh.jni.callbacks.ReplyCallback { flat -> handler.handle(Reply.from(flat)) },
733+
io.zenoh.jni.callbacks.Callback { handler.onClose() }
734+
)
735+
handler.receiver()
736+
}
651737
}
652738

653739
@Throws(ZError::class)
@@ -656,42 +742,84 @@ class Session private constructor(private val config: Config) : AutoCloseable {
656742
callback: Callback<Reply>,
657743
options: GetOptions
658744
) {
659-
return jniSession?.performGetWithCallback(
660-
selector,
661-
callback,
662-
options
663-
) ?: throw sessionClosedException
745+
val zSession = zSession ?: throw sessionClosedException
746+
wrapJNIExceptionAsZError {
747+
val sel = selector.into()
748+
zSession.sessionGet(
749+
sel.keyExpr.flat,
750+
sel.parameters?.toString(),
751+
options.timeout.toMillis(),
752+
options.target.toFlat(),
753+
options.consolidation.toFlat(),
754+
options.acceptReplies.toFlat(),
755+
options.qos.congestionControl.jni,
756+
options.qos.priority.jni,
757+
options.qos.express,
758+
options.payload?.into()?.inner,
759+
(options.encoding ?: Encoding.defaultEncoding()).toFlat(),
760+
options.attachment?.into()?.inner,
761+
io.zenoh.jni.callbacks.ReplyCallback { flat -> callback.run(Reply.from(flat)) },
762+
io.zenoh.jni.callbacks.Callback { }
763+
)
764+
}
664765
}
665766

666767
@Throws(ZError::class)
667768
internal fun resolvePut(keyExpr: KeyExpr, payload: IntoZBytes, putOptions: PutOptions) {
668-
jniSession?.run { performPut(keyExpr, payload, putOptions) }
769+
val zSession = zSession ?: return
770+
wrapJNIExceptionAsZError {
771+
val encoding = putOptions.encoding ?: Encoding.defaultEncoding()
772+
zSession.zSessionPut(
773+
keyExpr.flat,
774+
payload.into().inner,
775+
encoding.toFlat(),
776+
putOptions.congestionControl.jni,
777+
putOptions.priority.jni,
778+
putOptions.express,
779+
putOptions.attachment?.into()?.inner,
780+
putOptions.reliability.jni
781+
)
782+
}
669783
}
670784

671785
@Throws(ZError::class)
672786
internal fun resolveDelete(keyExpr: KeyExpr, deleteOptions: DeleteOptions) {
673-
jniSession?.run { performDelete(keyExpr, deleteOptions) }
787+
val zSession = zSession ?: return
788+
wrapJNIExceptionAsZError {
789+
zSession.zSessionDelete(
790+
keyExpr.flat,
791+
deleteOptions.congestionControl.jni,
792+
deleteOptions.priority.jni,
793+
deleteOptions.express,
794+
deleteOptions.attachment?.into()?.inner,
795+
deleteOptions.reliability.jni
796+
)
797+
}
674798
}
675799

676800
@Throws(ZError::class)
677801
internal fun zid(): ZenohId {
678-
return jniSession?.zid() ?: throw sessionClosedException
802+
val zSession = zSession ?: throw sessionClosedException
803+
return wrapJNIExceptionAsZError { ZenohId(zSession.zSessionZid()) }
679804
}
680805

681806
@Throws(ZError::class)
682807
internal fun getPeersId(): List<ZenohId> {
683-
return jniSession?.peersZid() ?: throw sessionClosedException
808+
val zSession = zSession ?: throw sessionClosedException
809+
return wrapJNIExceptionAsZError { zSession.zSessionPeersZid().map { ZenohId(it) } }
684810
}
685811

686812
@Throws(ZError::class)
687813
internal fun getRoutersId(): List<ZenohId> {
688-
return jniSession?.routersZid() ?: throw sessionClosedException
814+
val zSession = zSession ?: throw sessionClosedException
815+
return wrapJNIExceptionAsZError { zSession.zSessionRoutersZid().map { ZenohId(it) } }
689816
}
690817

691-
/** Launches the session through the jni session, returning the [Session] on success. */
818+
/** Launches the session, returning the [Session] on success. */
692819
@Throws(ZError::class)
693820
private fun launch(): Session {
694-
this.jniSession = JNISession.open(config)
821+
ZenohLoad
822+
this.zSession = wrapJNIExceptionAsZError { ZSession.zOpen(config.zConfig) }
695823
return this
696824
}
697825
}

0 commit comments

Comments
 (0)