KTOR-7435 Add serialization for SSE

[marychatte]

Subsystem
Client SSE, Server SSE

Motivation
KTOR-7435

[bjhham]

Please check the comments about including TypeInfo in the default config so that we can use different types in our sessions.

[bjhham]

Did you consider using overloads instead of the default deserialize argument here to allow the original API sans type argument when using strings?

By that, I mean something like...

public suspend fun HttpClient.sseSession(
    reconnectionTime: Duration? = null,
    showCommentEvents: Boolean? = null,
    showRetryEvents: Boolean? = null,
    block: HttpRequestBuilder.() -> Unit
) = sseSession({ it }, reconnectionTime, showCommentEvents, showRetryEvents, block)

public suspend fun <T : Any> HttpClient.sseSession(
    deserialize: (String) -> T,
    reconnectionTime: Duration? = null,
    showCommentEvents: Boolean? = null,
    showRetryEvents: Boolean? = null,
    block: HttpRequestBuilder.() -> Unit
)

This would make it so the default deserialization can't be used, however.

[bjhham]

I think maybe ignore this suggestion and just include typeInfo in the default lambda signature. We'll probably want some logic to ensure that when String is supplied as the type argument that we avoid any serialization, otherwise you'd end up with some weirdness with it expecting strings to be wrapped in quotes.

[bjhham]

Looks like we're missing the typeInfo here required to solve the general use-case for deserializing different types. I think it may need to be (TypeInfo) -> (String) -> Any so when using a StringFormat like Json, we can supply it like deserialize = { typeInfo -> Json.serializersModule.serializer(typeInfo).let { serializer -> {{ Json.decodeFromString(serializer, it) }} } }. Rather ugly, but we could provide some helpers somewhere.

[bjhham]

From the API side, this will need to use reified T and typeOf<T>() to pass to the session.

[bjhham]

Would like to see this work with different types on the same server.

[bjhham]

Would be nice if this was a data class, so instead of having:

incoming.collect { person ->
   println(person.data)
}

We could have

incoming.collect { (person) ->
    println(person)
}

[e5l]

I would keep SSESession without generic types and introduce a new type for the session with deserialization. It will be more explicit and simplify maintenance in the long run

[marychatte]

@e5l @bjhham Thanks for the comments, please take a look at the new commits

Right now, for the client-side:

1. Two types of sessions: SSESession and SSESessionWithDeserialization:

public interface SSESession : CoroutineScope {
    public val incoming: Flow<ServerSentEvent<String>>
}

public interface SSESessionWithDeserialization: SSESession {
    public val deserializer: (TypeInfo) -> (String) -> Any
}

2. The difference in builder functions is in the parameter deserialize. This parameter is required for all builder functions for SSESessionWithDeserialization: deserialize: (TypeInfo) -> (String) -> Any
3. Example of SSESessionWithDeserialization usage:

@Serializable
data class Customer(val id: Int, val firstName: String, val lastName: String)
@Serializable
data class Product(val name: String, val price: Int)

client.sse({
    url("$TEST_SERVER/sse/json")
}, deserialize = { typeInfo: TypeInfo ->
    { jsonString: String ->
         val serializer = Json.serializersModule.serializer(typeInfo.kotlinType!!)
         Json.decodeFromString(serializer, jsonString) ?: Exception()
     }
}) {
    incoming.collect { event ->
        if (customer) {
            val customer = deserialize<Customer>(event.data)
            assertEquals("Jet", customer?.firstName)
        } else {
           val product = deserialize<Product>(event.data)
           assertEquals(100, product?.price)
        }
   }
}

Example of SSESession usage:

val session = client.serverSentEventsSession("$TEST_SERVER/sse/hello")
val event: ServerSentEvent<String> = session.incoming.single()
assertEquals("0", event.id)

For the server-side:

1. Also two types of sessions, SSESession and SSESessionWithSerialization:

public interface SSESession : CoroutineScope {
    public val call: ApplicationCall

    public suspend fun send(event: ServerSentEvent<String>)
}

public interface SSESessionWithSerialization : SSESession {
    public val serializer: (TypeInfo) -> (Any) -> String
}
public suspend inline fun <reified T : Any> SSESessionWithSerialization.sendSerialized(event: ServerSentEvent<T>) {
    send(
        ServerSentEvent(
            event.data?.let {
                serializer(typeInfo<T>()).invoke(it)
            },
            event.event,
            event.id,
            event.retry,
            event.comments
        )
    )
}

2. Parameter serialize is required for all builder functions for SSESessionWithSerialization: serialize: (TypeInfo) -> (Any) -> String
3. Example of SSESession usage:

sse("/hello") {
      send(ServerSentEvent("world", event = "send", id = "100", retry = 1000, comments = "comment"))
}

Example of SSESessionWithSerialization usage:

class Person1(val age: Int)
class Person2(val number: Int)
sse(serialize = {
    typeInfo ->
    { data ->
        when (typeInfo.type) {
            Person1::class -> {
                "Age ${(data as Person1).age}"
            }

            Person2::class -> {
                "Number ${(data as Person2).number}"
            }

            else -> {
                data.toString()
            }
        }
    }
}) {
    sendSerialized(Person1(22))
    sendSerialized(Person2(123456))
}

Questions:

1. Is configuring the serialization/deserialization functions only through function parameters and not SSE config okay?
2. Are we agreed on the types for serialization/deserialization functions deserialize: (TypeInfo) -> (String) -> Any and serialize: (TypeInfo) -> (Any) -> String?

