bind method

@override Stream<String> bind(Stream<String> stream)

Transforms the provided stream.

Returns a new stream with events that are computed from events of the provided stream.

The StreamTransformer interface is completely generic, so it cannot say what subclasses do. Each StreamTransformer should document clearly how it transforms the stream (on the class or variable used to access the transformer), as well as any differences from the following typical behavior:

• When the returned stream is listened to, it starts listening to the input stream.
• Subscriptions of the returned stream forward (in a reasonable time) a StreamSubscription.pause call to the subscription of the input stream.
• Similarly, canceling a subscription of the returned stream eventually (in reasonable time) cancels the subscription of the input stream.

"Reasonable time" depends on the transformer and stream. Some transformers, like a "timeout" transformer, might make these operations depend on a duration. Others might not delay them at all, or just by a microtask.

Transformers are free to handle errors in any way. A transformer implementation may choose to propagate errors, or convert them to other events, or ignore them completely, but if errors are ignored, it should be documented explicitly.

Implementation

@override
Stream<String> bind(Stream<String> stream) {
  final framesPerWindow = math.max(
    1,
    (window.inMicroseconds / tick.inMicroseconds).ceil(),
  );
  // The backlog correction must act over a longer horizon than the lead it
  // is correcting toward, otherwise `arrival + (backlog - arrival·F)/F`
  // collapses to `backlog/F` — the feed-forward term cancels exactly and
  // the smoother degenerates into a fixed-interval backlog drainer.
  final correctionFrames = 2 * framesPerWindow;
  // Per-tick EMA coefficient for a `smoothing` time constant.
  final alpha = smoothing <= Duration.zero
      ? 1.0
      : 1 - math.exp(-tick.inMicroseconds / smoothing.inMicroseconds);

  final queue = Queue<String>();
  StreamSubscription<String>? upstream;
  Timer? timer;
  var sourceDone = false;
  // Units queued since the last tick; the raw signal for the rate estimate.
  var arrivedSinceTick = 0;
  // Units per tick, exponentially smoothed.
  var arrivalRate = 0.0;
  var releaseRate = 0.0;
  // Fractional carry, so a rate below one unit per tick still makes
  // progress instead of rounding away to nothing.
  var credit = 0.0;
  // Ticks left to fully drain once the source is done. Counting it down
  // bounds the tail at `window` without jolting the pace at the handover.
  var tailFrames = framesPerWindow;
  late StreamController<String> controller;

  void updateRates() {
    arrivalRate += alpha * (arrivedSinceTick - arrivalRate);
    arrivedSinceTick = 0;

    final lead = arrivalRate * framesPerWindow;
    final target = arrivalRate + (queue.length - lead) / correctionFrames;
    releaseRate += alpha * (target - releaseRate);
    if (releaseRate < 0) releaseRate = 0;

    if (sourceDone && queue.isNotEmpty) {
      tailFrames = math.max(1, tailFrames - 1);
      releaseRate = math.max(releaseRate, queue.length / tailFrames);
    }
  }

  void emitBudget() {
    if (queue.isEmpty) {
      credit = 0;
      return;
    }
    credit += releaseRate;
    final budget = math.min(credit.floor(), queue.length);
    if (budget <= 0) return;
    credit -= budget;
    final out = StringBuffer();
    for (var i = 0; i < budget; i++) {
      out.write(queue.removeFirst());
    }
    if (queue.isEmpty) credit = 0;
    controller.add(out.toString());
  }

  void stopTimer() {
    timer?.cancel();
    timer = null;
  }

  void onTick(Timer _) {
    updateRates();
    emitBudget();
    if (queue.isEmpty && sourceDone) {
      stopTimer();
      controller.close();
    }
  }

  void onData(String chunk) {
    if (chunk.isEmpty) return;
    if (atomic(chunk)) {
      queue.add(chunk);
      arrivedSinceTick++;
    } else {
      for (final grapheme in chunk.characters) {
        queue.add(grapheme);
        arrivedSinceTick++;
      }
    }
  }

  void onDone() {
    sourceDone = true;
    if (queue.isEmpty) {
      stopTimer();
      controller.close();
    }
  }

  controller = StreamController<String>(
    onListen: () {
      // The timer runs for the whole subscription, not just while the
      // queue is non-empty: the rate estimate has to keep decaying through
      // a decode pause, or the first chunk after the pause would be paced
      // against a stale, much higher rate and dump.
      timer = Timer.periodic(tick, onTick);
      upstream = stream.listen(
        onData,
        onError: (Object error, StackTrace stackTrace) {
          // Buffered text is dropped — it belongs to a run that failed —
          // but the timer keeps running: an error need not end the stream,
          // and stopping it here would strand anything sent afterwards.
          queue.clear();
          credit = 0;
          arrivalRate = 0;
          releaseRate = 0;
          controller.addError(error, stackTrace);
        },
        onDone: onDone,
      );
    },
    onCancel: () {
      stopTimer();
      final sub = upstream;
      upstream = null;
      return sub?.cancel();
    },
  );

  return controller.stream;
}