fix: aligning cigaretteSmokers example with csm version (#31)

* fix: aligning cigaretteSmokers example with csm version

* fix: fixing issue in cigaretteSmokers

Author: AlexZangerle

cigaretteSmokers/src/main/kotlin/ac/at/uibk/dps/dapr/cigarette/arbiter/ArbiterActorImpl.kt

@@ -8,48 +8,39 @@ import io.dapr.client.DaprClient
 import io.dapr.client.DaprClientBuilder
 import java.util.concurrent.TimeUnit
 import kotlin.time.Clock
-import kotlin.time.measureTime
-import kotlin.time.toJavaDuration
 
 class ArbiterActorImpl(runtimeContext: ActorRuntimeContext<ArbiterActorImpl>, actorId: ActorId) :
   AbstractActor(runtimeContext, actorId), ArbiterActor {
 
   val client: DaprClient = DaprClientBuilder().build()
 
-  val ingredients = listOf("0", "1", "2")
+  val ingredients = listOf(0, 1, 2)
   var count = 0
 
   var metricRegistry = CigaretteSmokers.provideMetricRegistry()
-  var counter = metricRegistry.counter("arbiter.rounds")
-  var eventTimer = metricRegistry.timer("event.latency")
-  var provideTimer = metricRegistry.timer("provide.duration")
 
   override fun provide(time: Long) {
-    val delta = measureTime {
-      var now = Clock.System.now()
-      val nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+    var now = Clock.System.now()
+    val nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+    val deltaNanos = (nowNanos - time).coerceAtLeast(0L)
 
-      val deltaNanos = (nowNanos - time).coerceAtLeast(0L)
+    metricRegistry.timer("event.latency").update(deltaNanos, TimeUnit.NANOSECONDS)
 
-      eventTimer.update(deltaNanos, TimeUnit.NANOSECONDS)
+    ++count
+    metricRegistry.counter("arbiter.rounds").inc()
 
-      count++
-      counter.inc()
+    val provide = ingredients.toMutableList()
+    provide.remove(provide.random())
 
-      val provide = ingredients.toMutableList()
-      provide.remove(provide.random())
+    now = Clock.System.now()
+    val epochNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
 
-      now = Clock.System.now()
-      val epochNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
-
-      client
-        .publishEvent(
-          "pubsub",
-          "provide",
-          mutableMapOf("ingredients" to provide, "time" to epochNanos),
-        )
-        .subscribe()
-    }
-    provideTimer.update(delta.toJavaDuration())
+    client
+      .publishEvent(
+        "pubsub",
+        "provided",
+        mutableMapOf("ingredients" to provide, "time" to epochNanos),
+      )
+      .subscribe()
   }
 }

cigaretteSmokers/src/main/kotlin/ac/at/uibk/dps/dapr/cigarette/arbiter/ArbiterSubscriber.kt

@@ -4,6 +4,7 @@ import io.dapr.Topic
 import io.dapr.actors.ActorId
 import io.dapr.actors.client.ActorClient
 import io.dapr.actors.client.ActorProxyBuilder
+import io.dapr.client.domain.CloudEvent
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.web.bind.annotation.PostMapping
 import org.springframework.web.bind.annotation.RequestBody
@@ -16,9 +17,9 @@ class ArbiterSubscriber {
   private val arbiterProxy =
     ActorProxyBuilder(ArbiterActor::class.java, ActorClient()).build(ActorId("arbiter-1"))
 
-  @Topic(name = "done", pubsubName = "pubsub")
-  @PostMapping("/done")
-  fun handleDone(@RequestBody body: Map<String, Any>) {
-    arbiterProxy.provide(body["data"] as Long)
+  @Topic(name = "finish", pubsubName = "pubsub")
+  @PostMapping("/finish")
+  fun finishSubscriber(@RequestBody event: CloudEvent<Long>) {
+    arbiterProxy.provide(event.data)
   }
 }

cigaretteSmokers/src/main/kotlin/ac/at/uibk/dps/dapr/cigarette/smoker/SmokerActor.kt

@@ -4,5 +4,5 @@ import io.dapr.actors.ActorType
 
 @ActorType(name = "SmokerActor")
 interface SmokerActor {
-  fun smoke(data: Map<String, Any>)
+  fun smoking(data: Map<String, Any>)
 }

cigaretteSmokers/src/main/kotlin/ac/at/uibk/dps/dapr/cigarette/smoker/SmokerActorImpl.kt

@@ -6,41 +6,46 @@ import io.dapr.actors.runtime.AbstractActor
 import io.dapr.actors.runtime.ActorRuntimeContext
 import io.dapr.client.DaprClient
 import io.dapr.client.DaprClientBuilder
+import java.security.SecureRandom
 import java.util.concurrent.TimeUnit
+import kotlin.random.Random
 import kotlin.time.Clock
-import kotlin.time.measureTime
-import kotlin.time.toJavaDuration
 
 class SmokerActorImpl(runtimeContext: ActorRuntimeContext<SmokerActorImpl>, val actorId: ActorId) :
   AbstractActor(runtimeContext, actorId), SmokerActor {
 
   val client: DaprClient = DaprClientBuilder().build()
 
   var metricRegistry = CigaretteSmokers.provideMetricRegistry()
-  var eventTimer = metricRegistry.timer("event.latency")
-  var smokeTimer = metricRegistry.timer("smoke.duration")
 
-  override fun smoke(data: Map<String, Any>) {
-    val delta = measureTime {
-      val ingredients = data["ingredients"] as List<String>
-      val time = data["time"] as Long
+  private val seedGenerator = SecureRandom()
+  private val threadRng =
+    object : ThreadLocal<Random>() {
+      override fun initialValue(): Random {
+        return Random(seedGenerator.nextLong())
+      }
+    }
 
-      if (!ingredients.contains(actorId.toString())) {
-        var now = Clock.System.now()
-        val nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+  override fun smoking(data: Map<String, Any>) {
+    var now = Clock.System.now()
+    val nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+    val deltaNanos = (nowNanos - data["time"] as Long).coerceAtLeast(0L)
 
-        val deltaNanos = (nowNanos - time).coerceAtLeast(0L)
+    metricRegistry.timer("event.latency").update(deltaNanos, TimeUnit.NANOSECONDS)
 
-        eventTimer.update(deltaNanos, TimeUnit.NANOSECONDS)
+    val ingredients = data["ingredients"] as List<*>
 
-        Thread.sleep(10)
+    if (!ingredients.contains(actorId.toString().toInt())) {
+      Thread.sleep(randomAround(10, 2).toLong())
 
-        now = Clock.System.now()
-        val epochNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+      now = Clock.System.now()
+      val epochNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
 
-        client.publishEvent("pubsub", "done", epochNanos).subscribe()
-      }
+      client.publishEvent("pubsub", "finish", epochNanos).subscribe()
     }
-    smokeTimer.update(delta.toJavaDuration())
+  }
+
+  fun randomAround(base: Int, delta: Int): Int {
+    return (base - delta..base + delta).random(threadRng.get())
   }
 }

cigaretteSmokers/src/main/kotlin/ac/at/uibk/dps/dapr/cigarette/smoker/SmokerSubscriber.kt

@@ -4,6 +4,7 @@ import io.dapr.Topic
 import io.dapr.actors.ActorId
 import io.dapr.actors.client.ActorClient
 import io.dapr.actors.client.ActorProxyBuilder
+import io.dapr.client.domain.CloudEvent
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.web.bind.annotation.PostMapping
 import org.springframework.web.bind.annotation.RequestBody
@@ -16,9 +17,9 @@ class SmokerSubscriber {
   private val smokerProxy =
     ActorProxyBuilder(SmokerActor::class.java, ActorClient()).build(ActorId("$id"))
 
-  @Topic(name = "provide", pubsubName = "pubsub")
-  @PostMapping("/provide")
-  fun handleProvide(@RequestBody body: Map<String, Any>) {
-    smokerProxy.smoke(body["data"] as Map<String, Any>)
+  @Topic(name = "provided", pubsubName = "pubsub")
+  @PostMapping("/provided")
+  fun provideSubscriber(@RequestBody event: CloudEvent<Map<String, Any>>) {
+    smokerProxy.smoking(event.data)
   }
 }