encodeSse function

Stream<List<int>> encodeSse(Stream<ServerSentEvent> events, { int heartbeatSeconds = 0, })

Encodes events into the text/event-stream wire format. A positive heartbeatSeconds interleaves comment lines (: keep-alive) so proxies keep the connection open — and so a dead client is detected on the next write.

Implementation

Stream<List<int>> encodeSse(
  Stream<ServerSentEvent> events, {
  int heartbeatSeconds = 0,
}) {
  final controller = StreamController<List<int>>();
  Timer? heartbeat;
  late StreamSubscription<ServerSentEvent> subscription;

  void startHeartbeat() {
    if (heartbeatSeconds <= 0) return;
    heartbeat = Timer.periodic(Duration(seconds: heartbeatSeconds), (_) {
      if (!controller.isClosed) {
        controller.add(utf8.encode(': keep-alive\n\n'));
      }
    });
  }

  controller.onListen = () {
    startHeartbeat();
    subscription = events.listen(
      (event) => controller.add(utf8.encode(encodeSseEvent(event))),
      onError: (Object error, StackTrace stackTrace) {
        // A source error terminates the stream: cancel the heartbeat (it
        // would otherwise keep writing into the closed connection forever)
        // and close after forwarding.
        heartbeat?.cancel();
        unawaited(subscription.cancel());
        controller.addError(error, stackTrace);
        controller.close();
      },
      onDone: controller.close,
    );
  };
  controller.onCancel = () async {
    heartbeat?.cancel();
    await subscription.cancel();
  };
  return controller.stream;
}