Real-Time Data Streaming con WebStreams API e ReadableStreams

Quando si riceve una risposta HTTP, il comportamento più comune è aspettare che arrivi completamente, poi lavorarci sopra. Per un JSON di pochi kilobyte funziona benissimo. Per un file da 50 megabyte, un feed di eventi continuo, o la risposta incrementale di un modello linguistico, questo approccio è inefficiente e a volte semplicemente impraticabile. La WebStreams API nasce proprio per gestire i dati come flusso continuo: si inizia a elaborare i primi byte mentre gli ultimi sono ancora in viaggio.

Il modello a stream: produttori, consumatori e backpressure

La WebStreams API ruota attorno a tre tipi di oggetti: ReadableStream, che rappresenta una sorgente di dati, WritableStream, che rappresenta una destinazione, e TransformStream, che si inserisce in mezzo trasformando i dati che vi passano attraverso. Questi tre elementi possono essere collegati in una pipeline, dove i dati fluiscono dalla sorgente alla destinazione attraverso zero o più trasformazioni.

Un concetto fondamentale in questo modello è la backpressure, ovvero la pressione a ritroso. Quando il consumatore non riesce a stare al passo con la velocità della sorgente, lo stream lo segnala automaticamente al produttore, che rallenta di conseguenza. Questo evita che i dati si accumulino in memoria senza controllo. È un meccanismo che nei sistemi di streaming tradizionali richiede implementazione manuale; qui è incorporato nel design della API.

ReadableStream: leggere i dati mentre arrivano

La proprietà body di una risposta fetch è già un ReadableStream. Normalmente si chiama response.json() o response.text(), che internamente consumano l'intero stream prima di restituire il valore. Per accedere ai dati incrementalmente si usa invece il reader, che permette di leggere un chunk alla volta non appena diventa disponibile.

const risposta = await fetch('/api/file-grande');
const reader = risposta.body.getReader();
const decoder = new TextDecoder();
let testo = '';

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  // value è un Uint8Array con i byte del chunk corrente
  testo += decoder.decode(value, { stream: true });
  console.log(`Ricevuti ${value.length} byte, totale: ${testo.length}`);
}

console.log('Stream completato');

Il metodo reader.read() restituisce una promise che si risolve con un oggetto contenente done e value. Quando done è true lo stream è esaurito. Ogni iterazione del ciclo processa un chunk non appena il browser lo rende disponibile, senza aspettare che il download sia completo. Questo è già sufficiente per mostrare progressi in tempo reale all'utente durante download lunghi.

Creare un ReadableStream personalizzato

Non è obbligatorio partire da una risposta fetch: è possibile creare stream personalizzati da qualsiasi sorgente di dati. Il costruttore di ReadableStream accetta un oggetto con metodi start, pull e cancel. Il metodo pull viene invocato automaticamente ogni volta che il consumatore è pronto per ricevere altri dati, implementando così la backpressure in modo naturale.

function creaStreamNumeri(fino) {
  let corrente = 0;

  return new ReadableStream({
    pull(controller) {
      if (corrente <= fino) {
        controller.enqueue(corrente++);
      } else {
        controller.close();
      }
    },
    cancel() {
      console.log('Stream annullato dal consumatore');
    }
  });
}
const stream = creaStreamNumeri(100);
const reader = stream.getReader();

let chunk;
while (!(chunk = await reader.read()).done) {
  console.log(chunk.value);
}

Il metodo controller.enqueue aggiunge un dato alla coda interna dello stream. Il metodo controller.close segnala che non arriveranno altri dati. L'API gestisce internamente la sincronizzazione: pull non viene chiamato di nuovo finché il consumatore non ha processato il chunk precedente.

TransformStream: trasformare i dati in pipeline

Un TransformStream è allo stesso tempo un WritableStream e un ReadableStream: riceve dati da un lato, li trasforma, e li emette dall'altro. È il pezzo ideale da inserire in mezzo a una pipeline. Un caso classico è la decodifica dei byte in testo, che il browser fornisce già pronta tramite TextDecoderStream.

const risposta = await fetch('/api/testo-lungo');

// Pipeline: bytes → testo → righe
const streamTesto = risposta.body
  .pipeThrough(new TextDecoderStream());
const reader = streamTesto.getReader();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  process.stdout.write(value); // o aggiorna il DOM
}

Si può creare un TransformStream personalizzato per qualsiasi trasformazione serva: parsing, filtraggio, aggregazione. Il costruttore accetta un oggetto con il metodo transform, che riceve ogni chunk in ingresso e decide cosa emettere a valle tramite il controller.

// Trasforma uno stream di testo in uno stream di righe complete
const splitterRighe = new TransformStream({
  buffer: '',
  transform(chunk, controller) {
    this.buffer += chunk;
    const righe = this.buffer.split('\n');
    this.buffer = righe.pop(); // l'ultima riga potrebbe essere incompleta
    for (const riga of righe) {
      controller.enqueue(riga);
    }
  },
  flush(controller) {
    if (this.buffer) controller.enqueue(this.buffer);
  }
});
const reader = risposta.body
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(splitterRighe)
  .getReader();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  console.log('Riga completa:', value);
}

Il metodo flush viene chiamato quando lo stream a monte si chiude, permettendo di emettere eventuali dati rimasti nel buffer interno prima di concludere.

Streaming delle risposte AI: un caso d'uso concreto

Le API dei modelli linguistici come quella di Anthropic restituiscono le risposte in streaming: i token arrivano uno dopo l'altro mentre il modello genera il testo. Senza streaming l'utente fisserebbe uno schermo bianco per diversi secondi; con lo streaming vede le parole apparire progressivamente, esattamente come avviene in qualsiasi chat AI moderna. Implementarlo richiede di consumare il body della risposta come stream di testo.

async function streamRispostaAI(prompt, onToken) {
  const risposta = await fetch('https://api.esempio.com/completions', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ prompt, stream: true })
  });

  const reader = risposta.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(splitterRighe) // il TransformStream definito sopra
    .getReader();
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    if (value.startsWith('data: ')) {
      const json = value.slice(6);
      if (json === '[DONE]') break;
      const evento = JSON.parse(json);
      const token = evento.choices?.[0]?.delta?.content ?? '';
      if (token) onToken(token);
    }
  }
}

// Utilizzo: aggiorna il DOM a ogni token ricevuto
await streamRispostaAI('Spiegami i Web Workers', token => {
  document.getElementById('output').textContent += token;
});

Questa implementazione combina tutto quello discusso finora: fetch con body in streaming, TextDecoderStream per la decodifica, un TransformStream personalizzato per dividere il testo in righe, e una callback che aggiorna il DOM incrementalmente. La backpressure garantisce che il browser non accumuli in memoria più dati di quanti il codice riesca a processare.