Websockets implementation

10 files changed, 547 insertions, 7 deletions

diff --git a/kvision-modules/kvision-common-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt b/kvision-modules/kvision-common-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
index 514fb07d..e89744f6 100644
--- a/kvision-modules/kvision-common-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
+++ b/kvision-modules/kvision-common-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
@@ -21,6 +21,8 @@
  */
 package pl.treksoft.kvision.remote
 
+import kotlinx.coroutines.channels.ReceiveChannel
+import kotlinx.coroutines.channels.SendChannel
 import kotlin.reflect.KClass
 
 enum class HttpMethod {
@@ -115,4 +117,13 @@ expect open class KVServiceManager<T : Any>(serviceClass: KClass<T>) {
     protected fun bind(
         function: T.(String?, String?) -> List<RemoteSelectOption>
     )
+
+    /**
+     * Binds a given function of the receiver as a web socket connection
+     * @param function a function of the receiver
+     */
+    protected inline fun <reified PAR1 : Any, reified PAR2 : Any> bind(
+        noinline function: suspend T.(ReceiveChannel<PAR1>, SendChannel<PAR2>) -> Unit,
+        route: String? = null
+    )
 }
diff --git a/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVRemoteAgent.kt b/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVRemoteAgent.kt
index 6054592e..3e538bd1 100644
--- a/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVRemoteAgent.kt
+++ b/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVRemoteAgent.kt
@@ -21,10 +21,18 @@
  */
 package pl.treksoft.kvision.remote
 
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.asDeferred
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ReceiveChannel
+import kotlinx.coroutines.channels.SendChannel
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
 import kotlinx.serialization.ImplicitReflectionSerializer
 import kotlinx.serialization.list
 import kotlinx.serialization.serializer
+import kotlinx.serialization.stringify
+import kotlin.js.console
 import kotlin.js.js
 import kotlin.reflect.KClass
 import kotlin.js.JSON as NativeJSON
@@ -366,6 +374,178 @@ open class KVRemoteAgent<T : Any>(val serviceManager: KVServiceManager<T>) : Rem
         }.asDeferred().await()
     }
 
