# How the Streaming Response Works in the OpenAPI Server

> Discover how the OpenAPI server leverages Spring's SseEmitter for efficient streaming responses. Learn about asynchronous event processing and keep-alive pings to prevent timeouts.

- Repository: [Junjie.M/dat](https://github.com/junjiem/dat)
- Tags: deep-dive
- Published: 2026-03-05

---

**The OpenAPI server implements streaming responses using Spring's `SseEmitter` to maintain an open HTTP connection, asynchronously processing agent events through `StreamAction` while sending periodic ping keep-alives to prevent timeouts.**

The `junjiem/dat` repository provides an AI data analysis tool that streams real-time responses through its OpenAPI server. Understanding how the streaming response works in the OpenAPI server is essential for building responsive client applications that consume agent outputs, SQL generation results, and incremental answers as they become available.

## Core Architecture of the Streaming Endpoint

The streaming endpoint is implemented in **`AskController`** at `/api/v1/ask/stream`. When a client POSTs an `AskRequest`, the controller creates a **`SseEmitter`** and returns it immediately, allowing the HTTP connection to stay open while events are pushed to the client.

### The AskController Entry Point

Located in [`dat-servers/dat-server-openapi/src/main/java/ai/dat/server/openapi/controller/AskController.java`](https://github.com/junjiem/dat/blob/main/dat-servers/dat-server-openapi/src/main/java/ai/dat/server/openapi/controller/AskController.java), the controller handles the initial request setup:

```java
@PostMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter askStream(@Valid @RequestBody AskRequest request) {
    String conversationId = (request.getConversationId() == null || request.getConversationId().isBlank())
            ? UUID.randomUUID().toString()
            : request.getConversationId();

    SseEmitter emitter = new SseEmitter();
    // start ping & stream processing (see full implementation in AskController)
    …
    return emitter;
}

```

The method generates a unique `conversationId` if not provided, initializes the `SseEmitter`, and returns it immediately to the client while background processing continues.

### Connection Keep-Alive with Scheduled Ping

To prevent HTTP timeouts during long-running agent operations, a `ScheduledExecutorService` (`pingScheduler`) sends a **ping event** every 10 seconds:

```java
ScheduledFuture<?> pingTask = pingScheduler.scheduleAtFixedRate(() -> {
    emitter.send(SseEmitter.event()
        .name(PING_EVENT)
        .data(Map.of(TIMESTAMP, System.currentTimeMillis(),
                    CONVERSATION_ID, conversationId)));
}, 0, 10, TimeUnit.SECONDS);

```

*Source:* [AskController – ping scheduler](https://github.com/junjiem/dat/blob/main/dat-servers/dat-server-openapi/src/main/java/ai/dat/server/openapi/controller/AskController.java#L84-L92)

If the ping fails (indicating the client has disconnected), the task is cancelled to prevent resource leaks.

## Asynchronous Stream Processing

The controller delegates actual stream processing to a background thread pool so the request-handling thread can return immediately.

### Background Thread Execution

A cached thread-pool (`streamExecutor`) runs `processStreamEvents(...)` asynchronously:

```java
streamExecutor.execute(() ->
    processStreamEvents(emitter, conversationId, request, histories, pingTask));

```

*Source:* [AskController – async processing](https://github.com/junjiem/dat/blob/main/dat-servers/dat-server-openapi/src/main/java/ai/dat/server/openapi/controller/AskController.java#L44-L48)

### The StreamAction and Event Iteration

The `processStreamEvents` method obtains a **`StreamAction`** from **`ProjectService`**:

```java
StreamAction action = runnerService.ask(conversationId,
    request.getAgentName(), request.getQuestion(), histories);

```

*Source:* [AskController – get StreamAction](https://github.com/junjiem/dat/blob/main/dat-servers/dat-server-openapi/src/main/java/ai/dat/server/openapi/controller/AskController.java#L73-L76)

`ProjectService` creates or re-uses a `ProjectRunner` that holds the agent state. The `StreamAction` implements `Iterable<StreamEvent>`, allowing the controller to iterate over the agent's output:

```java
for (StreamEvent event : action) {
    // translate each StreamEvent into an SSE payload
    sendStreamEvent(emitter, event, eventId, conversationId);
}

```

*Source:* [StreamAction iterator](https://github.com/junjiem/dat/blob/main/dat-core/src/main/java/ai/dat/core/agent/data/StreamAction.java#L15-L23)

## Mapping StreamEvents to SSE Messages

The `sendStreamEvent` method in `AskController` translates each `StreamEvent` into a specific SSE event name based on which data fields are present.

### Event Type Detection Logic

Each **`StreamEvent`** carries optional pieces of data (incremental answer text, semantic SQL, query results, tool calls, etc.). The controller maps these to specific SSE event names:

| SSE name | Triggered by |
|----------|--------------|
| `ping` | Scheduled ping task |
| `sql_generate` | `event.getSemanticSql()` |
| `semantic_to_sql` | `event.getQuerySql()` |
| `sql_execute` | `event.getQueryData()` |
| `agent_answer` | `event.getIncrementalContent()` |
| `agent_answer_end` | End of incremental answer sequence |
| `before_tool_execution` | Tool request (no result yet) |
| `tool_execution` | Tool result present |
| `hitl_ai_request` | Human-in-the-loop AI request |
| `hitl_tool_approval` | Human-in-the-loop tool approval |
| `error` | Exceptions during processing |
| `finished` | Normal or error termination |
| `other` | Any additional custom event |

*Source:* [sendStreamEvent logic](https://github.com/junjiem/dat/blob/main/dat-servers/dat-server-openapi/src/main/java/ai/dat/server/openapi/controller/AskController.java#L55-L99)

### Termination and Cleanup

After the loop finishes, a **`finished`** event is emitted with status `succeeded` (or `failed` if an exception occurred) and the SSE connection is closed via `emitter.complete()` or `completeWithError()`. The ping task is cancelled in the `finally` block to guarantee cleanup.

*Source:* [processStreamEvents – finalisation](https://github.com/junjiem/dat/blob/main/dat-servers/dat-server-openapi/src/main/java/ai/dat/server/openapi/controller/AskController.java#L35-L49)

## Client-Side Implementation Example

Clients consume the stream using standard SSE protocols. The following JavaScript example demonstrates how to handle the various event types:

```javascript
const payload = {
  conversationId: '',
  agentName: 'my-agent',
  question: 'Show the total sales per month',
  // … other fields if needed
};

fetch('/api/v1/ask/stream', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify(payload)
}).then(res => {
  const evtSource = new EventSource(res.url); // SSE URL returned by the server

  evtSource.addEventListener('ping', e => console.log('ping', JSON.parse(e.data)));
  evtSource.addEventListener('agent_answer', e => console.log('chunk', JSON.parse(e.data).answer));
  evtSource.addEventListener('agent_answer_end', e => console.log('answer finished'));
  evtSource.addEventListener('sql_execute', e => console.log('query result', JSON.parse(e.data).query_data));
  evtSource.addEventListener('error', e => console.error('stream error', JSON.parse(e.data)));
  evtSource.addEventListener('finished', e => {
    console.log('stream done', JSON.parse(e.data).status);
    evtSource.close();
  });
});

```

The client reconstructs the full answer from incremental `agent_answer` messages and handles HITL prompts (`hitl_ai_request`, `hitl_tool_approval`) similarly.

A typical SSE payload for an agent answer chunk looks like this:

```text
event: agent_answer
data: {"conversation_id":"c1f2…","timestamp":1719915600000,"answer_id":"a3d9…","answer":"We are analyzing the data for you …"}

```

## Key Source Files and Implementation Details

| File | Role | GitHub link |
|------|------|-------------|
| [`AskController.java`](https://github.com/junjiem/dat/blob/main/AskController.java) | Exposes `/stream` endpoint, creates `SseEmitter`, schedules ping, dispatches `StreamEvent`s. | [AskController](https://github.com/junjiem/dat/blob/main/dat-servers/dat-server-openapi/src/main/java/ai/dat/server/openapi/controller/AskController.java) |
| [`ProjectService.java`](https://github.com/junjiem/dat/blob/main/ProjectService.java) | Looks up or creates a `ProjectRunner` per conversation, forwards the ask request. | [ProjectService (openapi)](https://github.com/junjiem/dat/blob/main/dat-servers/dat-server-openapi/src/main/java/ai/dat/server/openapi/service/ProjectService.java) |
| [`StreamAction.java`](https://github.com/junjiem/dat/blob/main/StreamAction.java) | Implements `Iterable<StreamEvent>`; wraps the agent's processing pipeline. | [StreamAction](https://github.com/junjiem/dat/blob/main/dat-core/src/main/java/ai/dat/core/agent/data/StreamAction.java) |
| [`StreamEvent.java`](https://github.com/junjiem/dat/blob/main/StreamEvent.java) | Holds the optional data pieces (incremental answer, SQL, tool calls, etc.) that are turned into SSE messages. | [StreamEvent](https://github.com/junjiem/dat/blob/main/dat-core/src/main/java/ai/dat/core/agent/data/StreamEvent.java) |
| [`ProjectRunner.java`](https://github.com/junjiem/dat/blob/main/ProjectRunner.java) (used indirectly) | Executes the agent logic (`ask`, `userResponse`, `userApproval`). | [ProjectRunner](https://github.com/junjiem/dat/blob/main/dat-boot/src/main/java/ai/dat/boot/ProjectRunner.java) |

## Summary

- **SseEmitter** and **ScheduledExecutorService** maintain the HTTP connection through periodic ping events every 10 seconds.
- **ProjectService** and **ProjectRunner** drive the agent's execution, returning a **StreamAction** that implements `Iterable<StreamEvent>`.
- The controller iterates over **StreamEvent** objects and maps them to specific SSE event names based on which data fields are present.
- A cached thread pool processes the stream asynchronously, allowing the request thread to return the `SseEmitter` immediately.
- The connection closes with a `finished` event, and the ping task is cancelled in a `finally` block to prevent resource leaks.

## Frequently Asked Questions

### What protocol does the OpenAPI server use for streaming?

The server uses **Server-Sent Events (SSE)**, a standard HTTP-based protocol where the client opens a persistent connection and the server pushes events as they occur. This is implemented through Spring's `SseEmitter` class with `MediaType.TEXT_EVENT_STREAM_VALUE`, allowing real-time delivery of agent outputs without WebSocket complexity.

### How does the server prevent connection timeouts during long-running agent operations?

A `ScheduledExecutorService` sends a **ping event** every 10 seconds to keep the HTTP connection alive. If a ping fails (indicating client disconnection), the scheduled task is cancelled immediately. Additionally, all stream processing occurs in a background thread pool, ensuring the initial request thread returns quickly while the connection remains open for the duration of agent execution.

### What types of events can clients expect from the stream?

Clients receive various event types including `agent_answer` (incremental text chunks), `agent_answer_end` (completion signal), `sql_generate` (semantic SQL), `semantic_to_sql` (translated SQL), `sql_execute` (query results), `tool_execution` (tool results), `hitl_ai_request` and `hitl_tool_approval` (human-in-the-loop prompts), `error` (exceptions), and `finished` (stream termination).

### How is resource cleanup handled when a client disconnects?

The ping scheduler detects failed send attempts and cancels the keep-alive task. Additionally, the `processStreamEvents` method includes a `finally` block that guarantees the ping task is cancelled and the `SseEmitter` is completed (either normally or with error) when the stream ends, encounters an exception, or the client disconnects.