🎓 A1Learner
🔍
0 of 24 lessons completed

Dart Project Event Pipeline

An event pipeline moves events through a stream in stages. Each stage in this project filters, counts, classifies, or collects the events.

Events enter through a controller

Each event is a record pairing a name with a score. The events added before the loop are buffered, and the loop prints them in order.

import 'dart:async';

Future<void> main() async {
  var controller = StreamController<(String, int)>();
  controller.add(('login', 5));
  controller.add(('click', 12));
  controller.close();
  await for (var (name, score) in controller.stream) {
    print('$name: $score');
  }
}

A broadcast listener counts high events

The broadcast controller has sync set to true, so the listener counts as each event is added. Two of the three scores reach 10.

import 'dart:async';

Future<void> main() async {
  var controller = StreamController<(String, int)>.broadcast(sync: true);
  var high = 0;
  var sub = controller.stream.listen((event) {
    if (event.$2 >= 10) {
      high++;
    }
  });
  controller.add(('login', 5));
  controller.add(('click', 12));
  controller.add(('buy', 25));
  await controller.close();
  await sub.cancel();
  print('high: $high');
}

Where keeps and map reshapes

The where call keeps the events with a score of at least 10. The map call then writes each kept event as a name with its score in brackets.

import 'dart:async';

Future<void> main() async {
  var events = [('login', 5), ('click', 12), ('buy', 25)];
  var kept = Stream.fromIterable(events)
      .where((event) => event.$2 >= 10)
      .map((event) => '${event.$1} (${event.$2})');
  await for (var line in kept) {
    print(line);
  }
}

A switch expression classifies an event

The classify function reads the name and score out of the record with a pattern. A score of 20 or more is high, a score of at least 10 is medium, and any other score is low.

String classify((String, int) event) {
  return switch (event) {
    (var name, var score) when score >= 20 => '$name: high',
    (var name, var score) when score >= 10 => '$name: medium',
    (var name, _) => '$name: low',
  };
}

void main() {
  print(classify(('login', 5)));
  print(classify(('click', 12)));
  print(classify(('buy', 25)));
}

Example

import 'dart:async';

extension type Threshold(int value) {
  bool passes(int score) {
    return score >= value;
  }
}

String classify((String, int) event) {
  return switch (event) {
    (var name, var score) when score >= 20 => '$name: high',
    (var name, var score) when score >= 10 => '$name: medium',
    (var name, _) => '$name: low',
  };
}

Future<void> main() async {
  var threshold = Threshold(10);
  var controller = StreamController<(String, int)>.broadcast(sync: true);
  var highCount = 0;
  var counting = controller.stream.listen((event) {
    if (threshold.passes(event.$2)) {
      highCount++;
    }
  });
  var resultsFuture = controller.stream
      .where((event) => threshold.passes(event.$2))
      .map((event) => classify(event))
      .toList();
  controller.add(('login', 5));
  controller.add(('click', 12));
  controller.add(('buy', 25));
  controller.add(('view', 8));
  controller.add(('share', 15));
  await controller.close();
  var results = await resultsFuture;
  await counting.cancel();
  print('high: $highCount');
  for (var result in results) {
    print(result);
  }
}

Key points

  • Events travel as records that pair a name with a score.
  • A StreamController is the entrance where events are pushed into the pipeline.
  • A broadcast stream lets a counting listener and a collecting listener hear the same events.
  • A Threshold extension type gives the score limit a named type with a passes test.
  • The where stage keeps only the events that pass the threshold.
  • The map stage turns each kept event into a classified line with a switch expression.
  • Collecting the transformed stream into a list gathers every result before printing.

In this example, five events are pushed into a broadcast controller. The counting listener finds three events at or above the threshold of 10. The where and map stages keep those three events and classify each one, and the collected list prints click as medium, buy as high, and share as medium.

Note

The whole pipeline prints awaited results only, so its output is the same on every run.

Try it Yourself

Edit the code, then press Run.

Running is simulated · DartPad

Exercise

In the example pipeline, what happens to the login event, which has a score of 5?