Snippets to show basic use of "Dart Stream API" in case I forget. (written for dart-1.3.0-dev)

broadcast_basic.dart

import 'dart:async';

void main() {

  var data = [1,2,3,4,5];
  var stream = new Stream.fromIterable(data);
  var broadcastStream = stream.asBroadcastStream();

  broadcastStream.listen((value) => print("(3)stream.listen: $value")); 
  broadcastStream.first.then((value) => print("(3)stream.first: $value")); // 1 
  broadcastStream.last.then((value) => print("(3)stream.last: $value")); // 5
  broadcastStream.isEmpty.then((value) => print("(3)stream.isEmpty: $value")); // false
  broadcastStream.length.then((value) => print("(3)stream.length: $value")); // 5

}

broadcast_take.dart

import 'dart:async';

void main() {

  var data = [1,2,3,4,5];
  var stream = new Stream.fromIterable(data);
  var broadcastStream = stream.asBroadcastStream();

  broadcastStream
      .where((value) => value % 2 == 0) // divisible by 2
      .listen((value) => print("where: $value")); // where: 2
                                                  // where: 4
  broadcastStream
      .take(3) // takes only the first three elements
      .listen((value) => print("take: $value")); // take: 1
                                                 // take: 2
                                                 // take: 3

  broadcastStream
      .skip(3)  // skips the first three elements
      .listen((value) => print("skip: $value")); // skip: 4
                                                 // skip: 5

  broadcastStream
      .takeWhile((value) => value < 3) // take while true
      .listen((value) => print("takeWhile: $value")); // takeWhile: 1
                                                      // takeWhile: 2

  broadcastStream
      .skipWhile((value) => value < 3) // skip while true
      .listen((value) => print("skipWhile: $value")); // skipWhile: 4
                                                      // skipWhile: 5

}

broadcast_then.dart

import 'dart:async';

void main() {

  var data = [1,2,3,4,5];
  var stream = new Stream.fromIterable(data);
  var broadcastStream = stream.asBroadcastStream();

  broadcastStream
      .any((value) => value < 5)
      .then((result) => print("Any less than 5?: $result")); // true
    
  broadcastStream
      .every((value) => value < 5)
      .then((result) => print("All less than 5?: $result")); // false
    
  broadcastStream
      .contains(4)
      .then((result) => print("Contains 4?: $result")); // true
  
  broadcastStream
      .singleWhere((value) => value < 2)  // there is only one value less than 2
      .then((value) => print("single value: $value")); 
      // outputs: single value: 1

//  broadcastStream
//      .single  // will fail - there is more than one value in the stream
//      .then((value) => print("single value: $value"));
  
  broadcastStream
      .single  // will fail - there is more than one value in the stream
      .then((value) => print("single value: $value")) 
      .catchError((err) => print("Expected Error: $err")); // catch any error in the then()
      // output: Bad State: More than one element
}

fileread.dart

// import 'dart:async';
import 'dart:io';
import 'dart:convert';

void main() {

  File file = new File("fileread.dart");
  file.openRead()
      .transform(UTF8.decoder) // use a UTF8.decoder
      .listen((String data) => print(data), // output the data
          onError: (error) => print("Error, could not open file"),
          onDone: () => print("Finished reading data"));
}

stream_iterable.dart

import 'dart:async';

void main() {
  
  var data = [1,2,3,4,5]; // some sample data
  var stream = new Stream.fromIterable(data); // create the stream
  
  stream = new Stream.fromIterable([1,2,3,4,5]);
  stream.first.then((value) => print("stream.first: $value"));  // 1

  stream = new Stream.fromIterable([1,2,3,4,5]);
  stream.last.then((value) => print("stream.last: $value"));  // 5  

  stream = new Stream.fromIterable([1,2,3,4,5]);
  stream.isEmpty.then((value) => print("stream.isEmpty: $value"));  // false

  stream = new Stream.fromIterable([1,2,3,4,5]);
  stream.length.then((value) => print("stream.length: $value"));  // 5
  
}

subscription_cancel.dart

import 'dart:async';

void main() {
  var data = [1,2,3,4,5];
  Stream stream = new Stream.fromIterable(data);

  StreamSubscription subscription = stream.listen(null);

  subscription.onData((value) {
    print("listen: $value");
    if (value == 2) subscription.cancel(); // cancel the subscription
  });

  subscription.onError((err) => print("error: $err"));
  subscription.onDone(() => print("done"));  
}

subscription_subscribe.dart

import 'dart:async';

void main() {
  var data = [1,2,3,4,5];
  Stream stream = new Stream.fromIterable(data);

  // setup the handlers through the subscription's handler methods
  var subscription = stream.listen(
      (value) => print("listen: $value"),
      onError: (err) => print("error: $err"),
      onDone: () => print("done"));
}

transformer.dart

import 'dart:async';

void main() {
  
  var data = [1,2,3,4,5];
  Stream stream = new Stream.fromIterable(data);
  
  // define a stream transformer
  var transformer = new StreamTransformer.fromHandlers(handleData: (value, sink) {
    // create two new values from the original value
    sink.add("Message: $value");
    sink.add("Body: $value");
  });
  
  stream.transform(transformer).listen((value) => print("listen: $value"));
  
}