SSE 与 WebSocket¶

network-realtime 提供有界 SSE 协议解析、SSE/WebSocket 托管重连，以及 Android OkHttp / iOS NSURLSession WebSocket 实现。

SSE 增量解析¶

Transport 只需提供任意分片边界的文本 Flow：

val events: Flow<ServerSentEvent> = parseServerSentEvents(textChunks)

events.collect { event ->
    when (event.event) {
        "message" -> renderMessage(event.data)
        "done" -> finish()
    }
}

Parser 支持：

data: 多行合并。
event:、id:、retry:。
\r\n 标准化。
网络分片落在任意字符块边界。
以 : 开头的注释行。
最大行长与最大事件长度保护，异常输入不会无限增长内存。

ServerSentEvent：

data class ServerSentEvent(
    val data: String,
    val event: String? = null,
    val id: String? = null,
    val retryMillis: Long? = null,
)

只需要解释已有文本流时，直接使用 Parser。需要连接生命周期管理时，注入一个轻量 Transport：

val events = observeServerSentEvents(
    transport = NetworkSseTransport { url, headers ->
        eventStreamTransport.openTextStream(url, headers)
    },
    url = "https://events.example.com/v1/stream",
    headersProvider = { mapOf("Authorization" to "Bearer ${tokenStore.current()}") },
    reconnectPolicy = RealtimeReconnectPolicy(maxReconnectAttempts = 5),
)

托管 SSE 会在每次连接前重新调用 headersProvider，以免缓存过期令牌；重连自动携带 Last-Event-ID，包括正确处理仅含控制字段的事件块和空 id: 清除语义。服务端 retry: 会覆盖下一次指数退避时间，但仍受最大延迟约束。协议越界属于不可恢复输入，不会反复重连。

WebSocket¶

interface NetworkSocketConnection {
    val incoming: Flow<NetworkSocketFrame>
    suspend fun send(frame: NetworkSocketFrame): Boolean
    suspend fun close(code: Int = 1_000, reason: String = "")
}

fun interface NetworkSocketTransport {
    suspend fun connect(
        url: String,
        headers: Map<String, String>,
    ): NetworkSocketConnection
}

Frame 包括 Text、Binary 和 Closing。createPlatformNetworkSocketTransport() 在 Android 返回 OkHttp 实现，在 iOS 返回 NSURLSession WebSocketTask 实现；也可以注入 Harmony NetworkKit 或已有业务 Socket SDK。

class ChatRepository(
    private val transport: NetworkSocketTransport = createPlatformNetworkSocketTransport(),
) {
    suspend fun connect(token: String): NetworkSocketConnection = transport.connect(
        url = "wss://chat.example.com/v1/socket",
        headers = mapOf("Authorization" to "Bearer $token"),
    )
}

只关心入站帧时，可以让模块托管重连和应用层心跳：

val frames = observeNetworkSocketFrames(
    transport = createPlatformNetworkSocketTransport(),
    url = "wss://chat.example.com/v1/socket",
    headersProvider = { mapOf("Authorization" to "Bearer ${tokenStore.current()}") },
    reconnectPolicy = RealtimeReconnectPolicy(
        maxReconnectAttempts = 5,
        initialDelayMillis = 500,
        maxDelayMillis = 30_000,
        jitterRatio = 0.2,
    ),
    heartbeat = NetworkSocketHeartbeat(
        intervalMillis = 30_000,
        frame = NetworkSocketFrame.Text("ping"),
    ),
    onConnected = { connection ->
        connection.send(NetworkSocketFrame.Text("subscribe:orders"))
    },
)

每次重连都会重新执行 onConnected，适合恢复订阅。1000 正常关闭会结束 Flow；传输失败、流意外结束和异常关闭会按上限重连。需要持续发送任意业务消息时，仍可直接持有原始 NetworkSocketConnection。

连接生命周期¶

Socket 应由 Repository/Application Scope 持有，不要无条件绑到单个页面。
页面只收集 incoming，取消页面收集不一定意味着关闭共享连接。
明确提供 close()，并让平台 Transport 释放底层任务。
托管 API 已提供有上限的指数退避和抖动；业务可在收集 Flow 前结合网络状态暂停或恢复。
托管连接默认只允许公开的 HTTPS/WSS 主机，并拒绝 Header 换行注入；可信内网需显式传入放行策略。

选择协议¶

需求	推荐
服务端到客户端单向文本事件	SSE
双向消息/二进制帧	WebSocket
短任务状态、服务端无推送能力	`pollNetwork`
普通一次性请求	`NetworkCall`

Realtime 模块负责通用的心跳和重连机制；业务确认、消息持久化、出站队列和协议级保活响应仍留在 Repository 层。