Server-Sent Events com Ktor e Kotlin: SSE em produção

Resposta rápida: use Server-Sent Events (SSE) com Ktor quando o servidor precisa enviar atualizações contínuas ao cliente — progresso de processamento, notificações, status de pedidos, dashboards ou tokens de uma IA — sem a comunicação bidirecional de um WebSocket. No Ktor, instale o suporte a SSE, exponha uma rota sse, envie objetos ServerSentEvent e trate cancelamento, heartbeat e reconexão. Em produção, use SharedFlow ou um broker para distribuir eventos, autentique a requisição HTTP inicial e configure proxy e observabilidade para conexões longas.

SSE ocupa um espaço útil entre polling e WebSocket. O cliente abre uma requisição HTTP, o servidor responde com Content-Type: text/event-stream e mantém a conexão aberta enquanto envia eventos de texto. O protocolo é simples, funciona bem com infraestrutura HTTP tradicional e possui reconexão nativa no EventSource dos navegadores.

Este guia mostra a implementação completa com Kotlin e Ktor, desde a primeira rota até uma arquitetura com Flow, IDs retomáveis, autenticação, testes e múltiplas réplicas. Se você ainda está preparando routing e serialização, comece pelo tutorial de API REST com Ktor e volte para adicionar o canal de eventos.

Quando escolher SSE, WebSocket ou polling?

A escolha depende da direção e da frequência da comunicação.

Escolha SSE quando o cliente já envia comandos por REST e só precisa receber atualizações. Um painel pode criar uma exportação com POST /exports e acompanhar o progresso em GET /exports/{id}/events. Não há motivo para manter um protocolo bidirecional apenas para receber “20%, 40%, concluído”.

Prefira WebSockets com Ktor quando os dois lados enviam mensagens frequentes na mesma conexão, como em chat ou edição colaborativa. Use polling quando a atualização acontece raramente e alguns segundos de atraso não afetam a experiência.

Como o protocolo Server-Sent Events funciona?

Cada evento é um bloco de linhas UTF-8 separado por uma linha vazia:

id: pedido-42:17
event: pedido_atualizado
retry: 5000
data: {"pedidoId":"42","status":"EM_ROTA"}

Os campos mais úteis são:

• data: conteúdo do evento; pode ser JSON, mas continua sendo texto;
• event: nome usado para distinguir tipos de evento;
• id: identificador que ajuda o cliente a retomar o stream;
• retry: sugestão, em milissegundos, para a próxima tentativa de conexão;
• comentários iniciados por :: úteis como heartbeat sem evento de negócio.

SSE não define um schema de domínio. Você decide o formato de data e deve versioná-lo como qualquer contrato de API. Para modelos Kotlin mais complexos, aplique as práticas de kotlinx.serialization e polimorfismo.

Dependências do Ktor

Mantenha os artefatos Ktor na mesma versão e centralize essa versão no catalog do Gradle:

// build.gradle.kts

dependencies {
    implementation("io.ktor:ktor-server-core:<versao-atual>")
    implementation("io.ktor:ktor-server-netty:<versao-atual>")
    implementation("io.ktor:ktor-server-sse:<versao-atual>")
    implementation("io.ktor:ktor-server-content-negotiation:<versao-atual>")
    implementation("io.ktor:ktor-serialization-kotlinx-json:<versao-atual>")

    testImplementation("io.ktor:ktor-server-test-host:<versao-atual>")
    testImplementation(kotlin("test"))
}

Para consumir SSE com o Ktor Client, adicione o artefato correspondente:

dependencies {
    implementation("io.ktor:ktor-client-core:<versao-atual>")
    implementation("io.ktor:ktor-client-cio:<versao-atual>")
    implementation("io.ktor:ktor-client-sse:<versao-atual>")
}

Os placeholders evitam congelar uma combinação que ficará antiga. Use uma versão estável e compatível com Kotlin, engine e plugins do projeto.

Criando o primeiro endpoint SSE

Instale o plugin e exponha uma rota. O exemplo envia cinco atualizações de progresso:

import io.ktor.server.application.Application
import io.ktor.server.routing.routing
import io.ktor.server.sse.SSE
import io.ktor.server.sse.sse
import io.ktor.sse.ServerSentEvent
import kotlinx.coroutines.delay

