Make StreamSubscription be the active part of a stream.

tests/lib/async/stream_iterator_test.dart

+// Copyright (c) 2011, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+library timer_test;
+
+import 'dart:async';
+import '../../../pkg/unittest/lib/unittest.dart';
+
+main() {
+  // Test that the stream generates the integers 0..9.
+  testTenStream(String name, Stream<int> stream,
+                action(int count, StreamIterator iter)) {
+    test(name, () {
+      Function onDone = expectAsync0((){});
+      StreamIterator<int> iter = new StreamIterator<int>(stream);
+
+      void testNext(int expectedNext) {
+        if (expectedNext < 10) {
+          iter.moveNext().then((v) {
+            expect(v, isTrue);
+            expect(iter.current, equals(expectedNext));
+            testNext(expectedNext + 1);  // May change "current".
+            action(expectedNext, iter);
+          });
+        } else {
+          iter.moveNext().then((v) {
+            expect(v, isFalse);
+            onDone();
+          });
+        }
+      }
+      testNext(0);
+    });
+  }
+
+  testTenStream("fromIterable",
+                new Stream.fromIterable(new Iterable.generate(10, (x) => x)),
+                (count, iter) {});
+
+  testTenStream("periodic",
+                new Stream.periodic(const Duration(milliseconds: 5), (x) => x),
+                (count, iter) { if (count == 9) iter.cancel(); });
+
+  StreamController<int> c = new StreamController<int>();
+  c.add(0);
+  testTenStream("controller", c.stream,
+                (count, iter) {
+                    if (count < 9) {
+                      c.add(count + 1);
+                    } else {
+                      c.close();
+                    }
+                });
+}