Streaming & Reactive

This content was generated with the assistance of AI and should be verified against the official documentation of each tool and framework before being relied on in production.

This section’s bibliography lists the reference material consulted while preparing these pages.

Written against Spring AI 2.0.1, Spring Boot 4.1.x and Java 21 (verified 2026-09-29). A model needs seconds to write a long answer, and a user who stares at a blank page for that long assumes the page is broken. Streaming sends each piece of the answer to the browser as the model produces it. This page adds GET /api/chat/stream to the running DocsAssistant service, an endpoint that answers with server-sent events, and covers what has to be true for it to work well: the memory and tool behaviour while a stream is running, what happens when the reader leaves, how to fail cleanly half way through an answer, and how to keep blocking code off the wrong threads. Every class here was compiled against the 2.0.1 jars. The endpoints were run on a Spring context with a scripted ChatModel (no model server was available) and read with curl, and the two JUnit tests were run against the same scripted model.

What streaming changes

Streaming does not make the model faster. It changes when the user sees the first words. With the time to the first token, the number of tokens in the answer and the generation rate in tokens per second, the reader of a blocking call waits

while the reader of a stream waits only for the first text and then reads at the speed the model writes. For an illustrative 400-token answer at 50 tokens per second with s, is s against 0.8 s of waiting before the first chunk. The numbers are an example, not a measurement; measure your own model. The price is complexity: half an answer can be on its way when something fails, the reader can leave while the model is still writing, and every layer between the model and the browser (advisors, your controller, the servlet container, proxies) must pass chunks along instead of buffering them.

The streaming API

ChatClient chooses the mode when you write .call() or .stream(). Neither runs the model: the request goes out when a terminal method such as content() is invoked and, for a stream, only once somebody subscribes to the returned Flux. A Flux from stream() is therefore cold: subscribe twice and you pay for two model calls, and returning it without subscribing does nothing. Source: Spring AI: Chat Client API.

Table 1. What a streaming ChatClient call can return
Method	Element	Use it for
`stream().content()`	`String`, the next piece of text	Forwarding text to a browser. The pieces are arbitrary: a token, half a word or several words
`stream().chatResponse()`	`ChatResponse`, one per provider chunk	Metadata that travels with the chunks: model name, finish reason and, on providers that report it, usage (usually on the last chunk)
`stream().chatClientResponse()`	`ChatClientResponse`, the response plus the advisor context	Reading what advisors put in the context while streaming, for example the documents a RAG advisor retrieved
`ChatModel.stream(Prompt)`	`ChatResponse`	Bypassing `ChatClient` and its advisors altogether. In 2.0 this also bypasses the tool loop

There is no entity() on a stream: a JSON document is only valid once complete. To extract a record from a streamed reply, aggregate first and then convert, as on Structured Output.

StreamingChatService.java: tokens and chunks for one user’s conversation

package com.example.docsassistant;

import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.chat.memory.ChatMemory;
import org.springframework.ai.chat.model.ChatResponse;
import org.springframework.security.core.Authentication;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;

@Service
class StreamingChatService {

    private final ChatClient chatClient;
    private final ConversationIds conversationIds;

    StreamingChatService(ChatClient memoryChatClient, ConversationIds conversationIds) {
        this.chatClient = memoryChatClient;
        this.conversationIds = conversationIds;
    }

    /** The text of the answer, one chunk per element. Nothing runs until somebody subscribes. */
    Flux<String> tokens(Authentication authentication, String conversationLabel, String question) {
        String conversationId = this.conversationIds.forUser(authentication, conversationLabel);
        return this.chatClient.prompt()
                .user(question)
                .advisors(spec -> spec.param(ChatMemory.CONVERSATION_ID, conversationId))
                .stream()
                .content();
    }

    /** The same stream with the metadata: finish reason, model name and, on some providers, usage. */
    Flux<ChatResponse> chunks(Authentication authentication, String conversationLabel, String question) {
        String conversationId = this.conversationIds.forUser(authentication, conversationLabel);
        return this.chatClient.prompt()
                .user(question)
                .advisors(spec -> spec.param(ChatMemory.CONVERSATION_ID, conversationId))
                .stream()
                .chatResponse();
    }
}

