diff --git a/CHANGELOG.ko.md b/CHANGELOG.ko.md index 69f8be0..631f010 100644 --- a/CHANGELOG.ko.md +++ b/CHANGELOG.ko.md @@ -10,9 +10,23 @@ ## [Unreleased] -이번 릴리스에서 공개 API 는 추가·제거·변경이 **없다** — `0.0.1` 로 컴파일한 소비자는 그대로 -링크된다. 바뀐 것은 내장 MCP 툴이 **무엇을 답하는가**다. 이 툴들을 wrap·subclass·위임해서 -쓰고 있다면 이 절 끝의 마이그레이션 노트를 읽어야 한다. +콘솔 SSE 스트림의 wire 계약에 하나가 추가됐다(아래). 소비자가 컴파일하는 Java API 자체는 +`0.0.1` 에서 바뀌지 않았고 그대로 링크된다. 그 밖에 바뀐 것은 내장 MCP 툴이 **무엇을 답하는가**다. +이 툴들을 wrap·subclass·위임해서 쓰고 있다면 이 절 끝의 마이그레이션 노트를 읽어야 한다. + +### 추가 + +- `GET /platform/traces/stream` 이 초기 스냅샷보다 앞서 **`ready` 확인(ack) 프레임**을 보낸다. + 트래픽이 없는 플랫폼에서는 *"스트림이 열렸다"* 와 *"서버가 실제로 돌고 있다"* 가 클라이언트 + 눈에 구별되지 않는다 — 요청이 들어오기 전까지 아무것도 도착하지 않기 때문이다. 이 ack 가 그 + 간극을 메우며, 클라이언트가 연결 시점에만 알 수 있는 값을 싣는다: `platform` 과 + `platformVersion`(후자는 jar manifest 가 없는 레이아웃에서 임의의 대체 문자열 대신 `null`), + `tracesEnabled` / `metricsEnabled`("기록이 꺼져 있다" 와 "켜져 있는데 한가하다" 를 가른다 — 빈 + 스트림으로는 안 된다), `buffered`(**뒤따르는 `trace` 프레임이 싣는 행 수**이며 링이 보유한 행 + 수가 아니다), `tickMs`(침묵 감지 워치독이 가정 대신 계약을 기준으로 삼게 한다), 그리고 이 + 연결 시점의 `capacity`. `EventSource` 가 재연결할 때마다 다시 나가며, 클라이언트는 모르는 + 필드를 무시해야 한다. 기존 클라이언트에는 영향이 없다 — `EventSource` 는 리스너가 없는 named + 이벤트를 버린다. 문서는 `docs/guide/04-rest-api.ko.md`. ### 수정 diff --git a/CHANGELOG.md b/CHANGELOG.md index b734e2c..b03f597 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,10 +10,27 @@ While the version is `0.x`, the public API may change between minor releases. ## [Unreleased] -No public API was added, removed, or changed in this release: a consumer compiled -against `0.0.1` links unchanged. What changed is what the built-in MCP tools -*answer*. If you wrap, subclass, or delegate to them, read the migration note at -the end of this section. +One addition to the wire contract of the console SSE stream (below); the Java API a +consumer compiles against is otherwise unchanged from `0.0.1` and still links. What +else changed is what the built-in MCP tools *answer*. If you wrap, subclass, or +delegate to them, read the migration note at the end of this section. + +### Added + +- `GET /platform/traces/stream` now opens with a **`ready` acknowledgement frame**, + sent ahead of the initial snapshot. On a platform with no traffic, *"the stream is + open"* and *"the server is actually running"* are indistinguishable to a client — + nothing arrives until a request does. The ack closes that gap and carries what a + client can only learn at connect time: `platform` and `platformVersion` (the latter + `null` when running from a layout with no jar manifest, rather than a fabricated + placeholder), `tracesEnabled` / `metricsEnabled` (which separate "recording is off" + from "recording is on but idle" — an empty stream does not), `buffered` (**the rows + the following `trace` frame carries**, not the rows the ring holds), `tickMs` (so a + silence watchdog has a contract to size itself against instead of an assumption), + and `capacity` as of this connection. It repeats on every `EventSource` reconnect, + and clients must ignore fields they do not recognize. Existing clients are + unaffected — `EventSource` drops named events that have no listener. Documented in + `docs/guide/04-rest-api.md`. ### Fixed diff --git a/README.ko.md b/README.ko.md index 198863f..1dec4f9 100644 --- a/README.ko.md +++ b/README.ko.md @@ -164,7 +164,7 @@ curl -X POST localhost:8080/platform/modules \ |--------|------|------| | `GET` | `/platform/traces` | 요청 트레이스(최신순). 필터 `moduleId`·`errorsOnly`·`status`·`minLatencyMs`·`since`·`beforeSeq`·`limit` | | `GET` | `/platform/traces/metrics` | 모듈별 집계 메트릭(요청수·에러율·지연 p50/p95/p99). `protean.trace.metrics.enabled` 필요 | -| `GET` | `/platform/traces/stream` | 라이브 SSE 푸시(`trace`/`metrics`/`modules`/`summary` 이벤트) | +| `GET` | `/platform/traces/stream` | 라이브 SSE 푸시(`ready` ack 후 `trace`/`metrics`/`modules`/`summary` 이벤트) | ### 설정 REST — `/platform/config` (`ConfigAdminController`, `protean.admin.enabled`) | 메서드 | 경로 | 동작 | diff --git a/README.md b/README.md index d9e3f2b..342cec9 100644 --- a/README.md +++ b/README.md @@ -202,7 +202,7 @@ curl -X POST localhost:8080/platform/modules \ |--------|------|------| | `GET` | `/platform/traces` | Request traces (newest-first). Filters `moduleId`·`errorsOnly`·`status`·`minLatencyMs`·`since`·`beforeSeq`·`limit` | | `GET` | `/platform/traces/metrics` | Per-module aggregate metrics (count, error rate, latency p50/p95/p99). Requires `protean.trace.metrics.enabled` | -| `GET` | `/platform/traces/stream` | Live SSE push (`trace`/`metrics`/`modules`/`summary` events) | +| `GET` | `/platform/traces/stream` | Live SSE push (`ready` ack, then `trace`/`metrics`/`modules`/`summary` events) | ### Config REST — `/platform/config` (`ConfigAdminController`, `protean.admin.enabled`) | Method | Path | Action | diff --git a/docs/guide/04-rest-api.ko.md b/docs/guide/04-rest-api.ko.md index bf460fe..7048659 100644 --- a/docs/guide/04-rest-api.ko.md +++ b/docs/guide/04-rest-api.ko.md @@ -393,16 +393,36 @@ POST /platform/modules/cp-mod/approve?approver=alice ### GET `/platform/traces/stream` -라이브 푸시 스트림(SSE) — 콘솔이 5초 폴링 대신 쓴다. 한 연결이 네 종류의 named 이벤트를 멀티플렉싱하며, 새로 연 연결은 네 종류의 초기 스냅샷을 먼저 받은 뒤 증분 푸시(대략 1초마다)를 받는다. +라이브 푸시 스트림(SSE) — 콘솔이 5초 폴링 대신 쓴다. 연결은 `ready` 프레임으로 확인(ack)된 뒤 네 종류의 named 이벤트를 멀티플렉싱하며, 새로 연 연결은 네 종류의 초기 스냅샷을 먼저 받은 뒤 증분 푸시(대략 1초마다)를 받는다. - Produces: `text/event-stream` - named 이벤트: + - `ready` — 연결 확인(ack). 언제나 첫 프레임이다(아래 참고) - `trace` — 새 `RequestTrace` 델타 - `metrics` — `ModuleMetricsSnapshot[]` 갱신(모듈별 누적 집계, `protean.trace.metrics.enabled=true` 일 때만 채워짐) - `modules` — 현재 `ModuleStatus[]` - `summary` — `TraceSummary`: 콘솔 헤더용 **윈도** 집계(아래 참고) - `/platform/traces` 와 마찬가지로 스트림 연결 자체는 trace 기록에서 제외된다(자기-관측 방지). +`ready` 이벤트는 데이터 프레임이 아니라 **연결 확인 응답(ack)** 이다. 트래픽이 없는 플랫폼에서는 *"스트림이 열렸다"* 와 *"서버가 실제로 돌고 있다"* 가 클라이언트 눈에 똑같이 보이기 때문에 존재한다 — 요청이 들어오기 전까지는 아무것도 도착하지 않는다. 연결마다 한 번씩, `EventSource` 가 재연결할 때도 매번, 초기 `trace` 재생보다 앞서 나간다: + +``` +event: ready +data: {"platform":"protean","platformVersion":"0.1.0","tracesEnabled":true, + "metricsEnabled":false,"buffered":5,"tickMs":1000,"capacity":200} +``` + +여기에는 **연결 시점에 한 번 알아야 하고 이후 스트림으로는 알 수 없는** 값만 싣는다 — 주기적으로 바뀌는 값은 `metrics`·`modules`·`summary` 프레임의 몫이다: + +- `platform` — 구현체 식별자. 언제나 `"protean"`. 이 계약을 여러 구현이 공유하므로, 버전 숫자만 받은 콘솔이 그것이 누구의 버전인지 추측하지 않도록 함께 싣는다. **표시·진단용이며 동작을 분기하는 데 쓰지 않는다.** +- `platformVersion` — 이 라이브러리의 버전. jar manifest 의 `Implementation-Version` 에서 읽는다. manifest 가 없는 레이아웃(exploded classes — 테스트·IDE 실행)에서는 **`null`** 이다. 그 상황에서는 버전을 정말로 모르며, 임의의 대체 문자열은 "지금 어느 버전과 이야기하고 있나" 에 대한 지어낸 답이 된다. 필드 자체는 항상 있으므로 클라이언트가 "없음" 과 "모름" 을 구별할 일이 없다. +- `tracesEnabled` / `metricsEnabled` — `protean.trace.enabled` 와 `protean.trace.metrics.enabled`. **"기록이 꺼져 있다"** 와 **"켜져 있는데 아직 아무 일도 없었다"** 를 가른다. 빈 스트림만으로는 이 둘이 구별되지 않는다. +- `buffered` — 이 프레임 **다음에 오는 `trace` 프레임이 싣는 행 수**이며, 링이 보유한 행 수가 아니다. `protean.trace.capacity` 가 재생 상한 200 을 넘으면 둘이 달라진다. 숫자만 알리고 행은 주지 않는 것은 둘 다 안 하느니만 못하다. +- `tickMs` — 푸시 주기(`1000`). 클라이언트의 침묵 감지 워치독이 가정 대신 계약을 기준으로 자기 값을 정할 수 있게 한다. +- `capacity` — **이 연결 시점의** `protean.trace.capacity`. 클라이언트가 "이보다 과거는 서버에 없다" 를 말할 수 있다. LIVE 키이므로 이후 변경은 이미 연결된 클라이언트에 닿지 않으며, 재연결 때 새 값이 다시 나간다. + +이후 버전에서 `ready` 에 필드가 더해질 수 있다 — 클라이언트는 모르는 필드를 무시해야 한다. + ``` event: trace data: {"seq":13,"method":"GET","uri":"/cp/ping","status":200,"latencyMs":2,"moduleId":"cp-mod"} diff --git a/docs/guide/04-rest-api.md b/docs/guide/04-rest-api.md index 1657aca..ae9ccb2 100644 --- a/docs/guide/04-rest-api.md +++ b/docs/guide/04-rest-api.md @@ -393,16 +393,36 @@ Query per-module aggregate metrics (request count, error rate, latency percentil ### GET `/platform/traces/stream` -Live server-sent-events (SSE) push stream — the console uses it in place of 5-second polling. One connection multiplexes four named event types; a freshly opened connection first receives an initial snapshot of all four, then incremental pushes (roughly once per second). +Live server-sent-events (SSE) push stream — the console uses it in place of 5-second polling. A connection is acknowledged with a `ready` frame and then multiplexes four named event types; a freshly opened connection first receives an initial snapshot of all four, then incremental pushes (roughly once per second). - Produces: `text/event-stream` - Named events: + - `ready` — connection acknowledgement, always the first frame (see below) - `trace` — new `RequestTrace` deltas - `metrics` — a `ModuleMetricsSnapshot[]` update (per-module cumulative aggregate; populated only when `protean.trace.metrics.enabled=true`) - `modules` — the current `ModuleStatus[]` - `summary` — a `TraceSummary`: a **windowed** aggregate for the console header (see below) - Like `/platform/traces`, the stream connection itself is excluded from trace recording (no self-observation). +The `ready` event is an acknowledgement, not a data frame. It exists because *"the stream is open"* and *"the server is actually running"* look identical to a client on a quiet platform — nothing else arrives until traffic does. It is sent once per connection, including on every `EventSource` reconnect, ahead of the initial `trace` replay: + +``` +event: ready +data: {"platform":"protean","platformVersion":"0.1.0","tracesEnabled":true, + "metricsEnabled":false,"buffered":5,"tickMs":1000,"capacity":200} +``` + +It carries only what a client must know **once, at connect time** and cannot derive from the stream afterwards — anything that changes on a schedule belongs to the `metrics`/`modules`/`summary` frames instead: + +- `platform` — implementation identifier, always `"protean"`. Present so a console watching several implementations of this contract is not left reading a bare version number and guessing whose it is. **Display and diagnostics only — do not branch behaviour on it.** +- `platformVersion` — this library's version, read from the jar manifest (`Implementation-Version`). **`null`** when running from a layout that has no manifest (exploded classes: tests, IDE runs) — the version is genuinely unknown there, and a placeholder would be a fabricated answer. The field is always present, so a client never has to distinguish "absent" from "unknown". +- `tracesEnabled` / `metricsEnabled` — `protean.trace.enabled` and `protean.trace.metrics.enabled`. They separate **"recording is off"** from **"recording is on but nothing has happened yet"**, which an empty stream cannot. +- `buffered` — how many rows the `trace` frame that follows this one carries, **not** how many the ring holds. The two differ whenever `protean.trace.capacity` exceeds the 200-row replay cap; announcing a count while withholding the rows would be worse than announcing neither. +- `tickMs` — the push period (`1000`), so a client's silence watchdog has a contract to size itself against instead of assuming one. +- `capacity` — `protean.trace.capacity` **as of this connection**, letting a client say "nothing older than this is on the server". It is a live key, so a later change does not reach an already-connected client; the value is re-sent on reconnect. + +Fields may be added to `ready` in later versions — a client must ignore the ones it does not recognize. + ``` event: trace data: {"seq":13,"method":"GET","uri":"/cp/ping","status":200,"latencyMs":2,"moduleId":"cp-mod"} diff --git a/src/main/java/org/htcom/protean/web/StreamReady.java b/src/main/java/org/htcom/protean/web/StreamReady.java new file mode 100644 index 0000000..a87fa8a --- /dev/null +++ b/src/main/java/org/htcom/protean/web/StreamReady.java @@ -0,0 +1,63 @@ +/* + * Copyright (c) 2026 htjulia + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at https://mozilla.org/MPL/2.0/. + */ + +package org.htcom.protean.web; + +/** + * Connection acknowledgement: the first frame of every {@code /platform/traces/stream} connection + * ({@code event: ready}). It exists because "the stream is open" and "the server is actually running" + * are indistinguishable to a client on a quiet platform — nothing else arrives until traffic does. + * + *

