Streaming Responses

Streaming Responses

Veloce-TS supports streaming responses via AsyncGenerator handlers. Use @SSE() for Server-Sent Events and @Stream(contentType) for any other streaming format (CSV, NDJSON, binary, etc.).

@SSE() — Server-Sent Events

Mark an AsyncGenerator handler with @SSE() to stream events to the browser. The response is automatically set to text/event-stream with the correct no-cache headers.

Each yield becomes an SSE message:

import { Controller, Get, SSE } from 'veloce-ts';

@Controller('/notifications')
class NotificationController {
  @Get('/stream')
  @SSE()
  async* streamNotifications(): AsyncGenerator<{ data: string; event?: string; id?: string }> {
    yield { data: 'connected', event: 'open' };

    while (true) {
      const notification = await getNextNotification();
      yield {
        data: JSON.stringify(notification),
        event: 'notification',
        id: notification.id,
      };
    }
  }
}

SSE event shape:

Field	Required	Description
data	Yes	Event payload (string)
event	No	Event type (defaults to "message")
id	No	Event ID for reconnection
retry	No	Reconnect delay in ms

LLM streaming example

@Get('/chat')
@SSE()
async* chat(@Query('prompt') prompt: string): AsyncGenerator<{ data: string }> {
  const stream = await llm.stream(prompt);
  for await (const chunk of stream) {
    yield { data: chunk.text };
  }
  yield { data: '[DONE]', event: 'done' };
}

Client-side usage

const source = new EventSource('/notifications/stream');

source.addEventListener('notification', (e) => {
  const notification = JSON.parse(e.data);
  console.log(notification);
});

source.addEventListener('open', (e) => {
  console.log('Connected:', e.data);
});

@Stream(contentType) — Raw stream

For non-SSE streaming, use @Stream with any content type. Each yield sends a chunk directly to the client.

CSV export

import { Controller, Get, Stream } from 'veloce-ts';

@Controller('/export')
class ExportController {
  @Get('/users.csv')
  @Stream('text/csv')
  async* exportUsers(): AsyncGenerator<string> {
    yield 'id,name,email\n';
    for await (const user of db.select().from(users)) {
      yield `${user.id},${user.name},${user.email}\n`;
    }
  }
}

NDJSON stream

@Get('/feed.ndjson')
@Stream('application/x-ndjson')
async* streamFeed(): AsyncGenerator<string> {
  for await (const item of eventSource) {
    yield JSON.stringify(item) + '\n';
  }
}

Binary stream

@Get('/video/:id')
@Stream('video/mp4')
async* streamVideo(@Param('id') id: string): AsyncGenerator<Uint8Array> {
  const file = await openVideoFile(id);
  const reader = file.readable.getReader();
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    yield value;
  }
}

Combining with other decorators

@Get('/export.csv')
@Stream('text/csv')
@ResponseHeader('Content-Disposition', 'attachment; filename="users.csv"')
@UseMiddleware(authMiddleware)
async* exportUsers(): AsyncGenerator<string> {
  yield 'id,name\n';
  // ...
}

Error handling in streams

Throw inside the generator to terminate the stream. For SSE, the client will reconnect automatically after the connection closes.

@Get('/events')
@SSE()
async* streamEvents(): AsyncGenerator<{ data: string }> {
  try {
    for await (const event of eventSource) {
      yield { data: JSON.stringify(event) };
    }
  } catch (err) {
    yield { data: JSON.stringify({ error: 'Stream interrupted' }), event: 'error' };
  }
}

Next Steps

• Event Bus — in-process pub/sub for feeding SSE streams
• Interceptors
• Decorators Guide