maya/Rainytoken
1
0
Fork 0
Code Issues 4 Pull Requests Actions 38 Packages Projects Releases 28 Wiki Activity

feat(proxy): stream=true 实时透传上游 SSE 字节流

Browse Source 
- Ktor respondOutputStream 逐块转发上游 body，不再整包缓冲
- 流式/非流式按客户端 body 的 stream 字段分流
- openStreamingChat 入口正式接线
This commit is contained in:
Liuxinyu176 2026-10-08 23:52:12 +08:00
parent 072f7efa55
commit de95013164
1 changed files with 50 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
app/src/main/java/com/rainy/token/data/proxy/KtorLocalProxyServer.kt
View File

@ -14,6 +14,7 @@ import io.ktor.server.plugins.contentnegotiation.ContentNegotiation
import io.ktor.server.request.receiveText import io.ktor.server.request.receiveText
import io.ktor.server.response.respond import io.ktor.server.response.respond
import io.ktor.server.response.respondBytes import io.ktor.server.response.respondBytes
import io.ktor.server.response.respondOutputStream
import io.ktor.server.routing.get import io.ktor.server.routing.get
import io.ktor.server.routing.post import io.ktor.server.routing.post
import io.ktor.server.routing.routing import io.ktor.server.routing.routing
@ -129,22 +130,58 @@ class KtorLocalProxyServer @Inject constructor(
val pooled = pool.next(route.kind, route.region, conversationId) val pooled = pool.next(route.kind, route.region, conversationId)
val accountId = pooled?.accountId val accountId = pooled?.accountId
try { try {
val result: ProxyUpstreamResponse? = when (route.kind) { if (extractStream(body)) {
ProviderKind.WORKBUDDY_CN, ProviderKind.WORKBUDDY_INTL -> val stream: ProxyUpstreamStream? = when (route.kind) {
workBuddy.forwardChat(body, accountId, route.region) ProviderKind.WORKBUDDY_CN, ProviderKind.WORKBUDDY_INTL ->
workBuddy.openStreamingChat(body, accountId, route.region)
ProviderKind.TRAE_CN, ProviderKind.TRAE_INTL -> ProviderKind.TRAE_CN, ProviderKind.TRAE_INTL ->
trae.forwardChat(body, accountId, route.region) trae.openStreamingChat(body, accountId, route.region)
else -> sub2Api.forwardChat(body, accountId) else -> sub2Api.openStreamingChat(body, accountId)
} }
if (result == null) { if (stream == null) {
call.respond( call.respond(
HttpStatusCode.BadRequest, HttpStatusCode.BadRequest,
errorBody("${route.kind.displayName} 未配置或未登录，请先在设置中配置") errorBody("${route.kind.displayName} 未配置或未登录，请先在设置中配置")
) )
} else {
call.respondOutputStream(
contentType = contentTypeOf(stream.contentType),
status = HttpStatusCode(stream.status, "")
) {
try {
val buffer = ByteArray(8192)
val input = stream.input
while (true) {
val read = input.read(buffer)
if (read < 0) break
write(buffer, 0, read)
flush()
}
} finally {
stream.close()
}
}
}
} else { } else {
call.respondBytes(result.body, contentTypeOf(result.contentType), HttpStatusCode(result.status, "")) val result: ProxyUpstreamResponse? = when (route.kind) {
ProviderKind.WORKBUDDY_CN, ProviderKind.WORKBUDDY_INTL ->
workBuddy.forwardChat(body, accountId, route.region)
ProviderKind.TRAE_CN, ProviderKind.TRAE_INTL ->
trae.forwardChat(body, accountId, route.region)
else -> sub2Api.forwardChat(body, accountId)
}
if (result == null) {
call.respond(
HttpStatusCode.BadRequest,
errorBody("${route.kind.displayName} 未配置或未登录，请先在设置中配置")
)
} else {
call.respondBytes(result.body, contentTypeOf(result.contentType), HttpStatusCode(result.status, ""))
}
} }
} catch (e: IOException) { } catch (e: IOException) {
val detail = e.message ?: "未知错误" val detail = e.message ?: "未知错误"