+    /**
+     * Executes defined web socket connection
+     */
+    suspend inline fun <reified PAR1 : Any, reified PAR2 : Any> webSocket(
+        noinline function: suspend T.(ReceiveChannel<PAR1>, SendChannel<PAR2>) -> Unit,
+        noinline handler: suspend (SendChannel<PAR1>, ReceiveChannel<PAR2>) -> Unit
+    ) {
+        val (url, _) =
+            serviceManager.getCalls()[function.toString().replace("\\s".toRegex(), "")]
+                ?: throw IllegalStateException("Function not specified!")
+        val socket = Socket()
+        val requestChannel = Channel<PAR1>()
+        val responseChannel = Channel<PAR2>()
+        try {
+            coroutineScope {
+                socket.connect(getWebSocketUrl(url))
+                lateinit var responseJob: Job
+                lateinit var handlerJob: Job
+                val requestJob = launch {
+                    for (par1 in requestChannel) {
+                        val param = serialize(par1)
+                        val str = JSON.plain.stringify(JsonRpcRequest(0, url, listOf(param)))
+                        if (!socket.sendOrFalse(str)) break
+                    }
+                    responseJob.cancel()
+                    handlerJob.cancel()
+                    if (!requestChannel.isClosedForReceive) requestChannel.close()
+                    if (!responseChannel.isClosedForSend) responseChannel.close()
+                }
+                responseJob = launch {
+                    while (true) {
+                        val str = socket.receiveOrNull() ?: break
+                        val data = kotlin.js.JSON.parse<dynamic>(str).result
+                        val par2 = try {
+                            @Suppress("UNCHECKED_CAST")
+                            deserialize<PAR2>(data, PAR2::class.js.name)
+                        } catch (t: NotStandardTypeException) {
+                            try {
+                                @Suppress("UNCHECKED_CAST")
+                                tryDeserializeEnum(PAR2::class as KClass<Any>, data) as PAR2
+                            } catch (t: NotEnumTypeException) {
+                                JSON.nonstrict.parse(PAR2::class.serializer(), data)
+                            }
+                        }
+                        responseChannel.send(par2)
+                    }
+                    requestJob.cancel()
+                    handlerJob.cancel()
+                    if (!requestChannel.isClosedForReceive) requestChannel.close()
+                    if (!responseChannel.isClosedForSend) responseChannel.close()
+                }
+                handlerJob = launch {
+                    exceptionHelper {
+                        handler(requestChannel, responseChannel)
+                    }
+                    requestJob.cancel()
+                    responseJob.cancel()
+                    if (!requestChannel.isClosedForReceive) requestChannel.close()
+                    if (!responseChannel.isClosedForSend) responseChannel.close()
+                }
+            }
+        } catch (e: Exception) {
+            console.log(e)
+        }
+        if (!requestChannel.isClosedForReceive) requestChannel.close()
+        if (!responseChannel.isClosedForSend) responseChannel.close()
+        socket.close()
+    }
+
+    /**
+     * Executes defined web socket connection returning list objects
+     */
+    suspend inline fun <reified PAR1 : Any, reified PAR2 : Any> webSocket(
+        noinline function: suspend T.(ReceiveChannel<PAR1>, SendChannel<List<PAR2>>) -> Unit,
+        noinline handler: suspend (SendChannel<PAR1>, ReceiveChannel<List<PAR2>>) -> Unit
+    ) {
+        val (url, _) =
+            serviceManager.getCalls()[function.toString().replace("\\s".toRegex(), "")]
+                ?: throw IllegalStateException("Function not specified!")
+        val socket = Socket()
+        val requestChannel = Channel<PAR1>()
+        val responseChannel = Channel<List<PAR2>>()
+        try {
+            coroutineScope {
+                socket.connect(getWebSocketUrl(url))
+                lateinit var responseJob: Job
+                lateinit var handlerJob: Job
+                val requestJob = launch {
+                    for (par1 in requestChannel) {
+                        val param = serialize(par1)
+                        val str = JSON.plain.stringify(JsonRpcRequest(0, url, listOf(param)))
+                        if (!socket.sendOrFalse(str)) break
+                    }
+                    responseJob.cancel()
+                    handlerJob.cancel()
+                    if (!requestChannel.isClosedForReceive) requestChannel.close()
+                    if (!responseChannel.isClosedForSend) responseChannel.close()
+                }
+                responseJob = launch {
+                    while (true) {
+                        val str = socket.receiveOrNull() ?: break
+                        val data = kotlin.js.JSON.parse<dynamic>(str).result
+                        val par2 = try {
+                            deserializeList<PAR2>(data, PAR2::class.js.name)
+                        } catch (t: NotStandardTypeException) {
+                            try {
+                                @Suppress("UNCHECKED_CAST")
+                                tryDeserializeEnumList(PAR2::class as KClass<Any>, data) as List<PAR2>
+                            } catch (t: NotEnumTypeException) {
+                                JSON.nonstrict.parse(PAR2::class.serializer().list, data)
+                            }
+                        }
+                        responseChannel.send(par2)
+                    }
+                    requestJob.cancel()
+                    handlerJob.cancel()
+                    if (!requestChannel.isClosedForReceive) requestChannel.close()
+                    if (!responseChannel.isClosedForSend) responseChannel.close()
+                }
+                handlerJob = launch {
+                    exceptionHelper {
+                        handler(requestChannel, responseChannel)
+                    }
+                    requestJob.cancel()
+                    responseJob.cancel()
+                    if (!requestChannel.isClosedForReceive) requestChannel.close()
+                    if (!responseChannel.isClosedForSend) responseChannel.close()
+                }
+            }
+        } catch (e: Exception) {
+            console.log(e)
+        }
+        if (!requestChannel.isClosedForReceive) requestChannel.close()
+        if (!responseChannel.isClosedForSend) responseChannel.close()
+        socket.close()
+    }
+
+    /**
+     * @suppress internal function
+     */
+    suspend fun Socket.receiveOrNull(): String? {
+        return try {
+            this.receive()
+        } catch (e: SocketClosedException) {
+            console.log("Socket was closed: ${e.reason}")
+            null
+        }
+    }
+
+    /**
+     * @suppress internal function
+     */
+    fun Socket.sendOrFalse(str: String): Boolean {
+        return try {
+            this.send(str)
+            true
+        } catch (e: SocketClosedException) {
+            console.log("Socket was closed: ${e.reason}")
+            false
+        }
+    }
+
+    /**
+     * @suppress internal function
+     */
+    suspend fun exceptionHelper(block: suspend () -> Unit) {
+        try {
+            block()
+        } catch (e: Exception) {
+            console.log(e)
+        }
+    }
 
     /**
      * @suppress
diff --git a/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt b/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
index 4ca6a4a5..0cf72f76 100644
--- a/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
+++ b/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
@@ -21,6 +21,8 @@
  */
 package pl.treksoft.kvision.remote
 
