resolves #1961 [web connector] add streaming support for rich message

Signed-off-by: Julien Buret <jburet@gmail.com>

Author: zigzago

bot/api/service/src/main/kotlin/BotApiClientController.kt

@@ -130,8 +130,8 @@ internal class BotApiClientController(
                     }
                 } else {
                     val holder = setHolder(request.requestId)
-                    sendWithSse(request, lastConfiguration?.version, sendResponse)
                     holder.waitForResponse(sendResponse)
+                    sendWithSse(request, lastConfiguration?.version, sendResponse)
                 }
             }
     }
@@ -144,8 +144,8 @@ internal class BotApiClientController(
         if (pushHandler != null) {
             val holder = setHolder(request.requestId)
             logger.debug { "send request ${request.requestId}" }
-            pushHandler.invoke(mapper.writeValueAsString(request))
             holder.waitForResponse(sendResponse)
+            pushHandler.invoke(mapper.writeValueAsString(request))
         } else {
             if (request.configuration != true) {
                 error("no websocket handler for $apiKey and no webhook reachable")

bot/api/service/src/main/kotlin/WSHolder.kt

@@ -52,19 +52,25 @@ internal class WSHolder {
     private var latch: CountDownLatch = CountDownLatch(1)
 
     fun receive(response: ResponseData) {
+        logger.debug { "add to holder: $response" }
         this.response.add(response)
-        latch.countDown()
+        synchronized(response) {
+            latch.countDown()
+        }
     }
 
-    @Synchronized
     fun wait(): List<ResponseData> {
+        logger.debug { "start await" }
         latch.await(timeoutInSeconds, SECONDS)
-        val r = response.sortedBy { it.botResponse?.context?.date ?: Instant.now() }
-        logger.debug { r }
-        if (r.lastOrNull()?.botResponse?.context?.lastResponse == false) {
-            latch = CountDownLatch(1)
+        synchronized(response) {
+            val r = response.sortedBy { it.botResponse?.context?.date ?: Instant.now() }
+            logger.debug { "responses: $r" }
+            if (r.lastOrNull()?.botResponse?.context?.lastResponse == false) {
+                latch = CountDownLatch(1)
+            }
+
+            return r.filterNot { seen.contains(it) }.apply { seen.addAll(this) }
         }
-        return r.filterNot { seen.contains(it) }.apply { seen.addAll(this) }
     }
 
     fun waitForResponse(sendResponse: (ResponseData?) -> Unit = {}): Unit =

bot/connector-web/src/main/kotlin/WebConnector.kt

@@ -258,7 +258,7 @@ class WebConnector internal constructor(
                 streamedResponse = (event as? Action)?.metadata?.streamedResponse == true,
             )
         if (sseEnabled) {
-            // Uniquely identify each response, so they can be reconciliated between SSE and POST
+            // Uniquely identify each response, so they can be reconciled between SSE and POST
             callback.addMetadata(MetadataEvent.responseId(UUID.randomUUID(), applicationId))
         }
         controller.handle(

bot/connector-web/src/main/kotlin/WebConnectorCallback.kt

@@ -50,7 +50,7 @@ internal class WebConnectorCallback(
 
     fun addMetadata(metadata: MetadataEvent) {
         this.metadata[metadata.type] = metadata.value
-        if (metadata.isEndStreamMetadata()) {
+        if (metadata.isStreamMetadata()) {
             WebRequestInfosByEvent.get(eventId)?.clearStreamedResponse()
         }
     }
@@ -60,7 +60,7 @@ internal class WebConnectorCallback(
             actions.mapNotNull { a ->
                 val action =
                     if (a is SendSentence && metadata.hasStreamMetadata() && mergeStreamResponse) {
-                        a.withText(WebRequestInfosByEvent.getOrPut(eventId).addStreamedResponse(a.stringText))
+                        WebRequestInfosByEvent.getOrPut(eventId).addStreamedResponse(a, webConnectorType)
                     } else {
                         a
                     }

bot/connector-web/src/main/kotlin/WebRequestInfos.kt

@@ -16,10 +16,14 @@
 
 package ai.tock.bot.connector.web
 
+import ai.tock.bot.connector.ConnectorType
+import ai.tock.bot.connector.web.send.WebCarousel
+import ai.tock.bot.engine.action.SendSentence
 import com.google.common.cache.CacheBuilder
 import io.vertx.core.MultiMap
 import io.vertx.core.http.Cookie
 import io.vertx.core.http.HttpServerRequest
+import mu.KotlinLogging
 import java.util.concurrent.TimeUnit
 
 object WebRequestInfosByEvent {
@@ -40,7 +44,8 @@ object WebRequestInfosByEvent {
 data class WebRequestInfos(
     private val headers: MultiMap = MultiMap.caseInsensitiveMultiMap(),
     private val cookies: Set<Cookie> = emptySet(),
-    private val streamedResponse: StringBuilder = StringBuilder(),
+    @Volatile
+    private var streamedResponse: SendSentence? = null,
 ) {
     internal constructor(request: HttpServerRequest) : this(request.headers(), request.cookies())
 
@@ -50,14 +55,53 @@ data class WebRequestInfos(
 
     fun firstCookie(name: String): String? = cookies.firstOrNull { it.name == name }?.value
 
-    internal fun addStreamedResponse(response: String?): String {
-        if (response != null) {
-            streamedResponse.append(response)
-        }
-        return streamedResponse.toString()
+    internal fun addStreamedResponse(
+        response: SendSentence,
+        connectorType: ConnectorType,
+    ): SendSentence {
+        val s =
+            streamedResponse?.run {
+                val originalMessage = message(connectorType) as? WebMessage
+                if (originalMessage != null) {
+                    val newMessage = response.message(connectorType) as? WebMessage
+                    if (newMessage == null) {
+                        logger.warn { "no custom message in streamed message - but the previous message is custom - ignore" }
+                        return this
+                    }
+
+                    changeConnectorMessage(
+                        WebMessage(
+                            text = (originalMessage.text ?: "") + (newMessage.text ?: ""),
+                            buttons = originalMessage.buttons + newMessage.buttons,
+                            card =
+                                originalMessage.card?.let {
+                                    it.copy(
+                                        title = (it.title?.toString() ?: "") + (newMessage.card?.title?.toString() ?: ""),
+                                        subTitle = (it.subTitle?.toString() ?: "") + (newMessage.card?.subTitle?.toString() ?: ""),
+                                        file = newMessage.card?.file ?: it.file,
+                                        buttons = it.buttons + (newMessage.card?.buttons ?: emptyList()),
+                                    )
+                                } ?: newMessage.card,
+                            carousel = ((originalMessage.carousel?.cards ?: emptyList()) + (newMessage.carousel?.cards ?: emptyList())).takeUnless { it.isEmpty() }?.let { WebCarousel(it) },
+                            widget = newMessage.widget ?: originalMessage.widget,
+                            image = newMessage.image ?: originalMessage.image,
+                            version = newMessage.version,
+                            deepLink = newMessage.deepLink ?: originalMessage.deepLink,
+                            footnotes = originalMessage.footnotes + newMessage.footnotes,
+                        ),
+                    )
+                    this
+                } else {
+                    withText((text?.toString() ?: "") + (response.text?.toString() ?: ""))
+                }
+            } ?: response
+        streamedResponse = s
+        return s
     }
 
     internal fun clearStreamedResponse() {
-        streamedResponse.clear()
+        streamedResponse = null
     }
 }
+
+private val logger = KotlinLogging.logger {}

bot/engine/src/main/kotlin/engine/event/MetadataEvent.kt

@@ -40,7 +40,9 @@ class MetadataEvent(val type: String, val value: String, applicationId: String)
         const val STREAM_RESPONSE_METADATA = "TOCK_STREAM_RESPONSE"
     }
 
-    fun isEndStreamMetadata(): Boolean = type == STREAM_RESPONSE_METADATA && value != "true"
+    fun isEndStreamMetadata(): Boolean = isStreamMetadata() && value != "true"
+
+    fun isStreamMetadata(): Boolean = type == STREAM_RESPONSE_METADATA
 }
 
 fun Map<String, String>.hasStreamMetadata(): Boolean = this[STREAM_RESPONSE_METADATA] == "true"