Implementing workflow-cmd and workflow-engine events - WIP

infiniticio · Jan 23, 2024 · d8c13e4 · d8c13e4
1 parent 012fe00
commit d8c13e4
Show file tree

Hide file tree

Showing 33 changed files with 895 additions and 162 deletions.
diff --git a/infinitic-client/src/main/kotlin/io/infinitic/clients/InfiniticClient.kt b/infinitic-client/src/main/kotlin/io/infinitic/clients/InfiniticClient.kt
@@ -190,7 +190,7 @@ class InfiniticClient(
   ): CompletableFuture<Unit> =
       when (val handler = getProxyHandler(stub)) {
         is ExistingWorkflowProxyHandler -> {
-          val taskName =
+          val serviceName =
               taskClass?.let {
                 // Use NewTaskProxyHandler in case of use of @Name annotation
                 NewServiceProxyHandler(it, setOf(), TaskMeta()) { dispatcher }.serviceName
@@ -199,7 +199,7 @@ class InfiniticClient(
           dispatcher.retryTaskAsync(
               workflowName = handler.workflowName,
               requestBy = handler.requestBy,
-              serviceName = taskName,
+              serviceName = serviceName,
               taskStatus = taskStatus,
               taskId = taskId?.let { TaskId(it) },
           )

diff --git a/infinitic-client/src/main/kotlin/io/infinitic/clients/dispatcher/ClientDispatcher.kt b/infinitic-client/src/main/kotlin/io/infinitic/clients/dispatcher/ClientDispatcher.kt
@@ -227,6 +227,9 @@ internal class ClientDispatcher(
           workflowName = workflowName,
           workflowId = workflowId,
           emitterName = emitterName,
+          parentWorkflowId = null,
+          parentWorkflowName = null,
+          parentWorkflowMethodId = null,
           emittedAt = null,
       )
       // synchronously sent the message to get errors
@@ -286,6 +289,9 @@ internal class ClientDispatcher(
           workflowMethodId = workflowMethodId,
           workflowName = workflowName,
           workflowId = requestBy.workflowId,
+          parentWorkflowId = null,
+          parentWorkflowName = null,
+          parentWorkflowMethodId = null,
           emitterName = emitterName,
           emittedAt = null,
       )
@@ -297,7 +303,9 @@ internal class ClientDispatcher(
           workflowName = workflowName,
           workflowTag = requestBy.workflowTag,
           reason = WorkflowCancellationReason.CANCELED_BY_CLIENT,
-          emitterWorkflowId = null,
+          parentWorkflowId = null,
+          parentWorkflowName = null,
+          parentWorkflowMethodId = null,
           emitterName = emitterName,
           emittedAt = null,
       )
@@ -316,6 +324,9 @@ internal class ClientDispatcher(
           workflowName = workflowName,
           workflowId = requestBy.workflowId,
           emitterName = emitterName,
+          parentWorkflowId = null,
+          parentWorkflowName = null,
+          parentWorkflowMethodId = null,
           emittedAt = null,
       )
       msg.sendToAsync(WorkflowCmdTopic)
@@ -325,6 +336,9 @@ internal class ClientDispatcher(
       val msg = RetryWorkflowTaskByTag(
           workflowName = workflowName,
           workflowTag = requestBy.workflowTag,
+          parentWorkflowId = null,
+          parentWorkflowName = null,
+          parentWorkflowMethodId = null,
           emitterName = emitterName,
           emittedAt = null,
       )
@@ -346,6 +360,9 @@ internal class ClientDispatcher(
           workflowName = workflowName,
           workflowId = requestBy.workflowId,
           emitterName = emitterName,
+          parentWorkflowId = null,
+          parentWorkflowName = null,
+          parentWorkflowMethodId = null,
           emittedAt = null,
       )
       msg.sendToAsync(WorkflowCmdTopic)
@@ -357,6 +374,9 @@ internal class ClientDispatcher(
           workflowTag = requestBy.workflowTag,
           workflowMethodId = workflowMethodId,
           emitterName = emitterName,
+          parentWorkflowId = null,
+          parentWorkflowName = null,
+          parentWorkflowMethodId = null,
           emittedAt = null,
       )
       msg.sendToAsync(WorkflowTagTopic)
@@ -381,6 +401,9 @@ internal class ClientDispatcher(
           taskId = taskId,
           taskStatus = taskStatus,
           serviceName = serviceName,
+          parentWorkflowId = null,
+          parentWorkflowName = null,
+          parentWorkflowMethodId = null,
           emittedAt = null,
       )
       msg.sendToAsync(WorkflowCmdTopic)
@@ -394,6 +417,9 @@ internal class ClientDispatcher(
           taskStatus = taskStatus,
           serviceName = serviceName,
           emitterName = emitterName,
+          parentWorkflowId = null,
+          parentWorkflowName = null,
+          parentWorkflowMethodId = null,
           emittedAt = null,
       )
       msg.sendToAsync(WorkflowTagTopic)
@@ -700,6 +726,9 @@ internal class ClientDispatcher(
             channelTypes = handler.channelTypes,
             workflowName = handler.workflowName,
             workflowId = (handler.requestBy as RequestByWorkflowId).workflowId,
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterName,
             emittedAt = null,
         )
@@ -715,6 +744,8 @@ internal class ClientDispatcher(
             signalData = handler.signalData,
             channelTypes = handler.channelTypes,
             parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterName,
             emittedAt = null,
         )

diff --git a/infinitic-client/src/test/kotlin/io/infinitic/clients/InfiniticClientTests.kt b/infinitic-client/src/test/kotlin/io/infinitic/clients/InfiniticClientTests.kt
@@ -406,6 +406,9 @@ internal class InfiniticClientTests : StringSpec(
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = msg.workflowId,
             emitterName = emitterNameTest,
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emittedAt = null,
         )
 
@@ -466,6 +469,9 @@ internal class InfiniticClientTests : StringSpec(
             channelTypes = ChannelType.allFrom(String::class.java),
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = WorkflowId(id),
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -486,6 +492,9 @@ internal class InfiniticClientTests : StringSpec(
             channelTypes = ChannelType.allFrom(String::class.java),
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = WorkflowId(id),
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -505,6 +514,8 @@ internal class InfiniticClientTests : StringSpec(
             signalData = SignalData.from("a"),
             channelTypes = ChannelType.allFrom(String::class.java),
             parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -525,6 +536,8 @@ internal class InfiniticClientTests : StringSpec(
             signalData = SignalData.from("a"),
             channelTypes = ChannelType.allFrom(String::class.java),
             parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -550,6 +563,9 @@ internal class InfiniticClientTests : StringSpec(
             ),
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = WorkflowId(id),
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -575,6 +591,9 @@ internal class InfiniticClientTests : StringSpec(
             ),
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = WorkflowId(id),
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -593,6 +612,9 @@ internal class InfiniticClientTests : StringSpec(
             serviceName = null,
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = WorkflowId(id),
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -611,6 +633,9 @@ internal class InfiniticClientTests : StringSpec(
             serviceName = null,
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = WorkflowId(id),
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -627,6 +652,9 @@ internal class InfiniticClientTests : StringSpec(
             workflowMethodId = null,
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = WorkflowId(id),
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -643,6 +671,9 @@ internal class InfiniticClientTests : StringSpec(
             workflowMethodId = null,
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = WorkflowId(id),
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -709,6 +740,9 @@ internal class InfiniticClientTests : StringSpec(
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = WorkflowId(id),
             emitterName = emitterNameTest,
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emittedAt = null,
         )
       }
@@ -725,6 +759,9 @@ internal class InfiniticClientTests : StringSpec(
             workflowMethodId = null,
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = WorkflowId(id),
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -741,7 +778,9 @@ internal class InfiniticClientTests : StringSpec(
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowTag = WorkflowTag(tag),
             reason = WorkflowCancellationReason.CANCELED_BY_CLIENT,
-            emitterWorkflowId = null,
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -759,7 +798,9 @@ internal class InfiniticClientTests : StringSpec(
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowTag = WorkflowTag("foo"),
             reason = WorkflowCancellationReason.CANCELED_BY_CLIENT,
-            emitterWorkflowId = null,
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emitterName = emitterNameTest,
             emittedAt = null,
         )
@@ -778,6 +819,9 @@ internal class InfiniticClientTests : StringSpec(
             workflowName = WorkflowName(FakeWorkflow::class.java.name),
             workflowId = WorkflowId(deferred.id),
             emitterName = emitterNameTest,
+            parentWorkflowId = null,
+            parentWorkflowName = null,
+            parentWorkflowMethodId = null,
             emittedAt = null,
         )
       }

diff --git a/infinitic-common/src/main/kotlin/io/infinitic/common/tasks/data/TaskReturnValue.kt b/infinitic-common/src/main/kotlin/io/infinitic/common/tasks/data/TaskReturnValue.kt
@@ -22,16 +22,20 @@
  */
 package io.infinitic.common.tasks.data
 
+import com.github.avrokotlin.avro4k.Avro
+import com.github.avrokotlin.avro4k.AvroDefault
 import com.github.avrokotlin.avro4k.AvroNamespace
 import io.infinitic.common.data.ReturnValue
+import io.infinitic.common.data.methods.MethodName
 import kotlinx.serialization.SerialName
 import kotlinx.serialization.Serializable
 
 @Serializable
 @AvroNamespace("io.infinitic.tasks.data")
 data class TaskReturnValue(
-    val taskId: TaskId,
-    @SerialName("taskName") val serviceName: ServiceName,
-    val taskMeta: TaskMeta,
-    val returnValue: ReturnValue
+  val taskId: TaskId,
+  @SerialName("taskName") val serviceName: ServiceName,
+  @AvroDefault(Avro.NULL) val taskMethodName: MethodName?,
+  val taskMeta: TaskMeta,
+  val returnValue: ReturnValue
 )
diff --git a/infinitic-common/src/main/kotlin/io/infinitic/common/tasks/data/TaskTag.kt b/infinitic-common/src/main/kotlin/io/infinitic/common/tasks/data/TaskTag.kt
@@ -45,3 +45,5 @@ object TaskTagSerializer : KSerializer<TaskTag> {
 
   override fun deserialize(decoder: Decoder) = TaskTag(decoder.decodeString())
 }
+
+val Set<TaskTag>.set get() = map { it.tag }.toSet()