Section 12 — Streaming

Begin with the problem

Showing an answer while it is produced

A long model response feels slow if the user sees nothing until the end. Streaming delivers small pieces as they arrive, but the application must handle partial data, cancellation, and errors.

provider chunks → Flux → server response → browser

What you will learn

• Explain streaming versus one complete response.
• Use Flux without blocking the reactive path.
• Handle cancellation and partial chunks.
• Measure first-token and total latency separately.

Current official reference: Spring AI documentation for this topic. The examples below primarily preserve the stated 1.1.x course target. Where Spring AI 2.0 differs, the text must treat that behavior as version-specific rather than universal.

(Continues from Section 11. Target: Spring AI 1.1.x / Spring Boot 3.5.x.)

12.1 Why Streaming Is a Reactive-Stack Concern, Not Just an API Choice

Beginner primer: see the glossary’s streaming section if this is new — streaming means the model’s response is delivered incrementally, chunk by chunk, as it’s generated, rather than all at once after the full response is ready. If you haven’t used Project Reactor’s Flux/Mono types before, a Flux<T> is Reactor’s representation of a stream of zero-or-more asynchronous values over time — the reactive equivalent of an Iterable<T>, but pushed to you incrementally rather than pulled all at once.

.stream() on ChatClient returns Flux<T> — a real, backpressure-aware Publisher — and Spring AI’s provider implementations translate provider-specific chunked/SSE wire formats into that Flux correctly, including the parts that are easy to get wrong by hand: partial-chunk buffering, tool-call argument reassembly across multiple deltas, and proper completion/error signaling.

Real-world analogy — Live News Ticker vs. Printed Newspaper: .call() is the printed newspaper — wait for the whole thing to be typeset and printed before you get anything. .stream() is the live news ticker — words appear as they’re written, and critically, the ticker operator (the reactive pipeline) controls the pace so the display doesn’t get overwhelmed if it can’t keep up (backpressure) — versus a firehose that just dumps everything regardless of whether the reader is ready.

Analogy: The Live News Ticker vs. The Printed Newspaper Think of streaming options in terms of how information is read:

• The Printed Newspaper (.call()): You wait until the entire paper is compiled, printed, packed, and delivered to your doorstep. If the reporter writes a 10-page article, you sit in silence for hours waiting, receiving 0 feedback until the final package arrives.
• The News Ticker (.stream()): Text is projected onto the wall ticker symbol-by-symbol as the reporter types it in real time. You get instant updates, and the ticker speed adapts to how fast your eyes can blink (Backpressure).
• The Switchboard Buffer (Spring AI): If a sentence contains structural punctuation or a partial JSON tool request, Spring AI’s streaming assistant buffers those fragments in a temporary cache so you don’t read broken garbage.

📊 Visual Chart: Streaming Delta Chunk Reassembly Timeline

Here is how partial token payloads and fragmented tool arguments are merged before emission:

