Dart: how to append a transformer to an existing stream?

Question

I'm looking a for a way to programmatically add a transformer to an existing stream that's already being listen to.

Example:

Stream numbers = new Stream.fromIterable([0,1,2,3]);

numbers.listen((number) => print(number));

Now in response to some UI event, I'd like to modify this stream by adding a mapping transformer, as if I originally wrote:

numbers.where((number) => number % 2 == 0);

All existing listeners should from now own only receive even numbers, without interruption. How can this be done?

Answer 1

Instead of thinking about it like "how do I dynamically insert a transformer into a stream", one possible way is to think about it like "how do I dynamically control a transformer that I already injected".

Here's an example of using a StreamTransformer:

var onlySendEvenNumbers = false; // controlled by some UI event handler

var originalStream = makeStreamOfStuff();

originalStream = originalStream.transform(new StreamTransformer.fromHandlers(
  handleData: (int value, EventSink<int> sink) {
    if (onlySendEvenNumber) {
      if (value.isEven) {
        sink.add(value);
      }
    } else {
      sink.add(value);
    }
}));

originalStream.listen(print);  // listen on events like normal

Answer 2

One way I can think of doing that is filtering the Stream with a function that calls another function:

var filter = (n) => true;
Stream numbers = new String.fromIterable([0, 1, 2, 3]).where((n) => filter(n));

Then, when you want to change the filtering:

filter = (n) => n % 2 == 0;

A concrete example:

import 'dart:async';

main() {
  var filter = (n) => true;

  Stream numbers = new Stream.periodic(new Duration(seconds: 1), (n) => n)
      .where((n) => filter(n));

  numbers.listen((n) => print(n));

  new Future.delayed(new Duration(seconds: 4)).then((_) {
    filter = (n) => n % 2 == 0;
  });
}

This will print:

0
1
2
3
4
6
8
10
12

And so on, for even numbers only, after 4 seconds.