Merge branch '846-sse-support'

Author: vsct-jburet

bot/connector-web/pom.xml

@@ -33,6 +33,15 @@
             <artifactId>tock-bot-engine-jackson</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.litote.kmongo</groupId>
+            <artifactId>kmongo</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.litote.kmongo</groupId>
+            <artifactId>kmongo-async</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>ai.tock</groupId>
             <artifactId>tock-shared</artifactId>

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

@@ -25,6 +25,8 @@ import ai.tock.bot.connector.media.MediaCard
 import ai.tock.bot.connector.media.MediaCarousel
 import ai.tock.bot.connector.media.MediaFile
 import ai.tock.bot.connector.media.MediaMessage
+import ai.tock.bot.connector.web.channel.ChannelMongoDAO
+import ai.tock.bot.connector.web.channel.Channels
 import ai.tock.bot.connector.web.send.UrlButton
 import ai.tock.bot.connector.web.send.WebCard
 import ai.tock.bot.connector.web.send.WebCarousel
@@ -36,21 +38,27 @@ import ai.tock.bot.engine.event.Event
 import ai.tock.bot.engine.user.PlayerId
 import ai.tock.bot.engine.user.UserPreferences
 import ai.tock.shared.Executor
+import ai.tock.shared.booleanProperty
 import ai.tock.shared.injector
 import ai.tock.shared.jackson.mapper
 import ai.tock.shared.provide
+import com.fasterxml.jackson.databind.module.SimpleModule
+import com.fasterxml.jackson.databind.ser.std.ToStringSerializer
 import com.fasterxml.jackson.module.kotlin.readValue
 import io.vertx.core.http.HttpMethod
 import io.vertx.ext.web.RoutingContext
 import io.vertx.ext.web.handler.CorsHandler
 import mu.KotlinLogging
 
 internal const val WEB_CONNECTOR_ID = "web"
+
 /**
  * The web (REST) connector type.
  */
 val webConnectorType = ConnectorType(WEB_CONNECTOR_ID)
 