fun Application.module() {
    install(SSE)

    routing {
        sse("/exports/{id}/events") {
            val exportId = call.parameters["id"]
                ?: return@sse

            for (progress in listOf(0, 25, 50, 75, 100)) {
                send(
                    ServerSentEvent(
                        data = """{"exportId":"$exportId","progress":$progress}""",
                        event = "export_progress",
                        id = "$exportId:$progress",
                        retry = 5_000,
                    )
                )
                delay(1_000)
            }
        }
    }
}

Esse código demonstra o protocolo, mas um endpoint real não deve fabricar progresso dentro da rota. O processamento normalmente acontece em outro serviço, coroutine, fila ou worker. A rota apenas observa os eventos autorizados e os transmite para a conexão atual.

Também não construa JSON por interpolação em produção. Modele o payload e serialize com Json.encodeToString para escapar corretamente aspas, quebras de linha e caracteres Unicode.

Modelando eventos tipados

Um contrato explícito evita strings mágicas espalhadas pelo backend:

import kotlinx.serialization.SerialName
import kotlinx.serialization.Serializable

@Serializable
sealed interface ExportEvent {
    val exportId: String

    @Serializable
    @SerialName("progress")
    data class Progress(
        override val exportId: String,
        val percentage: Int,
    ) : ExportEvent

    @Serializable
    @SerialName("completed")
    data class Completed(
        override val exportId: String,
        val downloadUrl: String,
    ) : ExportEvent

    @Serializable
    @SerialName("failed")
    data class Failed(
        override val exportId: String,
        val code: String,
    ) : ExportEvent
}

O URL de download deve ter expiração curta ou exigir autorização. Não envie stack trace, caminho interno, token ou detalhes de infraestrutura em Failed. O cliente precisa de um código estável e de uma mensagem apropriada para a interface.

Distribuindo eventos com SharedFlow

Para uma única instância do serviço, SharedFlow fornece uma ponte natural entre o worker e as conexões SSE:

import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow

class ExportEventBus {
    private val mutableEvents = MutableSharedFlow<ExportEvent>(
        extraBufferCapacity = 64,
        onBufferOverflow = BufferOverflow.DROP_OLDEST,
    )

    val events: SharedFlow<ExportEvent> = mutableEvents.asSharedFlow()

    suspend fun publish(event: ExportEvent) {
        mutableEvents.emit(event)
    }
}

A rota filtra apenas o job solicitado e serializa cada atualização:

import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.collect
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json

fun Application.configureExportEvents(bus: ExportEventBus) {
    val json = Json {
        classDiscriminator = "type"
        encodeDefaults = true
    }

    fun eventId(event: ExportEvent): String = when (event) {
        is ExportEvent.Progress -> "${event.exportId}:progress:${event.percentage}"
        is ExportEvent.Completed -> "${event.exportId}:completed"
        is ExportEvent.Failed -> "${event.exportId}:failed:${event.code}"
    }

    install(SSE)

    routing {
        sse("/exports/{id}/events") {
            val exportId = call.parameters["id"]
                ?: return@sse

            bus.events
                .filter { it.exportId == exportId }
                .collect { event ->
                    send(
                        ServerSentEvent(
                            data = json.encodeToString<ExportEvent>(event),
                            event = event::class.simpleName,
                            id = eventId(event),
                            retry = 5_000,
                        )
                    )
                }
        }
    }
}

A coroutine que executa collect acompanha o ciclo de vida da requisição. Quando o cliente desconecta, o trabalho deve ser cancelado. Não capture CancellationException como erro genérico e não reinicie o loop; cancelamento é o encerramento esperado de uma conexão longa.

O que acontece com cliente lento?

Esse é um ponto de arquitetura, não apenas de sintaxe. Se o produtor gera cem eventos por segundo e o cliente consome dez, alguma camada terá de:

• bloquear o produtor;
• criar uma fila crescente;
• agregar atualizações;
• ou descartar eventos antigos.

Para progresso e dashboards, descartar estados intermediários costuma ser aceitável: o cliente precisa do estado mais recente, não de cada percentual. Para auditoria financeira ou comandos, descarte pode ser proibido — e SSE não deve ser o armazenamento oficial. Persista o evento e use o stream apenas como mecanismo de entrega.

Consumindo SSE com Ktor Client

No cliente Kotlin, instale o plugin SSE e colete os eventos recebidos:

