AI LLM Streaming

AI / LLM Streaming

Atmosphere 4.0 provides a first-class SPI for streaming AI/LLM responses token-by-token to browsers over WebSocket. The atmosphere-ai module defines the core interfaces, and optional adapter modules bridge popular AI frameworks.

Architecture

Browser  ←—WebSocket—→  @ManagedService  →  StreamingSession  →  AiStreamingAdapter
                                                    ↑
                                              Direct write to
                                            AtmosphereResource
                                         (bypasses broadcaster)

When a user sends a prompt, the @Message handler creates a StreamingSession and passes it to an adapter. The adapter calls the AI model's streaming API, forwarding each token to session.send(token). Tokens are written directly to the WebSocket connection — they bypass the broadcaster to avoid re-triggering @Message handlers.

Modules

Module	Artifact	Purpose
Core SPI	`atmosphere-ai`	`StreamingSession`, `AiStreamingAdapter`, `AiConfig`, built-in LLM client
Spring AI	`atmosphere-spring-ai`	Adapter for Spring AI `ChatClient` (Flux-based)
LangChain4j	`atmosphere-langchain4j`	Adapter for LangChain4j `StreamingChatLanguageModel`
Embabel	`atmosphere-embabel`	Adapter for Embabel Agent Framework (Kotlin)

Quick Start — Built-in LLM Client

The simplest path uses atmosphere-ai with its built-in OpenAiCompatibleClient. This works with OpenAI, Google Gemini, Ollama, Azure OpenAI, and any OpenAI-compatible endpoint — with zero additional dependencies.

1. Add the dependency

<dependency>
    <groupId>org.atmosphere</groupId>
    <artifactId>atmosphere-ai</artifactId>
    <version>4.0.0-SNAPSHOT</version>
</dependency>

2. Configure the LLM

Set environment variables:

Variable	Description	Default
`LLM_MODE`	`remote` (cloud API) or `local` (Ollama)	`remote`
`LLM_MODEL`	Model name: `gemini-2.5-flash`, `gpt-4o`, `llama3.2`, etc.	`gemini-2.5-flash`
`LLM_API_KEY`	API key (or `GEMINI_API_KEY` for Gemini)	—
`LLM_BASE_URL`	Override the endpoint (auto-detected if omitted)	auto

The endpoint is auto-detected from the model name:

gemini-* → https://generativelanguage.googleapis.com/v1beta/openai
gpt-* → https://api.openai.com/v1
local mode → http://localhost:11434/v1 (Ollama)

Or configure programmatically in Spring Boot:

@Configuration
public class LlmConfig {
    @Bean
    AiConfig.LlmSettings llmSettings(@Value("${llm.api-key}") String apiKey,
                                      @Value("${llm.model:gemini-2.5-flash}") String model) {
        return AiConfig.configure(apiKey, model, "remote", null);
    }
}

3. Create the server endpoint

@ManagedService(path = "/ai-chat")
public class AiChat {

    @Inject private AtmosphereResource resource;

    @Message
    public void onMessage(String prompt) {
        var settings = AiConfig.get();
        var session = StreamingSessions.start(resource);

        var request = ChatCompletionRequest.builder(settings.model())
                .system("You are a helpful assistant.")
                .user(prompt)
                .build();

        // Stream on a virtual thread — non-blocking
        Thread.startVirtualThread(() -> settings.client().streamChatCompletion(request, session));
    }
}

4. Connect from the browser

import { subscribeStreaming } from 'atmosphere.js';

const handle = await subscribeStreaming(atmosphere, {
  url: '/ai-chat',
  transport: 'websocket',
}, {
  onToken:    (token) => output.textContent += token,
  onProgress: (msg)   => status.textContent = msg,
  onComplete: ()      => console.log('Done'),
  onError:    (err)   => console.error(err),
});

handle.send('Explain virtual threads in Java 21');

Or with React:

import { useStreaming } from 'atmosphere.js/react';

