Simple stream examples:
Dart
x
19
19
1
import 'dart:async';2
3
void main() {4
var streamController = StreamController();5
6
streamController.stream7
.listen(8
(data) => print('Incoming data: $data'),9
onError: (err) => print('An error: $err'),10
onDone: () => print('Mission complete!'),11
cancelOnError: false,12
);13
14
streamController.sink.add('Some nice data');15
streamController.sink.add('...more data');16
streamController.sink.addError('Houston, we have a problem!');17
streamController.sink.add('...much more data');18
streamController.sink.close();19
}Dart
1
29
29
1
import 'dart:async';2
import 'dart:math';3
4
main() {5
StreamController<double> controller = StreamController<double>();6
StreamSubscription<double> streamSubscription;7
8
// The controllers stream can be accessed through the stream property9
Stream stream = controller.stream;10
11
// Add a value to the streams sink12
Random rand = Random();13
Timer.periodic(Duration(seconds: 1), (_) {14
controller.sink.add(rand.nextDouble() * 10);15
});16
17
// Add a subsciption18
streamSubscription = stream19
.listen((value) {20
print('Value from controller: $value');21
});22
23
24
// Cancel a subscription25
Timer(Duration(seconds: 10), () {26
streamSubscription.cancel();27
print('subscription.cancel()');28
});29
}A stream which continuously sent random values:
Dart
1
11
11
1
//Defining the stream which roll a dice every half second2
Stream<int> getRandomValues() async* {3
var random = Random();4
while (true) {5
await Future.delayed(Duration(milliseconds: 500));6
yield random.nextInt(6) + 1;7
}8
}9
10
//Listen to this function11
getRandomValues().listen((data) => print(data));- The function return a Stream. That means we can to subscribe to the stream.
- The
async*means run the function asynchronously. Execution will continue even after “returning” a value. -
yieldis a return function which doesn’t exit the function. Instead it continues executing the rest of the code after yield.
Stream examples with a StreamTransformer:
A Stream Transformer allows us to perform data transformations on a Stream. These transformations are then pushed back into the Stream to be received by all the listeners defined for that particular stream.
Example 1:
Dart
1
25
25
1
import 'dart:async';2
3
void main() async {4
final StreamController<String> controller = StreamController<String>.broadcast();5
6
final StreamTransformer transformer =7
StreamTransformer<String, String>.fromHandlers(8
handleData: (data, EventSink sink) {9
sink.add(data);10
});11
12
final Stream<String> stream = controller.stream;13
14
stream15
.map((value) => value)16
.transform(transformer)17
.listen((data) {18
print('listen: $data');19
}, onError: (err) => print(err));20
21
controller.add('foo');22
controller.add('baa');23
24
controller.close();25
}This transformer is made as the type StreamTransformer<String, String> which means it takes a String and convert it into a String. But because map cannot determine the return type of value it most assume that value are dynamic. So map returns a Stream<dynamic> which are not compatible with the transformer.
We can tell map what type it is going to return on runtime by writing .map<String>((value) => value). But in general we should not throw away the generic part of types.
Example 2:
Dart
1
35
35
1
import 'dart:async';2
3
void main() {4
var streamController = StreamController<int>();5
6
final streamTransformer = StreamTransformer<int, int>.fromHandlers(7
handleData: (int data, EventSink sink) {8
sink.add(data * 2);9
},10
handleError: (Object error, StackTrace stacktrace, EventSink sink) {11
sink.addError('Something went wrong: $error');12
}, 13
handleDone: (EventSink sink) {14
print('Controller mission complete!');15
sink.close();16
}17
);18
19
streamController.stream20
.transform(streamTransformer)21
.listen(22
(data) => print('Data $data'),23
onError: (err) => print('Got an error! $err'),24
onDone: () => print('Stream mission complete!'),25
cancelOnError: false,26
);27
28
streamController.sink.add(1);29
streamController.sink.add(2);30
streamController.sink.add(4);31
streamController.sink.addError('Houston, we have a problem!');32
streamController.sink.add(4);33
34
streamController.sink.close();35
}Dart
1
39
39
1
import 'dart:async';2
import 'dart:math';3
4
main() {5
final controller = StreamController<int>();6
7
var subscription;8
9
Random rand = Random();10
11
Timer.periodic(Duration(seconds: 1), (_) {12
controller.add(rand.nextInt(100));13
});14
15
Stream<int> output = controller.stream;16
17
final trans = StreamTransformer<int, dynamic>.fromHandlers(18
handleData: (number, sink) {19
if (number < 75) {20
sink.add(number);21
} else {22
sink.addError('Oh no. The $number is big,');23
}24
}25
);26
27
subscription = output28
.transform(trans)29
.listen(30
(data) => print(data),31
onError: (err) => print(err)32
);33
34
35
Timer(Duration(seconds: 10), () {36
subscription.cancel();37
print('subscription.cancel()');38
});39
}https://medium.com/flutter-community/flutter-stream-basics-for-beginners-eda23e44e32f