Source: Spring AI: Chat Client API.

The conversation ID is derived from the authenticated user exactly as on Chat Memory, and the memoryChatClient is the one built there, so a streamed answer is remembered like any other. Streaming needs the reactive stack (Reactor) even in a servlet application: the ChatClient reference states that imperative applications must add spring-boot-starter-webflux for stream(), and that reactive applications need the servlet starter for non-streaming calls, which then block. Source: Spring AI: Chat Client API, implementation notes.

sequenceDiagram participant B as Browser participant C as Controller participant A as Advisors: memory, tools participant M as Model API B->>C: GET /api/chat/stream?question=... C->>A: subscribe to the Flux A->>M: streaming request M-->>A: chunk 1 A-->>C: text C-->>B: data: chunk 1 M-->>A: chunk 2 A-->>C: text C-->>B: data: chunk 2 B-xC: connection closed C->>A: cancel A->>M: cancel, connection closed Note over A: memory keeps the question, not the partial answer

Figure 1. One streamed answer, and what happens when the reader leaves

Which web stack

Both stacks can serve a Flux as server-sent events, so the choice follows the rest of the application.

Table 2. Spring MVC or WebFlux for the streaming endpoint
	Spring MVC (servlet)	Spring WebFlux (reactive)
Return type	`Flux<String>`, `Flux<ServerSentEvent<T>>` or `SseEmitter`	`Flux<String>` or `Flux<ServerSentEvent<T>>`
Threads	Servlet threads, or virtual threads with `spring.threads.virtual.enabled=true`. A stream holds no servlet thread while it waits	A few event-loop threads. Blocking code on them stalls every request
Security context	Thread-local. Needs context propagation to reach tools running on Reactor threads (below)	Reactor context, read with `ReactiveSecurityContextHolder`
Choose it when	The service is otherwise imperative: JDBC, JPA, blocking REST clients	The service is already reactive end to end, or needs very many concurrent slow streams

The trade-offs of the reactive model are on Reactive Programming with Project Reactor, and the servlet and reactive security contexts on Spring Security. With both spring-boot-starter-web and spring-boot-starter-webflux on the classpath Spring Boot starts the servlet stack; set spring.main.web-application-type=reactive to run WebFlux.

A server-sent-events endpoint on Spring MVC

Return the Flux<String> from a controller method that produces text/event-stream and Spring writes each element as one event, flushing after every chunk.

DocsAssistantStreamController.java: GET /api/chat/stream

package com.example.docsassistant;

import org.springframework.http.MediaType;
import org.springframework.security.core.Authentication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

@RestController
class DocsAssistantStreamController {

    private final StreamingChatService chat;

    DocsAssistantStreamController(StreamingChatService chat) {
        this.chat = chat;
    }

    // GET /api/chat/stream?question=... : every String the Flux emits becomes one "data:" event
    @GetMapping(path = "/api/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    Flux<String> stream(@RequestParam String question,
            @RequestParam(defaultValue = "main") String conversation,
            Authentication authentication) {
        return this.chat.tokens(authentication, conversation, question);
    }
}

Source: Spring Framework: Asynchronous Requests (Spring MVC).

Run against the scripted model, curl -N printed one data: line per element, and an element that ended in a newline produced an extra, empty data: line:

data:Hel

data:lo

data: wor

data:ld
data:

data:next line

Source: WHATWG HTML Standard: Server-sent events.

That output shows a trap for real models. Tokens often start with a space, and the SSE specification says a parser removes one leading space after the colon, so a browser’s EventSource reads data: wor as wor, and the words of the answer run together. Do not send raw text. Send each chunk as JSON, which also gives you room for event names and for the final and error events below.

TypedStreamController.java: JSON chunks, a done event and a safe error event

package com.example.docsassistant;

import java.time.Duration;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.security.core.Authentication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

@RestController
class TypedStreamController {

