Respuestas en Streaming

Respuestas en Streaming

Veloce-TS soporta respuestas en streaming mediante handlers AsyncGenerator. Usa @SSE() para Server-Sent Events y @Stream(contentType) para cualquier otro formato de streaming (CSV, NDJSON, binario, etc.).

@SSE() — Server-Sent Events

Marca un handler AsyncGenerator con @SSE() para hacer streaming de eventos al navegador. La respuesta se establece automáticamente como text/event-stream con las cabeceras correctas de no-caché.

Cada yield se convierte en un mensaje SSE:

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

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

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

Forma del evento SSE:

Campo	Requerido	Descripción
data	Sí	Payload del evento (string)
event	No	Tipo de evento (por defecto "message")
id	No	ID del evento para reconexión
retry	No	Delay de reconexión en ms

Ejemplo de streaming LLM

@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' };
}

Uso en el cliente

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

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

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

@Stream(contentType) — Stream raw

Para streaming no SSE, usa @Stream con cualquier tipo de contenido. Cada yield envía un chunk directamente al cliente.

Exportación CSV

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

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

Stream NDJSON

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

Stream binario

@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;
  }
}

Combinación con otros decoradores

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

Manejo de errores en streams

Lanza dentro del generator para terminar el stream. Para SSE, el cliente se reconectará automáticamente tras cerrar la conexión.

@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 interrumpido' }), event: 'error' };
  }
}

Próximos Pasos

• Bus de Eventos — pub/sub en proceso para alimentar streams SSE
• Interceptores
• Guía de Decoradores