Flutter'da LLM Yanıtını Stream Etmek: Dio ile Kendi SSE İstemcinizi Yazmak

SSE kolay, zor olan aradaki katmanlar

Model yanıt üretiyor, kelimeler ekranda birer birer beliriyor ve bir yerde ��alışıyor yazıyor. Bir sonraki denemede aynı cümle tertemiz geliyor. Başka bir gün uygulama yirmi saniye boş ekran gösteriyor, sonra yanıtın tamamı tek seferde düşüyor. İki durumda da suç protokolde değil. Sorun ya istemcinin baytları nasıl okuduğunda ya da sunucuyla telefon arasındaki bir katmanda.

LayeredMindAI Mobile'da sohbet yanıtları Server-Sent Events ile stream ediliyor ve istemci hazır bir SSE paketi değil, doğrudan Dio üzerine yazıldı. Aşağıda önce protokolü, sonra Dart'ta bağımsız bir parser'ı, en son da mobil ağın ve sunucunun ona neler yaptığını anlatıyorum. Örnekler Dart 3.10, Dio 5.9.2 ve küçük bir Node sunucusuyla çalıştırıldı.

Protokol bir sayfaya sığar

SSE, WHATWG HTML standardının 9.2 numaralı bölümünde tanımlı ve kuralları bir solukta sayılır. Stream her zaman UTF-8'dir, başka bir karakter kodlaması seçmenin yolu yoktur. Satırlar CRLF, tek başına LF ya da tek başına CR ile biter. Boş satır olayı dispatch eder, yani tamamlayıp dinleyiciye iletir. İki noktayla başlayan satır yorumdur. Geri kalan her satır alan: değer biçimindedir ve iki noktadan sonraki tek bir boşluk değere dahil edilmez, ikinci boşluk edilir.

Kendi parser'ınızı yazarken en sık atlanan üç kural var. İlki, id değeri olay tamamlanmadan kesinleşmez. Sunucu id: 8 yazıp boş satırı gönderemeden bağlantı koparsa yeniden bağlanırken bildirilecek kimlik hâlâ 7'dir. İkincisi, stream boş satırla bitmeden kapanırsa yarım olay dispatch edilmez, atılır. Üçüncüsü, alan adları büyük-küçük harfe duyarlıdır. Data: ile başlayan satır hiçbir şey ifade etmez.

Yeniden bağlanan istemci son kimliği Last-Event-ID başlığıyla sunucuya bildirir. Tarayıcıdaki EventSource bunu kendiliğinden yapar, 200 dışında bir durum kodu ya da text/event-stream olmayan bir içerik türü görünce de bağlantıyı kalıcı olarak kapatır. Sunucu yeniden bağlanmayı kesmek isterse 204 No Content döner.

Neden hazır bir paket değil

EventSource yalnızca bir URL ve withCredentials seçeneği alır. Yöntem her zaman GET'tir, gövde gönderemez, Authorization başlığı ekleyemezsiniz. LLM sohbeti ise doğası gereği bir POST: prompt, konuşma kimliği, model ayarları ve ekler gövdede gider. GET ile POST arasındaki anlam farkı yeniden bağlanırken de önemli.

Otomatik yeniden bağlanma GET için zararsızdır. Aynı davranışı bir POST'a uygularsanız her kopuşta prompt'u baştan gönderir ve sunucuda ikinci, üçüncü bir üretim başlatırsınız. Bu hem boşa yakılan token hem de ekranda birbirine karışan iki yanıt demek. Yeniden bağlanma, hangi mesaja hangi kimlikten devam edileceğini bilen ve tekrarlanmasında sakınca olmayan ayrı bir istek olmalı.