    private static final Logger log = LoggerFactory.getLogger(TypedStreamController.class);

    record Chunk(String text) { }

    record Failure(String message) { }

    private final StreamingChatService chat;

    TypedStreamController(StreamingChatService chat) {
        this.chat = chat;
    }

    // Each chunk travels as JSON, so leading spaces and newlines in a token survive the browser's SSE parser
    @GetMapping(path = "/api/chat/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    Flux<ServerSentEvent<Object>> events(@RequestParam String question,
            @RequestParam(defaultValue = "main") String conversation,
            Authentication authentication) {
        Flux<ServerSentEvent<Object>> tokens = this.chat.tokens(authentication, conversation, question)
                .timeout(Duration.ofSeconds(30))   // no chunk for 30 s: cancel the model call and fail the stream
                .map(text -> ServerSentEvent.<Object>builder(new Chunk(text)).event("token").build());
        ServerSentEvent<Object> done = ServerSentEvent.<Object>builder(Map.of()).event("done").build();
        return tokens.concatWith(Flux.just(done))
                .onErrorResume(ex -> {
                    log.warn("Streaming answer failed", ex);   // the detail stays in the log ...
                    Failure failure = new Failure("The answer could not be completed.");   // ... the client gets a fixed text
                    return Flux.just(ServerSentEvent.<Object>builder(failure).event("error").build());
                });
    }
}

Source: Spring Framework: Asynchronous Requests (Spring MVC).

The same request now produces named events with the text safely inside a JSON string, the leading space and the newline intact, and a last done event that lets the client stop waiting:

event:token
data:{"text":"Hel"}

event:token
data:{"text":" wor"}

event:token
data:{"text":"ld\n"}

event:done
data:{}

Source: WHATWG HTML Standard: Server-sent events.

The browser side reads these with EventSource and JSON.parse(event.data).text. EventSource cannot send an Authorization header, so authenticate the stream with the session cookie (and CSRF protection where it applies), or use fetch with a streamed response body when you need a bearer token. If a reverse proxy sits in front, make sure it does not buffer the response (for nginx, X-Accel-Buffering: no on the response or proxy_buffering off), or the reader sees the answer arrive in one lump at the end.

A server-sent-events endpoint on WebFlux

In a WebFlux application the controller sends the same JSON token events as TypedStreamController, for the same reason: raw text would lose its leading spaces in EventSource. What differs is that the caller’s identity lives in the Reactor context instead of a thread-local, so it is read reactively and the service is called inside the pipeline:

ReactiveStreamController.java: the WebFlux flavour of GET /api/chat/stream

package com.example.docsassistant;

import org.springframework.context.annotation.Profile;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.ReactiveSecurityContextHolder;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

/** WebFlux flavour of GET /api/chat/stream: the identity comes from the Reactor context, not a thread-local. */
@RestController
@Profile("reactive")
class ReactiveStreamController {

    private final StreamingChatService chat;

    ReactiveStreamController(StreamingChatService chat) {
        this.chat = chat;
    }

    // The same JSON chunk as TypedStreamController: a raw String would lose its leading space in EventSource
    @GetMapping(path = "/api/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    Flux<ServerSentEvent<Object>> stream(@RequestParam String question) {
        return ReactiveSecurityContextHolder.getContext()
                .map(SecurityContext::getAuthentication)
                .flatMapMany(authentication -> this.chat.tokens(authentication, "main", question))
                .map(token -> ServerSentEvent.<Object>builder(new TypedStreamController.Chunk(token))
                        .event("token").build());
    }
}

Source: Spring Framework: Reactive Core.

The class is compiled, and it is marked @Profile("reactive") so that it cannot clash with the MVC controller in a build that has both. Nothing in StreamingChatService changes between the stacks. The WebFlux specifics are the blocking rules: everything that runs on the pipeline must be non-blocking or moved to another scheduler (see Virtual threads and blocking calls). The built-in advisors do their blocking work on a scheduler for streaming calls, and the reference says that scheduler can be configured on each advisor’s builder. Source: Spring AI: Chat Client API.

