sdk/tests/lib/async/stream_event_transformed_test.dart

// Copyright (c) 2013, 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.

import 'package:expect/expect.dart';
import 'package:async_helper/async_helper.dart';
import 'dart:async';
import 'event_helper.dart';

class DecrementingTransformerSink implements EventSink {
  final outSink;
  DecrementingTransformerSink(this.outSink);

  void add(int i) => outSink.add(i - 1);
  void addError(int e, [st]) => outSink.addError(e - 1, st);
  void close() => outSink.close();
}

class FutureWaitingTransformerSink implements EventSink {
  final outSink;
  final closeFuture;
  FutureWaitingTransformerSink(this.outSink, this.closeFuture);

  void add(Future future) {
    future.then(outSink.add);
  }

  void addError(Future e, [st]) {
    e.then((val) {
      outSink.addError(val, st);
    });
  }

  void close() {
    closeFuture.whenComplete(outSink.close);
  }
}

class ZoneTransformerSink implements EventSink {
  final outSink;
  ZoneTransformerSink(this.outSink);

  void add(_) {
    outSink.add(Zone.current);
  }

  void addError(_, [st]) {
    outSink.add(Zone.current);
  }

  void close() {
    outSink.add(Zone.current);
    outSink.close();
  }
}

class TypeChangingSink implements EventSink<int> {
  final EventSink<String> outSink;
  TypeChangingSink(this.outSink);

  void add(int data) {
    outSink.add(data.toString());
  }

  void addError(error, [st]) {
    outSink.addError(error, st);
  }

  void close() {
    outSink.close();
  }
}

class SinkTransformer<S, T> implements StreamTransformer<S, T> {
  final Function sinkMapper;
  SinkTransformer(this.sinkMapper);

  Stream<T> bind(Stream<S> stream) {
    return new Stream<T>.eventTransformed(stream, sinkMapper);
  }
}

get currentStackTrace {
  try {
    throw 0;
  } catch (e, st) {
    return st;
  }
}

// In most cases the callback will be 'asyncEnd'. Errors are reported
// asynchronously. We want to give them time to surface before reporting
// asynchronous tests as done.
void delayCycles(callback, int nbCycles) {
  if (nbCycles == 0) {
    callback();
    return;
  }
  Timer.run(() {
    delayCycles(callback, nbCycles - 1);
  });
}

