sseHandler method

@protected Future<void> sseHandler(HttpResponse res)

Implementation

@protected
Future<void> sseHandler(HttpResponse res) async {
  // Buffered, the response holds chunks back even through `flush()`, and
  // an event would only reach the client once 8 KB had piled up.
  res.bufferOutput = false;
  res.headers.contentType = ContentType.parse('text/event-stream');
  res.headers.add(HttpHeaders.cacheControlHeader, 'no-cache');
  res.headers.add(HttpHeaders.connectionHeader, 'keep-alive');
  res.headers.add(HttpHeaders.transferEncodingHeader, 'chunked');

  final finished = Completer<void>();
  void finish() {
    if (!finished.isCompleted) finished.complete();
  }

  // One write at a time: the response refuses a write while a flush is
  // in progress. A failed write means the client has gone — and so does
  // one that does not finish: once the client has left, dart:io neither
  // fails the flush nor completes `done`, it just never returns.
  var writes = Future<void>.value();
  void send(String text) {
    writes = writes.then((_) {
      res.add(utf8.encode(text));
      return res.flush().timeout(_sseHeartbeat);
    });
    writes.catchError((Object _) => finish());
  }

  final events = (data as Stream<dynamic>).listen(null);
  events
    ..onData((event) {
      try {
        send('data: ${jsonEncode(event)}\n\n');
      } catch (e, stack) {
        Logger.log('SSE event not sent: $e\n$stack', type: Logger.ERROR);
        finish();
        return;
      }
      // Holds the source back while a slow client catches up.
      events.pause(writes.catchError((Object _) {}));
    })
    ..onError((Object e, StackTrace stack) {
      Logger.log('SSE event source failed: $e\n$stack', type: Logger.ERROR);
      finish();
    })
    ..onDone(finish);

  // Keeps proxies from closing a quiet connection, and finds out when
  // the client has left even if no event is due.
  final heartbeat = Timer.periodic(_sseHeartbeat, (_) => send(':\n\n'));
  res.done.then((_) => finish(), onError: (Object _) => finish());

  await finished.future;
  heartbeat.cancel();
  await events.cancel();
  // Bounded for the same reason: a response to a client that has left
  // may never finish closing.
  await writes
      .catchError((Object _) {})
      .then((_) => closeQuietly(res))
      .timeout(_sseHeartbeat, onTimeout: () {});
}