function AiChat() {
  const { fullText, isStreaming, send } = useStreaming({
    request: { url: '/ai-chat', transport: 'websocket' },
  });

  return (
    <div>
      <button onClick={() => send('What is Atmosphere?')} disabled={isStreaming}>Ask</button>
      <p>{fullText}</p>
    </div>
  );
}

Wire Protocol

Every message is a JSON object written directly to the WebSocket:

{"type":"token","data":"Hello","sessionId":"abc-123","seq":1}
{"type":"token","data":" world","sessionId":"abc-123","seq":2}
{"type":"progress","data":"Thinking...","sessionId":"abc-123","seq":3}
{"type":"metadata","data":"{\"model\":\"gemini-2.5-flash\"}","sessionId":"abc-123","seq":4}
{"type":"complete","data":"","sessionId":"abc-123","seq":5}

Type	Description
`token`	A single token/chunk from the LLM
`progress`	A human-readable status update (e.g., "Searching documents...")
`metadata`	Structured metadata (model name, usage stats)
`complete`	Stream finished successfully
`error`	Stream failed — `data` contains the error message

The seq field is a monotonically increasing counter for deduplication on reconnect.

Core SPI — `StreamingSession`

The StreamingSession interface is the bridge between any AI framework and Atmosphere:

public interface StreamingSession extends AutoCloseable {
    String sessionId();
    void send(String token);           // push a token
    void sendMetadata(String key, Object value);  // structured metadata
    void progress(String message);     // status update
    void complete();                   // success
    void complete(String summary);     // success with summary
    void error(Throwable t);           // failure
    boolean isClosed();
}

Create one with StreamingSessions.start(resource). The DefaultStreamingSession implementation writes JSON directly to AtmosphereResource, bypassing the broadcaster.

Core SPI — `AiStreamingAdapter`

Implement this to bridge any AI framework:

public interface AiStreamingAdapter<T> {
    String name();
    void stream(T request, StreamingSession session);
}

Example for a custom framework:

public class MyAdapter implements AiStreamingAdapter<MyPrompt> {
    @Override public String name() { return "my-ai"; }

    @Override
    public void stream(MyPrompt request, StreamingSession session) {
        myModel.streamTokens(request,
            token -> session.send(token),
            ()    -> session.complete(),
            err   -> session.error(err));
    }
}

Adapter: Spring AI

Bridges Spring AI's ChatClient Flux-based streaming to StreamingSession.

<dependency>
    <groupId>org.atmosphere</groupId>
    <artifactId>atmosphere-spring-ai</artifactId>
    <version>4.0.0-SNAPSHOT</version>
</dependency>

@ManagedService(path = "/ai-chat")
public class AiChat {
    @Inject private AtmosphereResource resource;
    @Inject private ChatClient chatClient;

    private final SpringAiStreamingAdapter adapter = new SpringAiStreamingAdapter();

    @Message
    public void onMessage(String prompt) {
        var session = StreamingSessions.start(resource);
        adapter.stream(chatClient, prompt, session);
    }
}

The adapter subscribes to chatClient.prompt(prompt).stream().chatResponse() and pushes each ChatResponse token through the session.

Adapter: LangChain4j

Bridges LangChain4j's callback-based StreamingChatLanguageModel to StreamingSession.

<dependency>
    <groupId>org.atmosphere</groupId>
    <artifactId>atmosphere-langchain4j</artifactId>
    <version>4.0.0-SNAPSHOT</version>
</dependency>

@ManagedService(path = "/ai-chat")
public class AiChat {
    @Inject private AtmosphereResource resource;

    private final LangChain4jStreamingAdapter adapter = new LangChain4jStreamingAdapter();

    @Message
    public void onMessage(String prompt) {
        var session = StreamingSessions.start(resource);
        var model = OpenAiStreamingChatModel.builder()
                .baseUrl(AiConfig.get().baseUrl())
                .apiKey(AiConfig.get().client().apiKey())
                .modelName(AiConfig.get().model())
                .build();

        var chatRequest = ChatRequest.builder()
                .messages(List.of(UserMessage.from(prompt)))
                .build();

        Thread.startVirtualThread(() -> adapter.stream(model, chatRequest, session));
    }
}