+import kotlinx.coroutines.channels.ReceiveChannel
+import kotlinx.coroutines.channels.SendChannel
 import kotlin.reflect.KClass
 
 /**
@@ -138,6 +140,19 @@ actual open class KVServiceManager<T : Any> actual constructor(serviceClass: KCl
     }
 
     /**
+     * Binds a given web socket connetion with a function of the receiver.
+     * @param function a function of the receiver
+     * @param route a web socket route
+     */
+    protected actual inline fun <reified PAR1 : Any, reified PAR2 : Any> bind(
+        noinline function: suspend T.(ReceiveChannel<PAR1>, SendChannel<PAR2>) -> Unit,
+        route: String?
+    ) {
+        val routeDef = "route${this::class.simpleName}${counter++}"
+        calls[function.toString().replace("\\s".toRegex(), "")] = Pair("/kv/$routeDef", HttpMethod.POST)
+    }
+
+    /**
      * Returns the map of defined paths.
      */
     fun getCalls(): Map<String, Pair<String, HttpMethod>> = calls
diff --git a/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/Socket.kt b/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/Socket.kt
new file mode 100644
index 00000000..c547ce9b
--- /dev/null
+++ b/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/Socket.kt
@@ -0,0 +1,171 @@
+/*
+ * Copyright (c) 2017-present Robert Jaros
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+package pl.treksoft.kvision.remote
+
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import org.w3c.dom.CloseEvent
+import org.w3c.dom.ErrorEvent
+import org.w3c.dom.MessageEvent
+import org.w3c.dom.WebSocket
+import org.w3c.dom.WebSocket.Companion.CLOSED
+import org.w3c.dom.WebSocket.Companion.CLOSING
+import org.w3c.dom.WebSocket.Companion.OPEN
+import org.w3c.dom.events.Event
+import kotlin.coroutines.resume
+import kotlin.coroutines.suspendCoroutine
+
+//
+// Code taken from: https://discuss.kotlinlang.org/t/js-coroutines-experiements/8245/2
+//
+
+/**
+ * Websocket closed exception class.
+ */
+class SocketClosedException(val reason: String) : Throwable(reason)
+
+/**
+ * A websocket client implementation.
+ */
+class Socket {
+    private var eventQueue: Channel<Event> = Channel(Channel.UNLIMITED)
+    private lateinit var ws: WebSocket
+    val state: Short
+        get() = ws.readyState
+
+    private fun onWsEvent(event: Event) {
+        GlobalScope.launch { eventQueue.send(event) }
+    }
+
+    /**
+     * Connect to a websocket.
+     */
+    suspend fun connect(url: String, retryDelay: Long = 1000) {
+        while (true) {
+            val connected = suspendCoroutine<Boolean> { cont ->
+                while (eventQueue.poll() != null) {/*drain*/
+                }
+                ws = WebSocket(url)
+                ws.onopen = {
+                    ws.onclose = ::onWsEvent
+                    ws.onerror = ::onWsEvent
+                    cont.resume(true)
+                }
+                ws.onmessage = ::onWsEvent
+                ws.onerror = {
+                    logError(it)
+                }
+                ws.onclose = {
+                    cont.resume(false)
+                }
+            }
+            if (connected)
+                break
+            else
+                delay(retryDelay)
+        }
+    }
+
+    /**
+     * Receive a string from a websocket.
+     */
+    suspend fun receive(): String {
+        val event = eventQueue.receive()
+        return when (event) {
+            is MessageEvent -> {
+                event.data as String
+            }
+            is CloseEvent -> {
+                val reason = getReason(event.code)
+                throw SocketClosedException(reason)
+            }
+            is ErrorEvent -> {
+                logError(event)
+                close()
+                throw SocketClosedException(event.message)
+            }
+            else -> {
+                val reason = getReason(4001)
+                console.error(reason)
+                close()
+                throw SocketClosedException(reason)
+            }
+        }
+    }
+
+    /**
+     * Send string to a websocket.
+     */
+    fun send(obj: String) {
+        when {
+            isClosed() -> {
+                console.error(getReason(4002))
+                throw SocketClosedException(getReason(4002))
+            }
+            else -> ws.send(obj)
+        }
+    }
+
+    /**
+     * Close a websocket.
+     */
+    fun close(code: Short = 1000) {
+        when (state) {
+            OPEN -> ws.close(code, getReason(1000))
+        }
+    }
+
+    /**
+     * Returns if a websocket is closed.
+     */
+    fun isClosed(): Boolean {
+        return when (state) {
+            CLOSED, CLOSING -> true
+            else -> false
+        }
+    }
+
+    private fun logError(event: Event) = console.error("An error %o occurred when connecting to ${ws.url}", event)
+
+    private fun getReason(code: Short): String {
+        return when (code.toInt()) { // See http://tools.ietf.org/html/rfc6455#section-7.4.1
+            1000 -> "Normal closure"
+            1001 -> "An endpoint is \"going away\", such as a server going down or a browser having navigated away from a page."
+            1002 -> "An endpoint is terminating the connection due to a protocol error"
+            1003 -> "An endpoint is terminating the connection because it has received a type of data it cannot accept (e.g., an endpoint that understands only text data MAY send this if it receives a binary message)."
+            1004 -> "Reserved. The specific meaning might be defined in the future."
+            1005 -> "No status code was actually present."
+            1006 -> "The connection was closed abnormally, e.g., without sending or receiving a Close control frame"
+            1007 -> "An endpoint is terminating the connection because it has received data within a message that was not consistent with the type of the message (e.g., non-UTF-8 [http://tools.ietf.org/html/rfc3629] data within a text message)."
+            1008 -> "An endpoint is terminating the connection because it has received a message that \"violates its policy\". This reason is given either if there is no other sutible reason, or if there is a need to hide specific details about the policy."
+            1009 -> "An endpoint is terminating the connection because it has received a message that is too big for it to process."
+            1010 -> "An endpoint (client ) is terminating the connection because it has expected the server to negotiate one or more extension, but the server didn't return them in the response message of the WebSocket handshake. <br /> Specifically, the extensions that are needed are: "
+            1011 -> "A server is terminating the connection because it encountered an unexpected condition that prevented it from fulfilling the request."
+            1015 -> "The connection was closed due to a failure to perform a TLS handshake (e.g., the server certificate can't be verified)."
+            4001 -> "Unexpected event"
+            4002 -> "You are trying to use closed socket"
+            else -> "Unknown reason"
+        }
+    }
+}
diff --git a/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/Utils.kt b/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/Utils.kt
index 273c2a65..c4f4ed6d 100644
--- a/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/Utils.kt
+++ b/kvision-modules/kvision-remote/src/main/kotlin/pl/treksoft/kvision/remote/Utils.kt
@@ -25,6 +25,7 @@ import kotlinx.serialization.SerializationStrategy
 import kotlinx.serialization.context.SimpleModule
 import kotlinx.serialization.json.Json
 import pl.treksoft.kvision.types.JsonDateSerializer
