Tempo real (WebSocket e SSE)

import http from "node:http";

http.createServer((req, res) => {
  if (req.url !== "/eventos") {
    res.statusCode = 404;
    return res.end();
  }

  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    "X-Accel-Buffering": "no", // pede ao Nginx que não acumule a resposta
  });
  res.write("retry: 1000\n\n"); // o navegador reconecta após 1 s se a conexão cair

  // Continua de onde o cliente parou, se ele informar o último evento recebido
  let ultimo = Number(req.headers["last-event-id"] ?? 0);

  const eventos = setInterval(() => {
    ultimo++;
    res.write(`id: ${ultimo}\nevent: progresso\ndata: ${JSON.stringify({ pedido: 981, etapa: ultimo })}\n\n`);
  }, 300);
  const pulso = setInterval(() => res.write(": ping\n\n"), 15_000); // mantém a conexão viva

  req.on("close", () => {
    clearInterval(eventos);
    clearInterval(pulso);
  });
}).listen(3000);

import asyncio
import json

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()


@app.get("/eventos")
async def eventos(request: Request):
    # Continua de onde o cliente parou, se ele informar o último evento recebido
    ultimo = int(request.headers.get("last-event-id", 0))

    async def fluxo():
        nonlocal ultimo
        yield "retry: 1000\n\n"  # o navegador reconecta após 1 s se a conexão cair
        desde_o_pulso = 0.0
        while not await request.is_disconnected():
            ultimo += 1
            dados = json.dumps({"pedido": 981, "etapa": ultimo})
            yield f"id: {ultimo}\nevent: progresso\ndata: {dados}\n\n"
            await asyncio.sleep(0.3)
            desde_o_pulso += 0.3
            if desde_o_pulso >= 15:
                yield ": ping\n\n"  # mantém a conexão viva
                desde_o_pulso = 0

    return StreamingResponse(
        fluxo(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

package main

import (
	"fmt"
	"net/http"
	"strconv"
	"time"
)

func eventos(w http.ResponseWriter, r *http.Request) {
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming não suportado", http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("X-Accel-Buffering", "no")
	fmt.Fprint(w, "retry: 1000\n\n")
	flusher.Flush()

	// Continua de onde o cliente parou, se ele informar o último evento recebido
	ultimo, _ := strconv.Atoi(r.Header.Get("Last-Event-ID"))

	eventos := time.NewTicker(300 * time.Millisecond)
	pulso := time.NewTicker(15 * time.Second)
	defer eventos.Stop()
	defer pulso.Stop()

	for {
		select {
		case <-r.Context().Done(): // o cliente desconectou
			return
		case <-eventos.C:
			ultimo++
			fmt.Fprintf(w, "id: %d\nevent: progresso\ndata: {\"pedido\":981,\"etapa\":%d}\n\n", ultimo, ultimo)
			flusher.Flush()
		case <-pulso.C:
			fmt.Fprint(w, ": ping\n\n") // mantém a conexão viva
			flusher.Flush()
		}
	}
}

func main() {
	http.HandleFunc("GET /eventos", eventos)
	http.ListenAndServe(":3000", nil)
}

import com.sun.net.httpserver.HttpServer;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.Executors;

public class Eventos {
    public static void main(String[] args) throws IOException {
        var servidor = HttpServer.create(new InetSocketAddress(3000), 0);
        servidor.setExecutor(Executors.newVirtualThreadPerTaskExecutor()); // uma thread leve por conexão

        servidor.createContext("/eventos", ex -> {
            ex.getResponseHeaders().set("Content-Type", "text/event-stream");
            ex.getResponseHeaders().set("Cache-Control", "no-cache");
            ex.getResponseHeaders().set("X-Accel-Buffering", "no");
            ex.sendResponseHeaders(200, 0); // 0: tamanho desconhecido, envio em partes

            // Continua de onde o cliente parou, se ele informar o último evento recebido
            var cabecalho = ex.getRequestHeaders().getFirst("Last-Event-ID");
            int ultimo = cabecalho == null ? 0 : Integer.parseInt(cabecalho);

            try (var saida = ex.getResponseBody()) {
                saida.write("retry: 1000\n\n".getBytes(StandardCharsets.UTF_8));
                saida.flush();
                long ultimoPulso = System.currentTimeMillis();
                while (true) {
                    ultimo++;
                    var evento = "id: " + ultimo + "\nevent: progresso\ndata: {\"pedido\":981,\"etapa\":"
                        + ultimo + "}\n\n";
                    saida.write(evento.getBytes(StandardCharsets.UTF_8));
                    saida.flush();
                    Thread.sleep(300);
                    if (System.currentTimeMillis() - ultimoPulso >= 15_000) {
                        saida.write(": ping\n\n".getBytes(StandardCharsets.UTF_8)); // mantém a conexão viva
                        ultimoPulso = System.currentTimeMillis();
                    }
                }
            } catch (IOException | InterruptedException desconectou) {
                // o cliente fechou a conexão
            }
        });
        servidor.start();
    }
}

var app = WebApplication.Create(args);

app.MapGet("/eventos", async (HttpContext contexto, CancellationToken cancelamento) =>
{
    var resposta = contexto.Response;
    resposta.Headers.ContentType = "text/event-stream";
    resposta.Headers.CacheControl = "no-cache";
    resposta.Headers["X-Accel-Buffering"] = "no";

    // Continua de onde o cliente parou, se ele informar o último evento recebido
    int.TryParse(contexto.Request.Headers["Last-Event-ID"], out var ultimo);

    try
    {
        await resposta.WriteAsync("retry: 1000\n\n", cancelamento);
        await resposta.Body.FlushAsync(cancelamento);
        var ultimoPulso = DateTime.UtcNow;
        while (!cancelamento.IsCancellationRequested)
        {
            ultimo++;
            await resposta.WriteAsync(
                $"id: {ultimo}\nevent: progresso\ndata: {{\"pedido\":981,\"etapa\":{ultimo}}}\n\n", cancelamento);
            await resposta.Body.FlushAsync(cancelamento);
            await Task.Delay(300, cancelamento);
            if (DateTime.UtcNow - ultimoPulso >= TimeSpan.FromSeconds(15))
            {
                await resposta.WriteAsync(": ping\n\n", cancelamento); // mantém a conexão viva
                ultimoPulso = DateTime.UtcNow;
            }
        }
    }
    catch (OperationCanceledException)
    {
        // o cliente fechou a conexão
    }
});

app.Run("http://0.0.0.0:3000");