This is an ack, not a data frame. It carries only values a client must know once, at connect + * time and cannot derive from the stream afterwards; anything that changes on a schedule belongs to + * the {@code metrics}/{@code modules}/{@code summary} frames instead. + * + * @param platform implementation identifier, constant {@code "protean"}. Present so a console that + * watches several implementations does not have to read a bare version number and + * guess whose it is. Display/diagnostics only — clients must not branch behaviour on it. + * @param platformVersion this library's version, read from the jar manifest ({@code Implementation-Version}). + * {@code null} when running from a layout that has no manifest (exploded classes: tests, + * IDE runs) — the version is genuinely unknown there, and a placeholder string would be + * a fabricated answer to "which version am I talking to". + * @param tracesEnabled {@code protean.trace.enabled}: distinguishes "recording is off" from "recording is on + * but nothing has happened yet", which an empty stream cannot. + * @param metricsEnabled {@code protean.trace.metrics.enabled}: same distinction for the {@code metrics} frame. + * @param buffered how many rows the {@code trace} frame that follows this one carries — not how + * many the ring holds. The two differ whenever the ring is larger than the replay cap, and + * announcing a count while withholding the rows is worse than announcing neither. + * @param tickMs the push period, so a client's silence watchdog has a contract to size itself against + * instead of assuming one. + * @param capacity {@code protean.trace.capacity} as of this connection, letting a client say + * "nothing older than this is on the server". It is a live key, so a later change does not + * reach an already-connected client; the value is re-sent on reconnect. + */ +public record StreamReady( + String platform, + String platformVersion, + boolean tracesEnabled, + boolean metricsEnabled, + int buffered, + long tickMs, + int capacity +) { + + private static final String PLATFORM = "protean"; + + /** Resolved once: a jar manifest cannot change while the JVM runs. */ + private static final String VERSION = version(); + + private static String version() { + Package pkg = StreamReady.class.getPackage(); + return pkg == null ? null : pkg.getImplementationVersion(); + } + + /** Ack for a connection that is about to replay {@code buffered} trace rows. */ + public static StreamReady of(boolean tracesEnabled, boolean metricsEnabled, int buffered, long tickMs, int capacity) { + return new StreamReady(PLATFORM, VERSION, tracesEnabled, metricsEnabled, buffered, tickMs, capacity); + } +} diff --git a/src/main/java/org/htcom/protean/web/TraceStreamService.java b/src/main/java/org/htcom/protean/web/TraceStreamService.java index 45fd1b3..a63fa53 100644 --- a/src/main/java/org/htcom/protean/web/TraceStreamService.java +++ b/src/main/java/org/htcom/protean/web/TraceStreamService.java @@ -44,10 +44,11 @@ * the console; other polling consumers are their own choice). * *