main() {
  {
    // Simple test: use the SinkTransformer (using the Stream.eventTransformed
    // constructor) to transform a sequence of numbers. This is basically
    // similar to a map.
    asyncStart();
    new Stream.fromIterable([1, 2, 3])
        .transform(new SinkTransformer(
            (sink) => new DecrementingTransformerSink(sink)))
        .toList()
        .then((list) {
      Expect.listEquals([0, 1, 2], list);
      asyncEnd();
    });
  }

  {
    // Similar test as above: but this time also transform errors. Also
    // checks that the stack trace is correctly passed through.
    asyncStart();
    var controller;
    var events = [];
    var stackTrace = currentStackTrace;
    controller = new StreamController(onListen: () {
      controller.add(499);
      controller.addError(42, stackTrace);
      controller.close();
    });
    controller.stream
        .transform(new SinkTransformer(
            (sink) => new DecrementingTransformerSink(sink)))
        .listen((data) {
      events.add(data);
    }, onError: (e, st) {
      events.add(e);
      events.add(st);
    }, onDone: () {
      Expect.listEquals([498, 41, stackTrace], events);
      asyncEnd();
    });
  }

  {
    // Test that the output sink of the transformer can be used asynchronously.
    asyncStart();
    var controller;
    var events = [];
    var stackTrace = currentStackTrace;
    var completer1 = new Completer();
    var completer2 = new Completer();
    var completer3 = new Completer();
    var closeCompleter = new Completer();
    controller = new StreamController(onListen: () {
      controller.add(completer1.future);
      controller.addError(completer2.future, stackTrace);
      controller.add(completer3.future);
      controller.close();
    });
    controller.stream
        .transform(new SinkTransformer((sink) =>
            new FutureWaitingTransformerSink(sink, closeCompleter.future)))
        .listen((data) {
      events.add(data);
    }, onError: (e, st) {
      events.add(e);
      events.add(st);
    }, onDone: () {
      Expect.listEquals(["error2", stackTrace, "future3", "future1"], events);
      asyncEnd();
    });
    Timer.run(() {
      completer2.complete("error2");
      Timer.run(() {
        completer3.complete("future3");
        Timer.run(() {
          completer1.complete("future1");
          scheduleMicrotask(closeCompleter.complete);
        });
      });
    });
  }

  {
    // Test that the output sink of the transformer can be used asynchronously
    // and that events are paused if necessary.
    asyncStart();
    var controller;
    var events = [];
    var stackTrace = currentStackTrace;
    var completer1 = new Completer.sync();
    var completer2 = new Completer.sync();
    var completer3 = new Completer.sync();
    var closeCompleter = new Completer();
    controller = new StreamController(onListen: () {
      controller.add(completer1.future);
      controller.addError(completer2.future, stackTrace);
      controller.add(completer3.future);
      controller.close();
    });
    var subscription;
    completer1.future.then((_) {
      Expect.isTrue(subscription.isPaused);
    });
    completer2.future.then((_) {
      Expect.isTrue(subscription.isPaused);
    });
    completer3.future.then((_) {
      Expect.isTrue(subscription.isPaused);
    });
    subscription = controller.stream
        .transform(new SinkTransformer((sink) =>
            new FutureWaitingTransformerSink(sink, closeCompleter.future)))
        .listen((data) {
      Expect.isFalse(subscription.isPaused);
      events.add(data);
    }, onError: (e, st) {
      events.add(e);
      events.add(st);
    }, onDone: () {
      Expect.listEquals(["error2", stackTrace, "future3", "future1"], events);
      asyncEnd();
    });
    Timer.run(() {
      subscription.pause();
      completer2.complete("error2");
      Timer.run(() {
        subscription.resume();
        Timer.run(() {
          Expect.listEquals(["error2", stackTrace], events);
          subscription.pause();
          completer3.complete("future3");
          Timer.run(() {
            subscription.resume();
            Timer.run(() {
              Expect.listEquals(["error2", stackTrace, "future3"], events);
              subscription.pause();
              completer1.complete("future1");
              subscription.resume();
              scheduleMicrotask(closeCompleter.complete);
            });
          });
        });
      });
    });
  }

  {
    // Test that the output sink of the transformer reports errors when the
    // stream is already closed.
    asyncStart();
    var controller;
    var events = [];
    var stackTrace = currentStackTrace;
    var completer1 = new Completer();
    var completer2 = new Completer();
    var completer3 = new Completer();
    var closeCompleter = new Completer();
    controller = new StreamController(onListen: () {
      controller.add(completer1.future);
      controller.addError(completer2.future, stackTrace);
      controller.add(completer3.future);
      controller.close();
    });

    bool streamIsDone = false;
    int errorCount = 0;
    runZoned(() {
      controller.stream
          .transform(new SinkTransformer((sink) =>
              new FutureWaitingTransformerSink(sink, closeCompleter.future)))
          .listen((data) {
        events.add(data);
      }, onError: (e, st) {
        events.add(e);
        events.add(st);
      }, onDone: () {
        Expect.listEquals([], events);
        streamIsDone = true;
      });
    }, onError: (e) {
      Expect.isTrue(e is StateError);
      errorCount++;
    });
    closeCompleter.complete();
    Timer.run(() {
      Expect.isTrue(streamIsDone);
      // Each of the delayed completions should trigger an unhandled error
      // in the zone the stream was listened to.
      Timer.run(() {
        completer1.complete(499);
      });
      Timer.run(() {
        completer2.complete(42);
      });
      Timer.run(() {
        completer3.complete(99);
      });
      delayCycles(() {
        Expect.equals(3, errorCount);
        asyncEnd();
      }, 5);
    });
  }

  {
    // Test that the transformer is executed in the zone it was listened to.
    asyncStart();
    var stackTrace = currentStackTrace;
    var events = [];
    var controller;
    controller = new StreamController(onListen: () {
      // Events are added outside the zone.
      controller.add(499);
      controller.addError(42, stackTrace);
      controller.close();
    });
    Zone zone = Zone.current.fork();
    var stream = controller.stream.transform(
        new SinkTransformer((sink) => new ZoneTransformerSink(sink)));
    zone.run(() {
      stream.listen((data) {
        events.add(data);
      }, onDone: () {
        Expect.listEquals([zone, zone, zone], events);
        delayCycles(asyncEnd, 3);
      });
    });
  }

  {
    // Just make sure that the generic types are correct everywhere.
    asyncStart();
    new Stream.fromIterable([1, 2, 3])
        .transform(new SinkTransformer<int, String>(
            (sink) => new TypeChangingSink(sink)))
        .toList()
        .then((list) {
      Expect.listEquals(["1", "2", "3"], list);
      asyncEnd();
    });
  }
}