Micronaut Serialization + Kotlin Flow + SSE need some non-obvious configuration (docs needed?)

[mitushy]

This is not really an issue, but probably needs some better documentation - not sure if it belongs on the main SSE page or the Micronaut Serialization page or the Kotlin one. I'm leaning towards the SSE section.

I have a small Micronaut + Kotlin project where I wanted to give Micronaut Serialization a try. Everything worked as advertised, except for my SSE endpoints. Example:

    @Get("/time", produces = [MediaType.TEXT_EVENT_STREAM])
    fun time(): Flow<Event<Map<String, Long>>> = flow {
        while (true) {
            emit(Event.of(mapOf("t" to System.currentTimeMillis())))
            delay(1000)
        }
    }

This throws:

io.micronaut.http.codec.CodecException: Error encoding object [kotlinx.coroutines.flow.SafeFlow@794d5fb2] to JSON: No serializable introspection present for type SafeFlow. Consider adding Serdeable. Serializable annotate to type SafeFlow. Alternatively if you are not in control of the project's source code, you can use @SerdeImport(SafeFlow.class) to enable serialization of this type.
	at io.micronaut.json.codec.MapperMediaTypeCodec.encode(MapperMediaTypeCodec.java:234)
	at io.micronaut.http.server.codec.TextStreamCodec.encode(TextStreamCodec.java:158)
	at io.micronaut.http.server.netty.RoutingInBoundHandler.lambda$mapToHttpContent$7(RoutingInBoundHandler.java:1104)
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106)
	at io.micronaut.reactive.reactor.instrument.ReactorSubscriber.onNext(ReactorSubscriber.java:57)
	at io.micronaut.core.async.publisher.Publishers$JustPublisher$1.request(Publishers.java:563)
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:164)
	at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.request(FluxContextWrite.java:136)
	at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.request(FluxHide.java:152)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:144)
	at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.request(FluxHide.java:152)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:144)
	at reactor.core.publisher.StrictSubscriber.request(StrictSubscriber.java:138)
	at io.micronaut.http.netty.reactive.HandlerSubscriber.maybeRequestMore(HandlerSubscriber.java:222)
	at io.micronaut.http.netty.reactive.HandlerSubscriber.maybeStart(HandlerSubscriber.java:266)
	at io.micronaut.http.netty.reactive.HandlerSubscriber.provideSubscription(HandlerSubscriber.java:253)
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:566)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: io.micronaut.serde.exceptions.SerdeException: No serializable introspection present for type SafeFlow. Consider adding Serdeable. Serializable annotate to type SafeFlow. Alternatively if you are not in control of the project's source code, you can use @SerdeImport(SafeFlow.class) to enable serialization of this type.
	at io.micronaut.serde.support.serializers.ObjectSerializer$4.tryToFindSerializer(ObjectSerializer.java:209)
	at io.micronaut.serde.support.serializers.ObjectSerializer$RuntimeTypeSerializer.lambda$getSerializer$0(ObjectSerializer.java:264)
	at java.base/java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1708)
	at io.micronaut.serde.support.serializers.ObjectSerializer$RuntimeTypeSerializer.getSerializer(ObjectSerializer.java:262)
	at io.micronaut.serde.support.serializers.ObjectSerializer$RuntimeTypeSerializer.serialize(ObjectSerializer.java:230)
	at io.micronaut.serde.jackson.JacksonJsonMapper.writeValue(JacksonJsonMapper.java:108)
	at io.micronaut.serde.jackson.JacksonJsonMapper.writeValue0(JacksonJsonMapper.java:100)
	at io.micronaut.serde.jackson.JacksonJsonMapper.writeValue0(JacksonJsonMapper.java:94)
	at io.micronaut.serde.jackson.JacksonJsonMapper.writeValueAsBytes(JacksonJsonMapper.java:189)
	at io.micronaut.json.codec.MapperMediaTypeCodec.encode(MapperMediaTypeCodec.java:231)
	... 23 common frames omitted

The fix was to add

 implementation "org.jetbrains.kotlinx:kotlinx-coroutines-reactive:${coroutinesVersion}"

as described in this issue micronaut-kotlin/issues/63.

I couldn't find anything else about this on the docs - so we should probably document it somewhere 😄