AtmosphereStreamingResponseHandler converts LangChain4j's onNext/onComplete/onError callbacks to StreamingSession calls.

Adapter: Embabel Agent Framework

Bridges Embabel's OutputChannel pattern for agentic AI (Kotlin).

<dependency>
    <groupId>org.atmosphere</groupId>
    <artifactId>atmosphere-embabel</artifactId>
    <version>4.0.0-SNAPSHOT</version>
</dependency>

@ManagedService(path = "/ai-agent")
class AgentChat {
    @Inject lateinit var resource: AtmosphereResource

    @Message
    fun onMessage(prompt: String) {
        val session = StreamingSessions.start(resource)
        val channel = AtmosphereOutputChannel(session)
        // Agent progress events stream to the browser
        Thread.startVirtualThread { agentPlatform.runAgent(prompt, outputChannel = channel) }
    }
}

AtmosphereOutputChannel translates agent events (thinking, tool calls, results) into StreamingSession calls with appropriate progress / token / complete messages.

Client: atmosphere.js Streaming

Vanilla TypeScript

import { atmosphere, subscribeStreaming } from 'atmosphere.js';

const handle = await subscribeStreaming(atmosphere, {
  url: '/ai-chat',
  transport: 'websocket',
}, {
  onToken:    (token) => { /* append token to UI */ },
  onProgress: (msg)   => { /* show status */ },
  onMetadata: (meta)  => { /* model info, usage */ },
  onComplete: ()      => { /* done */ },
  onError:    (err)   => { /* handle error */ },
});

handle.send('Your prompt here');
handle.close(); // disconnect when done

React — `useStreaming`

import { useStreaming } from 'atmosphere.js/react';

function AiChat() {
  const { fullText, tokens, isStreaming, progress, metadata, error, send, reset } = useStreaming({
    request: { url: '/ai-chat', transport: 'websocket' },
  });

  return (
    <div>
      <button onClick={() => send('What is Atmosphere?')} disabled={isStreaming}>Ask</button>
      {isStreaming && <span>{progress ?? 'Generating…'}</span>}
      <p>{fullText}</p>
      <button onClick={reset}>Clear</button>
    </div>
  );
}

Vue — `useStreaming`

<script setup>
import { useStreaming } from 'atmosphere.js/vue';

const { fullText, isStreaming, send } = useStreaming(
  { url: '/ai-chat', transport: 'websocket' },
);
</script>

<template>
  <button @click="send('What is Atmosphere?')" :disabled="isStreaming">Ask</button>
  <p>{{ fullText }}</p>
</template>

Svelte — `createStreamingStore`

<script>
  import { createStreamingStore } from 'atmosphere.js/svelte';

  const { store: ai, send } = createStreamingStore(
    { url: '/ai-chat', transport: 'websocket' },
  );
</script>

<button on:click={() => send('What is Atmosphere?')} disabled={$ai.isStreaming}>Ask</button>
<p>{$ai.fullText}</p>

Samples

Sample	AI Framework	Source
AI Chat (built-in client)	`atmosphere-ai`	spring-boot-ai-chat
LangChain4j Chat	`atmosphere-langchain4j`	spring-boot-langchain4j-chat
Embabel Agent Chat	`atmosphere-embabel`	spring-boot-embabel-chat

Uh oh!

Uh oh!

AI LLM Streaming

AI / LLM Streaming

Architecture

Modules

Quick Start — Built-in LLM Client

1. Add the dependency

2. Configure the LLM

3. Create the server endpoint

4. Connect from the browser

Wire Protocol

Core SPI — StreamingSession

Core SPI — AiStreamingAdapter

Adapter: Spring AI

Adapter: LangChain4j

Adapter: Embabel Agent Framework

Client: atmosphere.js Streaming

Vanilla TypeScript

React — useStreaming

Vue — useStreaming

Svelte — createStreamingStore

Samples

See Also

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally

Core SPI — `StreamingSession`

Core SPI — `AiStreamingAdapter`

React — `useStreaming`

Vue — `useStreaming`

Svelte — `createStreamingStore`