import io.ktor.client.HttpClient
import io.ktor.client.engine.cio.CIO
import io.ktor.client.plugins.sse.SSE
import io.ktor.client.plugins.sse.sse

val client = HttpClient(CIO) {
    install(SSE)
}

suspend fun acompanharExportacao(exportId: String) {
    client.sse("https://api.exemplo.com/exports/$exportId/events") {
        incoming.collect { event ->
            println("tipo=${event.event}")
            println("id=${event.id}")
            println("data=${event.data}")
        }
    }
}

No Android, execute a coleta em um escopo com ciclo de vida definido, como o viewModelScope. Converta o payload para estado de UI e deixe o ViewModel decidir se deve reconectar. Evite abrir uma nova conexão a cada recomposição no Jetpack Compose.

No navegador, a API nativa é simples:

const source = new EventSource("/exports/42/events");

source.addEventListener("export_progress", (event) => {
  const payload = JSON.parse(event.data);
  console.log(payload.progress);
});

source.onerror = () => {
  console.log("Conexão interrompida; o navegador tentará reconectar.");
};

EventSource possui limitações importantes: a API padrão não permite configurar livremente o header Authorization. Para aplicações web, cookies seguros, uma sessão no mesmo domínio ou um ticket SSE curto trocado antes da conexão costumam ser melhores que colocar bearer token na query string.

Reconexão, IDs e Last-Event-ID

Uma rede móvel pode trocar de antena, entrar em túnel ou suspender o app. A conexão SSE vai cair. O sistema precisa assumir reconexões como parte normal do protocolo.

Quando o servidor envia id, clientes compatíveis podem informar o último identificador recebido em Last-Event-ID ao reconectar. Isso permite três estratégias:

1. estado atual: ignore o histórico e envie um snapshot novo;
2. replay curto: recupere os eventos posteriores ao último ID;
3. stream durável: leia a partir de uma sequência persistida em Kafka, Redis Streams ou banco.

Para progresso de exportação, o estado atual geralmente basta. Para uma tela de cotações, um snapshot seguido de deltas é melhor. Para eventos que não podem ser perdidos, persista-os antes de publicar.

Um ID útil é monotônico e estável, por exemplo exportId:sequence. Não use timestamp sozinho se dois eventos puderem compartilhar o mesmo milissegundo.

Heartbeat e conexões aparentemente ociosas

Proxies, CDNs e balanceadores encerram conexões sem tráfego após um período. Se o produto pode passar minutos sem evento, envie heartbeat periódico — normalmente um comentário SSE ou um evento técnico pequeno.

O heartbeat tem três funções:

• mantém a conexão ativa através da infraestrutura;
• detecta cliente desconectado mais cedo;
• ajuda a medir saúde e duração do stream.

Ele não deve virar ruído de negócio. Não persista heartbeat como notificação e não faça a UI reagir a ele. Ajuste o intervalo abaixo do timeout do menor componente da cadeia, sem criar milhares de mensagens desnecessárias.

Autenticação e autorização

SSE começa como uma requisição HTTP comum. Isso permite validar sessão, bearer token, tenant e acesso ao recurso antes de manter a conexão.

Com Ktor Authentication, proteja a rota da mesma forma que uma API privada:

authenticate("auth-jwt") {
    sse("/exports/{id}/events") {
        val principal = call.principal<JWTPrincipal>()
            ?: return@sse
        val exportId = call.parameters["id"]
            ?: return@sse

        val canRead = exportRepository.belongsToUser(
            exportId = exportId,
            userId = principal.payload.subject,
        )

        if (!canRead) return@sse

        // coleta e envio dos eventos autorizados
    }
}

O exemplo omite a resposta de erro para destacar a regra. No projeto real, responda 401 ou 403 antes de iniciar o stream. Depois que os headers e parte do corpo foram enviados, trocar para uma resposta JSON tradicional de erro já não é simples.

Nunca autorize apenas porque o cliente conhece o ID do recurso. Combine este fluxo com as práticas de Ktor Authentication e JWT e aplique rate limiting também à criação de conexões.

Escalando para várias réplicas

SharedFlow vive na memória de um processo. Em Kubernetes ou em qualquer deploy com mais de uma instância, o worker pode publicar na réplica A enquanto o usuário está conectado à réplica B.

