I’m trying to make a Stream modifier function which adds ‘latency’ to all items going through it:

extension SE<T> on Stream<T> {
  Stream<T> delayed(Duration latency) {
    return this.transform(StreamTransformer<T, T>.fromHandlers(
      handleData: (T element, EventSink<T> sink) {
        Future.delayed(latency).then((_) => sink.add(element));
      }
    ));
  }
}

void main() async {
  final s1 = Stream.periodic(Duration(seconds: 1), (i) => i);
  final s2 = Stream.periodic(Duration(seconds: 1), (i) => i).delayed(Duration(milliseconds: 500));
  
  s1.forEach((i) => print('$i'));
  s2.forEach((i) => print('  $i'));  
}

From the code above, you would expect it to print alternating increasing integers with 500ms between each line:

0
  0
1
  1
2
  2
...

However this is really not the case, as seen in this dartpad

Does anyone know why this is happening? Perhaps my current understanding of how async and Streams works in Dart isn’t enough.

Cheers! Dan