WebSockets
This feature adds WebSockets support to Ktor.WebSockets are a mechanism to keep a bi-directional real-time ordered connection betweenthe server and the client.Each message from this channel is called Frame: a frame can be a text or binary message,or a close or ping/pong message. Frames can be marked as incomplete or final.
This feature is defined in the class io.ktor.websocket.WebSockets
in the artifact io.ktor:ktor-websockets:$ktor_version
.
dependencies { implementation "io.ktor:ktor-websockets:$ktor_version"}
dependencies { implementation("io.ktor:ktor-websockets:$ktor_version")}
<project> … <dependencies> <dependency> <groupId>io.ktor</groupId> <artifactId>ktor-websockets</artifactId> <version>${ktor.version}</version> <scope>compile</scope> </dependency> </dependencies></project>
Table of contents:
Installing
In order to use the WebSockets functionality you first have to install it:
install(WebSockets)
You can adjust a few parameters when installing if required:
install(WebSockets) {
pingPeriod = Duration.ofSeconds(60) // Disabled (null) by default
timeout = Duration.ofSeconds(15)
maxFrameSize = Long.MAX_VALUE // Disabled (max value). The connection will be closed if surpassed this length.
masking = false
}
Usage
Once installed, you can define the webSocket
routes for the routing feature:
Instead of the short-lived normal route handlers, webSocket handlers are meant to be long-lived.And all the relevant WebSocket methods are suspended so that the function will be suspended ina non-blocking way while receiving or sending messages.
webSocket
methods receive a callback with a WebSocketSessioninstance as the receiver. That interface defines an incoming
(ReceiveChannel) property and an outgoing
(SendChannel)property, as well as a close
method. Check the full WebSocketSession for more information.
Usage as an suspend actor
routing {
webSocket("/") { // websocketSession
for (frame in incoming) {
when (frame) {
is Frame.Text -> {
val text = frame.readText()
outgoing.send(Frame.Text("YOU SAID: $text"))
if (text.equals("bye", ignoreCase = true)) {
close(CloseReason(CloseReason.Codes.NORMAL, "Client said BYE"))
}
}
}
}
}
}
An exception will be thrown while receiving a Frame if the client closes the connectionexplicitly or the TCP socket is closed. So even with a while (true)
loop, this shouldn’t bea leak.
Usage as a Channel
Since the incoming
property is a ReceiveChannel, you can use it with its stream-like interface:
routing {
webSocket("/") { // websocketSession
for (frame in incoming.mapNotNull { it as? Frame.Text }) {
val text = frame.readText()
outgoing.send(Frame.Text("YOU SAID $text"))
if (text.equals("bye", ignoreCase = true)) {
close(CloseReason(CloseReason.Codes.NORMAL, "Client said BYE"))
}
}
}
}
Interface
The WebSocketSession interface
You receive a WebSocketSession as the receiver (this), giving you direct accessto these members inside your webSocket handler.
interface WebSocketSession {
// Basic interface
val incoming: ReceiveChannel<Frame> // Incoming frames channel
val outgoing: SendChannel<Frame> // Outgoing frames channel
fun close(reason: CloseReason)
// Convenience method equivalent to `outgoing.send(frame)`
suspend fun send(frame: Frame) // Enqueue frame, may suspend if the outgoing queue is full. May throw an exception if the outgoing channel is already closed, so it is impossible to transfer any message.
// The call and the context
val call: ApplicationCall
val application: Application
// Modifiable properties for this request. Their initial value comes from the feature configuration.
var pingInterval: Duration?
var timeout: Duration
var masking: Boolean // Enable or disable masking output messages by a random xor mask.
var maxFrameSize: Long // Specifies frame size limit. The connection will be closed if violated
// Advanced
val closeReason: Deferred<CloseReason?>
suspend fun flush() // Flush all outstanding messages and suspend until all earlier sent messages will be written. Could be called at any time even after close. May return immediately if connection is already terminated.
fun terminate() // Initiate connection termination immediately. Termination may complete asynchronously.
}
If you need information about the connection. For example the client ip, you have accessto the call property. So you can do things like call.request.origin.host
insideyour websocket block.
The Frame interface
A frame is each packet sent and received at the WebSocket protocol level.There are two message types: TEXT and BINARY. And three control packets: CLOSE, PING, and PONG.Each packet has a payload buffer
. And for Text or Close messages, you cancall the readText
or readReason
to interpret that buffer.
enum class FrameType { TEXT, BINARY, CLOSE, PING, PONG }
sealed class Frame {
val fin: Boolean // Is this frame a final frame?
val frameType: FrameType // The Type of the frame
val buffer: ByteBuffer // Payload
val disposableHandle: DisposableHandle
class Binary : Frame
class Text : Frame {
fun readText(): String
}
class Close : Frame {
fun readReason(): CloseReason?
}
class Ping : Frame
class Pong : Frame
}
Testing
You can test WebSocket conversations by using the handleWebSocketConversation
method inside a withTestApplication
block.
test.kt
class MyAppTest {
@Test
fun testConversation() {
withTestApplication {
application.install(WebSockets)
val received = arrayListOf<String>()
application.routing {
webSocket("/echo") {
try {
while (true) {
val text = (incoming.receive() as Frame.Text).readText()
received += text
outgoing.send(Frame.Text(text))
}
} catch (e: ClosedReceiveChannelException) {
// Do nothing!
} catch (e: Throwable) {
e.printStackTrace()
}
}
}
handleWebSocketConversation("/echo") { incoming, outgoing ->
val textMessages = listOf("HELLO", "WORLD")
for (msg in textMessages) {
outgoing.send(Frame.Text(msg))
assertEquals(msg, (incoming.receive() as Frame.Text).readText())
}
assertEquals(textMessages, received)
}
}
}
}
FAQ
Standard Events: onConnect, onMessage, onClose and onError
How do the standard events from the WebSocket API maps to Ktor?
onConnect
happens at the start of the block.onMessage
happens after successfully reading a message (for example withincoming.receive()
) or using suspended iteration withfor(frame in incoming)
.onClose
happens when theincoming
channel is closed. That would complete the suspended iteration, or throw aClosedReceiveChannelException
when trying to receive a message`.onError
is equivalent to other other exceptions.
In both onClose
and onError
, the closeReason
property is set.
To illustrate this:
webSocket("/echo") {
println("onConnect")
try {
for (frame in incoming){
val text = (frame as Frame.Text).readText()
println("onMessage")
received += text
outgoing.send(Frame.Text(text))
}
} catch (e: ClosedReceiveChannelException) {
println("onClose ${closeReason.await()}")
} catch (e: Throwable) {
println("onError ${closeReason.await()}")
e.printStackTrace()
}
}
In this sample, the infinite loop is only exited with an exception is risen: either a ClosedReceiveChannelException
or another exception.