Examples refactor. (#7)

eclipse-zenoh · Sep 26, 2023 · 00a9e94 · 00a9e94
1 parent ad9d626
commit 00a9e94
Show file tree

Hide file tree

Showing 9 changed files with 70 additions and 73 deletions.
diff --git a/examples/src/main/kotlin/io.zenoh/ZDelete.kt b/examples/src/main/kotlin/io.zenoh/ZDelete.kt
@@ -21,8 +21,10 @@ fun main() {
     Session.open().onSuccess { session ->
         session.use {
             "demo/example/zenoh-kotlin-put".intoKeyExpr().onSuccess { keyExpr ->
-                println("Deleting resources matching '$keyExpr'...")
-                session.delete(keyExpr).res()
+                keyExpr.use {
+                    println("Deleting resources matching '$keyExpr'...")
+                    session.delete(keyExpr).res()
+                }
             }
         }
     }

diff --git a/examples/src/main/kotlin/io.zenoh/ZGet.kt b/examples/src/main/kotlin/io.zenoh/ZGet.kt
@@ -22,10 +22,9 @@ fun main() {
     val timeout = Duration.ofMillis(1000)
     Session.open().onSuccess { session ->
         session.use {
-            val selectorResult = "demo/example/**".intoSelector()
-            selectorResult.onSuccess { selector ->
+            "demo/example/**".intoSelector().onSuccess { selector ->
                 selector.use {
-                    val request = session.get(selector)
+                    session.get(selector)
                         .with { reply ->
                             if (reply is Reply.Success) {
                                 println("Received ('${reply.sample.keyExpr}': '${reply.sample.value}')")
@@ -36,11 +35,10 @@ fun main() {
                         }
                         .timeout(timeout)
                         .res()
-
-                    request.onSuccess {
-                        // Keep the session alive for the duration of the timeout.
-                        Thread.sleep(timeout.toMillis())
-                    }
+                        .onSuccess {
+                            // Keep the session alive for the duration of the timeout.
+                            Thread.sleep(timeout.toMillis())
+                        }
                 }
             }
         }

diff --git a/examples/src/main/kotlin/io.zenoh/ZPub.kt b/examples/src/main/kotlin/io.zenoh/ZPub.kt
@@ -15,34 +15,30 @@
 package io.zenoh
 
 import io.zenoh.keyexpr.intoKeyExpr
-import io.zenoh.publication.CongestionControl
-import io.zenoh.publication.Priority
 
 fun main() {
     println("Opening session...")
-    Session.open().onSuccess {
-        it.use { session ->
-            val keyExpressionResult = "demo/example/zenoh-kotlin-pub".intoKeyExpr()
-            keyExpressionResult.onSuccess { keyExpr ->
+    Session.open().onSuccess { session ->
+        session.use {
+            "demo/example/zenoh-kotlin-pub".intoKeyExpr().onSuccess { keyExpr ->
                 keyExpr.use {
                     println("Declaring publisher on '$keyExpr'...")
-                    session.declarePublisher(keyExpr).priority(Priority.REALTIME)
-                        .congestionControl(CongestionControl.DROP).res().onSuccess { pub ->
-                            pub.use {
-                                var idx = 0
-                                while (true) {
-                                    Thread.sleep(1000)
-                                    val payload = "Pub from Kotlin!"
-                                    println(
-                                        "Putting Data ('$keyExpr': '[${
-                                            idx.toString().padStart(4, ' ')
-                                        }] $payload')..."
-                                    )
-                                    pub.put(payload).res()
-                                    idx++
-                                }
+                    session.declarePublisher(keyExpr).res().onSuccess { pub ->
+                        pub.use {
+                            var idx = 0
+                            while (true) {
+                                Thread.sleep(1000)
+                                val payload = "Pub from Kotlin!"
+                                println(
+                                    "Putting Data ('$keyExpr': '[${
+                                        idx.toString().padStart(4, ' ')
+                                    }] $payload')..."
+                                )
+                                pub.put(payload).res()
+                                idx++
                             }
                         }
+                    }
                 }
             }
         }

diff --git a/examples/src/main/kotlin/io.zenoh/ZQueryable.kt b/examples/src/main/kotlin/io.zenoh/ZQueryable.kt
@@ -14,32 +14,25 @@
 
 package io.zenoh
 
+import io.zenoh.keyexpr.KeyExpr
 import io.zenoh.keyexpr.intoKeyExpr
 import io.zenoh.prelude.SampleKind
+import io.zenoh.queryable.Query
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.runBlocking
 import org.apache.commons.net.ntp.TimeStamp
 
 fun main() {
     Session.open().onSuccess { session ->
         session.use {
             "demo/example/zenoh-kotlin-queryable".intoKeyExpr().onSuccess { keyExpr ->
-                println("Declaring Queryable")
-                session.declareQueryable(keyExpr).res().onSuccess { queryable ->
-                    queryable.use {
-                        it.receiver?.let { receiverChannel -> //  The default receiver is a Channel we can process on a coroutine.
-                            runBlocking {
-                                val iterator = receiverChannel.iterator()
-                                while (iterator.hasNext()) {
-                                    iterator.next().use { query ->
-                                        val valueInfo = query.value?.let { value -> " with value '$value'" } ?: ""
-                                        println(">> [Queryable] Received Query '${query.selector}' $valueInfo")
-                                        query.reply(keyExpr)
-                                            .success("Queryable from Kotlin!")
-                                            .withKind(SampleKind.PUT)
-                                            .withTimeStamp(TimeStamp.getCurrentTime())
-                                            .res()
-                                            .onFailure { println(">> [Queryable ] Error sending reply: $it") }
-                                    }
+                keyExpr.use {
+                    println("Declaring Queryable")
+                    session.declareQueryable(keyExpr).res().onSuccess { queryable ->
+                        queryable.use {
+                            queryable.receiver?.let { receiverChannel -> //  The default receiver is a Channel we can process on a coroutine.
+                                runBlocking {
+                                    handleRequests(receiverChannel, keyExpr)
                                 }
                             }
                         }
@@ -49,3 +42,18 @@ fun main() {
         }
     }
 }
+
+private suspend fun handleRequests(
+    receiverChannel: Channel<Query>, keyExpr: KeyExpr
+) {
+    val iterator = receiverChannel.iterator()
+    while (iterator.hasNext()) {
+        iterator.next().use { query ->
+            val valueInfo = query.value?.let { value -> " with value '$value'" } ?: ""
+            println(">> [Queryable] Received Query '${query.selector}' $valueInfo")
+            query.reply(keyExpr).success("Queryable from Kotlin!").withKind(SampleKind.PUT)
+                .withTimeStamp(TimeStamp.getCurrentTime()).res()
+                .onFailure { println(">> [Queryable ] Error sending reply: $it") }
+        }
+    }
+}
diff --git a/examples/src/main/kotlin/io.zenoh/ZSub.kt b/examples/src/main/kotlin/io.zenoh/ZSub.kt
@@ -15,20 +15,16 @@
 package io.zenoh
 
 import io.zenoh.keyexpr.intoKeyExpr
-import io.zenoh.subscriber.Reliability
 import kotlinx.coroutines.runBlocking
 
 fun main() {
     println("Opening session...")
     Session.open().onSuccess { session ->
         session.use {
             "demo/example/**".intoKeyExpr().onSuccess { keyExpr ->
-                println("Declaring Subscriber on '$keyExpr'...")
-                session.declareSubscriber(keyExpr)
-                    .bestEffort()
-                    .reliability(Reliability.RELIABLE)
-                    .res()
-                    .onSuccess { subscriber ->
+                keyExpr.use {
+                    println("Declaring Subscriber on '$keyExpr'...")
+                    session.declareSubscriber(keyExpr).bestEffort().res().onSuccess { subscriber ->
                         subscriber.use {
                             runBlocking {
                                 val receiver = subscriber.receiver!!
@@ -38,9 +34,11 @@ fun main() {
                                     println(">> [Subscriber] Received ${sample.kind} ('${sample.keyExpr}': '${sample.value}')")
                                 }
                             }
+                        }
                     }
-               }
+                }
             }
         }
     }
 }
+
diff --git a/examples/src/main/kotlin/io.zenoh/ZSubThr.kt b/examples/src/main/kotlin/io.zenoh/ZSubThr.kt
@@ -15,7 +15,6 @@
 package io.zenoh
 
 import io.zenoh.keyexpr.intoKeyExpr
-import io.zenoh.subscriber.Reliability
 
 const val NANOS_TO_SEC = 1_000_000_000L
 var n = 50000L
@@ -53,21 +52,19 @@ fun report() {
 }
 
 fun main() {
-    val keyExpr = "test/thr".intoKeyExpr().getOrThrow()
-    println("Opening Session")
-    Session.open().onSuccess { session ->
-        session.use {
-            session.declareSubscriber(keyExpr)
-                .reliability(Reliability.RELIABLE)
-                .with {
-                    listener()
-                }
-                .res()
-                .onSuccess {
-                    while (readlnOrNull() != "q") {
-                        // Do nothing
+    "test/thr".intoKeyExpr().onSuccess {
+        it.use { keyExpr ->
+            println("Opening Session")
+            Session.open().onSuccess { it.use {
+                session -> session.declareSubscriber(keyExpr)
+                    .reliable()
+                    .with { listener() }
+                    .res()
+                    .onSuccess {
+                        while (readlnOrNull() != "q") { /* Do nothing */ }
                     }
                 }
+            }
         }
     }
     report()

diff --git a/zenoh-kotlin/src/main/kotlin/io/zenoh/Session.kt b/zenoh-kotlin/src/main/kotlin/io/zenoh/Session.kt
@@ -139,7 +139,6 @@ class Session private constructor(private val config: Config) : AutoCloseable {
      *         "demo/kotlin/sub".intoKeyExpr().onSuccess { keyExpr ->
      *             session.declareSubscriber(keyExpr)
      *                 .bestEffort()
-     *                 .reliability(Reliability.RELIABLE)
      *                 .res()
      *                 .onSuccess { subscriber ->
      *                     subscriber.use {

diff --git a/zenoh-kotlin/src/main/kotlin/io/zenoh/jni/JNISession.kt b/zenoh-kotlin/src/main/kotlin/io/zenoh/jni/JNISession.kt
@@ -38,7 +38,7 @@ import java.time.Duration
 import java.util.concurrent.atomic.AtomicLong
 
 /** Adapter class to handle the communication with the Zenoh JNI code for a [Session]. */
-internal class JNISession() {
+internal class JNISession {
 
     /* Pointer to the underlying Rust zenoh session. */
     private var sessionPtr: AtomicLong = AtomicLong(0)

diff --git a/zenoh-kotlin/src/main/kotlin/io/zenoh/subscriber/Subscriber.kt b/zenoh-kotlin/src/main/kotlin/io/zenoh/subscriber/Subscriber.kt
@@ -37,7 +37,6 @@ import kotlinx.coroutines.channels.Channel
  *         "demo/kotlin/sub".intoKeyExpr().onSuccess { keyExpr ->
  *             session.declareSubscriber(keyExpr)
  *                 .bestEffort()
- *                 .reliability(Reliability.RELIABLE)
  *                 .res()
  *                 .onSuccess { subscriber ->
  *                     subscriber.use {