[e5l]

Should we also split this class in 2 so String would be default?

[marychatte]

I split into ServerSentEvent and ParameterizedServerSentEvent<T>

[e5l]

Could you tell me why it's not just fun deserialize(TypeInfo, String) -> Any?

[marychatte]

Added Deserializer and Serializer interface:

public interface Deserializer {
    public fun deserialize(typeInfo: TypeInfo, data: String): Any
}

Do we want to make them functional because now it's used like:

client.sse(
    { url("$TEST_SERVER/sse/person") },
    object : Deserializer {
        override fun deserialize(typeInfo: TypeInfo, data: String): Any {
            return Person1(data)
    }
}) {
    incoming.single().apply {
        assertEquals("Name 0", deserialize<Person1>(data)?.name)
    }
}

[e5l]

I would avoid having generics on the sse method

[bjhham]

We can just overload send here to avoid the wordiness of sendSerialized.

[bjhham]

I think we can just use the lambda expression (typeInfo, data) -> String for the serializer field, since the argument provides context and this interface clutters up the API.

[marychatte]

@e5l suggested using an interface in case we would need to change the API or add something. But (typeInfo, data) -> String seems like final thing

[bjhham]

I'd go with something like:

sealed interface ServerSentEventMetadata<T> {
    val data: T?
    val event: String?
    val id: String?
    val retry: Long?
    val comments: String?
}

data class ServerSentEvent(...): ServerSentEventMetadata<String>
data class ServerSentEventParsed(...): ServerSentEventMetadata<T>

[bjhham]

With the removal of the default deserializer in the client config, I think it would simplify things to just go back to (String) -> E as the argument for the session and have the incoming return the parsed events from incoming.

[marychatte]

Do you mean have incoming: Flow<ServerSentEventParsed<*>>?

[bjhham]

I'd be interested in seeing an example with kotlinx-serialization-json, since this is probably the most common use-case for this feature. We ought to shape the API so that this is easy to implement.

[marychatte]

Here is used kotlinx-serialization-json

ktor/ktor-server/ktor-server-plugins/ktor-server-sse/common/test/io/ktor/server/sse/ServerSentEventsTest.kt

Line 291 in 9ce8ec5

return Json.encodeToString(serializer, data)

[e5l]

Let's think about naming. ServerSentEventParsed looks inconsistent.

[marychatte]

How about TypedServerSentEvent?

[e5l]

lgtm

[e5l]

could we avoid functional fields?

[marychatte]

@e5l It was an interface, but after #4363 (comment) I reverted to the functional field

[marychatte]

@e5l @bjhham Please, take a look. Right now, there are two important things to discuss:

- Do we want to introduce functional fields or interfaces for Deserializer and Serializer?

With interfaces it may be easier to change the API in the future, but it consists only of one function, and I don't see potential ways for changing. And with functional fields it's easier for users to implement. For example:

1. Interfaces

client.sse(
    { URL(...) },
    object : Deserializer {
        override fun deserialize(typeInfo: TypeInfo, data: String): Any {
            val serializer = Json.serializersModule.serializer(typeInfo.kotlinType!!)
            Json.decodeFromString(serializer, data) ?: Exception()
        } 
    }
) {
    ...
}

2. Functional fields

client.sse(
    { URL(...) },
    { typeInfo, data ->
        val serializer = Json.serializersModule.serializer(typeInfo.kotlinType!!)
        Json.decodeFromString(serializer, data) ?: Exception()
    }
) {
    ...
}

I prefer the functional fields approach

- What type should incoming channel of SSESessionWithDeserialization have?

Suggested variant:

 public val incoming: Flow<TypedServerSentEvent<String>>

So during the session to get the typed event, we will need to call deserialize function:

client.sse({
    url(...)
}, deserialize = {
    typeInfo, jsonString ->
    val serializer = Json.serializersModule.serializer(typeInfo.kotlinType!!)
    Json.decodeFromString(serializer, jsonString) ?: Exception()
}) {
    incoming.collect { event: TypedServerSentEvent<String> ->
        when (event.event) {
            "customer" -> {
                val customer: Customer? = deserialize<Customer>(event.data)
                ...
            }
            "product" -> {
                val product: Product? = deserialize<Product>(event.data)
                ...
            }
        }
    }
}

[bjhham]

LGTM 👍

It would be nice to have some bridge code for the JSON scenario where you don't have to worry about the deserialization args, though we don't really have a precedent for convenience libraries for specific plugins... maybe it could be introduced to the sample code in the registry, idk, food for thought.

Also, please rebase on 3.1 EAP

[e5l]

Please provide more descriptive API docs with usage examples, edge cases, and references

[e5l]

Please consider adding a reference on what SSE is on the wiki page

[e5l]

it also would be great to highlight where the receiver type is used in the example

[marychatte]

Updated KDoc to:
A session for handling Server-Sent Events (SSE) from a server.

Example of usage:

client.sse("http://localhost:8080/sse") { // `this` is `ClientSSESession`
    incoming.collect { event ->
        println("Id: ${event.id}")
        println("Event: ${event.event}")
        println("Data: ${event.data}")
    }
}

To learn more, see the SSE and the SSE specification.

[e5l]

@vnikolova, could you help us check KDocs?

[vnikolova]

The KDocs look clear - I made some minor suggestions.
I'm not sure if it's a common practice, but it might be useful to add comments in the code examples for added clarity.