Examples of Quasar in Kotlin (#7045)

kotlin-quasar/pom.xml

@@ -0,0 +1,120 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <groupId>com.baeldung</groupId>
+    <artifactId>kotlin-quasar</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>kotlin-quasar</name>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-stdlib-jdk8</artifactId>
+            <version>${kotlin.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-test</artifactId>
+            <version>${kotlin.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>co.paralleluniverse</groupId>
+            <artifactId>quasar-core</artifactId>
+            <version>${quasar.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>co.paralleluniverse</groupId>
+            <artifactId>quasar-actors</artifactId>
+            <version>${quasar.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>co.paralleluniverse</groupId>
+            <artifactId>quasar-reactive-streams</artifactId>
+            <version>${quasar.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>co.paralleluniverse</groupId>
+            <artifactId>quasar-kotlin</artifactId>
+            <version>${quasar.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.12</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <sourceDirectory>src/main/kotlin</sourceDirectory>
+        <testSourceDirectory>src/test/kotlin</testSourceDirectory>
+        <plugins>
+            <plugin>
+                <groupId>org.jetbrains.kotlin</groupId>
+                <artifactId>kotlin-maven-plugin</artifactId>
+                <version>${kotlin.version}</version>
+                <executions>
+                    <execution>
+                        <id>compile</id>
+                        <phase>compile</phase>
+                        <goals>
+                            <goal>compile</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>test-compile</id>
+                        <phase>test-compile</phase>
+                        <goals>
+                            <goal>test-compile</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <jvmTarget>1.8</jvmTarget>
+                </configuration>
+            </plugin>
+            <plugin>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <version>3.1.1</version>
+                <executions>
+                    <execution>
+                        <id>getClasspathFilenames</id>
+                        <goals>
+                            <goal>properties</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>2.22.1</version>
+                <configuration>
+                    <argLine>-Dco.paralleluniverse.fibers.verifyInstrumentation=true</argLine>
+                    <argLine>-javaagent:${co.paralleluniverse:quasar-core:jar}</argLine>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>exec-maven-plugin</artifactId>
+                <version>1.3.2</version>
+                <configuration>
+                    <workingDirectory>target/classes</workingDirectory>
+                    <executable>echo</executable>
+                    <arguments>
+                        <argument>-javaagent:${co.paralleluniverse:quasar-core:jar}</argument>
+                        <argument>-classpath</argument> <classpath/>
+                        <argument>com.baeldung.quasar.QuasarHelloWorldKt</argument>
+                    </arguments>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+    <properties>
+        <quasar.version>0.8.0</quasar.version>
+        <kotlin.version>1.3.31</kotlin.version>
+    </properties>
+</project>

kotlin-quasar/src/main/kotlin/com/baeldung/quasar/QuasarHelloWorld.kt

@@ -0,0 +1,19 @@
+package com.baeldung.quasar
+
+import co.paralleluniverse.fibers.Fiber
+import co.paralleluniverse.strands.SuspendableRunnable
+
+
+/**
+ * Entrypoint into the application
+ */
+fun main(args: Array<String>) {
+    class Runnable : SuspendableRunnable {
+        override fun run() {
+            println("Hello")
+        }
+    }
+    val result = Fiber<Void>(Runnable()).start()
+    result.join()
+    println("World")
+}

kotlin-quasar/src/test/kotlin/com/baeldung/quasar/ChannelsTest.kt

@@ -0,0 +1,155 @@
+package com.baeldung.quasar
+
+import co.paralleluniverse.fibers.Suspendable
+import co.paralleluniverse.kotlin.fiber
+import co.paralleluniverse.strands.channels.Channels
+import co.paralleluniverse.strands.channels.Selector
+import com.google.common.base.Function
+import org.junit.Test
+
+class ChannelsTest {
+    @Test
+    fun createChannel() {
+        Channels.newChannel<String>(0, // The size of the channel buffer
+                Channels.OverflowPolicy.BLOCK,  // The policy for when the buffer is full
+                true,  // Whether we should optimize for a single message producer
+                true) // Whether we should optimize for a single message consumer
+    }
+
+    @Test
+    fun blockOnMessage() {
+        val channel = Channels.newChannel<String>(0, Channels.OverflowPolicy.BLOCK, true, true)
+
+        fiber @Suspendable {
+            while (!channel.isClosed) {
+                val message = channel.receive()
+                println("Received: $message")
+            }
+            println("Stopped receiving messages")
+        }
+
+        channel.send("Hello")
+        channel.send("World")
+
+        channel.close()
+    }
+
+    @Test
+    fun selectReceiveChannels() {
+        val channel1 = Channels.newChannel<String>(0, Channels.OverflowPolicy.BLOCK, true, true)
+        val channel2 = Channels.newChannel<String>(0, Channels.OverflowPolicy.BLOCK, true, true)
+
+        fiber @Suspendable {
+            while (!channel1.isClosed && !channel2.isClosed) {
+                val received = Selector.select(Selector.receive(channel1), Selector.receive(channel2))
+
+                println("Received: $received")
+            }
+        }
+
+        fiber @Suspendable {
+            for (i in 0..10) {
+                channel1.send("Channel 1: $i")
+            }
+        }
+
+        fiber @Suspendable {
+            for (i in 0..10) {
+                channel2.send("Channel 2: $i")
+            }
+        }
+    }
+
+    @Test
+    fun selectSendChannels() {
+        val channel1 = Channels.newChannel<String>(0, Channels.OverflowPolicy.BLOCK, true, true)
+        val channel2 = Channels.newChannel<String>(0, Channels.OverflowPolicy.BLOCK, true, true)
+
+        fiber @Suspendable {
+            for (i in 0..10) {
+                Selector.select(
+                        Selector.send(channel1, "Channel 1: $i"),
+                        Selector.send(channel2, "Channel 2: $i")
+                )
+            }
+        }
+
+        fiber @Suspendable {
+            while (!channel1.isClosed) {
+                val msg = channel1.receive()
+                println("Read: $msg")
+            }
+        }
+
+        fiber @Suspendable {
+            while (!channel2.isClosed) {
+                val msg = channel2.receive()
+                println("Read: $msg")
+            }
+        }
+    }
+
+    @Test
+    fun tickerChannel() {
+        val channel = Channels.newChannel<String>(3, Channels.OverflowPolicy.DISPLACE)
+
+        for (i in 0..10) {
+            val tickerConsumer = Channels.newTickerConsumerFor(channel)
+            fiber @Suspendable {
+                while (!tickerConsumer.isClosed) {
+                    val message = tickerConsumer.receive()
+                    println("Received on $i: $message")
+                }
+                println("Stopped receiving messages on $i")
+            }
+        }
+
+        for (i in 0..50) {
+            channel.send("Message $i")
+        }
+
+        channel.close()
+    }
+
+
+    @Test
+    fun transformOnSend() {
+        val channel = Channels.newChannel<String>(0, Channels.OverflowPolicy.BLOCK, true, true)
+
+        fiber @Suspendable {
+            while (!channel.isClosed) {
+                val message = channel.receive()
+                println("Received: $message")
+            }
+            println("Stopped receiving messages")
+        }
+
+        val transformOnSend = Channels.mapSend(channel, Function<String, String> { msg: String? -> msg?.toUpperCase() })
+
+        transformOnSend.send("Hello")
+        transformOnSend.send("World")
+
+        channel.close()
+    }
+
+    @Test
+    fun transformOnReceive() {
+        val channel = Channels.newChannel<String>(0, Channels.OverflowPolicy.BLOCK, true, true)
+
+        val transformOnReceive = Channels.map(channel, Function<String, String> { msg: String? -> msg?.reversed() })
+
+        fiber @Suspendable {
+            while (!transformOnReceive.isClosed) {
+                val message = transformOnReceive.receive()
+                println("Received: $message")
+            }
+            println("Stopped receiving messages")
+        }
+
+
+        channel.send("Hello")
+        channel.send("World")
+
+        channel.close()
+    }
+}

kotlin-quasar/src/test/kotlin/com/baeldung/quasar/DataflowTest.kt

@@ -0,0 +1,35 @@
+package com.baeldung.quasar
+
+import co.paralleluniverse.strands.dataflow.Val
+import co.paralleluniverse.strands.dataflow.Var
+import org.junit.Assert
+import org.junit.Test
+import java.util.concurrent.TimeUnit
+
+class DataflowTest {
+    @Test
+    fun testValVar() {
+        val a = Var<Int>()
+        val b = Val<Int>()
+
+        val c = Var<Int> { a.get() + b.get() }
+        val d = Var<Int> { a.get() * b.get() }
+
+        // (a*b) - (a+b)
+        val initialResult = Val<Int> { d.get() - c.get() }
+        val currentResult = Var<Int> { d.get() - c.get() }
+
+        a.set(2)
+        b.set(4)
+
+        Assert.assertEquals(2, initialResult.get())
+        Assert.assertEquals(2, currentResult.get())
+
+        a.set(3)
+
+        TimeUnit.SECONDS.sleep(1)
+
+        Assert.assertEquals(2, initialResult.get())
+        Assert.assertEquals(5, currentResult.get())
+    }
+}

kotlin-quasar/src/test/kotlin/com/baeldung/quasar/PiAsyncTest.kt

@@ -0,0 +1,53 @@
+package com.baeldung.quasar
+
+import co.paralleluniverse.fibers.Fiber
+import co.paralleluniverse.fibers.FiberAsync
+import co.paralleluniverse.fibers.Suspendable
+import co.paralleluniverse.kotlin.fiber
+import co.paralleluniverse.strands.Strand
+import org.junit.Assert
+import org.junit.Test
+import java.math.BigDecimal
+import java.util.concurrent.TimeUnit
+
+interface PiCallback {
+    fun success(result: BigDecimal)
+    fun failure(error: Exception)
+}
+
+fun computePi(callback: PiCallback) {
+    println("Starting calculations")
+    TimeUnit.SECONDS.sleep(2)
+    println("Finished calculations")
+    callback.success(BigDecimal("3.14"))
+}
+
+class PiAsync : PiCallback, FiberAsync<BigDecimal, Exception>() {
+    override fun success(result: BigDecimal) {
+        asyncCompleted(result)
+    }
+
+    override fun failure(error: Exception) {
+        asyncFailed(error)
+    }
+
+    override fun requestAsync() {
+        computePi(this)
+    }
+}
+
+class PiAsyncTest {
+    @Test
+    fun testPi() {
+        val result = fiber @Suspendable {
+            val pi = PiAsync()
+            println("Waiting to get PI on: " + Fiber.currentFiber().name)
+            val result = pi.run()
+            println("Got PI")
+
+            result
+        }.get()
+
+        Assert.assertEquals(BigDecimal("3.14"), result)
+    }
+}

kotlin-quasar/src/test/kotlin/com/baeldung/quasar/SuspendableCallableTest.kt

@@ -0,0 +1,48 @@
+package com.baeldung.quasar
+
+import co.paralleluniverse.fibers.Fiber
+import co.paralleluniverse.fibers.Suspendable
+import co.paralleluniverse.kotlin.fiber
+import co.paralleluniverse.strands.SuspendableCallable
+import org.junit.Assert
+import org.junit.Test
+import java.util.concurrent.TimeUnit
+
+
+class SuspendableCallableTest {
+    @Test
+    fun createFiber() {
+        class Callable : SuspendableCallable<String> {
+            override fun run(): String {
+                println("Inside Fiber")
+                return "Hello"
+            }
+        }
+        val result = Fiber<String>(Callable()).start()
+
+        Assert.assertEquals("Hello", result.get())
+    }
+
+    @Test
+    fun createFiberLambda() {
+        val lambda: (() -> String) = {
+            println("Inside Fiber Lambda")
+            "Hello"
+        }
+        val result = Fiber<String>(lambda)
+        result.start()
+
+        Assert.assertEquals("Hello", result.get())
+    }
+
+    @Test
+    fun createFiberDsl() {
+        val result = fiber @Suspendable {
+            TimeUnit.SECONDS.sleep(5)
+            println("Inside Fiber DSL")
+            "Hello"
+        }
+
+        Assert.assertEquals("Hello", result.get())
+    }
+}

kotlin-quasar/src/test/kotlin/com/baeldung/quasar/SuspensableRunnableTest.kt

@@ -0,0 +1,47 @@
+package com.baeldung.quasar
+
+import co.paralleluniverse.fibers.Fiber
+import co.paralleluniverse.fibers.Suspendable
+import co.paralleluniverse.kotlin.fiber
+import co.paralleluniverse.strands.SuspendableRunnable
+import org.junit.Test
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
+
+
+class SuspensableRunnableTest {
+    @Test
+    fun createFiber() {
+        class Runnable : SuspendableRunnable {
+            override fun run() {
+                println("Inside Fiber")
+            }
+        }
+        val result = Fiber<Void>(Runnable()).start()
+        result.join()
+    }
+
+    @Test
+    fun createFiberLambda() {
+        val result = Fiber<Void> {
+            println("Inside Fiber Lambda")
+        }
+        result.start()
+        result.join()
+    }
+
+    @Test
+    fun createFiberDsl() {
+        fiber @Suspendable {
+            println("Inside Fiber DSL")
+        }.join()
+    }
+
+    @Test(expected = TimeoutException::class)
+    fun fiberTimeout() {
+        fiber @Suspendable {
+            TimeUnit.SECONDS.sleep(5)
+            println("Inside Fiber DSL")
+        }.join(2, TimeUnit.SECONDS)
+    }
+}