Streaming with chat memory

MessageChatMemoryAdvisor works with stream(). It loads the history before the model is called, and it stores the exchange when the stream completes, from the aggregated response. Run with the scripted model, a completed stream left the user message and the whole assistant answer in memory, and a stream cancelled after two chunks left the question alone:

Table 3. Memory after a completed and after a cancelled stream, from the test below
Stream ends by	Messages in the conversation
completing	`USER: What is streaming?` and `ASSISTANT: Spring AI streams`
cancelling after the first chunk	`USER: Count` only. Spring AI does not roll the question back

A conversation that ends up with a question and no answer is usually harmless, because the next request adds another user message, but if your provider or your prompts need strictly alternating roles, repair the conversation in a doOnCancel handler. The mechanics of the advisor, its order and its window are on Chat Memory.

Streaming with tools

ToolCallingAdvisor is registered for stream() as it is for call(). When the model answers with a tool call instead of text, the advisor runs the tool, adds the result to the history and calls the model again, and only the text of the final answer reaches your Flux. Since 2.0 there is no streamToolCallResponses option: the intermediate tool-call chunks are never streamed, because a memory advisor downstream could store a tool call without its response. If you need to see each round, take over the loop as described on Tool Calling Advisor & Tool Search.

ToolStreamingService.java: tools, tool context and memory on a stream

package com.example.docsassistant;

import java.util.Map;

import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.chat.memory.ChatMemory;
import org.springframework.security.core.Authentication;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;

@Service
class ToolStreamingService {

    private final ChatClient chatClient;
    private final ConversationIds conversationIds;
    private final DocsTools docsTools;
    private final IssueTools issueTools;

    ToolStreamingService(ChatClient memoryChatClient, ConversationIds conversationIds,
            DocsTools docsTools, IssueTools issueTools) {
        this.chatClient = memoryChatClient;
        this.conversationIds = conversationIds;
        this.docsTools = docsTools;
        this.issueTools = issueTools;
    }

    /** Streams the final answer; the tool round trips happen silently before the first text chunk. */
    Flux<String> answer(Authentication authentication, String conversationLabel, String question) {
        String conversationId = this.conversationIds.forUser(authentication, conversationLabel);
        return this.chatClient.prompt()
                .user(question)
                .tools(this.docsTools, this.issueTools)
                .toolContext(Map.of("requester", authentication.getName()))
                .advisors(spec -> spec.param(ChatMemory.CONVERSATION_ID, conversationId))
                .stream()
                .content();
    }
}

Source: Spring AI: Chat Client API.

Three facts about tool execution during a stream, all observed with the scripted model:

The model call and the tool run on Reactor threads, not on the servlet thread that accepted the request. In the run, the tool executed on a boundedElastic thread while the memory advisor was in the chain. A blocking tool is therefore acceptable, but the pool is bounded, so a tool that can block for a long time belongs behind a timeout.
A thread-local security context does not follow. With the defaults, openIssue guarded by @PreAuthorize("hasRole('MAINTAINER')") failed for a maintainer as well: the model received the tool result An Authentication object was not found in the SecurityContext. The guard failed closed, which is the safe direction, but the feature did not work. Setting spring.reactor.context-propagation=auto made Spring Boot enable Reactor’s automatic context propagation, and the same request then succeeded for the maintainer (a draft waiting for approval) and was answered with Access Denied for a plain user. The property is a Spring Boot switch backed by the Micrometer context-propagation library, which was already on the classpath of the test application.
The tool context still travels explicitly. toolContext(…) reaches the tool as on a blocking call, and it is the better carrier for values such as the requester’s name than any thread-local. It is not a substitute for authorisation: the check that a caller may open an issue stays in the tool, as on Tool Calling.

The ChatClient reference adds a caveat about observability: because tool calling is imperative, the ChatClient span and the tool spans of a streamed request may not be connected, and the outer span may stay incomplete. See Observability for what to expect in traces.

Cancellation and time limits