+import kotlin.browser.window
 import kotlin.js.Date
 
 /**
@@ -60,3 +61,14 @@ object JSON {
         return kotlin.js.JSON.parse(plain.stringify(serializer, this))
     }
 }
+
+/**
+ * Creates a websocket URL from current window.location and given path.
+ */
+fun getWebSocketUrl(url: String): String {
+    val location = window.location
+    val scheme = if (location.protocol == "https:") "wss" else "ws"
+    val port = if (location.port == "8088") ":8080"
+    else if (location.port != "0" && location.port != "") ":${location.port}" else ""
+    return "$scheme://${location.hostname}$port$url"
+}
diff --git a/kvision-modules/kvision-server-jooby/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt b/kvision-modules/kvision-server-jooby/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
index d1e64ce4..cfd9d467 100644
--- a/kvision-modules/kvision-server-jooby/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
+++ b/kvision-modules/kvision-server-jooby/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
@@ -25,6 +25,8 @@ import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import com.google.inject.Injector
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.channels.ReceiveChannel
+import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.launch
 import org.jooby.Kooby
 import org.jooby.Request
@@ -311,6 +313,19 @@ actual open class KVServiceManager<T : Any> actual constructor(val serviceClass:
     }
 
     /**
+     * Binds a given web socket connetion with a function of the receiver.
+     * @param function a function of the receiver
+     * @param route a route
+     */
+    protected actual inline fun <reified PAR1 : Any, reified PAR2 : Any> bind(
+        noinline function: suspend T.(ReceiveChannel<PAR1>, SendChannel<PAR2>) -> Unit,
+        route: String?
+    ) {
+        TODO("Not implemented in Jooby module")
+    }
+
+
+    /**
      * Binds a given function of the receiver as a select options source
      * @param function a function of the receiver
      */
