feat: aligning chameneos with csm version (#33)

Author: AlexZangerle

chameneos/src/main/kotlin/ac/at/uibk/dps/dapr/chameneos/chameneos/ChameneosActor.kt

@@ -6,7 +6,7 @@ import io.dapr.actors.ActorType
 interface ChameneosActor {
   fun request()
 
-  fun meet(data: Map<String, Any>)
+  fun matchMade(data: Map<String, Any>)
 
   fun change(data: Map<String, Any>)
 }

chameneos/src/main/kotlin/ac/at/uibk/dps/dapr/chameneos/chameneos/ChameneosActorImpl.kt

@@ -8,88 +8,65 @@ import io.dapr.client.DaprClientBuilder
 import java.util.concurrent.TimeUnit
 import kotlin.random.Random
 import kotlin.time.Clock
-import kotlin.time.measureTime
-import kotlin.time.toJavaDuration
 
 class ChameneosActorImpl(
   runtimeContext: ActorRuntimeContext<ChameneosActorImpl>,
   val actorId: ActorId,
 ) : AbstractActor(runtimeContext, actorId), ChameneosActor {
   val client = DaprClientBuilder().build()
 
-  var color = Random.nextInt(1, 4)
-
   var metricRegistry = Chameneos.provideMetricRegistry()
-
   var eventTimer = metricRegistry.timer("event.latency")
-  var requestTimer = metricRegistry.timer("request.duration")
-  var meetTimer = metricRegistry.timer("meet.duration")
-  var changeTimer = metricRegistry.timer("change.duration")
 
-  override fun request() {
-    val delta = measureTime {
-      val now = Clock.System.now()
-      val epochNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+  var color = Random.nextInt(1, 4)
 
-      client
-        .publishEvent(
-          "pubsub",
-          "request",
-          mapOf<String, Any>(
-            "requestor" to actorId.toString(),
-            "color" to color,
-            "time" to epochNanos,
-          ),
-        )
-        .subscribe()
-    }
-    requestTimer.update(delta.toJavaDuration())
+  override fun request() {
+    val now = Clock.System.now()
+    val epochNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+
+    client
+      .publishEvent(
+        "pubsub",
+        "requesting",
+        mapOf<String, Any>("id" to actorId.toString(), "color" to color, "time" to epochNanos),
+      )
+      .subscribe()
   }
 
-  override fun meet(data: Map<String, Any>) {
-    val delta = measureTime {
-      val time = data["time"] as Long
-      val partnerColor = data["color"] as Int
-      val partner = data["partner"] as String
+  override fun matchMade(data: Map<String, Any>) {
+    val now = Clock.System.now()
+    val nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+    val deltaNanos = (nowNanos - data["time"] as Long).coerceAtLeast(0L)
 
-      var now = Clock.System.now()
-      var nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+    eventTimer.update(deltaNanos, TimeUnit.NANOSECONDS)
 
-      val deltaNanos = (nowNanos - time).coerceAtLeast(0L)
+    val partnerColor = data["color"] as Int
 
-      eventTimer.update(deltaNanos, TimeUnit.NANOSECONDS)
+    color = if (color == partnerColor) color else (color xor partnerColor)
 
-      color = if (color == partnerColor) color else (color xor partnerColor)
+    client
+      .publishEvent(
+        "pubsub",
+        "change",
+        mapOf<String, Any>(
+          "partner" to data["partner"].toString(),
+          "color" to color,
+          "time" to nowNanos,
+        ),
+      )
+      .subscribe()
 
-      now = Clock.System.now()
-      nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
-      client
-        .publishEvent(
-          "pubsub",
-          "change",
-          mapOf<String, Any>("partner" to partner, "color" to color, "time" to nowNanos),
-        )
-        .subscribe()
-    }
-    meetTimer.update(delta.toJavaDuration())
     request()
   }
 
   override fun change(data: Map<String, Any>) {
-    val delta = measureTime {
-      val time = data["time"] as Long
-
-      val now = Clock.System.now()
-      val nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
-
-      val deltaNanos = (nowNanos - time).coerceAtLeast(0L)
-
-      eventTimer.update(deltaNanos, TimeUnit.NANOSECONDS)
+    val now = Clock.System.now()
+    val nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+    val deltaNanos = (nowNanos - data["time"] as Long).coerceAtLeast(0L)
 
-      color = data["color"] as Int
-    }
-    changeTimer.update(delta.toJavaDuration())
+    eventTimer.update(deltaNanos, TimeUnit.NANOSECONDS)
 
+    color = data["color"] as Int
     request()
   }
 }

chameneos/src/main/kotlin/ac/at/uibk/dps/dapr/chameneos/chameneos/ChameneosSubscriber.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
@@ -14,29 +15,22 @@ import org.springframework.web.bind.annotation.RestController
 class ChameneosSubscriber {
 
   private val id = System.getenv("CHAMENEOS_ID")
-  private val actorClient = ActorClient()
   private val chameneosProxy =
-    ActorProxyBuilder(ChameneosActor::class.java, actorClient).build(ActorId("$id"))
+    ActorProxyBuilder(ChameneosActor::class.java, ActorClient()).build(ActorId("$id"))
 
-  @Topic(name = "meet", pubsubName = "pubsub")
-  @PostMapping("/meet")
-  fun handleMeet(@RequestBody body: Map<String, Any>) {
-    val data = body["data"] as? Map<String, Any> ?: body
-    val initiator = data["initiator"] as? String ?: return
-
-    if (initiator == id) {
-      chameneosProxy.meet(data)
+  @Topic(name = "matchMade", pubsubName = "pubsub")
+  @PostMapping("/matchMade")
+  fun matchMadeSubscriber(@RequestBody event: CloudEvent<Map<String, Any>>) {
+    if (event.data["target"] == id) {
+      chameneosProxy.matchMade(event.data)
     }
   }
 
   @Topic(name = "change", pubsubName = "pubsub")
   @PostMapping("/change")
-  fun handleChange(@RequestBody body: Map<String, Any>) {
-    val data = body["data"] as? Map<String, Any> ?: body
-    val partner = data["partner"] as? String ?: return
-
-    if (partner == id) {
-      chameneosProxy.change(data)
+  fun changeSubscriber(@RequestBody event: CloudEvent<Map<String, Any>>) {
+    if (event.data["partner"] == id) {
+      chameneosProxy.change(event.data)
     }
   }
 }

chameneos/src/main/kotlin/ac/at/uibk/dps/dapr/chameneos/mall/MallActorImpl.kt

@@ -7,63 +7,49 @@ import io.dapr.actors.runtime.ActorRuntimeContext
 import io.dapr.client.DaprClientBuilder
 import java.util.concurrent.TimeUnit
 import kotlin.time.Clock
-import kotlin.time.measureTime
-import kotlin.time.toJavaDuration
 
 class MallActorImpl(runtimeContext: ActorRuntimeContext<MallActorImpl>, actorId: ActorId) :
   AbstractActor(runtimeContext, actorId), MallActor {
 
   val client = DaprClientBuilder().build()
 
-  var count = 0
-  var startTime = System.nanoTime()
-  var waiting: MutableList<String> = mutableListOf()
-
   var metricRegistry = Chameneos.provideMetricRegistry()
-  var counter = metricRegistry.counter("mall.meetings")
-  var eventTimer = metricRegistry.timer("event.latency")
-  var requestingTimer = metricRegistry.timer("request.duration")
-
-  override fun requesting(data: Map<String, Any>) {
-    val delta = measureTime {
-      val time = data["time"] as Long
-      val requestor = data["requestor"] as String
-      val requestorColor = data["color"] as Int
-
-      var now = Clock.System.now()
-      var nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
-
-      val deltaNanos = (nowNanos - time).coerceAtLeast(0L)
 
-      eventTimer.update(deltaNanos, TimeUnit.NANOSECONDS)
-
-      if (count == 0) startTime = System.nanoTime()
-
-      if (waiting.isEmpty()) {
-        waiting.add(requestor)
-      } else {
-        count++
-        counter.inc()
-
-        val waitingId = waiting.removeAt(0)
-
-        now = Clock.System.now()
-        nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+  var count = 0
+  var waiting: ArrayList<String> = arrayListOf()
 
-        client
-          .publishEvent(
-            "pubsub",
-            "meet",
-            mapOf<String, Any>(
-              "initiator" to waitingId,
-              "partner" to requestor,
-              "color" to requestorColor,
-              "time" to nowNanos,
-            ),
-          )
-          .subscribe()
-      }
+  override fun requesting(data: Map<String, Any>) {
+    var now = Clock.System.now()
+    var nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+    val deltaNanos = (nowNanos - data["time"] as Long).coerceAtLeast(0L)
+
+    metricRegistry.timer("event.latency").update(deltaNanos, TimeUnit.NANOSECONDS)
+
+    val id = data["id"] as String
+
+    if (waiting.isEmpty()) {
+      waiting.add(id)
+    } else {
+      ++count
+      metricRegistry.counter("mall.meetings").inc()
+
+      now = Clock.System.now()
+      nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
+
+      client
+        .publishEvent(
+          "pubsub",
+          "matchMade",
+          mapOf<String, Any>(
+            "target" to waiting[0],
+            "partner" to id,
+            "color" to data["color"] as Int,
+            "time" to nowNanos,
+          ),
+        )
+        .subscribe()
+
+      waiting.removeAt(0)
     }
-    requestingTimer.update(delta.toJavaDuration())
   }
 }

chameneos/src/main/kotlin/ac/at/uibk/dps/dapr/chameneos/mall/MallSubscriber.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
@@ -12,15 +13,12 @@ import org.springframework.web.bind.annotation.RestController
 @RestController
 @ConditionalOnProperty("app.role", havingValue = "mall")
 class MallSubscriber {
-  private val actorClient = ActorClient()
   private val mallProxy =
-    ActorProxyBuilder(MallActor::class.java, actorClient).build(ActorId("mall-1"))
+    ActorProxyBuilder(MallActor::class.java, ActorClient()).build(ActorId("mall-1"))
 
-  @Topic(name = "request", pubsubName = "pubsub")
-  @PostMapping("/request")
-  fun handleRequest(@RequestBody body: Map<String, Any>) {
-    val data = body["data"] as? Map<String, Any> ?: body
-
-    mallProxy.requesting(data)
+  @Topic(name = "requesting", pubsubName = "pubsub")
+  @PostMapping("/requesting")
+  fun requestSubscriber(@RequestBody event: CloudEvent<Map<String, Any>>) {
+    mallProxy.requesting(event.data)
   }
 }