Streaming gives you cancellation for free, provided nothing in the chain swallows it. When the browser closes the connection, Spring cancels the subscription, the cancel signal travels back through the advisors and the HTTP client closes the request to the model. In the run, a curl that disconnected after half a second made the scripted model stream log its cancellation, both for the Flux endpoint and for the imperative one below. Most providers stop generating when the connection closes, but confirm this for yours, and remember that tokens generated before the close are still billed.

Reactor operators let you set your own limits:

BoundedStreaming.java: a silence limit, a total limit and a size limit

package com.example.docsassistant;

import java.time.Duration;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.core.Authentication;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;

@Service
class BoundedStreaming {

    private static final Logger log = LoggerFactory.getLogger(BoundedStreaming.class);

    private final StreamingChatService chat;

    BoundedStreaming(StreamingChatService chat) {
        this.chat = chat;
    }

    Flux<String> tokens(Authentication authentication, String conversationLabel, String question) {
        return this.chat.tokens(authentication, conversationLabel, question)
                .timeout(Duration.ofSeconds(20))    // silence between two chunks: cancel upstream, signal TimeoutException
                .take(Duration.ofSeconds(90))       // hard cap on the whole answer: complete and cancel upstream
                .take(2_000)                        // hard cap on the number of chunks
                .doOnCancel(() -> log.info("Consumer went away; the model call was cancelled"))
                .doFinally(signal -> log.debug("Stream ended with {}", signal));
    }
}

Source: Project Reactor reference.

timeout(Duration) applies between two elements, including before the first, so a model that never answers fails the stream instead of holding a connection open. take(Duration) and take(long) end the stream normally and cancel what is upstream. doOnCancel is the place to record that the reader left, and doFinally runs for every ending.

Errors half way through an answer

Once the first chunk has been written, the response status is already 200 and cannot change. A failure in the middle of a stream can only be reported inside the stream, which is what the error event in TypedStreamController does. Two rules follow. Do not send the exception message to the client, since it may contain provider details, URLs or prompt fragments: log it, and send a fixed text. And do not retry in the middle: a retry would repeat text the user has already read. End the stream with the error event and let the client decide whether to send the question again. (The scripted model failed after two chunks; the client received both chunks and the fixed error event, and the exception detail appeared only in the server log.)

Virtual threads and blocking calls

Blocking code is normal in a servlet application and forbidden on an event loop. Java 21 virtual threads make the first case cheap: with spring.threads.virtual.enabled=true, Spring Boot runs request handling and its default task executors on virtual threads, so a blocking call() or a JDBC query in a tool no longer costs an expensive platform thread while it waits. Streaming is unchanged by this setting, since the Flux is driven by Reactor’s own schedulers. Reactor can also back its boundedElastic scheduler by virtual threads when started with -Dreactor.schedulers.defaultBoundedElasticOnVirtualThreads=true; in the run, a tool executed during a stream then ran on a loomBoundedElastic thread that reported isVirtual() == true, instead of on a platform thread named boundedElastic.

application.properties: the two switches this page uses

spring.threads.virtual.enabled=true
spring.reactor.context-propagation=auto

Source: Spring Boot: Common Application Properties.

When a blocking call sits inside a reactive pipeline, move it to Schedulers.boundedElastic() and keep the rest reactive. searchDocs blocks on a database query and an embedding call, so this endpoint looks up the sources first, sends them as an event, and only then starts the model call:

SourcesThenAnswerController.java: a blocking lookup on boundedElastic, then the stream

package com.example.docsassistant;

import java.time.Duration;
import java.util.List;

import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.security.core.Authentication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

@RestController
class SourcesThenAnswerController {

    private final StreamingChatService chat;
    private final DocsTools docsTools;

    SourcesThenAnswerController(StreamingChatService chat, DocsTools docsTools) {
        this.chat = chat;
        this.docsTools = docsTools;
    }