Há três desenhos comuns:

1. Broker Pub/Sub

Redis Pub/Sub, NATS ou RabbitMQ distribuem eventos para todas as réplicas interessadas. É simples e rápido, mas mensagens podem não ficar disponíveis para replay.

2. Stream durável

Kafka ou Redis Streams guardam sequência e offset. Essa opção facilita Last-Event-ID, recuperação e auditoria, ao custo de maior operação.

3. Banco + notificação

O estado fica no PostgreSQL e um canal de notificação avisa que algo mudou. A réplica busca o estado atual antes de enviar. Funciona bem quando a frequência é moderada e o banco já é a fonte de verdade.

Sticky session pode reduzir trocas de réplica durante a conexão, mas não resolve publicação cruzada nem replay. Trate afinidade como otimização, não como barramento.

Proxy, compressão e buffering

Uma rota SSE pode funcionar localmente e parecer travada em produção porque um proxy está acumulando bytes antes de repassá-los. Verifique:

• buffering de resposta desativado para a rota;
• timeout de leitura maior que o intervalo de heartbeat;
• HTTP keep-alive habilitado;
• limite de conexões por origem e por usuário;
• compressão testada, pois buffers de compressão podem atrasar eventos pequenos;
• cache desabilitado para text/event-stream;
• CDN compatível com streaming de longa duração.

Teste a cadeia completa: cliente → CDN → ingress → proxy → Ktor. Um curl -N ajuda a confirmar se cada evento chega imediatamente:

curl -N \
  -H 'Accept: text/event-stream' \
  https://api.exemplo.com/exports/42/events

A opção -N desativa o buffer de saída do próprio curl. Sem ela, você pode culpar o servidor por um atraso criado no cliente de teste.

Observabilidade para conexões longas

Métricas HTTP tradicionais podem enganar. Uma requisição SSE saudável pode durar horas, então “latência alta” não significa lentidão.

Acompanhe pelo menos:

• conexões abertas por instância, rota e tenant;
• novas conexões e reconexões por minuto;
• duração da conexão;
• eventos e bytes enviados;
• clientes lentos e descartes por backpressure;
• falhas de autenticação;
• encerramentos por timeout, cancelamento e erro;
• atraso entre criação do evento e entrega.

Crie um span de conexão com cuidado: spans de várias horas podem custar caro e atrapalhar o backend de tracing. Uma alternativa é medir abertura/fechamento e criar spans curtos para etapas relevantes. O guia de OpenTelemetry com Kotlin, Spring e Ktor ajuda a definir traces e métricas sem registrar payload sensível.

Testando uma rota SSE

Divida os testes em camadas.

Teste de domínio

Teste o produtor e o filtro sem HTTP:

@Test
fun `evento de outra exportacao nao deve chegar ao assinante`() = runTest {
    val bus = ExportEventBus()
    val received = async {
        bus.events
            .filter { it.exportId == "export-1" }
            .first()
    }

    bus.publish(ExportEvent.Progress("export-2", 10))
    bus.publish(ExportEvent.Progress("export-1", 20))

    assertEquals(
        ExportEvent.Progress("export-1", 20),
        received.await(),
    )
}

Teste HTTP

Com testApplication, abra o endpoint como stream, leia eventos e encerre o cliente após receber o caso esperado. Use timeout no teste para evitar pipeline pendurado.

Cubra também:

• usuário sem sessão recebe 401;
• usuário de outra conta recebe 403;
• evento possui id, event e JSON válido;
• desconexão cancela o coletor;
• heartbeat aparece antes do timeout esperado;
• replay respeita Last-Event-ID;
• cliente lento segue a política definida;
• falha do broker não cria loop agressivo.

Para dependências reais como PostgreSQL, Redis ou Kafka, aplique a estratégia de Testcontainers com Kotlin.

Erros comuns com Ktor SSE

Usar SSE para comunicação bidirecional

Se o cliente envia comandos contínuos, você terminará combinando SSE com dezenas de POSTs e uma coordenação difícil. Avalie WebSocket.

Criar uma coroutine global por conexão

Uma coroutine fora do ciclo de vida da chamada pode continuar coletando depois que o usuário foi embora. Mantenha o trabalho estruturado sob a requisição.

Ignorar backpressure

