Изменение выносит типы контента в :content-api и разделяет потоки событий.
This commit is contained in:
@@ -10,10 +10,10 @@ import io.ktor.client.request.setBody
|
||||
import io.ktor.http.ContentType
|
||||
import io.ktor.http.contentType
|
||||
import kotlinx.serialization.Serializable
|
||||
import pw.binom.agentik.proto.Content
|
||||
import pw.binom.agentik.content.Content
|
||||
import pw.binom.agentik.proto.Conversation
|
||||
import pw.binom.agentik.proto.Message
|
||||
import pw.binom.agentik.proto.MessageContext
|
||||
import pw.binom.agentik.content.MessageContext
|
||||
import kotlin.time.Instant
|
||||
|
||||
/**
|
||||
|
||||
@@ -13,11 +13,16 @@ import pw.binom.agentik.outbox.OnlineOutbox
|
||||
* HTTP-реализация [OnlineOutbox] (= [pw.binom.agentik.outbox.OnlineOutbox]),
|
||||
* ходящая в `:server`-фасад.
|
||||
*
|
||||
* **Endpoint**: [onlineEvents] → `GET {baseUrl}/conversations/{id}/online`
|
||||
* (live-only SSE, без `after`). Сервер не реплеит — подписка получает
|
||||
* только то, что эмитится после подключения. Reconnect-логика здесь не
|
||||
* нужна: при обрыве поток просто закрывается, а потерянные дельты
|
||||
* восстанавливаются из durable-истории ([HttpEventStore] + журнал).
|
||||
* **Endpoints**:
|
||||
* - [onlineEvents] (без аргумента) → `GET {baseUrl}/online` (live-only SSE,
|
||||
* все диалоги агента);
|
||||
* - [onlineEvents] с `conversationId` → `GET {baseUrl}/conversations/{id}/online`
|
||||
* (live-only SSE, один диалог).
|
||||
*
|
||||
* Сервер не реплеит — подписка получает только то, что эмитится после
|
||||
* подключения. Reconnect-логика здесь не нужна: при обрыве поток просто
|
||||
* закрывается, а потерянные дельты восстанавливаются из durable-истории
|
||||
* ([HttpEventStore] + журнал).
|
||||
*/
|
||||
internal class HttpOnlineOutbox(
|
||||
private val httpClient: HttpClient,
|
||||
@@ -26,8 +31,12 @@ internal class HttpOnlineOutbox(
|
||||
|
||||
private val agentUrl: String = baseUrl.trimEnd('/')
|
||||
|
||||
override fun onlineEvents(conversationId: String): Flow<OnlineEvent> = flow {
|
||||
val url = "$agentUrl/conversations/$conversationId/online"
|
||||
override fun onlineEvents(): Flow<OnlineEvent> = stream("$agentUrl/online")
|
||||
|
||||
override fun onlineEvents(conversationId: String): Flow<OnlineEvent> =
|
||||
stream("$agentUrl/conversations/$conversationId/online")
|
||||
|
||||
private fun stream(url: String): Flow<OnlineEvent> = flow {
|
||||
httpClient.prepareGet(url) { noReadTimeout() }
|
||||
.execute { response ->
|
||||
check(response.status == HttpStatusCode.OK) {
|
||||
|
||||
@@ -66,7 +66,7 @@ private fun testEvent(dateMs: Long): CommonEvent =
|
||||
CommonEvent.Conversation(
|
||||
date = Instant.fromEpochMilliseconds(dateMs),
|
||||
conversationId = "test",
|
||||
event = Event.End(date = Instant.fromEpochMilliseconds(dateMs)),
|
||||
event = Event.Interrupted(date = Instant.fromEpochMilliseconds(dateMs)),
|
||||
)
|
||||
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
|
||||
Reference in New Issue
Block a user