resposta em fluxo

Home > Pedido e resposta > Resposta em fluxo

Resposta em fluxo (streaming e Server-Sent Events) 🚧

Important

Em desenvolvimento (2.0). Ainda não está numa versão lançada e pode mudar.

Delphi e Lazarus · Nível: intermediário

Em uma frase: a rota manda a resposta aos pedaços enquanto ainda está rodando - progresso de uma tarefa longa, uma lista que vai sendo lida do banco, eventos para o navegador (EventSource) ou para um cliente MCP - e o cliente RAL recebe cada pedaço quando ele chega.

Você vai precisar: um servidor RAL 2.0 com um motor que envia ao vivo (tabela Motores); do lado do cliente, o TRALClient 2.0 ou qualquer cliente HTTP (navegador, curl -N).

Servidor: mandar aos pedaços

procedure TForm1.Progresso(ARequest: TRALRequest; AResponse: TRALResponse);
var
  vInt: Integer;
begin
  for vInt := 1 to 10 do
  begin
    // False: o cliente foi embora - pare de trabalhar
    if not AResponse.SendEvent(Format('{"passo":%d}', [vInt]), 'progresso') then
      Exit;
    Sleep(500);
  end;
end;

FServer.CreateRoute('progresso', {$IFDEF FPC}@{$ENDIF}Progresso).Streaming := True;

Método da resposta	O que faz
Send(texto), Send(buffer, tamanho), Send(stream)	manda um pedaço agora
SendEvent(dados, evento, id, retry)	manda um evento SSE (data:, event:, id:, retry:)
BeginStream	manda status e cabeçalhos já, sem pedaço (o EventSource "abre" na hora)
EndStream	termina o corpo antes de a rota acabar
Detach	mantém o fluxo aberto depois que a rota retorna (veja abaixo)

• O primeiro Send envia status e cabeçalhos. Escolha StatusCode, ContentType e cabeçalhos antes dele. Depois, Answer, StatusCode, AddHeader e AddCookie levantam ERALResponseCommitted. O corpo que a rota já tinha montado vai como primeiro pedaço.
• SendEvent (ou BeginStream sem tipo) faz a resposta text/event-stream e manda Cache-Control: no-cache e X-Accel-Buffering: no (o nginx deixa de segurar). Para NDJSON use Send com ContentType := rctAPPLICATIONNDJSON.
• Um cliente que sai não é erro: Send devolve False, e AResponse.Connected pergunta sem mandar nada.
• Um erro depois do primeiro pedaço: o SSE recebe um event: error e a conexão fecha sem o pedaço final - é assim que o HTTP avisa que o corpo foi cortado.
• Marque a rota com Streaming := True: o Sagui precisa disso para rodar a rota numa thread própria, e o Swagger/Postman mostram a rota como text/event-stream.

Fluxo que continua depois da rota (Detach)

procedure TForm1.Assinar(ARequest: TRALRequest; AResponse: TRALResponse);
begin
  FAssinantes.Add(AResponse.Detach); // IRALResponseStream, contado por referência
end;

// em outra thread, quando houver novidade:
for vInt := FAssinantes.Count - 1 downto 0 do
  if not FAssinantes[vInt].SendEvent(vNovidade) then
    FAssinantes.Delete(vInt); // o cliente foi embora

Enquanto um fluxo segue aberto, o servidor manda um comentário SSE (: ping) a cada StreamKeepAlive milissegundos, para proxies não derrubarem a conexão parada. Close termina o fluxo; Active := False fecha todos.

Propriedades do servidor

Propriedade	Padrão	O que é
StreamKeepAlive	15000	ms entre pings de um SSE parado
StreamSendTimeout	30000	ms que um Send espera um cliente que parou de ler
StreamMaxPending	1048576	bytes que um motor de "puxar" (Sagui) guarda antes de o Send esperar
OnStreamClosed	—	um fluxo fechou, com o motivo (terminou, cliente saiu, tempo, servidor parou, erro)

Cliente: receber aos pedaços

procedure TForm1.AoEvento(ASender: TObject; AResponse: TRALResponse;
  const AEvent: TRALSSEEvent; var ACancel: boolean);
begin
  // roda na thread do pedido: sincronize antes de mexer na tela
  TThread.Queue(nil, ...);
  ACancel := FParar; // True fecha a conexão
