Rewrite akka type urls for routing

lightbend · Nov 8, 2024 · ec9c22c · ec9c22c
1 parent 989e90e
commit ec9c22c
Show file tree

Hide file tree

Showing 6 changed files with 22 additions and 16 deletions.
diff --git a/sdk/java-sdk-protobuf/src/main/scala/kalix/javasdk/impl/AnySupport.scala b/sdk/java-sdk-protobuf/src/main/scala/kalix/javasdk/impl/AnySupport.scala
@@ -48,6 +48,10 @@ object AnySupport {
   def stripJsonTypeUrlPrefix(typeUrl: String): String =
     typeUrl.stripPrefix(AkkaJsonTypeUrlPrefix).stripPrefix(KalixJsonTypeUrlPrefix)
 
+  def replaceAkkaJsonPrefix(typeUrl: String): String =
+    if (typeUrl.startsWith(AkkaJsonTypeUrlPrefix)) KalixJsonTypeUrlPrefix + typeUrl.stripPrefix(AkkaJsonTypeUrlPrefix)
+    else typeUrl
+
   sealed abstract class Primitive[T: ClassTag] {
     val name: String = fieldType.name().toLowerCase(Locale.ROOT)
     val fullName: String = KalixPrimitive + name

diff --git a/sdk/java-sdk-spring/src/main/scala/kalix/javasdk/impl/action/ReflectiveActionRouter.scala b/sdk/java-sdk-spring/src/main/scala/kalix/javasdk/impl/action/ReflectiveActionRouter.scala
@@ -6,11 +6,12 @@ package kalix.javasdk.impl.action
 
 import akka.NotUsed
 import akka.stream.javadsl.Source
-import com.google.protobuf.any.{ Any => ScalaPbAny }
-import kalix.javasdk.action.{ Action, MessageEnvelope }
+import com.google.protobuf.any.{Any => ScalaPbAny}
+import kalix.javasdk.action.{Action, MessageEnvelope}
+import kalix.javasdk.impl.AnySupport
 import kalix.javasdk.impl.AnySupport.ProtobufEmptyTypeUrl
 import kalix.javasdk.impl.reflection.Reflect
-import kalix.javasdk.impl.{ CommandHandler, InvocationContext }
+import kalix.javasdk.impl.{CommandHandler, InvocationContext}
 
 // TODO: abstract away reactor dependency
 import reactor.core.publisher.Flux
@@ -34,7 +35,7 @@ class ReflectiveActionRouter[A <: Action](
         commandHandler.requestMessageDescriptor,
         message.metadata())
 
-    val inputTypeUrl = message.payload().asInstanceOf[ScalaPbAny].typeUrl
+    val inputTypeUrl = AnySupport.replaceAkkaJsonPrefix(message.payload().asInstanceOf[ScalaPbAny].typeUrl)
     val methodInvoker = commandHandler.lookupInvoker(inputTypeUrl)
 
     // lookup ComponentClient
@@ -79,7 +80,7 @@ class ReflectiveActionRouter[A <: Action](
           componentMethod.requestMessageDescriptor,
           message.metadata())
 
-      val inputTypeUrl = message.payload().asInstanceOf[ScalaPbAny].typeUrl
+      val inputTypeUrl = AnySupport.replaceAkkaJsonPrefix(message.payload().asInstanceOf[ScalaPbAny].typeUrl)
       componentMethod.lookupInvoker(inputTypeUrl) match {
         case Some(methodInvoker) =>
           val response = methodInvoker.invoke(action, context).asInstanceOf[Flux[Action.Effect[_]]]

diff --git a/...main/scala/kalix/javasdk/impl/eventsourcedentity/ReflectiveEventSourcedEntityRouter.scala b/...main/scala/kalix/javasdk/impl/eventsourcedentity/ReflectiveEventSourcedEntityRouter.scala
@@ -4,11 +4,12 @@
 
 package kalix.javasdk.impl.eventsourcedentity
 
-import com.google.protobuf.any.{ Any => ScalaPbAny }
-import com.google.protobuf.{ Any => JavaPbAny }
+import com.google.protobuf.any.{Any => ScalaPbAny}
+import com.google.protobuf.{Any => JavaPbAny}
 import kalix.javasdk.JsonSupport
 import kalix.javasdk.eventsourcedentity.CommandContext
 import kalix.javasdk.eventsourcedentity.EventSourcedEntity
+import kalix.javasdk.impl.AnySupport
 import kalix.javasdk.impl.CommandHandler
 import kalix.javasdk.impl.InvocationContext
 import kalix.javasdk.impl.JsonMessageCodec
@@ -70,7 +71,7 @@ class ReflectiveEventSourcedEntityRouter[S, E, ES <: EventSourcedEntity[S, E]](
         commandHandler.requestMessageDescriptor,
         commandContext.metadata())
 
-    val inputTypeUrl = command.asInstanceOf[ScalaPbAny].typeUrl
+    val inputTypeUrl = AnySupport.replaceAkkaJsonPrefix(command.asInstanceOf[ScalaPbAny].typeUrl)
     val methodInvoker = commandHandler
       .getInvoker(inputTypeUrl)
 

diff --git a/...dk-spring/src/main/scala/kalix/javasdk/impl/valueentity/ReflectiveValueEntityRouter.scala b/...dk-spring/src/main/scala/kalix/javasdk/impl/valueentity/ReflectiveValueEntityRouter.scala
@@ -5,9 +5,9 @@
 package kalix.javasdk.impl.valueentity
 
 import java.lang.reflect.ParameterizedType
-
-import com.google.protobuf.any.{ Any => ScalaPbAny }
+import com.google.protobuf.any.{Any => ScalaPbAny}
 import kalix.javasdk.JsonSupport
+import kalix.javasdk.impl.AnySupport
 import kalix.javasdk.impl.CommandHandler
 import kalix.javasdk.impl.InvocationContext
 import kalix.javasdk.valueentity.CommandContext
@@ -36,7 +36,7 @@ class ReflectiveValueEntityRouter[S, E <: ValueEntity[S]](
         commandHandler.requestMessageDescriptor,
         commandContext.metadata())
 
-    val inputTypeUrl = command.asInstanceOf[ScalaPbAny].typeUrl
+    val inputTypeUrl = AnySupport.replaceAkkaJsonPrefix(command.asInstanceOf[ScalaPbAny].typeUrl)
 
     commandHandler
       .getInvoker(inputTypeUrl)

diff --git a/sdk/java-sdk-spring/src/main/scala/kalix/javasdk/impl/view/ReflectiveViewRouter.scala b/sdk/java-sdk-spring/src/main/scala/kalix/javasdk/impl/view/ReflectiveViewRouter.scala
@@ -6,11 +6,10 @@ package kalix.javasdk.impl.view
 
 import java.lang.reflect.ParameterizedType
 import java.util.{ Map => JMap }
-
 import scala.jdk.CollectionConverters._
-
 import com.google.protobuf.any.{ Any => ScalaPbAny }
 import kalix.javasdk.JsonSupport
+import kalix.javasdk.impl.AnySupport
 import kalix.javasdk.impl.AnySupport.ProtobufEmptyTypeUrl
 import kalix.javasdk.impl.CommandHandler
 import kalix.javasdk.impl.ComponentDescriptorFactory
@@ -51,7 +50,7 @@ class ReflectiveViewRouter[S, V <: View[S]](
     val commandHandler = commandHandlerLookup(commandName)
 
     val anyEvent = event.asInstanceOf[ScalaPbAny]
-    val inputTypeUrl = anyEvent.typeUrl
+    val inputTypeUrl = AnySupport.replaceAkkaJsonPrefix(anyEvent.typeUrl)
     val methodInvoker = commandHandler.lookupInvoker(inputTypeUrl)
 
     methodInvoker match {

diff --git a/...java-sdk-spring/src/main/scala/kalix/javasdk/impl/workflow/ReflectiveWorkflowRouter.scala b/...java-sdk-spring/src/main/scala/kalix/javasdk/impl/workflow/ReflectiveWorkflowRouter.scala
@@ -4,7 +4,8 @@
 
 package kalix.javasdk.impl.workflow
 
-import com.google.protobuf.any.{ Any => ScalaPbAny }
+import com.google.protobuf.any.{Any => ScalaPbAny}
+import kalix.javasdk.impl.AnySupport
 import kalix.javasdk.impl.CommandHandler
 import kalix.javasdk.impl.InvocationContext
 import kalix.javasdk.workflow.AbstractWorkflow
@@ -36,7 +37,7 @@ class ReflectiveWorkflowRouter[S, W <: Workflow[S]](
         commandHandler.requestMessageDescriptor,
         commandContext.metadata())
 
-    val inputTypeUrl = command.asInstanceOf[ScalaPbAny].typeUrl
+    val inputTypeUrl = AnySupport.replaceAkkaJsonPrefix(command.asInstanceOf[ScalaPbAny].typeUrl)
 
     commandHandler
       .getInvoker(inputTypeUrl)