diff --git a/kvision-modules/kvision-server-ktor/build.gradle b/kvision-modules/kvision-server-ktor/build.gradle
index 62b5fe83..44be9171 100644
--- a/kvision-modules/kvision-server-ktor/build.gradle
+++ b/kvision-modules/kvision-server-ktor/build.gradle
@@ -10,6 +10,7 @@ dependencies {
     compile "org.jetbrains.kotlinx:kotlinx-coroutines-core:$coroutinesVersion"
     compile "io.ktor:ktor-server-core:$ktorVersion"
     compile "io.ktor:ktor-jackson:$ktorVersion"
+    compile "io.ktor:ktor-websockets:$ktorVersion"
     compile "com.google.inject:guice:$guiceVersion"
     compile "com.fasterxml.jackson.module:jackson-module-kotlin:$jacksonModuleKotlinVersion"
     testCompile "org.jetbrains.kotlin:kotlin-test:$kotlinVersion"
diff --git a/kvision-modules/kvision-server-ktor/src/main/kotlin/pl/treksoft/kvision/remote/KVModules.kt b/kvision-modules/kvision-server-ktor/src/main/kotlin/pl/treksoft/kvision/remote/KVModules.kt
index 956e7301..e2edc175 100644
--- a/kvision-modules/kvision-server-ktor/src/main/kotlin/pl/treksoft/kvision/remote/KVModules.kt
+++ b/kvision-modules/kvision-server-ktor/src/main/kotlin/pl/treksoft/kvision/remote/KVModules.kt
@@ -31,17 +31,25 @@ import io.ktor.application.ApplicationCallPipeline
 import io.ktor.application.call
 import io.ktor.application.install
 import io.ktor.features.ContentNegotiation
+import io.ktor.http.cio.websocket.Frame
 import io.ktor.http.content.defaultResource
 import io.ktor.http.content.resources
 import io.ktor.http.content.static
 import io.ktor.jackson.jackson
 import io.ktor.routing.routing
 import io.ktor.util.AttributeKey
+import io.ktor.util.KtorExperimentalAPI
+import io.ktor.websocket.WebSocketServerSession
+import io.ktor.websocket.WebSockets
+import kotlinx.coroutines.channels.ReceiveChannel
+import kotlinx.coroutines.channels.SendChannel
+import kotlin.coroutines.CoroutineContext
 
 fun Application.kvisionInit(vararg modules: Module) {
     install(ContentNegotiation) {
         jackson()
     }
+    install(WebSockets)
     routing {
         static("/") {
             resources("assets")
@@ -72,3 +80,50 @@ class MainModule(private val application: Application) : AbstractModule() {
         bind(Application::class.java).toInstance(application)
     }
 }
+
+class WsSessionModule(private val webSocketSession: WebSocketServerSession) :
+    AbstractModule() {
+    override fun configure() {
+        bind(WebSocketServerSession::class.java).toInstance(webSocketSession)
+    }
+}
+
+class DummyWsSessionModule() : AbstractModule() {
+    override fun configure() {
+        bind(WebSocketServerSession::class.java).toInstance(DummyWebSocketServerSession())
+    }
+}
+
+class DummyWebSocketServerSession : WebSocketServerSession {
+    override val call: ApplicationCall
+        get() = throw UnsupportedOperationException()
+    override val coroutineContext: CoroutineContext
+        get() = throw UnsupportedOperationException()
+    override val incoming: ReceiveChannel<Frame>
+        get() = throw UnsupportedOperationException()
+    override var masking: Boolean
+        get() = throw UnsupportedOperationException()
+        set(value) {
+            throw UnsupportedOperationException()
+        }
+    override var maxFrameSize: Long
+        get() = throw UnsupportedOperationException()
+        set(value) {
+            throw UnsupportedOperationException()
+        }
+    override val outgoing: SendChannel<Frame>
+        get() = throw UnsupportedOperationException()
+
+    @UseExperimental(KtorExperimentalAPI::class)
+    override suspend fun close(cause: Throwable?) {
+        throw UnsupportedOperationException()
+    }
+
+    override suspend fun flush() {
+        throw UnsupportedOperationException()
+    }
+
+    override fun terminate() {
+        throw UnsupportedOperationException()
+    }
+}
diff --git a/kvision-modules/kvision-server-ktor/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt b/kvision-modules/kvision-server-ktor/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
index d4985adf..0839a5c5 100644
--- a/kvision-modules/kvision-server-ktor/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
+++ b/kvision-modules/kvision-server-ktor/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
@@ -24,6 +24,10 @@ package pl.treksoft.kvision.remote
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import io.ktor.application.ApplicationCall
 import io.ktor.application.call
+import io.ktor.http.cio.websocket.CloseReason
+import io.ktor.http.cio.websocket.Frame
+import io.ktor.http.cio.websocket.close
+import io.ktor.http.cio.websocket.readText
 import io.ktor.request.receive
 import io.ktor.response.respond
 import io.ktor.routing.Route
@@ -33,7 +37,17 @@ import io.ktor.routing.options
 import io.ktor.routing.post
 import io.ktor.routing.put
 import io.ktor.util.pipeline.PipelineContext
+import io.ktor.websocket.WebSocketServerSession
+import io.ktor.websocket.webSocket
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ReceiveChannel
+import kotlinx.coroutines.channels.SendChannel
+import kotlinx.coroutines.channels.filterNotNull
+import kotlinx.coroutines.channels.map
+import kotlinx.coroutines.channels.mapNotNull
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import kotlin.reflect.KClass
@@ -55,6 +69,8 @@ actual open class KVServiceManager<T : Any> actual constructor(val serviceClass:
         mutableMapOf()
     val optionsRequests: MutableMap<String, suspend PipelineContext<Unit, ApplicationCall>.(Unit) -> Unit> =
         mutableMapOf()
+    val webSocketRequests: MutableMap<String, suspend WebSocketServerSession.() -> Unit> =
+        mutableMapOf()
 
     val mapper = jacksonObjectMapper()
     var counter: Int = 0
@@ -72,7 +88,7 @@ actual open class KVServiceManager<T : Any> actual constructor(val serviceClass:
     ) {
         val routeDef = route ?: "route${this::class.simpleName}${counter++}"
         addRoute(method, "/kv/$routeDef") {
-            val service = call.injector.getInstance(serviceClass.java)
+            val service = call.injector.createChildInjector(DummyWsSessionModule()).getInstance(serviceClass.java)
             val jsonRpcRequest = if (method == HttpMethod.GET) {
                 JsonRpcRequest(call.request.queryParameters["id"]?.toInt() ?: 0, "", listOf())
             } else {
@@ -113,7 +129,7 @@ actual open class KVServiceManager<T : Any> actual constructor(val serviceClass:
             throw UnsupportedOperationException("GET method is only supported for methods without parameters")
         val routeDef = route ?: "route${this::class.simpleName}${counter++}"
         addRoute(method, "/kv/$routeDef") {
-            val service = call.injector.getInstance(serviceClass.java)
+            val service = call.injector.createChildInjector(DummyWsSessionModule()).getInstance(serviceClass.java)
             val jsonRpcRequest = call.receive<JsonRpcRequest>()
             if (jsonRpcRequest.params.size == 1) {
                 val param = getParameter<PAR>(jsonRpcRequest.params[0])
@@ -160,7 +176,7 @@ actual open class KVServiceManager<T : Any> actual constructor(val serviceClass:
             throw UnsupportedOperationException("GET method is only supported for methods without parameters")
         val routeDef = route ?: "route${this::class.simpleName}${counter++}"
         addRoute(method, "/kv/$routeDef") {
-            val service = call.injector.getInstance(serviceClass.java)
+            val service = call.injector.createChildInjector(DummyWsSessionModule()).getInstance(serviceClass.java)
             val jsonRpcRequest = call.receive<JsonRpcRequest>()
             if (jsonRpcRequest.params.size == 2) {
                 val param1 = getParameter<PAR1>(jsonRpcRequest.params[0])
@@ -208,7 +224,7 @@ actual open class KVServiceManager<T : Any> actual constructor(val serviceClass:
             throw UnsupportedOperationException("GET method is only supported for methods without parameters")
         val routeDef = route ?: "route${this::class.simpleName}${counter++}"
         addRoute(method, "/kv/$routeDef") {
-            val service = call.injector.getInstance(serviceClass.java)
+            val service = call.injector.createChildInjector(DummyWsSessionModule()).getInstance(serviceClass.java)
             val jsonRpcRequest = call.receive<JsonRpcRequest>()
             @Suppress("MagicNumber")
             if (jsonRpcRequest.params.size == 3) {
@@ -258,7 +274,7 @@ actual open class KVServiceManager<T : Any> actual constructor(val serviceClass:
             throw UnsupportedOperationException("GET method is only supported for methods without parameters")
         val routeDef = route ?: "route${this::class.simpleName}${counter++}"
         addRoute(method, "/kv/$routeDef") {
-            val service = call.injector.getInstance(serviceClass.java)
+            val service = call.injector.createChildInjector(DummyWsSessionModule()).getInstance(serviceClass.java)
             val jsonRpcRequest = call.receive<JsonRpcRequest>()
             @Suppress("MagicNumber")
             if (jsonRpcRequest.params.size == 4) {
@@ -310,7 +326,7 @@ actual open class KVServiceManager<T : Any> actual constructor(val serviceClass:
             throw UnsupportedOperationException("GET method is only supported for methods without parameters")
         val routeDef = route ?: "route${this::class.simpleName}${counter++}"
         addRoute(method, "/kv/$routeDef") {
-            val service = call.injector.getInstance(serviceClass.java)
+            val service = call.injector.createChildInjector(DummyWsSessionModule()).getInstance(serviceClass.java)
             val jsonRpcRequest = call.receive<JsonRpcRequest>()
             @Suppress("MagicNumber")
             if (jsonRpcRequest.params.size == 5) {
@@ -348,6 +364,51 @@ actual open class KVServiceManager<T : Any> actual constructor(val serviceClass:
     }
 
     /**
+     * Binds a given web socket connetion with a function of the receiver.
+     * @param function a function of the receiver
+     * @param route a route
+     */
+    protected actual inline fun <reified PAR1 : Any, reified PAR2 : Any> bind(
+        noinline function: suspend T.(ReceiveChannel<PAR1>, SendChannel<PAR2>) -> Unit,
+        route: String?
+    ) {
+        val routeDef = "route${this::class.simpleName}${counter++}"
+        webSocketRequests["/kv/$routeDef"] = {
+            val wsInjector = call.injector.createChildInjector(WsSessionModule(this))
+            val service = wsInjector.getInstance(serviceClass.java)
+            val requestChannel = incoming.mapNotNull { it as? Frame.Text }.map {
+                val jsonRpcRequest = getParameter<JsonRpcRequest>(it.readText())
+                if (jsonRpcRequest.params.size == 1) {
+                    getParameter<PAR1>(jsonRpcRequest.params[0])
+                } else {
+                    null
+                }
+            }.filterNotNull()
+            val responseChannel = Channel<PAR2>()
+            val session = this
+            coroutineScope {
+                launch {
+                    for (p in responseChannel) {
+                        val text = mapper.writeValueAsString(
+                            JsonRpcResponse(
+                                id = 0,
+                                result = mapper.writeValueAsString(p)
+                            )
+                        )
+                        outgoing.send(Frame.Text(text))
+                    }
+                    session.close(CloseReason(CloseReason.Codes.NORMAL, ""))
+                    session.close()
+                }
+                launch {
+                    function.invoke(service, requestChannel, responseChannel)
+                    if (!responseChannel.isClosedForReceive) responseChannel.close()
+                }
+            }
+        }
+    }
+
+    /**
      * Binds a given function of the receiver as a select options source
      * @param function a function of the receiver
      */
@@ -357,7 +418,7 @@ actual open class KVServiceManager<T : Any> actual constructor(val serviceClass:
     ) {
         val routeDef = "route${this::class.simpleName}${counter++}"
         addRoute(HttpMethod.POST, "/kv/$routeDef") {
-            val service = call.injector.getInstance(serviceClass.java)
+            val service = call.injector.createChildInjector(DummyWsSessionModule()).getInstance(serviceClass.java)
             val jsonRpcRequest = call.receive<JsonRpcRequest>()
             if (jsonRpcRequest.params.size == 2) {
                 val param1 = getParameter<String?>(jsonRpcRequest.params[0])
@@ -431,4 +492,9 @@ fun <T : Any> Route.applyRoutes(serviceManager: KVServiceManager<T>) {
     serviceManager.optionsRequests.forEach { (path, handler) ->
         options(path, handler)
     }
+    serviceManager.webSocketRequests.forEach { (path, handler) ->
+        this.webSocket(path) {
+            handler()
+        }
+    }
 }
diff --git a/kvision-modules/kvision-server-spring-boot/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt b/kvision-modules/kvision-server-spring-boot/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
index 66dc4b99..66429acf 100644
--- a/kvision-modules/kvision-server-spring-boot/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
+++ b/kvision-modules/kvision-server-spring-boot/src/main/kotlin/pl/treksoft/kvision/remote/KVServiceManager.kt
@@ -25,6 +25,8 @@ import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.channels.ReceiveChannel
+import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.launch
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -392,6 +394,18 @@ actual open class KVServiceManager<T : Any> actual constructor(val serviceClass:
     }
 
     /**
+     * Binds a given web socket connetion with a function of the receiver.
+     * @param function a function of the receiver
+     * @param route a route
+     */
+    protected actual inline fun <reified PAR1 : Any, reified PAR2 : Any> bind(
+        noinline function: suspend T.(ReceiveChannel<PAR1>, SendChannel<PAR2>) -> Unit,
+        route: String?
+    ) {
+        TODO("Not implemented in Spring Boot module")
+    }
+
+    /**
      * Binds a given function of the receiver as a select options source
      * @param function a function of the receiver
      */