+private val sseEnabled = booleanProperty("tock_web_sse", false)
+
 class WebConnector internal constructor(
     val applicationId: String,
     val path: String
@@ -62,18 +70,53 @@ class WebConnector internal constructor(
 
     private val executor: Executor get() = injector.provide()
 
+    private val channels = Channels(ChannelMongoDAO)
+
+    private val webMapper = mapper.copy().registerModule(
+        SimpleModule().apply {
+            //fallback for serializing CharSequence
+            addSerializer(CharSequence::class.java, ToStringSerializer())
+        }
+    )
+
     override fun register(controller: ConnectorController) {
+
         controller.registerServices(path) { router ->
             logger.debug("deploy web connector services for root path $path ")
 
             router.route(path)
                 .handler(
                     CorsHandler.create("*")
                         .allowedMethod(HttpMethod.POST)
+                        .run {
+                            if (sseEnabled) allowedMethod(HttpMethod.GET) else this
+                        }
                         .allowedHeader("Access-Control-Allow-Origin")
                         .allowedHeader("Content-Type")
                         .allowedHeader("X-Requested-With")
                 )
+            if (sseEnabled) {
+                router.route(path + "/sse")
+                    .handler { context ->
+                        try {
+                            val userId = context.queryParams()["userId"]
+                            val response = context.response()
+                            response.isChunked = true
+                            response.headers().add("Content-Type", "text/event-stream;charset=UTF-8")
+                            response.headers().add("Connection", "keep-alive")
+                            response.headers().add("Cache-Control", "no-cache")
+                            val channelId = channels.register(userId) { webConnectorResponse ->
+                                response.write("event: message\n")
+                                response.write("data: ${webMapper.writeValueAsString(webConnectorResponse)}\n\n")
+                            }
+                            response.closeHandler {
+                                channels.unregister(channelId)
+                            }
+                        } catch (t: Throwable) {
+                            context.fail(t)
+                        }
+                    }
+            }
             router.post(path)
                 .handler { context ->
                     try {
@@ -96,7 +139,7 @@ class WebConnector internal constructor(
         try {
             logger.debug { "Web request input : $body" }
             val request: WebConnectorRequest = mapper.readValue(body)
-            val callback = WebConnectorCallback(applicationId, request.locale, context)
+            val callback = WebConnectorCallback(applicationId = applicationId, locale = request.locale, context = context, webMapper = webMapper)
             controller.handle(request.toEvent(applicationId), ConnectorData(callback))
         } catch (t: Throwable) {
             BotRepository.requestTimer.throwable(t, timerData)
@@ -110,6 +153,7 @@ class WebConnector internal constructor(
         val c = callback as? WebConnectorCallback
         c?.addAction(event)
         if (event is Action) {
+            channels.send(event)
             if (event.metadata.lastAnswer) {
                 c?.sendResponse()
             }
@@ -141,7 +185,7 @@ class WebConnector internal constructor(
                     WebMessage(card = WebCard(
                         title = message.title,
                         subTitle = message.subTitle,
-                        file = message.file?.url?.let{MediaFile(message.file?.url as String, message.file?.name as String)},
+                        file = message.file?.url?.let { MediaFile(message.file?.url as String, message.file?.name as String) },
                         buttons = message.actions.map { UrlButton(it.title.toString(), it.url.toString()) }
                     ))
                 }
@@ -150,7 +194,7 @@ class WebConnector internal constructor(
                         WebCard(
                             title = mediaCard.title,
                             subTitle = mediaCard.subTitle,
-                            file = mediaCard.file?.url?.let{MediaFile(mediaCard.file?.url as String, mediaCard.file?.name as String)},
+                            file = mediaCard.file?.url?.let { MediaFile(mediaCard.file?.url as String, mediaCard.file?.name as String) },
                             buttons = mediaCard.actions.map { button -> UrlButton(button.title.toString(), button.url.toString()) }
                         )
                     }))

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

@@ -16,13 +16,11 @@
 
 package ai.tock.bot.connector.web
 
-import com.fasterxml.jackson.databind.module.SimpleModule
-import com.fasterxml.jackson.databind.ser.std.ToStringSerializer
 import ai.tock.bot.connector.ConnectorCallbackBase
 import ai.tock.bot.engine.action.Action
 import ai.tock.bot.engine.action.SendSentence
 import ai.tock.bot.engine.event.Event
-import ai.tock.shared.jackson.mapper
+import com.fasterxml.jackson.databind.ObjectMapper
 import io.vertx.ext.web.RoutingContext
 import mu.KotlinLogging
 import java.util.Locale
@@ -32,16 +30,11 @@ internal class WebConnectorCallback(
     applicationId: String,
     val locale: Locale,
     private val context: RoutingContext,
-    private val actions: MutableList<Action> = CopyOnWriteArrayList()
+    private val actions: MutableList<Action> = CopyOnWriteArrayList(),
+    private val webMapper: ObjectMapper
 ) : ConnectorCallbackBase(applicationId, webConnectorType) {
 
     private val logger = KotlinLogging.logger {}
-    private val webMapper = mapper.copy().registerModule(
-        SimpleModule().apply {
-            //fallback for serializing CharSequence
-            addSerializer(CharSequence::class.java, ToStringSerializer())
-        }
-    )
 
     fun addAction(event: Event) {
         if (event is Action) {

bot/connector-web/src/main/kotlin/channel/Channel.kt

@@ -0,0 +1,20 @@
+/*
+ * Copyright (C) 2017/2020 e-voyageurs technologies
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.tock.bot.connector.web.channel
+
+import java.util.UUID
+
+internal data class Channel(val uuid: UUID, val userId: String, val onAction: ChannelCallback)

bot/connector-web/src/main/kotlin/channel/ChannelCallback.kt

@@ -0,0 +1,20 @@
+/*
+ * Copyright (C) 2017/2020 e-voyageurs technologies
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.tock.bot.connector.web.channel
+
+import ai.tock.bot.connector.web.WebConnectorResponse
+
+internal typealias ChannelCallback = (webConnectorResponse: WebConnectorResponse) -> Unit

bot/connector-web/src/main/kotlin/channel/ChannelDAO.kt

@@ -0,0 +1,21 @@
+/*
+ * Copyright (C) 2017/2020 e-voyageurs technologies
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.tock.bot.connector.web.channel
+
+internal interface ChannelDAO {
+    fun listenChanges(listener: (channelEvent: ChannelEvent) -> Unit)
+    fun save(channelEvent: ChannelEvent)
+}

bot/connector-web/src/main/kotlin/channel/ChannelEvent.kt

@@ -0,0 +1,20 @@
+/*
+ * Copyright (C) 2017/2020 e-voyageurs technologies
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.tock.bot.connector.web.channel
+
+import ai.tock.bot.connector.web.WebConnectorResponse
+
+internal data class ChannelEvent(val recipientId: String, val webConnectorResponse: WebConnectorResponse)

bot/connector-web/src/main/kotlin/channel/ChannelMongoDAO.kt

@@ -0,0 +1,77 @@
+/*
+ * Copyright (C) 2017/2020 e-voyageurs technologies
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.tock.bot.connector.web.channel
+
+import ai.tock.shared.TOCK_BOT_DATABASE
+import ai.tock.shared.error
+import ai.tock.shared.injector
+import ai.tock.shared.watch
+import com.github.salomonbrys.kodein.instance
+import com.mongodb.client.MongoCollection
+import com.mongodb.client.MongoDatabase
+import com.mongodb.client.model.CreateCollectionOptions
+import mu.KotlinLogging
+import org.litote.kmongo.getCollection
+import org.litote.kmongo.reactivestreams.getCollectionOfName
+import org.litote.kmongo.save
+
+internal object ChannelMongoDAO : ChannelDAO {
+    private val collectionName = "web_channel_event"
+    private val asyncDatabase: com.mongodb.reactivestreams.client.MongoDatabase by injector.instance(
+        TOCK_BOT_DATABASE
+    )
+    private val database: MongoDatabase by injector.instance(
+        TOCK_BOT_DATABASE
+    )
+    private val logger = KotlinLogging.logger {}
+    private val asyncWebChannelResponseCol: com.mongodb.reactivestreams.client.MongoCollection<ChannelEvent>
+    private val webChannelResponseCol: MongoCollection<ChannelEvent>
+
+    private fun MongoDatabase.collectionExists(collectionName: String): Boolean =
+        listCollectionNames().contains(collectionName)
+
+    init {
+        if(!database.collectionExists(collectionName)) {
+            try {
+                database
+                    .createCollection(
+                        collectionName,
+                        CreateCollectionOptions()
+                            .capped(true)
+                            .sizeInBytes(100000000)
+                            .maxDocuments(50000)
+                    )
+            } catch (ex: Exception) {
+                logger.error(ex)
+            }
+        }
+        asyncWebChannelResponseCol = asyncDatabase.getCollectionOfName(collectionName)
+        webChannelResponseCol = database.getCollection<ChannelEvent>(collectionName)
+    }
+
+    override fun listenChanges(listener: (channelEvent: ChannelEvent) -> Unit) {
+        asyncWebChannelResponseCol.watch {
+            val channelEvent = it.fullDocument
+            if (channelEvent != null)
+                listener(channelEvent)
+        }
+    }
+
+    override fun save(channelEvent: ChannelEvent) {
+        webChannelResponseCol.save(channelEvent)
+    }
+
+}