end;

FClient.Get('progresso', {$IFDEF FPC}@{$ENDIF}AoEvento, vResposta);
vResposta.Free;

No TRALClient	O que é
OnServerEvent	cada evento de uma resposta text/event-stream (Data, Event, Id, Retry)
OnReceiveData	cada pedaço cru, já descomprimido
KeepStreamBody	guarda também o corpo em ResponseText (padrão: não guarda; a memória fica plana)
Get/Post com um dos dois	o tratador vale só para aquela chamada

• Só uma resposta 2xx é entregue aos pedaços; um erro chega inteiro, como sempre.
• Uma chamada que já entregou um pedaço nunca é reenviada (nem para outra BaseURL): entregar o começo duas vezes é pior que falhar.
• ACancel := True fecha a conexão e a chamada falha com rteCancelled.
• RequestTimeout passa a ser o tempo sem dado: um SSE de uma hora não cai, desde que chegue algo (os pings contam).
• gzip e deflate são descomprimidos pedaço a pedaço; por isso, numa chamada em fluxo o cliente só oferece esses dois no Accept-Encoding.

Motores

Motor	Servidor	Cliente
Indy	ao vivo	ao vivo
fpHTTP	ao vivo	ao vivo
mORMot2	ao vivo em smThreads com o mORMot2 que tem BeginChunkedResponse; no fim em smAsync/smHttpSys	ao vivo
Sagui	ao vivo (marque a rota Streaming)	—
CGI	ao vivo no Lazarus; no fim no Delphi	—
netHTTP	—	no fim
UniGUI, MsQuic, Kwik, OkHttp	no fim	no fim

"No fim" quer dizer que funciona com o mesmo código, mas os pedaços chegam todos juntos quando a rota termina. Detach só existe nos motores que enviam ao vivo.

Limitar quantos fluxos ficam abertos

Cada fluxo segura uma conexão (e, na maioria dos motores, uma thread). O TRALConcurrencyPlugin tem MaxStreams: passado o número, um novo fluxo recebe 503 antes de mandar qualquer byte. 0 (padrão) não limita.

Antes de ligar CompressStreams

O TRALCompressPlugin só comprime fluxos com CompressStreams := True (padrão: desligado), em gzip ou deflate, descarregando o compressor a cada Send para o pedaço sair na hora. Antes de ligar:

1. Pedaço pequeno pode crescer. Cada descarga acrescenta uns 5 bytes; pedaços de poucas dezenas de bytes saem maiores. A partir de algumas centenas, e com mensagens parecidas (JSON-RPC), comprime bem.
2. Memória por fluxo aberto. Cerca de 30 KB de compressor enquanto o fluxo durar: mil SSEs abertos são uns 30 MB só disso.
3. Intermediários que seguram resposta comprimida até o fim - antivírus de estação, proxies corporativos, algumas CDNs: o SSE "trava" e chega tudo no fechamento. É o problema mais comum, e não aparece no teste local.
4. Vazamento pelo tamanho (BREACH). Num fluxo que mistura um segredo (token, chave) com texto que um atacante consegue influenciar, o tamanho de cada pedaço comprimido, visível mesmo sob TLS, revela o segredo aos poucos.
5. Só gzip e deflate; zstd e brotli comprimem o corpo inteiro, nunca um fluxo.
6. O que não é problema: um cliente RAL antigo, um navegador e o EventSource leem o resultado normalmente.

Ligue numa rede própria, sem proxy no meio e sem segredo misturado com texto de fora. Não ligue em API pública atrás de CDN ou para clientes em redes corporativas.

Erros comuns

Sintoma	Causa
ERALResponseCommitted	Answer, StatusCode ou AddHeader depois do primeiro Send
os eventos chegam todos juntos	motor que entrega no fim, ou um proxy segurando (veja o item 3 acima)
500 "cannot be encrypted"	a cifra de corpo do RAL não serve para fluxo
503 num fluxo	MaxStreams do TRALConcurrencyPlugin atingido
Detach levanta	o motor não envia ao vivo
no Sagui os pedaços chegam só no fim (e um fluxo grande espera StreamSendTimeout)	a rota não está marcada Streaming

Veja também

• Pedido e resposta
• Cliente
• Qual motor escolher?
• Plugins
• Internos dos motores