Only change RPC response topic with supported topic hot swap

beemobot · Dec 4, 2023 · e69b2e8 · e69b2e8
1 parent 13d265e
commit e69b2e8
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 8 deletions.
diff --git a/latte/src/main/java/gg/beemo/latte/broker/BrokerClient.kt b/latte/src/main/java/gg/beemo/latte/broker/BrokerClient.kt
@@ -155,8 +155,10 @@ abstract class BrokerClient(
         return topics[topic]?.keys?.get(key)
     }
 
-    internal fun createResponseTopic(topic: String): String = "$topic.responses"
-    internal fun createResponseKey(key: String): String = "$key.response"
+    internal fun toResponseTopic(topic: String): String =
+        if (connection.supportsTopicHotSwap) "$topic.responses" else topic
+
+    internal fun toResponseKey(key: String): String = "$key.response"
 
     private fun onTopicMessage(
         topic: String,

diff --git a/latte/src/main/java/gg/beemo/latte/broker/BrokerMessage.kt b/latte/src/main/java/gg/beemo/latte/broker/BrokerMessage.kt
@@ -6,4 +6,9 @@ data class BrokerMessage<T>(
     val key: String,
     val value: T,
     val headers: BaseBrokerMessageHeaders
-)
+) {
+
+    val messageId: String
+        get() = headers.messageId
+
+}
diff --git a/latte/src/main/java/gg/beemo/latte/broker/BrokerSubclients.kt b/latte/src/main/java/gg/beemo/latte/broker/BrokerSubclients.kt
@@ -186,8 +186,8 @@ class RpcClient<RequestT, ResponseT>(
     private val requestConsumer = client.consumer(topic, key, options, requestType, responseIsNullable) {
         val result = callback(it)
         val responseProducer = client.producer(
-            client.createResponseTopic(topic),
-            client.createResponseKey(key),
+            client.toResponseTopic(topic),
+            client.toResponseKey(key),
             options,
             responseType,
             responseIsNullable,
@@ -197,7 +197,7 @@ class RpcClient<RequestT, ResponseT>(
             result,
             services = setOf(it.headers.sourceService),
             instances = setOf(it.headers.sourceInstance),
-            inReplyTo = it.headers.messageId,
+            inReplyTo = it.messageId,
         )
         responseProducer.destroy()
     }
@@ -230,8 +230,8 @@ class RpcClient<RequestT, ResponseT>(
             val messageId = AtomicReference<String?>(null)
 
             val responseConsumer = client.consumer(
-                client.createResponseTopic(topic),
-                client.createResponseKey(key),
+                client.toResponseTopic(topic),
+                client.toResponseKey(key),
                 options,
                 responseType,
                 responseIsNullable,