    @GetMapping(path = "/api/chat/answer-with-sources", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    Flux<ServerSentEvent<Object>> answer(@RequestParam String question, Authentication authentication) {
        // searchDocs blocks (JDBC and an embedding call), so it must never run on an event-loop thread
        Mono<List<String>> sources = Mono
                .fromCallable(() -> this.docsTools.searchDocs(question, null).stream()
                        .map(DocsTools.DocHit::sourcePath)
                        .distinct()
                        .toList())
                .subscribeOn(Schedulers.boundedElastic());

        Flux<ServerSentEvent<Object>> tokens = this.chat.tokens(authentication, "main", question)
                .timeout(Duration.ofSeconds(30))
                .map(text -> ServerSentEvent.<Object>builder(text).event("token").build());

        // the sources event goes out first; the model call starts only after the lookup finished
        return sources.flatMapMany(paths -> Flux.concat(
                Flux.just(ServerSentEvent.<Object>builder(paths).event("sources").build()),
                tokens));
    }
}

Source: Project Reactor reference.

If the application is servlet-based and you would rather write straight-line code, an SseEmitter filled by a virtual thread does the same job without operators. The one thing to get right is how the stream is consumed:

BlockingSseController.java: plain blocking code on a virtual thread

package com.example.docsassistant;

import java.io.IOException;
import java.util.stream.Stream;

import org.springframework.http.MediaType;
import org.springframework.security.core.Authentication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

/** Imperative style: a virtual thread reads the stream with plain blocking code. */
@RestController
class BlockingSseController {

    private final StreamingChatService chat;

    BlockingSseController(StreamingChatService chat) {
        this.chat = chat;
    }

    @GetMapping(path = "/api/chat/stream-blocking", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    SseEmitter stream(@RequestParam String question, Authentication authentication) {
        SseEmitter emitter = new SseEmitter(60_000L);
        Thread.ofVirtual().name("sse-answer-", 0).start(() -> {
            // toStream() gives a Stream whose close() cancels the model call; toIterable() would not
            try (Stream<String> tokens = this.chat.tokens(authentication, "main", question).toStream()) {
                tokens.forEach(token -> {
                    try {
                        emitter.send(SseEmitter.event().name("token").data(new TypedStreamController.Chunk(token)));
                    }
                    catch (IOException ex) {
                        throw new IllegalStateException("client went away", ex);   // leaves the loop, closes the stream
                    }
                });
                emitter.complete();
            }
            catch (RuntimeException ex) {
                emitter.completeWithError(ex);
            }
        });
        return emitter;
    }
}

Source: Spring Framework: Asynchronous Requests (Spring MVC).

toStream() returns a Stream whose close() cancels the subscription, and the try-with-resources block closes it when the loop ends or a send fails after the reader has gone. toIterable() would leave the model call running. In the run, a client that disconnected during this endpoint’s stream made the scripted model log its cancellation.

Testing a stream

A stream is tested with StepVerifier from reactor-test, and a scripted ChatModel replaces the provider. The two tests below check the order of the chunks, what memory holds after a completed and after a cancelled stream, and that cancelling reaches the model. More on stubs and the ChatModel traps (such as returning ToolCallingChatOptions from getOptions()) is on Testing.

StreamingChatServiceTest.java: order, memory and cancellation

package com.example.docsassistant;

import static org.assertj.core.api.Assertions.assertThat;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

import org.junit.jupiter.api.Test;
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.messages.AssistantMessage;
import org.springframework.ai.chat.model.ChatModel;
import org.springframework.ai.chat.model.ChatResponse;
import org.springframework.ai.chat.model.Generation;
import org.springframework.ai.chat.prompt.Prompt;
import org.springframework.security.authentication.TestingAuthenticationToken;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

class StreamingChatServiceTest {

    /** Streams the given chunks with a small delay and records whether the consumer cancelled. */
    static final class ScriptedModel implements ChatModel {

        final AtomicBoolean cancelled = new AtomicBoolean();
        private final List<String> chunks;

        ScriptedModel(String... chunks) {
            this.chunks = List.of(chunks);
        }

        @Override
        public ChatResponse call(Prompt prompt) {
            throw new UnsupportedOperationException("only streaming is scripted");
        }