sequenceDiagram
    autonumber
    participant Provider as Remote Model Endpoint (SSE)
    participant Spring as Spring AI Helper (Stateful Buffer)
    participant App as Subscriber client (Flux)

    Provider->>Spring: Chunk 1: "data: {'delta': {'content': 'The CAP'}}"
    Spring->>App: Emit: "The CAP"

    Provider->>Spring: Chunk 2: "data: {'delta': {'content': ' theorem'}}"
    Spring->>App: Emit: " theorem"

    rect rgb(240, 255, 240)
        Note over Spring: Tool calling request arrives split
        Provider->>Spring: Chunk 3: "data: {'delta': {'tool_calls': [{'args': '{\x22city\x22:'}]}}"
        Note over Spring: State Buffer holds: {"city":
        Provider->>Spring: Chunk 4: "data: {'delta': {'tool_calls': [{'args': '\x22Bengaluru\x22}'}]}}"
        Note over Spring: Reassembles complete JSON arguments:<br>{"city": "Bengaluru"}
    end

    Spring->>App: Emit fully assembled ToolCall event
    Provider->>Spring: Signal [STREAM COMPLETE]
    Spring->>App: Emit: onComplete()

---

12.2 Token Streaming — The Response Shape

Flux<String> tokenStream = chatClient.prompt()
        .user("Write a detailed explanation of the CAP theorem")
        .stream()
        .content();

tokenStream.subscribe(
        token -> sendToClient(token),      // onNext
        error -> handleStreamError(error), // onError
        () -> closeConnection()            // onComplete
);

For SSE delivery to a browser client in a WebFlux controller:

@RestController
public class ChatStreamController {

    private final ChatClient chatClient;

    @GetMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> streamChat(@RequestParam String message) {
        return chatClient.prompt()
                .user(message)
                .stream()
                .content()
                .doOnError(e -> log.error("Stream error", e))
                .onErrorResume(e -> Flux.just("[Error: unable to complete response]"));
    }
}

Spring WebFlux automatically wraps a Flux<String> return type with produces = TEXT_EVENT_STREAM_VALUE into proper SSE framing (data: ...\n\n per event) — you don’t hand-construct SSE protocol formatting yourself.

---

12.3 Internal Mechanism — Provider SSE to Flux<ChatResponse>

OpenAiChatModel.stream(Prompt prompt)
        │
        ▼
1. HTTP request sent with "stream": true, response consumed as
   Server-Sent Events over the WebClient reactive pipeline (NOT
   RestClient — streaming specifically requires the reactive
   WebClient underneath, even if your blocking .call() path for
   the SAME ChatModel instance uses RestClient)
        │
        ▼
2. Each SSE "data:" line is a partial JSON chunk (a "delta") —
   OpenAI's format: {"choices":[{"delta":{"content":"Hello"}}]}
   repeated per token/token-group
        │
        ▼
3. OpenAiStreamFunctionCallingHelper (or equivalent per-provider
   class) accumulates PARTIAL TOOL CALL ARGUMENTS across multiple
   chunks — a tool call's JSON arguments frequently arrive split
   across several deltas character-by-character or in small
   fragments; this reassembly is real, non-trivial state machine
   logic Spring AI handles for you, because naively parsing each
   chunk's tool-call fragment independently would fail
        │
        ▼
4. Each accumulated delta mapped to a ChatResponse, emitted as one
   element of the Flux<ChatResponse>
        │
        ▼
5. Advisor chain's StreamAroundAdvisor wraps this Flux — e.g.,
   MessageChatMemoryAdvisor's stream variant must buffer the FULL
   streamed response internally to persist it as one complete
   AssistantMessage to memory once the stream completes, even
   though it's passing tokens through to the subscriber incrementally
        │
        ▼
6. Flux<String> (via .content()) or Flux<ChatResponse> (full metadata)
   delivered to your subscriber

The tool-call-argument-reassembly detail in step 3 is worth remembering specifically: it’s the kind of correctness-critical, easy-to-get-subtly-wrong logic that’s a genuine reason to use Spring AI’s streaming abstraction rather than hand-rolling SSE parsing against a provider’s raw API — a naive implementation that JSON-parses each individual delta chunk will simply fail on tool-calling responses, since most individual chunks aren’t valid JSON on their own.

---

12.4 Blocking vs. Streaming vs. Reactive — Precise Distinctions

Mode	Return type	Thread behavior	When
Blocking	String/T via .call()	Blocks calling thread until full response received	MVC controllers, batch jobs, CLI tools, background workers
Streaming (non-reactive consumption)	Flux<String>, .subscribe()’d and manually bridged (e.g., to a SseEmitter in an MVC app)	Non-blocking on the Flux subscription itself, but MVC controller threads still tie up a request-handling thread for the connection duration	MVC apps wanting token-by-token UX without full WebFlux migration — a valid, common hybrid pattern
Reactive	Flux<String> returned directly from a WebFlux @RestController method	Fully non-blocking end-to-end, event-loop threads freed between emissions	High-concurrency chat UIs, WebFlux-native applications

A very common production pattern: an otherwise-MVC (Spring Web MVC, not WebFlux) application still wants token streaming to the browser. This works via SseEmitter, bridging the reactive Flux to a servlet-based emitter:

@GetMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter streamChat(@RequestParam String message) {
    SseEmitter emitter = new SseEmitter(0L); // no timeout; manage lifecycle explicitly

    chatClient.prompt().user(message).stream().content()
            .subscribe(
                    token -> {
                        try {
                            emitter.send(token);
                        } catch (IOException e) {
                            emitter.completeWithError(e);
                        }
                    },
                    emitter::completeWithError,
                    emitter::complete
            );

    return emitter;
}

This is a legitimate, common production bridge — you get streaming UX without a full-application migration to WebFlux, at the cost of the MVC request-handling thread being tied up for the connection’s duration (still meaningfully better than a fully blocking .call(), since the thread isn’t blocked waiting on the model, just holding the HTTP connection open while the reactive pipeline pushes data through).

---

12.5 Cancellation

Disposable subscription = chatClient.prompt()
        .user(longRunningQuery)
        .stream()
        .content()
        .subscribe(this::sendToken);

// user navigates away / closes connection / explicit cancel button:
subscription.dispose();

dispose() propagates a cancellation signal up through the reactive pipeline — critically, this should also cancel the underlying HTTP connection to the model provider, not just stop delivering tokens to a now-absent subscriber, since letting the provider-side generation continue after the client disconnected wastes real API cost for tokens nobody will ever see. WebClient-backed streaming implementations propagate cancellation to the underlying HTTP exchange correctly by default; verify this holds for any custom ChatModel wrapper you build (Section 4’s round-robin/routing wrappers, for instance) — a wrapper that doesn’t correctly propagate Flux cancellation semantics can silently leak this behavior and keep burning provider tokens after a user has left.

In a WebFlux controller, cancellation on client disconnect is handled automatically — Spring’s reactive web stack detects the disconnected client and cancels the returned Flux’s subscription for you.

---

12.6 Backpressure

Beginner note: backpressure is the mechanism by which a slow consumer can tell a fast producer to slow down, rather than being overwhelmed with more data than it can handle — a core concept in reactive programming.

Token generation rate from the model is typically the bottleneck (not consumption rate), so in most chat-UI scenarios backpressure is rarely the binding constraint — but it becomes relevant when the consumer is slow: writing to a slow downstream sink (a database, a slow client connection), or when many concurrent streams compete for limited resources.

chatClient.prompt().user(query).stream().content()
        .onBackpressureBuffer(1000, dropped -> log.warn("Dropped token due to backpressure"))
        .subscribe(this::processToken);

For high-concurrency scenarios (many simultaneous streaming chat sessions), the practical backpressure concern usually isn’t the Flux mechanics themselves (Reactor handles this correctly by default) but connection/resource exhaustion at scale — bound your WebClient’s connection pool explicitly rather than relying on defaults, since many concurrent long-lived SSE streams held open simultaneously is a really different resource profile than typical short-lived HTTP request/response traffic:

@Bean
public WebClient.Builder streamingWebClientBuilder() {
    ConnectionProvider provider = ConnectionProvider.builder("streaming-pool")
            .maxConnections(500)
            .maxIdleTime(Duration.ofSeconds(30))
            .build();
    HttpClient httpClient = HttpClient.create(provider);
    return WebClient.builder().clientConnector(new ReactorClientHttpConnector(httpClient));
}

---

12.7 Monitoring Streaming Calls

Streaming complicates observability versus blocking calls: a single logical “request” now spans an open-ended duration with many emitted events, and standard request-duration metrics need adjustment — measure time-to-first-token (latency-critical for perceived responsiveness) separately from total stream duration (throughput-critical for cost/capacity planning):

AtomicLong firstTokenTime = new AtomicLong(-1);
long startTime = System.nanoTime();

chatClient.prompt().user(query).stream().content()
        .doOnNext(token -> {
            if (firstTokenTime.compareAndSet(-1, System.nanoTime())) {
                long ttft = (firstTokenTime.get() - startTime) / 1_000_000;
                meterRegistry.timer("chat.stream.time_to_first_token").record(ttft, TimeUnit.MILLISECONDS);
            }
        })
        .doOnComplete(() -> {
            long totalMs = (System.nanoTime() - startTime) / 1_000_000;
            meterRegistry.timer("chat.stream.total_duration").record(totalMs, TimeUnit.MILLISECONDS);
        })
        .subscribe(this::sendToken);

Section 13 covers the full Micrometer/ObservationRegistry integration; the point here is specifically that streaming needs distinct metrics from blocking calls, not a reuse of the same “request duration” histogram — conflating them produces misleading dashboards (a fast time-to-first-token with a long total duration looks identical to a slow response in an aggregate-duration-only view, but represents very different user experience and cost profiles).

---

12.8 Common Mistakes

1. Hand-rolling SSE parsing against a provider’s raw API instead of using Spring AI’s Flux<ChatResponse> — you’ll likely mishandle partial tool-call argument reassembly (§12.3).
2. Not propagating cancellation to the underlying HTTP connection on client disconnect in custom ChatModel wrappers, wasting provider tokens/cost on abandoned streams.
3. Using unbounded SseEmitter timeouts without an explicit lifecycle/cleanup strategy in MVC-bridge patterns, risking connection leaks under client-disconnect edge cases.
4. Measuring only aggregate request duration for streaming calls, missing the time-to-first-token signal that actually drives perceived UX quality.
5. Not bounding the WebClient connection pool for high-concurrency streaming scenarios, hitting resource exhaustion under load that wouldn’t show up in low-concurrency testing.
6. Using .call() inside a WebFlux reactive chain — reintroduced from Section 3, worth repeating here specifically because it defeats the entire purpose of choosing a reactive stack.

---

12.9 Debugging

logging:
  level:
    org.springframework.web.reactive.function.client: DEBUG # raw WebClient SSE traffic
    org.springframework.ai.chat.client: DEBUG

For “streaming works locally but hangs/times out in production,” check load-balancer/proxy SSE support explicitly — some infrastructure (certain proxy configurations, older load balancer settings) buffers responses by default, defeating streaming’s incremental-delivery benefit entirely even though the application code is correct; this is an infrastructure configuration issue (disable response buffering for the streaming route, verify Connection: keep-alive and proxy timeout settings accommodate long-lived SSE connections) far more often than a Spring AI bug.

---

12.10 Interview Questions

1. Why does .stream() require WebClient internally even for a ChatModel whose blocking .call() path uses RestClient?
2. Explain the tool-call-argument-reassembly problem in streaming responses and why naively parsing each SSE chunk independently fails.
3. What’s the difference in thread-blocking behavior between a WebFlux-native streaming endpoint and an MVC-bridged SseEmitter pattern?
4. Why should dispose()-triggered cancellation propagate to the underlying provider HTTP connection, and what’s the cost implication if it doesn’t?
5. Why is time-to-first-token a distinct, important metric from total stream duration, and what UX/cost signals does each represent?
6. Under what circumstances does backpressure actually become a binding constraint in a typical LLM streaming scenario, given token generation is usually the bottleneck?
7. What infrastructure-level issue commonly causes “streaming works locally but not in production,” and how would you diagnose it?
8. How does MessageChatMemoryAdvisor’s stream variant handle persisting a complete conversation turn to memory while still delivering tokens incrementally to the subscriber?
9. What’s the risk of not bounding a WebClient connection pool for high-concurrency streaming scenarios specifically, versus typical request/response traffic?
10. Describe how you’d bridge a WebFlux Flux<String> streaming response into an MVC application without a full reactive-stack migration.
11. Why is hand-rolling SSE parsing against a provider’s raw streaming API risky compared to using Spring AI’s abstraction?
12. What does SseEmitter timeout of 0L mean, and why might a production system need explicit lifecycle management despite that setting?
13. How would you instrument time-to-first-token using Micrometer for a streaming chat endpoint?
14. What’s the correct behavior when a client disconnects mid-stream in a WebFlux controller, and does Spring handle this automatically?
15. Why does conflating streaming and blocking request-duration metrics into one histogram produce misleading dashboards?
16. What reactive operator would you use to handle a slow downstream consumer without unbounded memory growth in a token stream?
17. Explain why .call() inside a reactive WebFlux chain defeats the purpose of the reactive stack, concretely in terms of thread pool behavior.
18. How would you test that your streaming pipeline correctly reassembles a tool call whose arguments arrive split across multiple SSE chunks?
19. What load-balancer/proxy configuration issue commonly defeats SSE streaming even when application code is correct?
20. Why is Flux<ChatResponse> (versus Flux<String>) sometimes the better choice for a streaming consumer, and what extra information does it carry per-emission?

---

12.11 Best Practices Checklist

• Never hand-roll SSE parsing against a provider’s raw API — use Spring AI’s Flux<ChatResponse>/Flux<String>.
• Verify custom ChatModel wrappers correctly propagate stream cancellation to the underlying HTTP connection.
• Measure time-to-first-token and total stream duration as distinct metrics.
• Bound WebClient connection pools explicitly for high-concurrency streaming workloads.
• Verify infrastructure (load balancers, proxies) doesn’t buffer SSE responses before shipping a streaming feature to production.
• Use the SseEmitter bridge pattern deliberately when full WebFlux migration isn’t justified, understanding its thread-tie-up trade-off versus true reactive endpoints.

---

12.12 Key Takeaways

• Streaming is a real reactive-stack integration, not a thin wrapper — Spring AI handles really tricky correctness details like partial tool-call reassembly that are easy to get subtly wrong hand-rolled.
• Blocking, MVC-bridged streaming, and true reactive streaming are three distinct operational profiles with different thread-blocking and resource characteristics — choose deliberately based on your application’s actual stack.
• Cancellation should propagate all the way to the provider connection to avoid wasting cost on abandoned streams.
• Time-to-first-token and total duration are separate, both-necessary observability signals for streaming — don’t collapse them into one metric.
• Production streaming failures are disproportionately likely to be infrastructure (proxy buffering, connection pool exhaustion) rather than application code issues.

---

End of Section 12. Next: Section 13 — Observability (Micrometer, OpenTelemetry, Tracing, Metrics, Prometheus, Grafana, Logging, Correlation IDs).