LayeredMindAI Mobile'da kimlik doğrulama, token yenileme, hata normalleştirme ve loglama zaten Dio interceptor'larında duruyor, SSL pinning de aynı istemcide. Kendi HTTP istemcisini getiren bir SSE paketi bunların hepsinin yanından geçer. Stream aynı Dio istemcisinden geçince süresi dolmuş bir access token stream'de de diğer isteklerdeki gibi ele alınır. Geriye üç iş kalıyor: CancelToken ile iptal, durum kodlarını kendi kurallarınızla yorumlamak ve baytları olaylara çeviren küçük bir parser.

Parser: satırdan olaya

Parser'ı Dio'dan ve Flutter'dan bağımsız tutmak işin yarısı. Satır alır, olay döndürür, saf Dart olduğu için dart test ile saniyeler içinde sınanır.

1import 'dart:convert';2 3class SseEvent {4  const SseEvent(this.type, this.data, this.id);5  final String type;6  final String data;7  final String id;8 9  @override10  String toString() => 'SseEvent($type, ${jsonEncode(data)}, id=$id)';11}12 13/// WHATWG HTML §9.2.6'daki yorumlama kurallarının satır düzeyindeki hâli.14class SseParser {15  final _data = StringBuffer();16  var _type = '';17  var _idBuffer = '';18 19  /// Yalnızca bir olay dağıtıldığında güncellenir. Yeniden bağlanırken20  /// `Last-Event-ID` başlığına bu değer yazılır.21  String lastEventId = '';22 23  /// Sunucu `retry:` gönderdiyse milisaniye cinsinden bekleme süresi.24  int? retryMs;25 26  SseEvent? addLine(String line) {27    if (line.isEmpty) return _dispatch();28    if (line.startsWith(':')) return null; // yorum, heartbeat29 30    final colon = line.indexOf(':');31    final field = colon == -1 ? line : line.substring(0, colon);32    var value = colon == -1 ? '' : line.substring(colon + 1);33    if (value.startsWith(' ')) value = value.substring(1);34 35    switch (field) {36      case 'data':37        _data38          ..write(value)39          ..write('\n');40      case 'event':41        _type = value;42      case 'id':43        if (!value.contains('\u0000')) _idBuffer = value;44      case 'retry':45        if (RegExp(r'^[0-9]+$').hasMatch(value)) retryMs = int.parse(value);46    }47    return null;48  }49 50  SseEvent? _dispatch() {51    lastEventId = _idBuffer;52    if (_data.isEmpty) {53      _type = '';54      return null;55    }56    var data = _data.toString();57    if (data.endsWith('\n')) data = data.substring(0, data.length - 1);58    final event = SseEvent(_type.isEmpty ? 'message' : _type, data, lastEventId);59    _data.clear();60    _type = '';61    return event;62  }63}64 65/// Bayt akışını olay akışına çevirir.66Stream<SseEvent> sseEvents(Stream<List<int>> bytes, [SseParser? parser]) {67  final p = parser ?? SseParser();68  return const LineSplitter()69      .bind(const Utf8Decoder(allowMalformed: true).bind(bytes))70      .expand((line) {71        final event = p.addLine(line);72        return event == null ? const <SseEvent>[] : [event];73      });74}

_idBuffer ile lastEventId ayrımı standarttaki kimlik kuralının karşılığı. Tampon her id: satırında değişir ama dışarıya yalnızca boş satırda, yani olay tamamlanınca yansır. lastEventId dispatch'ten sonra sıfırlanmaz, bu yüzden kendi kimliği olmayan bir olay bir öncekinin kimliğini taşır. Testlerin bir kısmı doğrudan standarttaki örnek stream'lerden geliyor. En işe yarayanı, Türkçe bir satırı mümkün olan her bayt sınırından ikiye bölüp aynı sonucu bekleyen test:

1  test('Türkçe metin her bayt sınırından bölünse de bozulmaz', () async {2    final bytes = b('data: {"t":"Çalışıyor, ğüşİı"}\n\n');3    for (var cut = 1; cut < bytes.length; cut++) {4      final events = await parse([bytes.sublist(0, cut), bytes.sublist(cut)]);5      expect(events.single.data, '{"t":"Çalışıyor, ğüşİı"}', reason: 'kesim $cut');6    }7  });8 9  test('id yalnızca olay tamamlanınca kesinleşir', () async {10    final p = SseParser();11    await parse([b('id: 7\ndata: a\n\nid: 8\ndata: yarım')], p);12    expect(p.lastEventId, '7');13  });