        @Override
        public Flux<ChatResponse> stream(Prompt prompt) {
            return Flux.fromIterable(this.chunks)
                    .delayElements(Duration.ofMillis(20))
                    .map(text -> new ChatResponse(List.of(new Generation(new AssistantMessage(text)))))
                    .doOnCancel(() -> this.cancelled.set(true));
        }
    }

    private final TestingAuthenticationToken alice = new TestingAuthenticationToken("alice", "n/a", "ROLE_USER");

    private StreamingChatService service(ScriptedModel model, ChatMemory memory) {
        ChatClient client = ChatClient.builder(model)
                .defaultAdvisors(MessageChatMemoryAdvisor.builder(memory).build())
                .build();
        return new StreamingChatService(client, new ConversationIds());
    }

    @Test
    void chunksArriveInOrderAndMemoryKeepsTheWholeAnswer() {
        ChatMemory memory = MessageWindowChatMemory.builder().build();
        StreamingChatService service = service(new ScriptedModel("Spring ", "AI ", "streams"), memory);

        StepVerifier.create(service.tokens(this.alice, "main", "What is streaming?"))
                .expectNext("Spring ", "AI ", "streams")
                .verifyComplete();

        String conversationId = new ConversationIds().forUser(this.alice, "main");
        assertThat(memory.get(conversationId)).extracting(m -> m.getText())
                .containsExactly("What is streaming?", "Spring AI streams");
    }

    @Test
    void cancellingStopsTheModelStreamAndLeavesOnlyTheQuestionInMemory() {
        ScriptedModel model = new ScriptedModel("one ", "two ", "three ", "four ");
        ChatMemory memory = MessageWindowChatMemory.builder().build();

        StepVerifier.create(service(model, memory).tokens(this.alice, "main", "Count"))
                .expectNext("one ")
                .thenCancel()
                .verify(Duration.ofSeconds(5));

        String conversationId = new ConversationIds().forUser(this.alice, "main");
        assertThat(model.cancelled).isTrue();
        assertThat(memory.get(conversationId)).extracting(m -> m.getText()).containsExactly("Count");
    }
}

Source: Project Reactor reference.

Both tests passed against the 2.0.1 jars. Avoid block() in tests of a reactive service: it hides threading problems that StepVerifier and the scripted delays expose.

Streaming to other channels

The same Flux<String> can feed transports other than SSE, such as a WebSocket, a chat channel or a voice response. The chunking rules change with the channel (a chat message cannot be edited on every token, a speech synthesiser needs whole sentences), and those integrations are covered separately in Streaming Chat APIs and the Conversational Channels sub-section.

Common problems

Table 4. Symptoms of a broken stream
Symptom	Likely cause	Fix
The answer arrives all at once	A proxy or gateway buffers the response, or a filter collects the body	Turn off buffering for the path; check `curl -N` against the application directly
Words run together in the browser	Raw text in `data:` lines and a leading space in the token	Send JSON chunks (`TypedStreamController`)
The endpoint returns nothing and no model call is made	The `Flux` was not returned to the framework, or it was assembled and never subscribed	Return the `Flux`; do not call `stream()` and drop the result
The model is called twice	The `Flux` was subscribed twice, for example by logging it with `block()` or `collectList()`	Subscribe once; use `share()` or `cache()` only when you really need two consumers
A tool guarded by `@PreAuthorize` denies everyone during a stream	The thread-local security context is not propagated to Reactor threads	`spring.reactor.context-propagation=auto`, and pass identity explicitly in the tool context
`IllegalStateException` from `block()` on a reactive thread	Blocking on an event-loop thread	Use `subscribeOn(Schedulers.boundedElastic())` or run the code on a servlet or virtual thread

Streaming & Reactive

What streaming changes

The streaming API

Which web stack

A server-sent-events endpoint on Spring MVC

A server-sent-events endpoint on WebFlux

Streaming with chat memory

Streaming with tools

Cancellation and time limits

Errors half way through an answer

Virtual threads and blocking calls

Testing a stream

Streaming to other channels

Common problems

References