The recording hot path ({@code RequestTraceFilter.record()}) is untouched: this reads the store - * out-of-band on a timer. One SSE connection multiplexes three named events — {@code trace} (incremental, - * only when there is something new), {@code metrics} and {@code modules} (full snapshots each tick, which - * also keeps the connection warm through proxies). A fresh connection receives an initial snapshot of all - * three immediately so it is never blank until the first tick. + * out-of-band on a timer. One SSE connection multiplexes four named events — {@code trace} (incremental, + * only when there is something new), {@code metrics}, {@code modules} and {@code summary} (full snapshots + * each tick, which also keeps the connection warm through proxies). A fresh connection is acknowledged with + * a {@link StreamReady} {@code ready} frame and then receives an initial snapshot of all four immediately, + * so it is never blank until the first tick. * *

Same gating as the admin controllers ({@code @Profile("!worker")} + {@code protean.admin.enabled}), so * it is absent on workers and when the management surface is opted out. @@ -90,7 +91,9 @@ public TraceStreamService(TraceStore store, TraceMetrics metrics, ModulePlatform /** * Opens a live SSE stream. Timeout is disabled (0) because the stream is meant to stay open; if a proxy - * drops it anyway, the browser {@code EventSource} reconnects and gets a fresh initial snapshot. + * drops it anyway, the browser {@code EventSource} reconnects and gets a fresh {@code ready} ack plus a + * fresh initial snapshot — the ack repeats per connection precisely so a reconnecting client re-learns + * the connect-time facts rather than carrying stale ones across the gap. */ public SseEmitter open() { SseEmitter emitter = new SseEmitter(0L); @@ -101,8 +104,11 @@ public SseEmitter open() { }); emitter.onError(ex -> emitters.remove(emitter)); try { + // Read the backlog before the ack so `buffered` can report the rows actually replayed below. + List initial = store.recent(INITIAL_TRACES, null); List mods = moduleStatuses(); - send(emitter, "trace", store.recent(INITIAL_TRACES, null)); + send(emitter, "ready", ready(initial.size())); + send(emitter, "trace", initial); send(emitter, "metrics", metrics.snapshots()); send(emitter, "modules", mods); send(emitter, "summary", summary(mods)); @@ -114,6 +120,17 @@ public SseEmitter open() { return emitter; } + /** + * Connection ack for a client about to be replayed {@code buffered} rows. Every value is read here, at + * connect time, because all four config keys are live — a value cached at startup would be a stale answer + * to a question the client only gets to ask once. + */ + private StreamReady ready(int buffered) { + ProteanProperties.Trace trace = props.getTrace(); + return StreamReady.of(trace.isEnabled(), trace.getMetrics().isEnabled(), buffered, TICK_MS, + trace.getCapacity()); + } + /** One push cycle: advance the cursor by the newest delta, then fan the deltas + fresh snapshots out. */ private void tick() { try { diff --git a/src/test/java/org/htcom/protean/TraceStreamE2ETest.java b/src/test/java/org/htcom/protean/TraceStreamE2ETest.java index d686ba1..27574ec 100644 --- a/src/test/java/org/htcom/protean/TraceStreamE2ETest.java +++ b/src/test/java/org/htcom/protean/TraceStreamE2ETest.java @@ -8,6 +8,8 @@ package org.htcom.protean; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.Test; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.web.server.LocalServerPort; @@ -26,6 +28,7 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -34,8 +37,8 @@ /** * Real-JVM e2e for the console SSE push transport ({@code GET /platform/traces/stream}). Verifies the - * multiplexed initial snapshot (trace/metrics/modules named events) and that the server-side ticker keeps - * pushing without any client polling. + * connection ack ({@code ready}) and the multiplexed initial snapshot that follows it, the ack's payload + * contract, and that the server-side ticker keeps pushing without any client polling. */ @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) class TraceStreamE2ETest { @@ -51,36 +54,64 @@ static void props(DynamicPropertyRegistry r) { int port; private final HttpClient client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(5)).build(); + private static final ObjectMapper MAPPER = new ObjectMapper(); @Test - void stream_emits_initial_snapshot_and_keeps_ticking() throws Exception { - // Collect enough events to cover the synchronous initial snapshot (trace+metrics+modules = 3) plus at - // least one later ticker cycle, proving push continues with no client polling. - List events = readSseEventNames(6, Duration.ofSeconds(5)); - - // Initial snapshot delivers all three multiplexed event types. - assertThat(events).contains("trace", "metrics", "modules"); - // The first three are exactly the initial snapshot, in order. - assertThat(events.subList(0, 3)).containsExactly("trace", "metrics", "modules"); + void stream_acks_then_emits_initial_snapshot_and_keeps_ticking() throws Exception { + // Collect enough frames to cover the synchronous ack + initial snapshot (ready+trace+metrics+modules + // +summary = 5) plus at least one later ticker cycle, proving push continues with no client polling. + List events = readSse(8, Duration.ofSeconds(5)).stream().map(Map.Entry::getKey).toList(); + + // The ack comes first, then the four snapshot frames, in this exact order — the console reads the ack + // to tell "the server is live" from "the socket merely opened", so anything ahead of it defeats that. + assertThat(events.subList(0, 5)).containsExactly("ready", "trace", "metrics", "modules", "summary"); // The ticker pushed at least one more cycle after the initial snapshot (no polling involved). - assertThat(events.size()).isGreaterThan(3); + assertThat(events.size()).isGreaterThan(5); + // The ack is per connection, not per tick. + assertThat(events.stream().filter("ready"::equals)).hasSize(1); + } + + @Test + void ready_ack_carries_the_connect_time_contract() throws Exception { + List> frames = readSse(2, Duration.ofSeconds(5)); + + JsonNode ready = MAPPER.readTree(frames.get(0).getValue()); + assertEquals("protean", ready.path("platform").asText()); + // Present but null when running from exploded classes (no jar manifest to read the version from); the + // field must still exist so a client never has to distinguish "absent" from "unknown". + assertThat(ready.has("platformVersion")).isTrue(); + assertThat(ready.path("tracesEnabled").asBoolean()).isTrue(); + assertThat(ready.path("metricsEnabled").asBoolean()).isFalse(); // protean.trace.metrics.enabled default + assertEquals(1_000L, ready.path("tickMs").asLong()); // the watchdog contract + assertEquals(200, ready.path("capacity").asInt()); // protean.trace.capacity default + + // buffered counts the rows of the trace frame that follows, not the rows held by the ring. Announcing a + // count the client does not then receive is the failure this assertion exists to catch. + JsonNode traces = MAPPER.readTree(frames.get(1).getValue()); + assertEquals(traces.size(), ready.path("buffered").asInt()); } - /** Opens the SSE stream and returns the {@code event:} names in arrival order, up to {@code expected} or the deadline. */ - private List readSseEventNames(int expected, Duration deadline) throws Exception { + /** + * Opens the SSE stream and returns {@code (event, data)} pairs in arrival order, up to {@code expected} or + * the deadline. + */ + private List> readSse(int expected, Duration deadline) throws Exception { HttpRequest req = HttpRequest.newBuilder(URI.create("http://localhost:" + port + "/platform/traces/stream")) .header("Accept", "text/event-stream").GET().build(); HttpResponse resp = client.send(req, HttpResponse.BodyHandlers.ofInputStream()); assertEquals(200, resp.statusCode()); assertThat(resp.headers().firstValue("Content-Type").orElse("")).contains("text/event-stream"); - List names = new ArrayList<>(); + List> frames = new ArrayList<>(); CompletableFuture reader = CompletableFuture.runAsync(() -> { try (BufferedReader br = new BufferedReader(new InputStreamReader(resp.body(), StandardCharsets.UTF_8))) { - String line; - while (names.size() < expected && (line = br.readLine()) != null) { + String line, name = null; + while (frames.size() < expected && (line = br.readLine()) != null) { if (line.startsWith("event:")) { - names.add(line.substring("event:".length()).trim()); + name = line.substring("event:".length()).trim(); + } else if (line.startsWith("data:") && name != null) { + frames.add(Map.entry(name, line.substring("data:".length()).trim())); + name = null; } } } catch (Exception ignored) { @@ -92,6 +123,6 @@ private List readSseEventNames(int expected, Duration deadline) throws E reader.cancel(true); } resp.body().close(); - return names; + return frames; } }