Add chat-ui-htmx module: streaming chat page on Spring AI 2.0.2 with htmx and its SSE extension (vendored), MessageChatMemoryAdvisor keyed by the HTTP session, a Stop button, and the traps measured: EventSource strips one leading space per data line, model text must be HTML-escaped, a cancelled stream left the upstream connection open in 7 of 8 runs until an OkHttp interceptor calls Call.cancel (8 of 8 closed after), and the SDK push source ignores back-pressure (limitRate changes only the demand; a bounded buffer works). The model is a stub OpenAI-compatible SSE server written for the module; no real provider was run.
Co-Authored-By: Claude Sonnet 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_017ic4b6Kg9RWSTJzyz6KP9X
This commit is contained in:
Claude Sonnet 5.5
@@ -33,5 +33,6 @@ Runnable companion code for the Spring AI articles on [ankurm.com](https://ankur
|
|||||||
| [`image-gen/`](image-gen) | `ImageModel` and `ImagePrompt` with `OpenAiImageOptions` (what the real `OpenAiImageModel` puts on the wire, a per-call options object that silently resets the model, a `List<ImageMessage>` prompt that throws), a prompt gate with `ModerationModel` that fails closed, and safe storage in MinIO: sniffed content type, byte and pixel caps, server-chosen keys, a fetcher for provider URLs, presigned reads. A per-image price table from the vendors' pages dated 2026-10-28. No image provider called: the `ImageModel` and `ModerationModel` are hand-written fakes and a mock server; the MinIO tests use a real server through Testcontainers. Spring Boot 4.1.2, Spring AI 2.0.2, Java 25. | [Image Generation and Vision in Spring AI: ImageModel, Prompt Options and Safe Storage](https://ankurm.com/spring-ai-image-generation-imagemodel-s3-minio-safe-storage/) |
|
| [`image-gen/`](image-gen) | `ImageModel` and `ImagePrompt` with `OpenAiImageOptions` (what the real `OpenAiImageModel` puts on the wire, a per-call options object that silently resets the model, a `List<ImageMessage>` prompt that throws), a prompt gate with `ModerationModel` that fails closed, and safe storage in MinIO: sniffed content type, byte and pixel caps, server-chosen keys, a fetcher for provider URLs, presigned reads. A per-image price table from the vendors' pages dated 2026-10-28. No image provider called: the `ImageModel` and `ModerationModel` are hand-written fakes and a mock server; the MinIO tests use a real server through Testcontainers. Spring Boot 4.1.2, Spring AI 2.0.2, Java 25. | [Image Generation and Vision in Spring AI: ImageModel, Prompt Options and Safe Storage](https://ankurm.com/spring-ai-image-generation-imagemodel-s3-minio-safe-storage/) |
|
||||||
| [`batch-llm/`](batch-llm) | 100,000 synthetic support tickets classified by a `ChatClient` inside a Spring Batch 6 chunk step: a Resilience4j rate limiter and retry around the model call (and where Spring Batch's own retry and skip belong instead), replies that are not valid JSON skipped and recorded, a restart after `kill -9` in a second JVM that ends with a byte-identical result, and a cost ledger row committed with every chunk. **The model is a deterministic simulated `ChatModel`, not a language model; token counts are estimates and prices are configurable placeholders.** Spring Boot 4.1.2, Spring Batch 6.0.6, Spring AI 2.0.2, Resilience4j 2.4.0, Java 25. | [Spring AI + Spring Batch: Classifying 100,000 Support Tickets with an LLM](https://ankurm.com/spring-ai-spring-batch-llm-ticket-classification/) |
|
| [`batch-llm/`](batch-llm) | 100,000 synthetic support tickets classified by a `ChatClient` inside a Spring Batch 6 chunk step: a Resilience4j rate limiter and retry around the model call (and where Spring Batch's own retry and skip belong instead), replies that are not valid JSON skipped and recorded, a restart after `kill -9` in a second JVM that ends with a byte-identical result, and a cost ledger row committed with every chunk. **The model is a deterministic simulated `ChatModel`, not a language model; token counts are estimates and prices are configurable placeholders.** Spring Boot 4.1.2, Spring Batch 6.0.6, Spring AI 2.0.2, Resilience4j 2.4.0, Java 25. | [Spring AI + Spring Batch: Classifying 100,000 Support Tickets with an LLM](https://ankurm.com/spring-ai-spring-batch-llm-ticket-classification/) |
|
||||||
| [`docker-model-runner/`](docker-model-runner) | The Spring AI OpenAI starter pointed at Docker Model Runner and at Ollama by changing only `base-url` and the model name (no `/v1` is appended: the path is used exactly as given), plus a benchmark runner: time to first token, decode tokens/second, container memory from `docker stats` and tool-call success over 10 runs. Measured with the real `docker/model-runner` v1.2.9 CPU image and Ollama 0.40.13 on one shared 2-vCPU sandbox, Qwen2.5-0.5B Q4_K_M; no GPU, and Docker Desktop's `docker model` path was not used. Spring Boot 4.1.2, Spring AI 2.0.2, Java 25. | [Spring AI with Docker Model Runner vs Ollama: Local LLMs Benchmarked on the Same Laptop](https://ankurm.com/spring-ai-docker-model-runner-vs-ollama/) |
|
| [`docker-model-runner/`](docker-model-runner) | The Spring AI OpenAI starter pointed at Docker Model Runner and at Ollama by changing only `base-url` and the model name (no `/v1` is appended: the path is used exactly as given), plus a benchmark runner: time to first token, decode tokens/second, container memory from `docker stats` and tool-call success over 10 runs. Measured with the real `docker/model-runner` v1.2.9 CPU image and Ollama 0.40.13 on one shared 2-vCPU sandbox, Qwen2.5-0.5B Q4_K_M; no GPU, and Docker Desktop's `docker model` path was not used. Spring Boot 4.1.2, Spring AI 2.0.2, Java 25. | [Spring AI with Docker Model Runner vs Ollama: Local LLMs Benchmarked on the Same Laptop](https://ankurm.com/spring-ai-docker-model-runner-vs-ollama/) |
|
||||||
|
| [`chat-ui-htmx/`](chat-ui-htmx) | A streaming chat page with no JavaScript of our own: htmx and its SSE extension (vendored) post the message and append each token from a `Flux<ServerSentEvent<String>>` over `ChatClient` with `MessageChatMemoryAdvisor` keyed by the HTTP session. Measured traps: EventSource strips one leading space per data line, model text must be HTML-escaped, a cancelled stream left the upstream connection open in 7 of 8 runs until an OkHttp interceptor made it `call.cancel()` (closed in 8 of 8 after), and the SDK's push source ignores back-pressure (`limitRate` changes nothing; a bounded buffer does). Checked in headless Chromium. **The model is a stub OpenAI-compatible SSE server I wrote, not a language model; no real provider was run.** Spring Boot 4.1.2, Spring AI 2.0.2, htmx 2.0.11, Java 25. | [Streaming AI Chat UI with Spring AI, SSE and htmx (No React Needed)](https://ankurm.com/spring-ai-streaming-chat-ui-sse-htmx/) |
|
||||||
|
|
||||||
Upgrading from Spring AI 1.x: [migration guide](https://ankurm.com/spring-ai-1-to-2-migration-guide/).
|
Upgrading from Spring AI 1.x: [migration guide](https://ankurm.com/spring-ai-1-to-2-migration-guide/).
|
||||||
|
|||||||
@@ -0,0 +1,55 @@
|
|||||||
|
# chat-ui-htmx
|
||||||
|
|
||||||
|
Companion code for [Streaming AI Chat UI with Spring AI, SSE and htmx (No React Needed)](https://ankurm.com/spring-ai-streaming-chat-ui-sse-htmx/), part of the [Spring AI series](../README.md) on ankurm.com.
|
||||||
|
|
||||||
|
A chat page with no JavaScript of our own: htmx and its SSE extension (both vendored under `src/main/resources/static/vendor/`, no CDN) post the message, open an `EventSource`, and append each token to the page. The server side is one `ChatClient` with `MessageChatMemoryAdvisor` (conversation id taken from the HTTP session), a `Flux<ServerSentEvent<String>>` endpoint, a Stop button, and the traps found on the way: a leading space that EventSource strips, model text that must be HTML-escaped, a cancel that does not close the upstream connection, and a model-side stream that ignores back-pressure.
|
||||||
|
|
||||||
|
**The model is a STUB.** `OpenAiStubServer` (`src/main/java/com/ankurm/chatui/stub/`) is a small OpenAI-compatible `POST /v1/chat/completions` server that streams SSE chunks with a controllable delay and answers from a template built from the request it received. It does no inference. The real Spring AI `OpenAiChatModel` talks to it over real HTTP. A real provider was **not** run; to try one: `CHAT_BASE_URL=https://api.openai.com/v1 CHAT_MODEL=<model> OPENAI_API_KEY=<key> java -jar target/chat-ui-htmx-1.0.0.jar` (the base URL must end in `/v1`: the starter does not append it).
|
||||||
|
|
||||||
|
| Component | Version |
|
||||||
|
|---|---|
|
||||||
|
| Spring Boot | 4.1.2 |
|
||||||
|
| Spring AI | 2.0.2 (OpenAI starter, openai-java 4.80.0, OkHttp 4.12.0) |
|
||||||
|
| htmx / htmx-ext-sse | 2.0.11 / 2.2.4 (vendored, sha256 in `static/vendor/SHA256SUMS`) |
|
||||||
|
| Java | 25 (LTS) |
|
||||||
|
|
||||||
|
## Run
|
||||||
|
|
||||||
|
```bash
|
||||||
|
mvn -DskipTests package
|
||||||
|
mvn -q dependency:build-classpath -Dmdep.outputFile=target/cp.txt
|
||||||
|
java -cp "target/classes:$(cat target/cp.txt)" com.ankurm.chatui.stub.OpenAiStubServer 38181 20 & # port, ms between chunks
|
||||||
|
java -jar target/chat-ui-htmx-1.0.0.jar # http://localhost:38180/
|
||||||
|
scripts/run-all.sh # regenerates output/ (about 5 minutes; step 2 needs Playwright for Python + Chromium)
|
||||||
|
```
|
||||||
|
|
||||||
|
Stub directives (put them at the start of the message): `count N` streams N one-word chunks, `big N` streams N chunks of 32 KiB, `html ...` streams text containing markup. Anything else is echoed with the number of messages and earlier user turns it saw. `POST /stub/config?delayMs=N` changes the delay, `GET /stub/stats` shows counters.
|
||||||
|
|
||||||
|
App switches: `--chat.cancel-fix=false` (turn the cancel fix off), `--chat.backpressure=none|limitRate|slow|bounded`, `--chat.memory.max-messages=20`.
|
||||||
|
|
||||||
|
## Files
|
||||||
|
|
||||||
|
| File | What |
|
||||||
|
|---|---|
|
||||||
|
| `ChatController.java` | `POST /chat` (HTML fragment), `GET /chat/stream` (SSE), `POST /chat/stop`, `POST /chat/new` |
|
||||||
|
| `ChatConfig.java` | `MessageWindowChatMemory` + `ChatClient` with `MessageChatMemoryAdvisor` |
|
||||||
|
| `CancelOnCloseConfig.java` | OkHttp interceptor that makes a cancelled SSE stream close its connection |
|
||||||
|
| `stub/OpenAiStubServer.java` | the stub model server |
|
||||||
|
| `static/index.html` | the whole front end |
|
||||||
|
| `ChatFlowTest`, `OkHttpCloseReproTest` | tests that also write `output/01`-`05` |
|
||||||
|
| `scripts/browser-test.py` | headless Chromium driving the real page (`output/06`) |
|
||||||
|
| `scripts/cancel-upstream.sh`, `backpressure.sh`, `slow_client.py` | the cancel and back-pressure experiments |
|
||||||
|
|
||||||
|
## Captured output
|
||||||
|
|
||||||
|
| File | What |
|
||||||
|
|---|---|
|
||||||
|
| `output/01-sse-transcript.txt` | raw SSE lines for one turn, and the text with and without the leading-space padding |
|
||||||
|
| `output/02-memory.txt` | same session remembers, a new session does not |
|
||||||
|
| `output/03-escaping.txt` | markup in model output arrives escaped |
|
||||||
|
| `output/04-cancel-with-fix.txt` | socket closed by the client: Flux cancelled, upstream closed |
|
||||||
|
| `output/05-okhttp-close-vs-cancel.txt` | plain OkHttp: `response.close()` leaks, `call.cancel()` does not |
|
||||||
|
| `output/06-browser.txt`, `07-stop-logs.txt` | the page in Chromium, and the server and stub logs of the Stop click |
|
||||||
|
| `output/08-cancel-upstream.txt` | 8 aborted streams without and with the fix |
|
||||||
|
| `output/09-backpressure.txt` | slow client, four ways to shape the Flux |
|
||||||
|
| `output/10-long-stream.txt` | a 35 s answer with no async timeout configured |
|
||||||
@@ -0,0 +1,24 @@
|
|||||||
|
$ GET /chat/stream?message=hello world (raw lines, first 14 and last 5)
|
||||||
|
event:token
|
||||||
|
data: [stub]
|
||||||
|
|
||||||
|
event:token
|
||||||
|
data: saw
|
||||||
|
|
||||||
|
event:token
|
||||||
|
data: 2
|
||||||
|
|
||||||
|
event:token
|
||||||
|
data: messages.
|
||||||
|
|
||||||
|
event:token
|
||||||
|
data: Earlier
|
||||||
|
...
|
||||||
|
data: world
|
||||||
|
<blank>
|
||||||
|
event:done
|
||||||
|
data:
|
||||||
|
<blank>
|
||||||
|
event names seen: [token, done]
|
||||||
|
joined token text as EventSource delivers it: [stub] saw 2 messages. Earlier user turns: none. You said: hello world
|
||||||
|
same text if the server did not pad: [stub]saw2messages.Earlieruserturns:none.Yousaid:helloworld
|
||||||
@@ -0,0 +1,4 @@
|
|||||||
|
session A, turn 1: [stub] saw 2 messages. Earlier user turns: none. You said: My name is Asha and I like tea
|
||||||
|
session A, turn 2: [stub] saw 4 messages. Earlier user turns: My name is Asha and I like tea. You said: What did I just tell you?
|
||||||
|
session B, turn 1: [stub] saw 2 messages. Earlier user turns: none. You said: What did I just tell you?
|
||||||
|
session A, turn 3: [stub] saw 6 messages. Earlier user turns: My name is Asha and I like tea | What did I just tell you?. You said: And now?
|
||||||
@@ -0,0 +1 @@
|
|||||||
|
data sent to the browser: Here is <b>bold</b> and <img src=x onerror=alert(1)> & done
|
||||||
@@ -0,0 +1,4 @@
|
|||||||
|
client read 40000 bytes then closed the socket
|
||||||
|
Flux cancelled on the server: 1
|
||||||
|
stub saw the upstream connection close: 1 (within 50 ms)
|
||||||
|
stub chunks written for this request: 5 of 400 requested
|
||||||
@@ -0,0 +1,4 @@
|
|||||||
|
plain OkHttp 4.12.0, 10 runs each, stub streaming 60 chunks at 20 ms, reader thread blocked in a read
|
||||||
|
response.close() threw: java.lang.IllegalStateException: Unbalanced enter/exit
|
||||||
|
response.close() from another thread: upstream connection closed in 0 of 10 runs
|
||||||
|
call.cancel() from another thread: upstream connection closed in 10 of 10 runs
|
||||||
@@ -0,0 +1,18 @@
|
|||||||
|
== turn 1 (session A)
|
||||||
|
A1: '[stub] saw 2 messages. Earlier user turns: none. You said: My name is Asha and I like tea'
|
||||||
|
== turn 2 (session A, same cookie)
|
||||||
|
A2: '[stub] saw 4 messages. Earlier user turns: My name is Asha and I like tea. You said: What did I just tell you?'
|
||||||
|
cookies: ['JSESSIONID']
|
||||||
|
== turn 1 (session B, fresh browser context, no cookie)
|
||||||
|
B1: '[stub] saw 2 messages. Earlier user turns: none. You said: What did I just tell you?'
|
||||||
|
== New conversation button in session A, then ask again
|
||||||
|
A3: '[stub] saw 2 messages. Earlier user turns: none. You said: Do you remember me?'
|
||||||
|
== model output containing markup
|
||||||
|
rendered text: 'Here is <b>bold</b> and <img src=x onerror=alert(1)> & done'
|
||||||
|
<b> elements created inside the answer: 0
|
||||||
|
<img> elements created inside the answer: 0
|
||||||
|
alert dialogs fired: 0
|
||||||
|
== stop button
|
||||||
|
text at click : 'w1 w2 w3 w4 w5 w6 w7 w8 w9 w10 w11'
|
||||||
|
text 1.5 s on : 'w1 w2 w3 w4 w5 w6 w7 w8 w9 w10 w11 w12'
|
||||||
|
grew after stop: False
|
||||||
@@ -0,0 +1,6 @@
|
|||||||
|
-- app log (the turn started by the Stop-button test)
|
||||||
|
turn 6 start conversation=def8b65c-b58a-4677-b149-e71575426ebd message=count 200
|
||||||
|
turn 6 CANCELLED by client after 14 tokens
|
||||||
|
-- stub log
|
||||||
|
stub: request #6 model=stub-gpt stream=true messages=6 roles=system,user,assistant,user,assistant,user latest="count 200"
|
||||||
|
stub: stream #6 client closed the connection after 15 chunks
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
== chunk delay 20 ms, 8 runs, app args: --chat.cancel-fix=false
|
||||||
|
run 1: upstream closed
|
||||||
|
run 2: upstream LEFT OPEN
|
||||||
|
run 3: upstream LEFT OPEN
|
||||||
|
run 4: upstream LEFT OPEN
|
||||||
|
run 5: upstream LEFT OPEN
|
||||||
|
run 6: upstream LEFT OPEN
|
||||||
|
run 7: upstream LEFT OPEN
|
||||||
|
run 8: upstream LEFT OPEN
|
||||||
|
upstream closed: 1 of 8, left open: 7 of 8
|
||||||
|
|
||||||
|
== chunk delay 20 ms, 8 runs, app args: (none)
|
||||||
|
run 1: upstream closed
|
||||||
|
run 2: upstream closed
|
||||||
|
run 3: upstream closed
|
||||||
|
run 4: upstream closed
|
||||||
|
run 5: upstream closed
|
||||||
|
run 6: upstream closed
|
||||||
|
run 7: upstream closed
|
||||||
|
run 8: upstream closed
|
||||||
|
upstream closed: 8 of 8, left open: 0 of 8
|
||||||
@@ -0,0 +1,45 @@
|
|||||||
|
== chat.backpressure=none (stub delay 0 ms, 600 chunks x 32 KiB, client reads 64 KiB every 0.2 s)
|
||||||
|
t= 1.0s client got 0 KiB stub: chunksWritten=0 finished=0 abortedByClient=0 app heap after GC: used 34669K
|
||||||
|
t= 2.8s client got 197 KiB stub: chunksWritten=452 finished=0 abortedByClient=0 app heap after GC: used 134096K
|
||||||
|
t= 5.5s client got 896 KiB stub: chunksWritten=603 finished=1 abortedByClient=0 app heap after GC: used 86306K
|
||||||
|
t= 8.9s client got 1832 KiB stub: chunksWritten=603 finished=1 abortedByClient=0 app heap after GC: used 109343K
|
||||||
|
t=13.3s client got 2967 KiB stub: chunksWritten=603 finished=1 abortedByClient=0 app heap after GC: used 116565K
|
||||||
|
-- app log (demand + outcome):
|
||||||
|
213 web-side demand request(1)
|
||||||
|
213 model-side demand request(1)
|
||||||
|
1 CANCELLED by client after 213 tokens
|
||||||
|
|
||||||
|
== chat.backpressure=limitRate (stub delay 0 ms, 600 chunks x 32 KiB, client reads 64 KiB every 0.2 s)
|
||||||
|
t= 1.0s client got 0 KiB stub: chunksWritten=0 finished=0 abortedByClient=0 app heap after GC: used 42630K
|
||||||
|
t= 2.8s client got 242 KiB stub: chunksWritten=603 finished=1 abortedByClient=0 app heap after GC: used 86598K
|
||||||
|
t= 5.2s client got 896 KiB stub: chunksWritten=603 finished=1 abortedByClient=0 app heap after GC: used 86745K
|
||||||
|
t= 8.5s client got 1768 KiB stub: chunksWritten=603 finished=1 abortedByClient=0 app heap after GC: used 109571K
|
||||||
|
t=12.9s client got 2967 KiB stub: chunksWritten=603 finished=1 abortedByClient=0 app heap after GC: used 116491K
|
||||||
|
-- app log (demand + outcome):
|
||||||
|
213 web-side demand request(1)
|
||||||
|
35 model-side demand request(6)
|
||||||
|
1 model-side demand request(8)
|
||||||
|
1 CANCELLED by client after 213 tokens
|
||||||
|
|
||||||
|
== chat.backpressure=slow (stub delay 0 ms, 600 chunks x 32 KiB, client reads 64 KiB every 0.2 s)
|
||||||
|
t= 1.0s client got 0 KiB stub: chunksWritten=0 finished=0 abortedByClient=0 app heap after GC: used 40942K
|
||||||
|
t= 2.6s client got 72 KiB stub: chunksWritten=218 finished=0 abortedByClient=0 app heap after GC: used 146705K
|
||||||
|
t= 5.3s client got 787 KiB stub: chunksWritten=603 finished=1 abortedByClient=0 app heap after GC: used 111009K
|
||||||
|
t= 8.6s client got 1659 KiB stub: chunksWritten=603 finished=1 abortedByClient=0 app heap after GC: used 109188K
|
||||||
|
t=13.0s client got 2858 KiB stub: chunksWritten=603 finished=1 abortedByClient=0 app heap after GC: used 119593K
|
||||||
|
-- app log (demand + outcome):
|
||||||
|
221 model-side demand request(1)
|
||||||
|
220 web-side demand request(1)
|
||||||
|
1 CANCELLED by client after 220 tokens
|
||||||
|
|
||||||
|
== chat.backpressure=bounded (stub delay 0 ms, 600 chunks x 32 KiB, client reads 64 KiB every 0.2 s)
|
||||||
|
t= 1.0s client got 0 KiB stub: chunksWritten=0 finished=0 abortedByClient=0 app heap after GC: used 61061K
|
||||||
|
t= 2.7s client got 133 KiB stub: chunksWritten=256 finished=0 abortedByClient=0 app heap after GC: used 58420K
|
||||||
|
t= 5.1s client got 787 KiB stub: chunksWritten=256 finished=0 abortedByClient=1 app heap after GC: used 58420K
|
||||||
|
t= 8.5s client got 1659 KiB stub: chunksWritten=256 finished=0 abortedByClient=1 app heap after GC: used 30251K
|
||||||
|
t=12.8s client got 2858 KiB stub: chunksWritten=256 finished=0 abortedByClient=1 app heap after GC: used 30381K
|
||||||
|
-- app log (demand + outcome):
|
||||||
|
144 web-side demand request(1)
|
||||||
|
1 model-side demand request(unbounded)
|
||||||
|
1 error reactor.core.Exceptions$OverflowException: The receiver is overrun by more signals than expected (bounded queue...)
|
||||||
|
|
||||||
@@ -0,0 +1,7 @@
|
|||||||
|
$ curl -sN "localhost:38180/chat/stream?message=count 700" (700 tokens x 50 ms = 35 s)
|
||||||
|
<blank>
|
||||||
|
event:done
|
||||||
|
data:
|
||||||
|
<blank>
|
||||||
|
elapsed: 38 s
|
||||||
|
turn 1 complete, 700 tokens
|
||||||
@@ -0,0 +1 @@
|
|||||||
|
stream failed: IllegalArgumentException: conversationId cannot be null
|
||||||
@@ -0,0 +1,2 @@
|
|||||||
|
turn 1 'warm' completed, turn 2 'count 300' cancelled after a few tokens, then turn 3:
|
||||||
|
[stub] saw 5 messages. Earlier user turns: warm | count 300. You said: What do you remember?
|
||||||
@@ -0,0 +1,73 @@
|
|||||||
|
<?xml version="1.0" encoding="UTF-8"?>
|
||||||
|
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||||
|
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||||
|
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||||
|
<modelVersion>4.0.0</modelVersion>
|
||||||
|
|
||||||
|
<parent>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-starter-parent</artifactId>
|
||||||
|
<version>4.1.2</version>
|
||||||
|
<relativePath/>
|
||||||
|
</parent>
|
||||||
|
|
||||||
|
<groupId>com.ankurm</groupId>
|
||||||
|
<artifactId>chat-ui-htmx</artifactId>
|
||||||
|
<version>1.0.0</version>
|
||||||
|
<name>chat-ui-htmx</name>
|
||||||
|
<description>Streaming chat UI: Spring AI ChatClient to Server-Sent Events to htmx, with per-session memory, cancel and back-pressure.</description>
|
||||||
|
|
||||||
|
<properties>
|
||||||
|
<java.version>25</java.version>
|
||||||
|
<spring-ai.version>2.0.2</spring-ai.version>
|
||||||
|
</properties>
|
||||||
|
|
||||||
|
<dependencyManagement>
|
||||||
|
<dependencies>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.ai</groupId>
|
||||||
|
<artifactId>spring-ai-bom</artifactId>
|
||||||
|
<version>${spring-ai.version}</version>
|
||||||
|
<type>pom</type>
|
||||||
|
<scope>import</scope>
|
||||||
|
</dependency>
|
||||||
|
</dependencies>
|
||||||
|
</dependencyManagement>
|
||||||
|
|
||||||
|
<dependencies>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-starter-webmvc</artifactId>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.ai</groupId>
|
||||||
|
<artifactId>spring-ai-starter-model-openai</artifactId>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-starter-test</artifactId>
|
||||||
|
<scope>test</scope>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>io.projectreactor</groupId>
|
||||||
|
<artifactId>reactor-test</artifactId>
|
||||||
|
<scope>test</scope>
|
||||||
|
</dependency>
|
||||||
|
</dependencies>
|
||||||
|
|
||||||
|
<build>
|
||||||
|
<plugins>
|
||||||
|
<plugin>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||||
|
</plugin>
|
||||||
|
<plugin>
|
||||||
|
<groupId>org.apache.maven.plugins</groupId>
|
||||||
|
<artifactId>maven-surefire-plugin</artifactId>
|
||||||
|
<configuration>
|
||||||
|
<argLine>-Duser.timezone=UTC -Dstdout.encoding=UTF-8 -Dfile.encoding=UTF-8</argLine>
|
||||||
|
</configuration>
|
||||||
|
</plugin>
|
||||||
|
</plugins>
|
||||||
|
</build>
|
||||||
|
</project>
|
||||||
Executable
+25
@@ -0,0 +1,25 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# A slow client against a fast model: curl reads 300 KB/s of a 600 x 32 KB stream (19.6 MB) while the
|
||||||
|
# stub writes with no delay. Samples how far the stub got and how much heap the app holds.
|
||||||
|
# Usage: backpressure.sh <none|limitRate|slow> [extra app args]
|
||||||
|
source "$(dirname "$0")/lib.sh"; set +e
|
||||||
|
MODE=${1:-none}; shift
|
||||||
|
stop_all; start_stub 0; start_app --chat.backpressure="$MODE" --chat.slow-millis=5 "$@"
|
||||||
|
echo "== chat.backpressure=$MODE (stub delay 0 ms, 600 chunks x 32 KiB, client reads 64 KiB every 0.2 s)"
|
||||||
|
echo 0 > "$RUN/recv.bytes"
|
||||||
|
python3 scripts/slow_client.py "http://localhost:$APP_PORT/chat/stream?message=big%20600&turn=1" "$RUN/recv.bytes" &
|
||||||
|
CURL=$!
|
||||||
|
T0=$(date +%s.%N)
|
||||||
|
for pause in 1 1 2 3 4; do
|
||||||
|
sleep "$pause"
|
||||||
|
st=$(curl -s "localhost:$STUB_PORT/stub/stats" | sed 's/requests=[0-9]* //')
|
||||||
|
recv=$(cat "$RUN/recv.bytes")
|
||||||
|
now=$(date +%s.%N)
|
||||||
|
jcmd "$(cat $RUN/app.pid)" GC.run >/dev/null 2>&1
|
||||||
|
heap=$(jcmd "$(cat $RUN/app.pid)" GC.heap_info 2>/dev/null | grep -o "used [0-9]*K" | head -1)
|
||||||
|
printf "t=%4.1fs client got %5d KiB stub: %s app heap after GC: %s\n" "$(echo "$now - $T0" | bc)" $((recv/1024)) "$st" "$heap"
|
||||||
|
done
|
||||||
|
wait $CURL 2>/dev/null
|
||||||
|
echo "-- app log (demand + outcome):"
|
||||||
|
grep "demand\|CANCELLED\|complete\|error" $RUN/app.log | sed "s/turn 1 //" | sort | uniq -c | sort -rn | head -6
|
||||||
|
stop_all
|
||||||
Executable
+67
@@ -0,0 +1,67 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Drives the real page in headless Chromium (Playwright for Python) against the running app + stub.
|
||||||
|
Prints a transcript to stdout. Needs: pip install playwright && playwright install chromium."""
|
||||||
|
import sys, time, urllib.request
|
||||||
|
from playwright.sync_api import sync_playwright
|
||||||
|
|
||||||
|
APP = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:38180"
|
||||||
|
STUB = sys.argv[2] if len(sys.argv) > 2 else "http://localhost:38181"
|
||||||
|
|
||||||
|
def stub(path):
|
||||||
|
return urllib.request.urlopen(STUB + path, data=b"" if "config" in path else None).read().decode().strip()
|
||||||
|
|
||||||
|
def ask(page, text):
|
||||||
|
n = page.locator(".msg.bot").count()
|
||||||
|
page.fill("input[name=message]", text)
|
||||||
|
page.click("form button")
|
||||||
|
page.wait_for_selector(f".msg.bot >> nth={n}")
|
||||||
|
return int(page.locator(".msg.bot span[id^=out-]").last.get_attribute("id")[4:])
|
||||||
|
|
||||||
|
def wait_done(page, n, timeout=20000):
|
||||||
|
page.wait_for_function(f"!document.querySelector('#stop-{n}')", timeout=timeout)
|
||||||
|
|
||||||
|
def bot_text(page, n):
|
||||||
|
return page.locator(f"#out-{n}").inner_text()
|
||||||
|
|
||||||
|
with sync_playwright() as p:
|
||||||
|
b = p.chromium.launch()
|
||||||
|
ctx = b.new_context()
|
||||||
|
page = ctx.new_page()
|
||||||
|
dialogs = []
|
||||||
|
page.on("dialog", lambda d: (dialogs.append(d.message), d.dismiss()))
|
||||||
|
page.goto(APP + "/")
|
||||||
|
print("== turn 1 (session A)")
|
||||||
|
n = ask(page, "My name is Asha and I like tea"); wait_done(page, n)
|
||||||
|
print("A1:", repr(bot_text(page, n)))
|
||||||
|
print("== turn 2 (session A, same cookie)")
|
||||||
|
n = ask(page, "What did I just tell you?"); wait_done(page, n)
|
||||||
|
print("A2:", repr(bot_text(page, n)))
|
||||||
|
print("cookies:", [c["name"] for c in ctx.cookies()])
|
||||||
|
print("== turn 1 (session B, fresh browser context, no cookie)")
|
||||||
|
ctx2 = b.new_context(); page2 = ctx2.new_page(); page2.goto(APP + "/")
|
||||||
|
n = ask(page2, "What did I just tell you?"); wait_done(page2, n)
|
||||||
|
print("B1:", repr(bot_text(page2, n)))
|
||||||
|
print("== New conversation button in session A, then ask again")
|
||||||
|
page.click("text=New conversation"); time.sleep(0.5)
|
||||||
|
n = ask(page, "Do you remember me?"); wait_done(page, n)
|
||||||
|
print("A3:", repr(bot_text(page, n)))
|
||||||
|
print("== model output containing markup")
|
||||||
|
n = ask(page, "html please"); wait_done(page, n)
|
||||||
|
print("rendered text:", repr(bot_text(page, n)))
|
||||||
|
print("<b> elements created inside the answer:", page.locator(f"#out-{n} b").count())
|
||||||
|
print("<img> elements created inside the answer:", page.locator(f"#out-{n} img").count())
|
||||||
|
print("alert dialogs fired:", len(dialogs))
|
||||||
|
print("== stop button")
|
||||||
|
stub("/stub/config?delayMs=50")
|
||||||
|
n = ask(page, "count 200")
|
||||||
|
page.wait_for_function(f"document.querySelector('#out-{n}').innerText.length > 30")
|
||||||
|
before = bot_text(page, n)
|
||||||
|
t0 = time.time(); page.click(f"#stop-{n}")
|
||||||
|
page.wait_for_function(f"!document.querySelector('#stop-{n}')")
|
||||||
|
time.sleep(1.5)
|
||||||
|
after = bot_text(page, n)
|
||||||
|
print("text at click :", repr(before))
|
||||||
|
print("text 1.5 s on :", repr(after))
|
||||||
|
print("grew after stop:", len(after.replace(' [stopped]','')) > len(before.replace(' [stopped]','')) + 6)
|
||||||
|
stub("/stub/config?delayMs=20")
|
||||||
|
b.close()
|
||||||
Executable
+18
@@ -0,0 +1,18 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# Aborts a streaming request from the client side N times and asks the stub what happened to the
|
||||||
|
# upstream HTTP stream: closed (abortedByClient) or left open. Usage: cancel-upstream.sh <chunk-delay-ms> <runs> [app args]
|
||||||
|
source "$(dirname "$0")/lib.sh"; set +e
|
||||||
|
DELAY=${1:-20}; RUNS=${2:-10}; shift 2
|
||||||
|
stop_all; start_stub "$DELAY"; start_app "$@"
|
||||||
|
echo "== chunk delay ${DELAY} ms, ${RUNS} runs, app args: ${*:-(none)}"
|
||||||
|
closed=0; open=0
|
||||||
|
for r in $(seq 1 "$RUNS"); do
|
||||||
|
b=$(curl -s "localhost:$STUB_PORT/stub/stats" | sed 's/.*abortedByClient=//')
|
||||||
|
curl -sN --max-time 1 "localhost:$APP_PORT/chat/stream?message=big%20400&turn=$r" >/dev/null
|
||||||
|
sleep 2
|
||||||
|
a=$(curl -s "localhost:$STUB_PORT/stub/stats" | sed 's/.*abortedByClient=//')
|
||||||
|
if [ "$a" -gt "$b" ]; then closed=$((closed+1)); res=closed; else open=$((open+1)); res="LEFT OPEN"; fi
|
||||||
|
echo "run $r: upstream $res"
|
||||||
|
done
|
||||||
|
echo "upstream closed: $closed of $RUNS, left open: $open of $RUNS"
|
||||||
|
stop_all
|
||||||
Executable
+25
@@ -0,0 +1,25 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# Shared helpers: JDK 25, ports, start/stop of the stub and the app. Stop is by recorded PID only.
|
||||||
|
set -euo pipefail
|
||||||
|
cd "$(dirname "${BASH_SOURCE[0]}")/.."
|
||||||
|
export JAVA_HOME="${CHATUI_JAVA_HOME:-/tmp/tools/jdk-25.0.5+8}"
|
||||||
|
export PATH="$JAVA_HOME/bin:$PATH"
|
||||||
|
APP_PORT=${APP_PORT:-38180}
|
||||||
|
STUB_PORT=${STUB_PORT:-38181}
|
||||||
|
RUN=target/run; mkdir -p "$RUN" output
|
||||||
|
|
||||||
|
build() { [ -f target/cp.txt ] || mvn -B -q dependency:build-classpath -Dmdep.outputFile=target/cp.txt >/dev/null
|
||||||
|
mvn -B -q -DskipTests package >/dev/null; }
|
||||||
|
|
||||||
|
start_stub() { # $1 = delay ms
|
||||||
|
setsid nohup sh -c 'echo $$ > "$1"; shift; exec java "$@"' sh "$RUN/stub.pid" -cp "target/classes:$(cat target/cp.txt)" \
|
||||||
|
com.ankurm.chatui.stub.OpenAiStubServer "$STUB_PORT" "${1:-20}" > "$RUN/stub.log" 2>&1 < /dev/null &
|
||||||
|
for i in $(seq 1 30); do curl -s -o /dev/null "localhost:$STUB_PORT/stub/stats" && return; sleep 0.5; done; }
|
||||||
|
|
||||||
|
start_app() { # extra args are passed to the app (e.g. --chat.backpressure=limitRate)
|
||||||
|
setsid nohup sh -c 'echo $$ > "$1"; shift; exec java "$@"' sh "$RUN/app.pid" -jar target/chat-ui-htmx-1.0.0.jar \
|
||||||
|
--server.port="$APP_PORT" "$@" > "$RUN/app.log" 2>&1 < /dev/null &
|
||||||
|
for i in $(seq 1 60); do curl -s -o /dev/null "localhost:$APP_PORT/" && return; sleep 1; done; }
|
||||||
|
|
||||||
|
stop_pid() { [ -f "$RUN/$1.pid" ] && kill "$(cat "$RUN/$1.pid")" 2>/dev/null || true; rm -f "$RUN/$1.pid"; sleep 1; }
|
||||||
|
stop_all() { stop_pid app; stop_pid stub; }
|
||||||
Executable
+25
@@ -0,0 +1,25 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# Regenerates every file in output/. Needs JDK 25, Maven, curl, python3 and (for step 2) Playwright for
|
||||||
|
# Python with Chromium. Timing-dependent figures (back-pressure samples, cancel counts) differ run to run.
|
||||||
|
source "$(dirname "$0")/lib.sh"; set +e
|
||||||
|
echo "[1/6] unit + integration tests (write output/01..05)"
|
||||||
|
mvn -B -q test 2>&1 | grep "Tests run:\|FAIL\|ERROR" | tail -3
|
||||||
|
build
|
||||||
|
echo "[2/6] real browser (headless Chromium): memory, escaping, Stop button -> output/06-browser.txt, 07-stop-logs.txt"
|
||||||
|
stop_all; start_stub 20; start_app
|
||||||
|
python3 scripts/browser-test.py "http://localhost:$APP_PORT" "http://localhost:$STUB_PORT" > output/06-browser.txt 2>&1
|
||||||
|
sleep 12 # let the stub finish or abort the 200-token stream
|
||||||
|
{ echo "-- app log (the turn started by the Stop-button test)"; grep -E "CANCELLED|count 200" "$RUN/app.log"
|
||||||
|
echo "-- stub log"; grep -E "count 200|stream #.* (finished|client closed)" "$RUN/stub.log" | tail -2; } > output/07-stop-logs.txt
|
||||||
|
stop_all
|
||||||
|
echo "[3/6] client abort, upstream closed or left open? 8 runs each -> output/08-cancel-upstream.txt"
|
||||||
|
{ scripts/cancel-upstream.sh 20 8 --chat.cancel-fix=false; echo; scripts/cancel-upstream.sh 20 8; } 2>&1 | grep -v "^Picked\|Done" > output/08-cancel-upstream.txt
|
||||||
|
echo "[4/6] slow client, four ways to shape the Flux -> output/09-backpressure.txt"
|
||||||
|
for m in none limitRate slow bounded; do scripts/backpressure.sh "$m"; echo; done 2>&1 | grep -v "^Picked\|Done" > output/09-backpressure.txt
|
||||||
|
echo "[5/6] a 35 s answer with no async timeout configured -> output/10-long-stream.txt"
|
||||||
|
stop_all; start_stub 50; start_app
|
||||||
|
{ echo '$ curl -sN "localhost:'$APP_PORT'/chat/stream?message=count 700" (700 tokens x 50 ms = 35 s)'
|
||||||
|
s=$(date +%s); curl -sN "localhost:$APP_PORT/chat/stream?message=count%20700&turn=1" | tail -n 4 | sed 's/^$/<blank>/'
|
||||||
|
echo "elapsed: $(( $(date +%s) - s )) s"; grep -E "complete|error" "$RUN/app.log"; } > output/10-long-stream.txt 2>&1
|
||||||
|
stop_all
|
||||||
|
echo "[6/6] done"
|
||||||
Executable
+22
@@ -0,0 +1,22 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""A deliberately slow SSE reader on a raw socket: recv() up to CHUNK bytes every INTERVAL seconds
|
||||||
|
(default 64 KiB / 0.2 s, about 320 KB/s at most) and keep the running byte count in a file so another
|
||||||
|
process can sample it. Usage: slow_client.py URL COUNTER_FILE [CHUNK] [INTERVAL] [SECONDS]"""
|
||||||
|
import socket, sys, time, urllib.parse
|
||||||
|
url, counter = sys.argv[1], sys.argv[2]
|
||||||
|
chunk = int(sys.argv[3]) if len(sys.argv) > 3 else 64 * 1024
|
||||||
|
interval = float(sys.argv[4]) if len(sys.argv) > 4 else 0.2
|
||||||
|
limit = float(sys.argv[5]) if len(sys.argv) > 5 else 14.0
|
||||||
|
u = urllib.parse.urlparse(url)
|
||||||
|
s = socket.create_connection((u.hostname, u.port))
|
||||||
|
s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 64 * 1024)
|
||||||
|
s.sendall(f"GET {u.path}?{u.query} HTTP/1.1\r\nHost: {u.hostname}\r\nAccept: text/event-stream\r\n\r\n".encode())
|
||||||
|
total, t0 = 0, time.time()
|
||||||
|
while time.time() - t0 < limit:
|
||||||
|
data = s.recv(chunk)
|
||||||
|
if not data:
|
||||||
|
break
|
||||||
|
total += len(data)
|
||||||
|
open(counter, "w").write(str(total))
|
||||||
|
time.sleep(interval)
|
||||||
|
s.close()
|
||||||
@@ -0,0 +1,51 @@
|
|||||||
|
package com.ankurm.chatui;
|
||||||
|
|
||||||
|
import okhttp3.Interceptor;
|
||||||
|
import okhttp3.MediaType;
|
||||||
|
import okhttp3.Response;
|
||||||
|
import okhttp3.ResponseBody;
|
||||||
|
import okio.BufferedSource;
|
||||||
|
import org.springframework.ai.openai.http.okhttp.OpenAiHttpClientBuilderCustomizer;
|
||||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Makes a cancelled stream actually close its HTTP connection.
|
||||||
|
*
|
||||||
|
* Spring AI 2.0.2's OkHttp client closes a streaming response with {@code response.body().close()}.
|
||||||
|
* When that runs on a different thread from the one blocked reading the body, OkHttp 4.12 throws
|
||||||
|
* {@code IllegalStateException: Unbalanced enter/exit} and leaves the connection open, so the model
|
||||||
|
* server keeps generating. {@code Call.cancel()} is the call OkHttp documents for this and is safe from
|
||||||
|
* any thread, so this interceptor wraps SSE response bodies to call it first.
|
||||||
|
* Switch off with {@code chat.cancel-fix=false} to reproduce the leak (scripts/cancel-upstream.sh).
|
||||||
|
*/
|
||||||
|
@Configuration
|
||||||
|
@ConditionalOnProperty(name = "chat.cancel-fix", havingValue = "true", matchIfMissing = true)
|
||||||
|
public class CancelOnCloseConfig {
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
OpenAiHttpClientBuilderCustomizer cancelOnClose() {
|
||||||
|
return builder -> builder.interceptor(chain -> {
|
||||||
|
Response response = chain.proceed(chain.request());
|
||||||
|
ResponseBody body = response.body();
|
||||||
|
if (body == null || body.contentType() == null || !"event-stream".equals(body.contentType().subtype())) {
|
||||||
|
return response;
|
||||||
|
}
|
||||||
|
var call = chain.call();
|
||||||
|
return response.newBuilder().body(new ResponseBody() {
|
||||||
|
@Override public MediaType contentType() { return body.contentType(); }
|
||||||
|
@Override public long contentLength() { return body.contentLength(); }
|
||||||
|
@Override public BufferedSource source() { return body.source(); }
|
||||||
|
@Override public void close() {
|
||||||
|
call.cancel();
|
||||||
|
try {
|
||||||
|
body.close();
|
||||||
|
} catch (RuntimeException ignored) {
|
||||||
|
// the call is already cancelled; closing the body can only complain
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}).build();
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,28 @@
|
|||||||
|
package com.ankurm.chatui;
|
||||||
|
|
||||||
|
import org.springframework.ai.chat.client.ChatClient;
|
||||||
|
import org.springframework.ai.chat.client.advisor.MessageChatMemoryAdvisor;
|
||||||
|
import org.springframework.ai.chat.memory.ChatMemory;
|
||||||
|
import org.springframework.ai.chat.memory.MessageWindowChatMemory;
|
||||||
|
import org.springframework.ai.chat.model.ChatModel;
|
||||||
|
import org.springframework.beans.factory.annotation.Value;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
|
||||||
|
@Configuration
|
||||||
|
public class ChatConfig {
|
||||||
|
|
||||||
|
/** In-memory window of the last N messages per conversation id (the default repository is a map). */
|
||||||
|
@Bean
|
||||||
|
ChatMemory chatMemory(@Value("${chat.memory.max-messages:20}") int maxMessages) {
|
||||||
|
return MessageWindowChatMemory.builder().maxMessages(maxMessages).build();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
ChatClient chatClient(ChatModel model, ChatMemory memory) {
|
||||||
|
return ChatClient.builder(model)
|
||||||
|
.defaultSystem("You are a concise assistant.")
|
||||||
|
.defaultAdvisors(MessageChatMemoryAdvisor.builder(memory).build())
|
||||||
|
.build();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,143 @@
|
|||||||
|
package com.ankurm.chatui;
|
||||||
|
|
||||||
|
import jakarta.servlet.http.HttpSession;
|
||||||
|
import java.time.Duration;
|
||||||
|
import java.util.UUID;
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
import org.springframework.ai.chat.client.ChatClient;
|
||||||
|
import org.springframework.ai.chat.memory.ChatMemory;
|
||||||
|
import org.springframework.beans.factory.annotation.Value;
|
||||||
|
import org.springframework.http.MediaType;
|
||||||
|
import org.springframework.http.codec.ServerSentEvent;
|
||||||
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
import org.springframework.web.bind.annotation.PostMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RequestParam;
|
||||||
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
|
import org.springframework.web.util.HtmlUtils;
|
||||||
|
import org.springframework.web.util.UriUtils;
|
||||||
|
import reactor.core.publisher.BufferOverflowStrategy;
|
||||||
|
import reactor.core.publisher.Flux;
|
||||||
|
|
||||||
|
import static java.nio.charset.StandardCharsets.UTF_8;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Three endpoints and no JavaScript of our own:
|
||||||
|
* POST /chat returns an HTML fragment (the user bubble and an empty assistant bubble that opens an
|
||||||
|
* EventSource), GET /chat/stream is the Server-Sent Events stream, POST /chat/stop swaps the stream
|
||||||
|
* element out so htmx closes the EventSource.
|
||||||
|
*/
|
||||||
|
@RestController
|
||||||
|
public class ChatController {
|
||||||
|
|
||||||
|
private static final Logger log = LoggerFactory.getLogger("chat");
|
||||||
|
static final String CID = "conversationId";
|
||||||
|
|
||||||
|
private final ChatClient chatClient;
|
||||||
|
private final AtomicInteger turns = new AtomicInteger();
|
||||||
|
/** Test hook: how many streams were cancelled by the client (also logged). */
|
||||||
|
public final AtomicInteger cancelled = new AtomicInteger();
|
||||||
|
/** none | limitRate | slow | bounded: how the Flux is shaped between the model and the socket. */
|
||||||
|
private final String backpressure;
|
||||||
|
private final int slowMillis;
|
||||||
|
|
||||||
|
public ChatController(ChatClient chatClient,
|
||||||
|
@Value("${chat.backpressure:none}") String backpressure,
|
||||||
|
@Value("${chat.slow-millis:0}") int slowMillis) {
|
||||||
|
this.chatClient = chatClient;
|
||||||
|
this.backpressure = backpressure;
|
||||||
|
this.slowMillis = slowMillis;
|
||||||
|
}
|
||||||
|
|
||||||
|
/** The conversation id lives in the HTTP session, so a new browser (new cookie) is a new conversation. */
|
||||||
|
static String conversationId(HttpSession session) {
|
||||||
|
Object id = session.getAttribute(CID);
|
||||||
|
if (id == null) {
|
||||||
|
id = UUID.randomUUID().toString();
|
||||||
|
session.setAttribute(CID, id);
|
||||||
|
}
|
||||||
|
return id.toString();
|
||||||
|
}
|
||||||
|
|
||||||
|
@PostMapping(value = "/chat", produces = MediaType.TEXT_HTML_VALUE)
|
||||||
|
public String send(@RequestParam String message) {
|
||||||
|
int turn = turns.incrementAndGet();
|
||||||
|
String url = "/chat/stream?turn=" + turn + "&message=" + UriUtils.encodeQueryParam(message, UTF_8);
|
||||||
|
return """
|
||||||
|
<div class="msg user">%s</div>
|
||||||
|
<div class="msg bot">
|
||||||
|
<span id="out-%d"></span>
|
||||||
|
<span id="sse-%d" hx-ext="sse" sse-connect="%s" sse-swap="token"
|
||||||
|
sse-close="done" hx-target="#out-%d" hx-swap="beforeend">
|
||||||
|
<span sse-swap="done" hx-target="#stop-%d" hx-swap="delete"></span>
|
||||||
|
</span>
|
||||||
|
<button id="stop-%d" type="button" class="stop"
|
||||||
|
hx-post="/chat/stop?turn=%d" hx-target="#sse-%d" hx-swap="outerHTML">Stop</button>
|
||||||
|
</div>
|
||||||
|
""".formatted(HtmlUtils.htmlEscape(message), turn, turn,
|
||||||
|
HtmlUtils.htmlEscape(url), turn, turn, turn, turn, turn);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Replaces the stream element; htmx cleans the removed element up and closes its EventSource. */
|
||||||
|
@PostMapping(value = "/chat/stop", produces = MediaType.TEXT_HTML_VALUE)
|
||||||
|
public String stop(@RequestParam int turn) {
|
||||||
|
return "<em id=\"sse-" + turn + "\"> [stopped]</em>"
|
||||||
|
+ "<span id=\"stop-" + turn + "\" hx-swap-oob=\"delete\"></span>";
|
||||||
|
}
|
||||||
|
|
||||||
|
@PostMapping("/chat/new")
|
||||||
|
public String newConversation(HttpSession session) {
|
||||||
|
session.removeAttribute(CID);
|
||||||
|
return "ok";
|
||||||
|
}
|
||||||
|
|
||||||
|
@GetMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
|
||||||
|
public Flux<ServerSentEvent<String>> stream(@RequestParam String message,
|
||||||
|
@RequestParam(defaultValue = "0") int turn,
|
||||||
|
HttpSession session) {
|
||||||
|
String cid = conversationId(session);
|
||||||
|
AtomicLong tokens = new AtomicLong();
|
||||||
|
AtomicLong requested = new AtomicLong();
|
||||||
|
|
||||||
|
Flux<String> content = chatClient.prompt()
|
||||||
|
.user(message)
|
||||||
|
.advisors(a -> a.param(ChatMemory.CONVERSATION_ID, cid))
|
||||||
|
.stream()
|
||||||
|
.content()
|
||||||
|
.doOnRequest(n -> log.info("turn {} model-side demand request({})", turn, n == Long.MAX_VALUE ? "unbounded" : n));
|
||||||
|
|
||||||
|
// Optional shaping between the model and the socket (see scripts/backpressure.sh). "model-side"
|
||||||
|
// demand is logged above this switch and "web-side" demand below it.
|
||||||
|
content = switch (backpressure) {
|
||||||
|
case "limitRate" -> content.limitRate(8);
|
||||||
|
// A push source (the SDK reader thread) cannot be slowed by demand; bound the queue and
|
||||||
|
// fail the slow client instead of buffering without limit.
|
||||||
|
case "bounded" -> content.onBackpressureBuffer(16, BufferOverflowStrategy.ERROR);
|
||||||
|
case "slow" -> content.delayElements(Duration.ofMillis(slowMillis));
|
||||||
|
default -> content;
|
||||||
|
};
|
||||||
|
|
||||||
|
return content
|
||||||
|
.doOnRequest(n -> {
|
||||||
|
requested.addAndGet(n);
|
||||||
|
log.info("turn {} web-side demand request({})", turn, n == Long.MAX_VALUE ? "unbounded" : n);
|
||||||
|
})
|
||||||
|
.map(token -> {
|
||||||
|
tokens.incrementAndGet();
|
||||||
|
// Model text is untrusted: htmx inserts the data as HTML, so escape it here.
|
||||||
|
// A browser strips ONE leading space from every data line (SSE spec), and Spring writes
|
||||||
|
// "data:" with no space of its own, so a token like " world" would lose its space.
|
||||||
|
// Prefix every line with a space that the client strips again.
|
||||||
|
String data = " " + HtmlUtils.htmlEscape(token).replace("\n", "\n ");
|
||||||
|
return ServerSentEvent.builder(data).event("token").build();
|
||||||
|
})
|
||||||
|
.concatWith(Flux.just(ServerSentEvent.<String>builder("").event("done").build()))
|
||||||
|
.doOnSubscribe(s -> log.info("turn {} start conversation={} message={}", turn, cid, message))
|
||||||
|
.doOnCancel(() -> cancelled.incrementAndGet())
|
||||||
|
.doOnCancel(() -> log.info("turn {} CANCELLED by client after {} tokens", turn, tokens.get()))
|
||||||
|
.doOnComplete(() -> log.info("turn {} complete, {} tokens", turn, tokens.get()))
|
||||||
|
.doOnError(e -> log.info("turn {} error {}", turn, e.toString()));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,11 @@
|
|||||||
|
package com.ankurm.chatui;
|
||||||
|
|
||||||
|
import org.springframework.boot.SpringApplication;
|
||||||
|
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||||
|
|
||||||
|
@SpringBootApplication
|
||||||
|
public class ChatUiApplication {
|
||||||
|
public static void main(String[] args) {
|
||||||
|
SpringApplication.run(ChatUiApplication.class, args);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,196 @@
|
|||||||
|
package com.ankurm.chatui.stub;
|
||||||
|
|
||||||
|
import com.sun.net.httpserver.HttpExchange;
|
||||||
|
import com.sun.net.httpserver.HttpServer;
|
||||||
|
import java.io.IOException;
|
||||||
|
import java.io.OutputStream;
|
||||||
|
import java.net.InetSocketAddress;
|
||||||
|
import java.nio.charset.StandardCharsets;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.concurrent.Executors;
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
|
import java.util.regex.Matcher;
|
||||||
|
import java.util.regex.Pattern;
|
||||||
|
import tools.jackson.databind.JsonNode;
|
||||||
|
import tools.jackson.databind.json.JsonMapper;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* STUB, NOT A LANGUAGE MODEL. A small OpenAI-compatible server for POST /v1/chat/completions that
|
||||||
|
* streams Server-Sent Events chunks ("data: {chat.completion.chunk}" ... "data: [DONE]") with a
|
||||||
|
* controllable delay between chunks. Its "answers" are templates built from the request it received,
|
||||||
|
* so a test can see exactly which messages the client sent (that is how conversation memory is
|
||||||
|
* observed). It does no inference.
|
||||||
|
*
|
||||||
|
* <pre>
|
||||||
|
* java -cp ... com.ankurm.chatui.stub.OpenAiStubServer [port=38181] [delayMs=20]
|
||||||
|
* POST /stub/config?delayMs=N change the delay between chunks at runtime
|
||||||
|
* GET /stub/stats request, chunk and aborted-stream counters
|
||||||
|
* latest user message "count N" streams N one-word chunks (w1 w2 ... wN)
|
||||||
|
* latest user message "big N" streams N chunks of 32 KB each (to make a slow reader visible)
|
||||||
|
* latest user message "html..." streams text containing markup, to show why the UI escapes it
|
||||||
|
* anything else "[stub] saw N messages. Earlier user turns: a | b. You said: c."
|
||||||
|
* </pre>
|
||||||
|
*/
|
||||||
|
public class OpenAiStubServer {
|
||||||
|
|
||||||
|
private static final JsonMapper JSON = JsonMapper.builder().build();
|
||||||
|
private static final Pattern COUNT = Pattern.compile("^count (\\d+).*", Pattern.DOTALL);
|
||||||
|
private static final Pattern BIG = Pattern.compile("^big (\\d+).*", Pattern.DOTALL);
|
||||||
|
private static final String KB32 = "x".repeat(32 * 1024);
|
||||||
|
|
||||||
|
private static volatile long delayMs = 20;
|
||||||
|
private static final AtomicInteger requests = new AtomicInteger();
|
||||||
|
private static final AtomicLong chunksWritten = new AtomicLong();
|
||||||
|
private static final AtomicInteger aborted = new AtomicInteger();
|
||||||
|
private static final AtomicInteger finished = new AtomicInteger();
|
||||||
|
|
||||||
|
public static void main(String[] args) throws Exception {
|
||||||
|
start(args.length > 0 ? Integer.parseInt(args[0]) : 38181, args.length > 1 ? Long.parseLong(args[1]) : 20);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Starts the stub (also used in-process by the tests). */
|
||||||
|
public static HttpServer start(int port, long delay) throws IOException {
|
||||||
|
delayMs = delay;
|
||||||
|
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", port), 0);
|
||||||
|
server.setExecutor(Executors.newVirtualThreadPerTaskExecutor());
|
||||||
|
server.createContext("/stub/config", ex -> {
|
||||||
|
String q = ex.getRequestURI().getQuery();
|
||||||
|
if (q != null && q.startsWith("delayMs=")) {
|
||||||
|
delayMs = Long.parseLong(q.substring(8));
|
||||||
|
}
|
||||||
|
reply(ex, 200, "delayMs=" + delayMs + "\n");
|
||||||
|
});
|
||||||
|
server.createContext("/stub/stats", ex -> reply(ex, 200,
|
||||||
|
"requests=" + requests + " chunksWritten=" + chunksWritten
|
||||||
|
+ " finished=" + finished + " abortedByClient=" + aborted + "\n"));
|
||||||
|
server.createContext("/v1/chat/completions", OpenAiStubServer::completions);
|
||||||
|
server.start();
|
||||||
|
System.out.println("stub: listening on " + port + " delayMs=" + delayMs);
|
||||||
|
return server;
|
||||||
|
}
|
||||||
|
|
||||||
|
public static void setDelayMs(long ms) {
|
||||||
|
delayMs = ms;
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Counters for tests: {requests, chunksWritten, finished, abortedByClient}. */
|
||||||
|
public static long[] counters() {
|
||||||
|
return new long[] {requests.get(), chunksWritten.get(), finished.get(), aborted.get()};
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void reply(HttpExchange ex, int code, String body) throws IOException {
|
||||||
|
byte[] b = body.getBytes(StandardCharsets.UTF_8);
|
||||||
|
ex.sendResponseHeaders(code, b.length);
|
||||||
|
try (OutputStream os = ex.getResponseBody()) {
|
||||||
|
os.write(b);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void completions(HttpExchange ex) throws IOException {
|
||||||
|
int id = requests.incrementAndGet();
|
||||||
|
JsonNode body = JSON.readTree(ex.getRequestBody());
|
||||||
|
String model = body.path("model").asText("?");
|
||||||
|
boolean stream = body.path("stream").asBoolean(false);
|
||||||
|
boolean usage = body.path("stream_options").path("include_usage").asBoolean(false);
|
||||||
|
List<String> userTurns = new ArrayList<>();
|
||||||
|
StringBuilder roles = new StringBuilder();
|
||||||
|
int n = 0;
|
||||||
|
for (JsonNode m : body.path("messages")) {
|
||||||
|
n++;
|
||||||
|
String role = m.path("role").asText();
|
||||||
|
roles.append(roles.isEmpty() ? "" : ",").append(role);
|
||||||
|
if (role.equals("user")) {
|
||||||
|
userTurns.add(text(m.path("content")));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
String latest = userTurns.isEmpty() ? "" : userTurns.getLast();
|
||||||
|
List<String> earlier = userTurns.size() > 1 ? userTurns.subList(0, userTurns.size() - 1) : List.of();
|
||||||
|
System.out.println("stub: request #" + id + " model=" + model + " stream=" + stream
|
||||||
|
+ " messages=" + n + " roles=" + roles + " latest=\"" + latest + "\"");
|
||||||
|
if (!stream) {
|
||||||
|
reply(ex, 400, "{\"error\":{\"message\":\"stub only streams\"}}");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
List<String> tokens = new ArrayList<>();
|
||||||
|
Matcher c = COUNT.matcher(latest);
|
||||||
|
if (c.matches()) {
|
||||||
|
for (int i = 1; i <= Integer.parseInt(c.group(1)); i++) {
|
||||||
|
tokens.add((i == 1 ? "" : " ") + "w" + i);
|
||||||
|
}
|
||||||
|
} else if (BIG.matcher(latest).matches()) {
|
||||||
|
Matcher m = BIG.matcher(latest);
|
||||||
|
m.matches();
|
||||||
|
for (int i = 1; i <= Integer.parseInt(m.group(1)); i++) {
|
||||||
|
tokens.add(KB32);
|
||||||
|
}
|
||||||
|
} else if (latest.startsWith("html")) {
|
||||||
|
for (String w : "Here is <b>bold</b> and <img src=x onerror=alert(1)> & done".split("(?= )")) {
|
||||||
|
tokens.add(w);
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
String text = "[stub] saw " + n + " messages. Earlier user turns: "
|
||||||
|
+ (earlier.isEmpty() ? "none" : String.join(" | ", earlier)) + ". You said: " + latest;
|
||||||
|
for (String w : text.split("(?= )")) {
|
||||||
|
tokens.add(w);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
ex.getResponseHeaders().add("Content-Type", "text/event-stream");
|
||||||
|
ex.getResponseHeaders().add("Cache-Control", "no-cache");
|
||||||
|
ex.sendResponseHeaders(200, 0); // chunked
|
||||||
|
long created = System.currentTimeMillis() / 1000;
|
||||||
|
String cid = "chatcmpl-stub-" + id;
|
||||||
|
int sent = 0;
|
||||||
|
try (OutputStream os = ex.getResponseBody()) {
|
||||||
|
write(os, chunk(cid, created, model, "{\"role\":\"assistant\",\"content\":\"\"}", null));
|
||||||
|
for (String t : tokens) {
|
||||||
|
if (delayMs > 0) {
|
||||||
|
Thread.sleep(delayMs);
|
||||||
|
}
|
||||||
|
write(os, chunk(cid, created, model, "{\"content\":" + JSON.writeValueAsString(t) + "}", null));
|
||||||
|
sent++;
|
||||||
|
}
|
||||||
|
write(os, chunk(cid, created, model, "{}", "stop"));
|
||||||
|
if (usage) {
|
||||||
|
write(os, "{\"id\":\"" + cid + "\",\"object\":\"chat.completion.chunk\",\"created\":" + created
|
||||||
|
+ ",\"model\":\"" + model + "\",\"choices\":[],\"usage\":{\"prompt_tokens\":" + n
|
||||||
|
+ ",\"completion_tokens\":" + sent + ",\"total_tokens\":" + (n + sent) + "}}");
|
||||||
|
}
|
||||||
|
os.write("data: [DONE]\n\n".getBytes(StandardCharsets.UTF_8));
|
||||||
|
os.flush();
|
||||||
|
finished.incrementAndGet();
|
||||||
|
System.out.println("stub: stream #" + id + " finished after " + sent + " chunks");
|
||||||
|
} catch (IOException e) {
|
||||||
|
aborted.incrementAndGet();
|
||||||
|
System.out.println("stub: stream #" + id + " client closed the connection after " + sent + " chunks");
|
||||||
|
} catch (InterruptedException e) {
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static String text(JsonNode content) {
|
||||||
|
if (content.isString()) {
|
||||||
|
return content.asString();
|
||||||
|
}
|
||||||
|
StringBuilder sb = new StringBuilder();
|
||||||
|
for (JsonNode part : content) {
|
||||||
|
sb.append(part.path("text").asString(""));
|
||||||
|
}
|
||||||
|
return sb.toString();
|
||||||
|
}
|
||||||
|
|
||||||
|
private static String chunk(String id, long created, String model, String delta, String finish) {
|
||||||
|
return "{\"id\":\"" + id + "\",\"object\":\"chat.completion.chunk\",\"created\":" + created
|
||||||
|
+ ",\"model\":\"" + model + "\",\"choices\":[{\"index\":0,\"delta\":" + delta
|
||||||
|
+ ",\"finish_reason\":" + (finish == null ? "null" : "\"" + finish + "\"") + "}]}";
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void write(OutputStream os, String json) throws IOException {
|
||||||
|
os.write(("data: " + json + "\n\n").getBytes(StandardCharsets.UTF_8));
|
||||||
|
os.flush();
|
||||||
|
chunksWritten.incrementAndGet();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,22 @@
|
|||||||
|
# The OpenAI starter used as a generic OpenAI-compatible client. Defaults point at the STUB server in
|
||||||
|
# this module (com.ankurm.chatui.stub.OpenAiStubServer). To use a real provider:
|
||||||
|
# CHAT_BASE_URL=https://api.openai.com/v1 CHAT_MODEL=<model> OPENAI_API_KEY=<key>
|
||||||
|
# base-url must END IN /v1: the starter does not append it, the SDK adds only /chat/completions.
|
||||||
|
server:
|
||||||
|
port: ${SERVER_PORT:38180}
|
||||||
|
spring:
|
||||||
|
ai:
|
||||||
|
openai:
|
||||||
|
base-url: ${CHAT_BASE_URL:http://localhost:38181/v1}
|
||||||
|
api-key: ${OPENAI_API_KEY:not-needed}
|
||||||
|
# SDK default read timeout is 60 s; a slow model can exceed it.
|
||||||
|
timeout: 5m
|
||||||
|
chat:
|
||||||
|
options:
|
||||||
|
model: ${CHAT_MODEL:stub-gpt}
|
||||||
|
logging:
|
||||||
|
pattern:
|
||||||
|
console: "%msg%n"
|
||||||
|
level:
|
||||||
|
root: warn
|
||||||
|
chat: info
|
||||||
@@ -0,0 +1,29 @@
|
|||||||
|
<!doctype html>
|
||||||
|
<html lang="en">
|
||||||
|
<head>
|
||||||
|
<meta charset="utf-8">
|
||||||
|
<title>Chat UI with Spring AI, SSE and htmx</title>
|
||||||
|
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||||
|
<script src="/vendor/htmx.min.js"></script>
|
||||||
|
<script src="/vendor/htmx-ext-sse.js"></script>
|
||||||
|
<style>
|
||||||
|
body { font: 16px/1.5 system-ui, sans-serif; max-width: 42rem; margin: 2rem auto; padding: 0 1rem; }
|
||||||
|
#log { min-height: 12rem; }
|
||||||
|
.msg { padding: .5rem .75rem; margin: .5rem 0; border-radius: .5rem; white-space: pre-wrap; }
|
||||||
|
.user { background: #e8eefc; }
|
||||||
|
.bot { background: #f4f5f7; }
|
||||||
|
.stop { float: right; }
|
||||||
|
form { display: flex; gap: .5rem; }
|
||||||
|
input[name=message] { flex: 1; padding: .5rem; }
|
||||||
|
</style>
|
||||||
|
</head>
|
||||||
|
<body>
|
||||||
|
<h1>Chat</h1>
|
||||||
|
<div id="log"></div>
|
||||||
|
<form hx-post="/chat" hx-target="#log" hx-swap="beforeend" hx-on::after-request="this.reset()">
|
||||||
|
<input name="message" autocomplete="off" placeholder="Say something" required autofocus>
|
||||||
|
<button>Send</button>
|
||||||
|
</form>
|
||||||
|
<p><button hx-post="/chat/new" hx-swap="none" hx-on::after-request="document.getElementById('log').innerHTML=''">New conversation</button></p>
|
||||||
|
</body>
|
||||||
|
</html>
|
||||||
@@ -0,0 +1,2 @@
|
|||||||
|
d6fdc75f204e6bdefa99b69bf1e6d4ac69b8a364f77929f45c13476b4000f717 htmx.min.js
|
||||||
|
3b5992a541619babefc4c169505af474df5c3039da51e59b96ccf9241ecd61d2 htmx-ext-sse.js
|
||||||
@@ -0,0 +1,290 @@
|
|||||||
|
/*
|
||||||
|
Server Sent Events Extension
|
||||||
|
============================
|
||||||
|
This extension adds support for Server Sent Events to htmx. See /www/extensions/sse.md for usage instructions.
|
||||||
|
|
||||||
|
*/
|
||||||
|
|
||||||
|
(function() {
|
||||||
|
/** @type {import("../htmx").HtmxInternalApi} */
|
||||||
|
var api
|
||||||
|
|
||||||
|
htmx.defineExtension('sse', {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Init saves the provided reference to the internal HTMX API.
|
||||||
|
*
|
||||||
|
* @param {import("../htmx").HtmxInternalApi} api
|
||||||
|
* @returns void
|
||||||
|
*/
|
||||||
|
init: function(apiRef) {
|
||||||
|
// store a reference to the internal API.
|
||||||
|
api = apiRef
|
||||||
|
|
||||||
|
// set a function in the public API for creating new EventSource objects
|
||||||
|
if (htmx.createEventSource == undefined) {
|
||||||
|
htmx.createEventSource = createEventSource
|
||||||
|
}
|
||||||
|
},
|
||||||
|
|
||||||
|
getSelectors: function() {
|
||||||
|
return ['[sse-connect]', '[data-sse-connect]', '[sse-swap]', '[data-sse-swap]']
|
||||||
|
},
|
||||||
|
|
||||||
|
/**
|
||||||
|
* onEvent handles all events passed to this extension.
|
||||||
|
*
|
||||||
|
* @param {string} name
|
||||||
|
* @param {Event} evt
|
||||||
|
* @returns void
|
||||||
|
*/
|
||||||
|
onEvent: function(name, evt) {
|
||||||
|
var parent = evt.target || evt.detail.elt
|
||||||
|
switch (name) {
|
||||||
|
case 'htmx:beforeCleanupElement':
|
||||||
|
var internalData = api.getInternalData(parent)
|
||||||
|
// Try to remove remove an EventSource when elements are removed
|
||||||
|
var source = internalData.sseEventSource
|
||||||
|
if (source) {
|
||||||
|
api.triggerEvent(parent, 'htmx:sseClose', {
|
||||||
|
source,
|
||||||
|
type: 'nodeReplaced',
|
||||||
|
})
|
||||||
|
internalData.sseEventSource.close()
|
||||||
|
}
|
||||||
|
|
||||||
|
return
|
||||||
|
|
||||||
|
// Try to create EventSources when elements are processed
|
||||||
|
case 'htmx:afterProcessNode':
|
||||||
|
ensureEventSourceOnElement(parent)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
/// ////////////////////////////////////////////
|
||||||
|
// HELPER FUNCTIONS
|
||||||
|
/// ////////////////////////////////////////////
|
||||||
|
|
||||||
|
/**
|
||||||
|
* createEventSource is the default method for creating new EventSource objects.
|
||||||
|
* it is hoisted into htmx.config.createEventSource to be overridden by the user, if needed.
|
||||||
|
*
|
||||||
|
* @param {string} url
|
||||||
|
* @returns EventSource
|
||||||
|
*/
|
||||||
|
function createEventSource(url) {
|
||||||
|
return new EventSource(url, { withCredentials: true })
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* registerSSE looks for attributes that can contain sse events, right
|
||||||
|
* now hx-trigger and sse-swap and adds listeners based on these attributes too
|
||||||
|
* the closest event source
|
||||||
|
*
|
||||||
|
* @param {HTMLElement} elt
|
||||||
|
*/
|
||||||
|
function registerSSE(elt) {
|
||||||
|
// Add message handlers for every `sse-swap` attribute
|
||||||
|
if (api.getAttributeValue(elt, 'sse-swap')) {
|
||||||
|
// Find closest existing event source
|
||||||
|
var sourceElement = api.getClosestMatch(elt, hasEventSource)
|
||||||
|
if (sourceElement == null) {
|
||||||
|
// api.triggerErrorEvent(elt, "htmx:noSSESourceError")
|
||||||
|
return null // no eventsource in parentage, orphaned element
|
||||||
|
}
|
||||||
|
|
||||||
|
// Set internalData and source
|
||||||
|
var internalData = api.getInternalData(sourceElement)
|
||||||
|
var source = internalData.sseEventSource
|
||||||
|
|
||||||
|
var sseSwapAttr = api.getAttributeValue(elt, 'sse-swap')
|
||||||
|
var sseEventNames = sseSwapAttr.split(',')
|
||||||
|
|
||||||
|
for (var i = 0; i < sseEventNames.length; i++) {
|
||||||
|
const sseEventName = sseEventNames[i].trim()
|
||||||
|
const listener = function(event) {
|
||||||
|
// If the source is missing then close SSE
|
||||||
|
if (maybeCloseSSESource(sourceElement)) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// If the body no longer contains the element, remove the listener
|
||||||
|
if (!api.bodyContains(elt)) {
|
||||||
|
source.removeEventListener(sseEventName, listener)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// swap the response into the DOM and trigger a notification
|
||||||
|
if (!api.triggerEvent(elt, 'htmx:sseBeforeMessage', event)) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
swap(elt, event.data)
|
||||||
|
api.triggerEvent(elt, 'htmx:sseMessage', event)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Register the new listener
|
||||||
|
api.getInternalData(elt).sseEventListener = listener
|
||||||
|
source.addEventListener(sseEventName, listener)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Add message handlers for every `hx-trigger="sse:*"` attribute
|
||||||
|
if (api.getAttributeValue(elt, 'hx-trigger')) {
|
||||||
|
// Find closest existing event source
|
||||||
|
var sourceElement = api.getClosestMatch(elt, hasEventSource)
|
||||||
|
if (sourceElement == null) {
|
||||||
|
// api.triggerErrorEvent(elt, "htmx:noSSESourceError")
|
||||||
|
return null // no eventsource in parentage, orphaned element
|
||||||
|
}
|
||||||
|
|
||||||
|
// Set internalData and source
|
||||||
|
var internalData = api.getInternalData(sourceElement)
|
||||||
|
var source = internalData.sseEventSource
|
||||||
|
|
||||||
|
var triggerSpecs = api.getTriggerSpecs(elt)
|
||||||
|
triggerSpecs.forEach(function(ts) {
|
||||||
|
if (ts.trigger.slice(0, 4) !== 'sse:') {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
var listener = function (event) {
|
||||||
|
if (maybeCloseSSESource(sourceElement)) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if (!api.bodyContains(elt)) {
|
||||||
|
source.removeEventListener(ts.trigger.slice(4), listener)
|
||||||
|
}
|
||||||
|
// Trigger events to be handled by the rest of htmx
|
||||||
|
htmx.trigger(elt, ts.trigger, event)
|
||||||
|
htmx.trigger(elt, 'htmx:sseMessage', event)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Register the new listener
|
||||||
|
api.getInternalData(elt).sseEventListener = listener
|
||||||
|
source.addEventListener(ts.trigger.slice(4), listener)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* ensureEventSourceOnElement creates a new EventSource connection on the provided element.
|
||||||
|
* If a usable EventSource already exists, then it is returned. If not, then a new EventSource
|
||||||
|
* is created and stored in the element's internalData.
|
||||||
|
* @param {HTMLElement} elt
|
||||||
|
* @param {number} retryCount
|
||||||
|
* @returns {EventSource | null}
|
||||||
|
*/
|
||||||
|
function ensureEventSourceOnElement(elt, retryCount) {
|
||||||
|
if (elt == null) {
|
||||||
|
return null
|
||||||
|
}
|
||||||
|
|
||||||
|
// handle extension source creation attribute
|
||||||
|
if (api.getAttributeValue(elt, 'sse-connect')) {
|
||||||
|
var sseURL = api.getAttributeValue(elt, 'sse-connect')
|
||||||
|
if (sseURL == null) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
ensureEventSource(elt, sseURL, retryCount)
|
||||||
|
}
|
||||||
|
|
||||||
|
registerSSE(elt)
|
||||||
|
}
|
||||||
|
|
||||||
|
function ensureEventSource(elt, url, retryCount) {
|
||||||
|
var source = htmx.createEventSource(url)
|
||||||
|
|
||||||
|
source.onerror = function(err) {
|
||||||
|
// Log an error event
|
||||||
|
api.triggerErrorEvent(elt, 'htmx:sseError', { error: err, source })
|
||||||
|
|
||||||
|
// If parent no longer exists in the document, then clean up this EventSource
|
||||||
|
if (maybeCloseSSESource(elt)) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Otherwise, try to reconnect the EventSource
|
||||||
|
if (source.readyState === EventSource.CLOSED) {
|
||||||
|
retryCount = retryCount || 0
|
||||||
|
retryCount = Math.max(Math.min(retryCount * 2, 128), 1)
|
||||||
|
var timeout = retryCount * 500
|
||||||
|
window.setTimeout(function() {
|
||||||
|
ensureEventSourceOnElement(elt, retryCount)
|
||||||
|
}, timeout)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
source.onopen = function(evt) {
|
||||||
|
api.triggerEvent(elt, 'htmx:sseOpen', { source })
|
||||||
|
|
||||||
|
if (retryCount && retryCount > 0) {
|
||||||
|
const childrenToFix = elt.querySelectorAll("[sse-swap], [data-sse-swap], [hx-trigger], [data-hx-trigger]")
|
||||||
|
for (let i = 0; i < childrenToFix.length; i++) {
|
||||||
|
registerSSE(childrenToFix[i])
|
||||||
|
}
|
||||||
|
// We want to increase the reconnection delay for consecutive failed attempts only
|
||||||
|
retryCount = 0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
api.getInternalData(elt).sseEventSource = source
|
||||||
|
|
||||||
|
|
||||||
|
var closeAttribute = api.getAttributeValue(elt, "sse-close");
|
||||||
|
if (closeAttribute) {
|
||||||
|
// close eventsource when this message is received
|
||||||
|
source.addEventListener(closeAttribute, function() {
|
||||||
|
api.triggerEvent(elt, 'htmx:sseClose', {
|
||||||
|
source,
|
||||||
|
type: 'message',
|
||||||
|
})
|
||||||
|
source.close()
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* maybeCloseSSESource confirms that the parent element still exists.
|
||||||
|
* If not, then any associated SSE source is closed and the function returns true.
|
||||||
|
*
|
||||||
|
* @param {HTMLElement} elt
|
||||||
|
* @returns boolean
|
||||||
|
*/
|
||||||
|
function maybeCloseSSESource(elt) {
|
||||||
|
if (!api.bodyContains(elt)) {
|
||||||
|
var source = api.getInternalData(elt).sseEventSource
|
||||||
|
if (source != undefined) {
|
||||||
|
api.triggerEvent(elt, 'htmx:sseClose', {
|
||||||
|
source,
|
||||||
|
type: 'nodeMissing',
|
||||||
|
})
|
||||||
|
source.close()
|
||||||
|
// source = null
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @param {HTMLElement} elt
|
||||||
|
* @param {string} content
|
||||||
|
*/
|
||||||
|
function swap(elt, content) {
|
||||||
|
api.withExtensions(elt, function(extension) {
|
||||||
|
content = extension.transformResponse(content, null, elt)
|
||||||
|
})
|
||||||
|
|
||||||
|
var swapSpec = api.getSwapSpecification(elt)
|
||||||
|
var target = api.getTarget(elt)
|
||||||
|
api.swap(target, content, swapSpec, { contextElement: elt })
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
function hasEventSource(node) {
|
||||||
|
return api.getInternalData(node).sseEventSource != null
|
||||||
|
}
|
||||||
|
})()
|
||||||
File diff suppressed because one or more lines are too long
@@ -0,0 +1,219 @@
|
|||||||
|
package com.ankurm.chatui;
|
||||||
|
|
||||||
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
|
|
||||||
|
import com.ankurm.chatui.stub.OpenAiStubServer;
|
||||||
|
import java.io.BufferedReader;
|
||||||
|
import java.io.InputStreamReader;
|
||||||
|
import java.net.CookieManager;
|
||||||
|
import java.net.URI;
|
||||||
|
import java.net.URLEncoder;
|
||||||
|
import java.net.http.HttpClient;
|
||||||
|
import java.net.http.HttpRequest;
|
||||||
|
import java.net.http.HttpResponse;
|
||||||
|
import java.nio.charset.StandardCharsets;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
|
import org.junit.jupiter.api.BeforeAll;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.beans.factory.annotation.Value;
|
||||||
|
import org.springframework.boot.test.context.SpringBootTest;
|
||||||
|
import org.springframework.test.context.DynamicPropertyRegistry;
|
||||||
|
import org.springframework.test.context.DynamicPropertySource;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The whole chain on a random port: HTTP -> Spring MVC -> ChatClient + memory -> OpenAiChatModel ->
|
||||||
|
* STUB server. Transcripts land in output/01-sse-transcript.txt, 02-memory.txt, 03-escaping.txt and
|
||||||
|
* 04-cancel-with-fix.txt.
|
||||||
|
*/
|
||||||
|
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
|
||||||
|
class ChatFlowTest {
|
||||||
|
|
||||||
|
@BeforeAll
|
||||||
|
static void stub() {
|
||||||
|
StubSupport.ensureStarted();
|
||||||
|
}
|
||||||
|
|
||||||
|
@DynamicPropertySource
|
||||||
|
static void props(DynamicPropertyRegistry r) {
|
||||||
|
r.add("spring.ai.openai.base-url", () -> "http://localhost:" + StubSupport.PORT + "/v1");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Value("${local.server.port}")
|
||||||
|
int port;
|
||||||
|
@Autowired
|
||||||
|
ChatController controller;
|
||||||
|
|
||||||
|
/** One SSE event as a browser's EventSource would see it. */
|
||||||
|
record Event(String name, String data) {
|
||||||
|
}
|
||||||
|
|
||||||
|
HttpClient browser() {
|
||||||
|
return HttpClient.newBuilder().cookieHandler(new CookieManager()).build();
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Reads the raw stream and returns (raw lines, events parsed per the WHATWG rules). */
|
||||||
|
List<String> raw = new ArrayList<>();
|
||||||
|
|
||||||
|
List<Event> stream(HttpClient client, String message) throws Exception {
|
||||||
|
raw = new ArrayList<>();
|
||||||
|
var req = HttpRequest.newBuilder(URI.create("http://localhost:" + port + "/chat/stream?turn=1&message="
|
||||||
|
+ URLEncoder.encode(message, StandardCharsets.UTF_8))).header("Accept", "text/event-stream").build();
|
||||||
|
var res = client.send(req, HttpResponse.BodyHandlers.ofInputStream());
|
||||||
|
List<Event> events = new ArrayList<>();
|
||||||
|
String name = "message";
|
||||||
|
StringBuilder data = new StringBuilder();
|
||||||
|
boolean has = false;
|
||||||
|
try (var in = new BufferedReader(new InputStreamReader(res.body(), StandardCharsets.UTF_8))) {
|
||||||
|
String line;
|
||||||
|
while ((line = in.readLine()) != null) {
|
||||||
|
raw.add(line);
|
||||||
|
if (line.isEmpty()) {
|
||||||
|
if (has) {
|
||||||
|
events.add(new Event(name, data.toString()));
|
||||||
|
}
|
||||||
|
name = "message";
|
||||||
|
data.setLength(0);
|
||||||
|
has = false;
|
||||||
|
} else if (line.startsWith("event:")) {
|
||||||
|
name = strip(line.substring(6));
|
||||||
|
} else if (line.startsWith("data:")) {
|
||||||
|
if (has) {
|
||||||
|
data.append('\n');
|
||||||
|
}
|
||||||
|
data.append(strip(line.substring(5)));
|
||||||
|
has = true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return events;
|
||||||
|
}
|
||||||
|
|
||||||
|
/** The SSE spec: remove ONE leading space from the field value. */
|
||||||
|
static String strip(String v) {
|
||||||
|
return v.startsWith(" ") ? v.substring(1) : v;
|
||||||
|
}
|
||||||
|
|
||||||
|
static String text(List<Event> events) {
|
||||||
|
StringBuilder sb = new StringBuilder();
|
||||||
|
events.stream().filter(e -> e.name().equals("token")).forEach(e -> sb.append(e.data()));
|
||||||
|
return sb.toString();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void sseFramesAndTheLeadingSpaceRule() throws Exception {
|
||||||
|
var t = new Transcript("01-sse-transcript.txt");
|
||||||
|
var events = stream(browser(), "hello world");
|
||||||
|
t.line("$ GET /chat/stream?message=hello world (raw lines, first 14 and last 5)");
|
||||||
|
raw.stream().limit(14).forEach(t::line);
|
||||||
|
t.line("...");
|
||||||
|
raw.subList(raw.size() - 5, raw.size()).forEach(l -> t.line(l.isEmpty() ? "<blank>" : l));
|
||||||
|
String joined = text(events);
|
||||||
|
t.line("event names seen: " + events.stream().map(Event::name).distinct().toList());
|
||||||
|
t.line("joined token text as EventSource delivers it: " + joined);
|
||||||
|
// What the browser would show if the server had NOT put a space in front of every data line:
|
||||||
|
String naive = events.stream().filter(e -> e.name().equals("token")).map(e -> strip(e.data()))
|
||||||
|
.reduce("", String::concat);
|
||||||
|
t.line("same text if the server did not pad: " + naive);
|
||||||
|
assertThat(joined).isEqualTo("[stub] saw 2 messages. Earlier user turns: none. You said: hello world");
|
||||||
|
assertThat(naive).doesNotContain(" ");
|
||||||
|
assertThat(events.getLast().name()).isEqualTo("done");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void memoryFollowsTheHttpSession() throws Exception {
|
||||||
|
var t = new Transcript("02-memory.txt");
|
||||||
|
var a = browser();
|
||||||
|
t.line("session A, turn 1: " + text(stream(a, "My name is Asha and I like tea")));
|
||||||
|
t.line("session A, turn 2: " + text(stream(a, "What did I just tell you?")));
|
||||||
|
var b = browser();
|
||||||
|
String other = text(stream(b, "What did I just tell you?"));
|
||||||
|
t.line("session B, turn 1: " + other);
|
||||||
|
String third = text(stream(a, "And now?"));
|
||||||
|
t.line("session A, turn 3: " + third);
|
||||||
|
assertThat(third).contains("saw 6 messages").contains("My name is Asha and I like tea | What did I just tell you?");
|
||||||
|
assertThat(other).contains("saw 2 messages").contains("Earlier user turns: none");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void modelTextIsEscapedBeforeItReachesInnerHtml() throws Exception {
|
||||||
|
var t = new Transcript("03-escaping.txt");
|
||||||
|
var events = stream(browser(), "html please");
|
||||||
|
String joined = text(events);
|
||||||
|
t.line("data sent to the browser: " + joined);
|
||||||
|
assertThat(joined).contains("<b>bold</b>").contains("<img src=x onerror=alert(1)>");
|
||||||
|
assertThat(joined).doesNotContain("<b>").doesNotContain("<img");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void closingTheConnectionCancelsTheFluxAndTheUpstreamStream() throws Exception {
|
||||||
|
var t = new Transcript("04-cancel-with-fix.txt");
|
||||||
|
OpenAiStubServer.setDelayMs(50);
|
||||||
|
long[] before = OpenAiStubServer.counters();
|
||||||
|
int cancelledBefore = controller.cancelled.get();
|
||||||
|
try (var socket = new java.net.Socket("localhost", port)) {
|
||||||
|
socket.getOutputStream().write(("GET /chat/stream?turn=9&message=big%20400 HTTP/1.1\r\nHost: localhost\r\n\r\n")
|
||||||
|
.getBytes(StandardCharsets.US_ASCII));
|
||||||
|
socket.getInputStream().readNBytes(40_000);
|
||||||
|
} // closing the socket is the "client abort"
|
||||||
|
long t0 = System.currentTimeMillis();
|
||||||
|
long[] after = before;
|
||||||
|
for (int i = 0; i < 100 && after[3] == before[3]; i++) {
|
||||||
|
Thread.sleep(50);
|
||||||
|
after = OpenAiStubServer.counters();
|
||||||
|
}
|
||||||
|
long took = System.currentTimeMillis() - t0;
|
||||||
|
OpenAiStubServer.setDelayMs(20);
|
||||||
|
t.line("client read 40000 bytes then closed the socket");
|
||||||
|
t.line("Flux cancelled on the server: " + (controller.cancelled.get() - cancelledBefore));
|
||||||
|
t.line("stub saw the upstream connection close: " + (after[3] - before[3]) + " (within " + took + " ms)");
|
||||||
|
t.line("stub chunks written for this request: " + (after[1] - before[1]) + " of 400 requested");
|
||||||
|
assertThat(controller.cancelled.get() - cancelledBefore).isEqualTo(1);
|
||||||
|
assertThat(after[3] - before[3]).isEqualTo(1);
|
||||||
|
assertThat(after[1] - before[1]).isLessThan(400);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
org.springframework.ai.chat.client.ChatClient chatClient;
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void aMemoryAdvisorWithoutAConversationIdFails() {
|
||||||
|
var t = new Transcript("11-missing-conversation-id.txt");
|
||||||
|
try {
|
||||||
|
chatClient.prompt().user("hi").stream().content().blockLast();
|
||||||
|
t.line("no exception");
|
||||||
|
} catch (RuntimeException e) {
|
||||||
|
t.line("stream failed: " + e.getClass().getSimpleName() + ": " + e.getMessage());
|
||||||
|
assertThat(e.getMessage()).contains("conversationId cannot be null");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void whatMemoryKeepsWhenTheClientCancelsMidAnswer() throws Exception {
|
||||||
|
var t = new Transcript("12-memory-after-cancel.txt");
|
||||||
|
OpenAiStubServer.setDelayMs(30);
|
||||||
|
var client = browser();
|
||||||
|
// turn 1: ask for 300 tokens over a raw socket that shares the session cookie, read a little, hang up
|
||||||
|
var warm = HttpRequest.newBuilder(URI.create("http://localhost:" + port + "/chat/stream?message=warm")).build();
|
||||||
|
client.send(warm, HttpResponse.BodyHandlers.discarding()); // sets JSESSIONID in the client's cookie jar
|
||||||
|
String cookie = client.cookieHandler().map(h -> {
|
||||||
|
try {
|
||||||
|
return h.get(URI.create("http://localhost:" + port + "/"), java.util.Map.of()).get("Cookie").getFirst();
|
||||||
|
} catch (Exception e) {
|
||||||
|
throw new IllegalStateException(e);
|
||||||
|
}
|
||||||
|
}).orElseThrow();
|
||||||
|
try (var socket = new java.net.Socket("localhost", port)) {
|
||||||
|
socket.getOutputStream().write(("GET /chat/stream?message=count%20300 HTTP/1.1\r\nHost: localhost\r\nCookie: "
|
||||||
|
+ cookie + "\r\n\r\n").getBytes(StandardCharsets.US_ASCII));
|
||||||
|
socket.getInputStream().readNBytes(600);
|
||||||
|
}
|
||||||
|
Thread.sleep(1500);
|
||||||
|
String next = text(stream(client, "What do you remember?"));
|
||||||
|
OpenAiStubServer.setDelayMs(20);
|
||||||
|
t.line("turn 1 'warm' completed, turn 2 'count 300' cancelled after a few tokens, then turn 3:");
|
||||||
|
t.line(next);
|
||||||
|
assertThat(next).contains("Earlier user turns: warm | count 300");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,77 @@
|
|||||||
|
package com.ankurm.chatui;
|
||||||
|
|
||||||
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
|
|
||||||
|
import com.ankurm.chatui.stub.OpenAiStubServer;
|
||||||
|
import okhttp3.MediaType;
|
||||||
|
import okhttp3.OkHttpClient;
|
||||||
|
import okhttp3.Request;
|
||||||
|
import okhttp3.RequestBody;
|
||||||
|
import org.junit.jupiter.api.BeforeAll;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* No Spring AI, no SDK: plain OkHttp 4.12 against the stub. A reader thread is blocked reading the
|
||||||
|
* SSE body; another thread then closes the response, or cancels the call. Which one closes the socket?
|
||||||
|
*/
|
||||||
|
class OkHttpCloseReproTest {
|
||||||
|
|
||||||
|
@BeforeAll
|
||||||
|
static void stub() {
|
||||||
|
StubSupport.ensureStarted();
|
||||||
|
}
|
||||||
|
|
||||||
|
int run(boolean cancelInsteadOfClose, Transcript t) throws Exception {
|
||||||
|
var http = new OkHttpClient();
|
||||||
|
int closed = 0;
|
||||||
|
for (int i = 0; i < 10; i++) {
|
||||||
|
long[] before = OpenAiStubServer.counters();
|
||||||
|
var req = new Request.Builder().url("http://localhost:" + StubSupport.PORT + "/v1/chat/completions")
|
||||||
|
.post(RequestBody.create("{\"model\":\"m\",\"stream\":true,\"messages\":[{\"role\":\"user\","
|
||||||
|
+ "\"content\":\"count 60\"}]}", MediaType.get("application/json"))).build();
|
||||||
|
var call = http.newCall(req);
|
||||||
|
var response = call.execute();
|
||||||
|
var source = response.body().source();
|
||||||
|
var reader = new Thread(() -> {
|
||||||
|
try {
|
||||||
|
while (source.readUtf8Line() != null) {
|
||||||
|
// keep reading
|
||||||
|
}
|
||||||
|
} catch (Exception ignored) {
|
||||||
|
// closed under us
|
||||||
|
}
|
||||||
|
});
|
||||||
|
reader.start();
|
||||||
|
Thread.sleep(150);
|
||||||
|
try {
|
||||||
|
if (cancelInsteadOfClose) {
|
||||||
|
call.cancel();
|
||||||
|
} else {
|
||||||
|
response.close();
|
||||||
|
}
|
||||||
|
} catch (RuntimeException e) {
|
||||||
|
if (i == 0) {
|
||||||
|
t.line(" response.close() threw: " + e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Thread.sleep(500);
|
||||||
|
if (OpenAiStubServer.counters()[3] - before[3] == 1) {
|
||||||
|
closed++;
|
||||||
|
}
|
||||||
|
reader.join(2000);
|
||||||
|
}
|
||||||
|
return closed;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void closeFromAnotherThreadLeaksCancelDoesNot() throws Exception {
|
||||||
|
var t = new Transcript("05-okhttp-close-vs-cancel.txt");
|
||||||
|
t.line("plain OkHttp 4.12.0, 10 runs each, stub streaming 60 chunks at 20 ms, reader thread blocked in a read");
|
||||||
|
int viaClose = run(false, t);
|
||||||
|
t.line("response.close() from another thread: upstream connection closed in " + viaClose + " of 10 runs");
|
||||||
|
int viaCancel = run(true, t);
|
||||||
|
t.line("call.cancel() from another thread: upstream connection closed in " + viaCancel + " of 10 runs");
|
||||||
|
assertThat(viaCancel).isEqualTo(10);
|
||||||
|
assertThat(viaClose).isLessThanOrEqualTo(2);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,24 @@
|
|||||||
|
package com.ankurm.chatui;
|
||||||
|
|
||||||
|
import com.ankurm.chatui.stub.OpenAiStubServer;
|
||||||
|
import java.io.IOException;
|
||||||
|
|
||||||
|
/** Starts the in-process stub once per JVM on port 38193. */
|
||||||
|
final class StubSupport {
|
||||||
|
static final int PORT = 38193;
|
||||||
|
private static boolean started;
|
||||||
|
|
||||||
|
static synchronized void ensureStarted() {
|
||||||
|
if (!started) {
|
||||||
|
try {
|
||||||
|
OpenAiStubServer.start(PORT, 20);
|
||||||
|
} catch (IOException e) {
|
||||||
|
throw new IllegalStateException(e);
|
||||||
|
}
|
||||||
|
started = true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private StubSupport() {
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,30 @@
|
|||||||
|
package com.ankurm.chatui;
|
||||||
|
|
||||||
|
import java.io.IOException;
|
||||||
|
import java.nio.file.Files;
|
||||||
|
import java.nio.file.Path;
|
||||||
|
import java.nio.file.StandardOpenOption;
|
||||||
|
|
||||||
|
/** Appends lines to output/NN-name.txt (the file is truncated by the first call of a JVM run). */
|
||||||
|
final class Transcript {
|
||||||
|
private final Path file;
|
||||||
|
|
||||||
|
Transcript(String name) {
|
||||||
|
this.file = Path.of("output", name);
|
||||||
|
try {
|
||||||
|
Files.createDirectories(file.getParent());
|
||||||
|
Files.writeString(file, "");
|
||||||
|
} catch (IOException e) {
|
||||||
|
throw new IllegalStateException(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
void line(String s) {
|
||||||
|
System.out.println(s);
|
||||||
|
try {
|
||||||
|
Files.writeString(file, s + "\n", StandardOpenOption.APPEND);
|
||||||
|
} catch (IOException e) {
|
||||||
|
throw new IllegalStateException(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user