1$ dart test200:00 +11: All tests passed!

Bayt sınırı harf sınırı değildir

Ağ baytları istediği yerden bölünmüş teslim eder. UTF-8'de ç iki bayttır (C3 A7) ve TCP bu iki baytı ayrı parçalara koymakta hiçbir sakınca görmez. Deneme sunucusu her olayı iki yazmaya böldüğünde Dio'nun verdiği 32 parçanın 11'i bir harfin ortasında bitti. Her parçayı ayrı decode eden kod bunu şöyle yaşar:

1import 'dart:convert';2 3void main() {4  final bytes = utf8.encode('data: Şimdi çalışıyor\n\n');5  // Ağ, "ç" harfinin iki baytını iki ayrı parçaya bölmüş olsun.6  final cut = bytes.indexOf(0xC3) + 1;7  final chunks = [bytes.sublist(0, cut), bytes.sublist(cut)];8  print('parçalar: ${chunks.map((c) => c.length).toList()}');9 10  try {11    print(chunks.map((c) => utf8.decode(c)).join());12  } on FormatException catch (e) {13    print('utf8.decode: ${e.message}');14  }15  print('allowMalformed: ${chunks.map((c) => utf8.decode(c, allowMalformed: true)).join().trim()}');16}

1$ dart run bin/naive.dart2parçalar: [14, 14]3utf8.decode: Unfinished UTF-8 octet sequence4allowMalformed: data: Şimdi ��alışıyor

allowMalformed kapalıyken exception, açıkken iki replacement character (U+FFFD) alırsınız. Baytları String.fromCharCodes ile karaktere çevirmek de Türkçe karakter bozulmasının tanıdık görüntüsünü üretir. Çare, decoder'ı parçalara değil stream'in kendisine uygulamak. Utf8Decoder yarım kalan baytları bir sonraki parçaya saklar. Ardından gelen LineSplitter SSE'nin istediğiyle birebir örtüşür: CR, LF ve CRLF'yi satır sonu sayar, bir parçanın sonunda kalan CR'den sonra gelen LF'yi sonraki parçada yutar.

Parser'daki allowMalformed: true bilinçli bir seçim. Standart, stream'in UTF-8 decode algoritmasıyla decode edilmesini ister ve bu algoritma bozuk diziyi hata fırlatmadan replacement character'a çevirir. Aynı algoritma baştaki BOM'u da atar. Dart'ın decoder'ı bunu kendisi yapıyor, testlerden biri bunu da doğruluyor.

İptal neden üç saniye sürer

Parser'ı async* ile yazmak akla gelen ilk yol: await for ile satırları oku, olayları yield et. Bu sürüm bütün testleri geçer. Sorun iptalde ortaya çıkar. Kaynağın üç saniye sustuğu, dinleyicinin birinci saniyede vazgeçtiği bir deneme:

1import 'dart:async';2 3// Ağdan 3 saniye hiç bayt gelmiyor. Dinleyici 1. saniyede vazgeçiyor.4Future<void> measure(String name, Stream<String> Function(Stream<String>) wrap) async {5  final sw = Stopwatch()..start();6  final source = StreamController<String>(7      onCancel: () => print('$name: kaynak ${sw.elapsedMilliseconds} ms\'de kapandı'));8  Timer(const Duration(seconds: 3), () => source.add('data: geç kalan'));9  final sub = wrap(source.stream).listen((_) {});10  await Future<void>.delayed(const Duration(seconds: 1));11  await sub.cancel();12  print('$name: cancel() ${sw.elapsedMilliseconds} ms\'de döndü');13}14 15Stream<String> withAsyncStar(Stream<String> lines) async* {16  await for (final line in lines) {17    yield line.toUpperCase();18  }19}20 21Future<void> main() async {22  await measure('async*', withAsyncStar);23  await measure('expand', (s) => s.expand((line) => [line.toUpperCase()]));24}

1$ dart run bin/iptal.dart2async*: kaynak 3006 ms'de kapandı3async*: cancel() 3008 ms'de döndü4expand: kaynak 1003 ms'de kapandı5expand: cancel() 1003 ms'de döndü

Bir async* generator'ı await for içinde beklerken subscription'ı iptal ederseniz generator kaynaktan bir sonraki olay gelene kadar uyanmaz ve iptal ancak o zaman tamamlanır. Kaynak bir HTTP yanıtıysa ağ sustuğu sürece soket açık kalır. Daha sinsi olanı şu: await for döngüsü bir hata aldığında önce subscription'ı iptal eder, hatayı ancak iptal bitince teslim eder. Stream'e iki saniyelik timeout koyduğunuzda TimeoutException üçüncü saniyede gelir. Aynı dönüşüm expand ile yazılınca iptal anında kaynağa ulaşıyor. Yukarıdaki sseEvents bu yüzden async* değil.

Dio ile bağlamak

1class ChatStream {2  ChatStream(this.dio);3 4  final Dio dio;5  final parser = SseParser();6  CancelToken? _cancel;7  String? messageId;8 9  Stream<SseEvent> start(String prompt) =>10      _open('POST', '/chat', {'prompt': prompt});11 12  /// Kopan ya da arka planda bırakılan akışa kaldığı yerden bağlanır.13  Stream<SseEvent> resume() => _open('GET', '/chat/$messageId/stream', null);14 15  /// Bağlantıyı bırakır ama üretimi durdurmaz.16  void detach() => _cancel?.cancel('detach');17 18  /// Kullanıcı "durdur"a bastı: önce ekranı serbest bırak, sonra sunucuya söyle.19  Future<void> stop() async {20    _cancel?.cancel('stop');21    if (messageId != null) await dio.post<void>('/chat/$messageId/stop');22  }23 24  Stream<SseEvent> _open(String method, String path, Object? body) async* {25    final cancel = _cancel = CancelToken();26    final Response<ResponseBody> res;27    try {28      res = await dio.request<ResponseBody>(29        path,30        data: body,31        cancelToken: cancel,32        options: Options(33          method: method,34          responseType: ResponseType.stream,35          headers: {36            'Accept': 'text/event-stream',37            if (parser.lastEventId.isNotEmpty) 'Last-Event-ID': parser.lastEventId,38          },39        ),40      );41    } on DioException catch (e) {42      final data = e.response?.data;43      if (e.type == DioExceptionType.badResponse && data is ResponseBody) {44        // Akış modunda hata gövdesi de bir akıştır, okumadan göremezsiniz.45        throw ChatHttpException(46            e.response!.statusCode!, await utf8.decodeStream(data.stream));47      }48      rethrow;49    }50 51    final type = res.headers.value(Headers.contentTypeHeader) ?? '';52    if (!type.startsWith('text/event-stream')) {53      cancel.cancel();54      throw ChatHttpException(res.statusCode ?? 0, 'beklenmeyen içerik türü: $type');55    }56    messageId ??= res.headers.value('x-message-id');57    yield* sseEvents(res.data!.stream, parser);58  }59}

Koddan okunmayan üç ayrıntı var. Birincisi, ResponseType.stream ile yapılan istekte sunucu 401 dönerse Dio yine DioException fırlatır, ama e.response.data bir ResponseBody olur ve hata gövdesi de bir stream'dir. Okumazsanız sorunun süresi dolmuş token mı yoksa dolan kota mı olduğunu bilemezsiniz. İkincisi, içerik türü kontrol edilmeli. Yanlış yapılandırılmış bir proxy ya da bakım sayfası 200 ile HTML dönebilir ve parser bunu tek olay üretmeden sessizce yutar. Üçüncüsü, stream'in ortasında kopan bağlantı her zaman DioException olarak gelmiyor. Deneme sunucusu bağlantıyı yarıda kestiğinde Dio'nun IO adaptörü dart:io'nun HttpException'ını olduğu gibi iletti:

1== bağlantı koparsa2[istemci +   44ms] olay id=0 "Yanıt "3[istemci +   70ms] olay id=1 "parça "4[istemci +  163ms] HttpException: HttpException: Connection closed while receiving data, uri = http://127.0.0.1:8787/chat?dropAfter=4&gap=30, 4 olay, lastEventId=35 6== Last-Event-ID ile devam7[istemci +    7ms] olay id=4 "ğ, "8[istemci +   25ms] olay id=5 "ü, "9[istemci +  324ms] olay id=15 "<done>"10[istemci +  325ms] akış bitti, 12 olay: ğ, ü, ş, ı, İ, ö ve ç harfleri de sağlam.<done>

Sunucu kopuştan hemen önce id: 4 olayını yazmıştı ama olay istemciye ulaşmadı, lastEventId 3'te kaldığı için de resume() 4'ten devam etti. Hiçbir parça kaybolmadı, hiçbiri iki kez gelmedi.

Mobilde bağlantı sizin değil

Arka plana geçiş

iOS arka plana geçen uygulamayı kısa süre içinde askıya alır ve askıdaki uygulama soketten okuyamaz. Android'de süreç bir süre daha çalışabilir, ama Doze ve App Standby kısıtları devreye girince ağ erişimi kesilir, süreç de her an sonlandırılabilir. Kırk saniyelik bir yanıtın ortasında kullanıcının bir bildirime bakması yeter. Bağlantıyı açık bırakıp şansınızı deneyebilirsiniz ya da Flutter 3.13 ile gelen AppLifecycleListener ile uygulama gizlendiğinde bağlantıyı kendiniz bırakıp ön plana dönünce kaldığınız kimlikten devam edebilirsiniz. Ben ikincisini tercih ederim, çünkü o zaman kopuş ile arka plan aynı koda iner: detach() ve resume().

1== arka plana geçiş2[istemci +   39ms] olay id=0 "Yanıt "3[istemci +  136ms] olay id=1 "parça "4[istemci +  136ms] uygulama arka plana geçti, bağlantı bırakıldı5[istemci +  139ms] DioException.cancel, 2 olay, lastEventId=16 7== 2 sn sonra ön plana dönüş8[istemci +    8ms] olay id=2 "parça "9[istemci +   14ms] olay id=3 "geliyor: "10[istemci +   84ms] olay id=15 "<done>"11[istemci +   86ms] akış bitti, 14 olay: parça geliyor: ğ, ü, ş, ı, İ, ö ve ç harfleri de sağlam.<done>

Bunun çalışması için sunucunun üretimi bağlantıya değil mesaja bağlaması şart. Deneme sunucusunda her istek bir iş başlatıyor ve bağlantı o işin yalnızca dinleyicisi. Bağlantı kopunca dinleyici ayrılıyor ama üretim sürüyor, yeni bağlantı da Last-Event-ID'den sonrasını tekrar oynatıp canlı stream'e katılıyor. Yukarıdaki örnekte iki saniyelik arada üretim bitmişti, dönüşte kalan on dört olay bir anda geldi.

Ağ değişimi

Wi-Fi'den hücresel ağa geçişte açık TCP bağlantısı yeni arayüze taşınmaz. Eski bağlantı kimi zaman hemen hata verir, kimi zaman tek kelime etmeden susar. Susan bağlantıyı yakalamanın güvenilir tek yolu sessizliği ölçmek. connectivity_plus gibi bir paketin bildirdiği ağ değişimini zaman aşımını beklemeden yeniden bağlanmak için tetik olarak kullanabilirsiniz, ama ağ değişmeden de bağlantı ölebilir.

Zaman aşımı

Dio'nun belgeleri receiveTimeout için iki şey söylüyor: bağlantı kurulup ilk yanıt baytları gelene kadar geçen süre ve aktarım sırasında iki bayt olayı arasındaki süre. Toplam süreden söz etmiyor. Stream'de bu tam aranan şey, çünkü üç dakika süren bir yanıt parçalar arasında uzun sessizlik olmadıkça kısa bir receiveTimeout ile de sorunsuz akar. Ölçünce yine de bir boşluk çıktı:

Dio 5.9.2'nin stream'i işleyen kodu sayacı ilk veri parçasıyla kuruyor. Sunucu başlıkları hemen gönderip ilk token'ı bekletirse, ki model düşünürken olan tam olarak budur, aradaki süreyi hiçbir şey saymıyor. Bu yüzden iki ayrı sınır kullanmayı doğru buluyorum. receiveTimeout bağlantının canlı olup olmadığını ölçer ve heartbeat aralığının birkaç katı olur, olay stream'ine uygulanan Stream.timeout ise modelin cevap vermediğine karar verir. Yorum satırları olay üretmediği için heartbeat bu ikinci sayacı sıfırlamaz, olması gereken de bu.

Kullanıcı durdurduğunda

Durdur düğmesi iki iş yapar ve sıraları önemli. Önce CancelToken ile bağlantıyı bırakın ki arayüz hemen serbest kalsın, sonra sunucuya ayrı bir istekle üretimi kesmesini söyleyin. Yalnızca bağlantıyı kesmek yetmez, çünkü bu tasarımda kopuş bilerek üretimi durdurmuyor. Fark sunucu logunda iki satırla görünüyor:

1[sunucu +14401ms] m3: dinleyici ayrıldı, üretim sürüyor2[sunucu +14403ms] POST /chat/m3/stop last-event-id=-3[sunucu +14403ms] m3: kullanıcı durdurdu, üretim kesildi

İptal edilen stream DioException.cancel ile biter. Bunu arayüzde hata olarak göstermeyin. Kullanıcının kendi isteği.

Sunucu ve nginx: tampon nerede

Sunucunun ilk işi başlıkları beklemeden göndermek: Content-Type: text/event-stream, Cache-Control: no-cache ve arkada nginx varsa X-Accel-Buffering: no. Node'da res.flushHeaders() başlıkları ilk token'ı beklemeden yollar. Aşağıdaki sunucu bu yazıdaki tasarımın özeti: üretim mesaja bağlı, bağlantı yalnızca dinleyici, durdurma ayrı bir istek.

1import http from 'node:http';2 3const jobs = new Map(); // mesaj kimliği -> { events, done, listeners }4 5function sse(req, res, job) {6  res.writeHead(200, {7    'Content-Type': 'text/event-stream',8    'Cache-Control': 'no-cache',9    'X-Accel-Buffering': 'no',10    'X-Message-Id': job.id,11  });12  res.flushHeaders();13  const send = (ev) => {14    // Her olay tek write: heartbeat iki yazmanın arasına giremez.15    res.write(`id: ${ev.id}\n${ev.event ? `event: ${ev.event}\n` : ''}data: ${JSON.stringify(ev.data)}\n\n`);16    if (ev.event) res.end();17  };18  const last = Number(req.headers['last-event-id'] ?? -1);19  job.events.filter((ev) => ev.id > last).forEach(send);20  if (job.done) return;21  job.listeners.add(send);22  const hb = setInterval(() => res.write(': ping\n\n'), 15_000);23  res.on('close', () => { // req.on('close') değil24    clearInterval(hb);25    job.listeners.delete(send); // üretim sürer, yalnızca dinleyici gider26  });27}28 29function startJob(tokens) {30  const job = { id: `m${jobs.size + 1}`, events: [], done: false, listeners: new Set() };31  const emit = (ev) => { job.events.push(ev); job.listeners.forEach((l) => l(ev)); };32  const timers = tokens.map((t, i) => setTimeout(() => emit({ id: i, data: { t } }), i * 100));33  timers.push(setTimeout(() => { job.done = true; emit({ id: tokens.length, event: 'done', data: {} }); },34    tokens.length * 100));35  job.stop = () => {36    timers.forEach(clearTimeout); // gerçek sunucuda: LLM isteğini AbortController ile kes37    job.done = true;38    emit({ id: job.events.length, event: 'stopped', data: {} });39  };40  jobs.set(job.id, job);41  return job;42}43 44http.createServer(async (req, res) => {45  let body = '';46  for await (const chunk of req) body += chunk;47  const [, id, action] = req.url.match(/^\/chat\/(m\d+)\/(stream|stop)$/) ?? [];48  if (req.method === 'POST' && req.url === '/chat') {49    return sse(req, res, startJob(['Merhaba, ', 'ğüşıöç ', 'tamam.']));50  }51  if (jobs.has(id) && action === 'stream') return sse(req, res, jobs.get(id));52  if (jobs.has(id) && action === 'stop') jobs.get(id).stop();53  res.writeHead(jobs.has(id) ? 204 : 404).end();54}).listen(8787);

nginx'te proxy_buffering varsayılan olarak açık. Belgelere göre açıkken yanıt tamponlara alınıyor, kapalıyken istemciye eşzamanlı aktarılıyor ve aynı ayar yanıt başlığındaki X-Accel-Buffering ile yanıt başına değiştirilebiliyor. nginx 1.29.8 ile yapılan ölçümde olaylar 150 ms arayla üretildi:

Tamponlama bu ölçümde olayları tek başına geciktirmedi. Sorun gzip ile birleşince çıktı. Sıkıştırıcı iyi oran için veri biriktirir, tamponlu modda nginx ona flush etmesini söylemez ve bütün yanıt en sona yığılır. Tamponlama kapanınca her parça anında sıkıştırılıp gönderiliyor. Aynı tuzak uygulama katmanında da var. Node'da gzip stream'ini her olaydan sonra flush() etmezseniz nginx olmadan da aynı tabloyu görürsünüz. nginx yalnızca text/html yanıtları kendiliğinden sıkıştırır, yani bu tablo biri gzip_types listesine text/event-stream eklediği gün ortaya çıkar. Kuralım basit. SSE yanıtı sıkıştırılmaz, sıkıştırılacaksa her olaydan sonra flush edilir ve X-Accel-Buffering: no her durumda gönderilir.

İki ayar daha önemli. proxy_read_timeout varsayılan olarak 60 saniye ve toplam süreye değil, iki okuma arasındaki süreye uygulanıyor. Standart eski proxy'lere karşı yaklaşık 15 saniyede bir yorum satırı göndermeyi öneriyor, bu aralık nginx'i de rahatlıkla canlı tutar. proxy_ignore_client_abort ise varsayılan olarak kapalı: telefon bağlantıyı bırakınca nginx de sunucuyla bağlantısını kapatır ve sunucunuz res.on('close') olayını alır. Heartbeat'i zamanlayıcıyla yazıyorsanız olay sınırına dikkat edin. Bir olayı birden fazla write ile gönderen kodda zamanlayıcı iki yazmanın arasına düşerse : ping satırı JSON'un ortasına girer. Deneme sunucusunun ilk hâli bu hatayı yaptı ve istemci FormatException: Unterminated string verdi.

Sonuç

Her kopuşu, her iptali ve her zaman aşımını tek bir soruya bağlayın: bu mesaj sunucuda hâlâ üretiliyor mu? Cevap evetse Last-Event-ID ile dönersiniz, hayırsa durdurmayı kullanıcı kendisi istemiştir. Bağlantıyı mesajın kendisi sanan bir tasarım mobil ağda er geç yanıt kaybeder. Parser ise işin en kolay kısmı, yeter ki baytları stream üzerinde decode edin ve iptali bekletmeyin.