Fila sem limite transforma cliente lento em vazamento de memória. Defina agregação, buffer ou descarte de maneira explícita.

Colocar token na URL

Query strings aparecem em logs, histórico, analytics e proxies. Prefira cookie seguro, header em clientes que suportam ou ticket curto de conexão.

Não enviar ID

Sem IDs, retomada e diagnóstico ficam mais difíceis. Mesmo quando o sistema envia apenas snapshot, uma sequência ajuda a observar duplicatas e ordem.

Tratar desconexão como erro crítico

Clientes móveis desconectam. Cancelamento esperado não deve gerar alerta de incidente nem stack trace em nível ERROR.

Confiar apenas em memória com múltiplas réplicas

Eventos publicados em outra instância não chegarão ao cliente. Use um broker ou consulte uma fonte compartilhada.

Esquecer o proxy

Buffering e timeout são as causas clássicas do “funciona local, não funciona em produção”. Valide o caminho externo antes do lançamento.

Checklist de produção

Antes do deploy, confirme:

1. SSE é realmente unidirecional e adequado ao caso;
2. autenticação e autorização acontecem antes do stream;
3. payload possui contrato e versão compatível;
4. eventos relevantes têm ID estável;
5. reconexão e replay estão definidos;
6. heartbeat é menor que o timeout da infraestrutura;
7. proxy não faz buffering da resposta;
8. buffer e backpressure possuem limite;
9. múltiplas réplicas compartilham eventos por broker ou armazenamento;
10. cancelamento encerra coleta e recursos;
11. métricas distinguem duração saudável de falha;
12. logs não contêm token nem payload sensível;
13. há limite de conexões por usuário ou tenant;
14. testes cobrem desconexão, autorização e cliente lento;
15. um endpoint de estado permite recuperar o snapshot após perda de eventos.

Perguntas frequentes

SSE funciona com JSON?

Sim. O campo data é texto, e JSON é o formato mais comum para payload estruturado. Serialize com kotlinx.serialization e trate o schema como contrato versionado.

SSE substitui WebSocket?

Não. SSE é excelente para servidor → cliente. WebSocket é melhor quando ambos enviam mensagens frequentes pela mesma conexão.

Posso usar SSE em Android e Compose?

Sim. Consuma com Ktor Client em um escopo controlado pelo ViewModel e exponha o estado para a UI. Não abra a conexão dentro de uma recomposição sem gerenciamento de ciclo de vida.

Como autenticar EventSource no navegador?

A API padrão não aceita header Authorization arbitrário. Use cookie HttpOnly e Secure quando a arquitetura permitir, ou troque uma sessão autenticada por um ticket SSE curto e de uso limitado. Evite token permanente na URL.

SSE funciona atrás de Nginx ou CDN?

Funciona se streaming for suportado e o buffering estiver desativado para a rota. Ajuste timeouts, heartbeat, cache e limites de conexão e teste a infraestrutura real.

Preciso salvar todos os eventos?

Não para todo caso. Progresso pode usar apenas estado atual. Eventos que não podem ser perdidos precisam de armazenamento durável; o stream não deve ser a única fonte de verdade.

Ktor SSE reconecta o cliente automaticamente?

O comportamento depende do cliente. EventSource no navegador possui reconexão automática; clientes Kotlin devem aplicar uma política controlada, com atraso e limite, além de respeitar o ciclo de vida da aplicação.

Conclusão

Server-Sent Events com Ktor é uma solução enxuta para streaming unidirecional em Kotlin. A primeira rota exige pouco código, mas a qualidade de produção depende das decisões ao redor: contrato tipado, IDs retomáveis, heartbeat, cancelamento estruturado, autorização por recurso, backpressure, broker entre réplicas e configuração correta do proxy.

Comece com um caso em que SSE encaixa naturalmente — progresso de exportação, status de pedido ou dashboard. Exponha também um endpoint REST de snapshot, trate reconexão como rotina e meça conexões abertas e atraso de entrega. Quando os requisitos passarem a exigir comandos frequentes nos dois sentidos, migre conscientemente para WebSocket em vez de forçar o protocolo.

Para completar a arquitetura, conecte este guia ao backend com Ktor, à autenticação JWT, ao Ktor Client resiliente e à observabilidade com OpenTelemetry. Assim, o stream deixa de ser uma demo de conexão aberta e vira uma parte operável da API.