// 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.

// Test the basic StreamController and StreamController.broadcast.
library stream_controller_async_test;

import 'dart:async';

import 'package:expect/expect.dart';
import 'package:async_helper/async_minitest.dart';

import 'event_helper.dart';
import 'stream_state_helper.dart';

void cancelSub(StreamSubscription sub) {
  sub.cancel();
}

testController() {
  // Test fold
  test("StreamController.fold", () {
    StreamController c = new StreamController();
    Stream stream = c.stream.asBroadcastStream(onCancel: cancelSub);
    stream.fold<dynamic>(0, (a, b) => a + b).then(expectAsync((int v) {
      Expect.equals(42, v);
    }));
    c.add(10);
    c.add(32);
    c.close();
  });

  test("StreamController.fold throws", () {
    StreamController c = new StreamController();
    Stream stream = c.stream.asBroadcastStream(onCancel: cancelSub);
    Future<int?>.value(stream.fold(0, (a, b) {
      throw "Fnyf!";
    })).catchError(expectAsync((error) {
      Expect.equals("Fnyf!", error);
    }));
    c.add(42);
  });
}

testSingleController() {
  test("Single-subscription StreamController.fold", () {
    StreamController c = new StreamController();
    Stream stream = c.stream;
    stream.fold<dynamic>(0, (a, b) => a + b).then(expectAsync((int v) {
      Expect.equals(42, v);
    }));
    c.add(10);
    c.add(32);
    c.close();
  });

  test("Single-subscription StreamController.fold throws", () {
    StreamController c = new StreamController();
    Stream stream = c.stream;
    Future<int?>.value(stream.fold(0, (a, b) {
      throw "Fnyf!";
    })).catchError(expectAsync((e) {
      Expect.equals("Fnyf!", e);
    }));
    c.add(42);
  });

  test(
      "Single-subscription StreamController events are buffered when"
      " there is no subscriber", () {
    StreamController<int> c = new StreamController();
    EventSink sink = c.sink;
    var stream = c.stream;
    int counter = 0;
    sink.add(1);
    sink.add(2);
    sink.close();
    stream.listen((data) {
      counter += data;
    }, onDone: expectAsync(() {
      Expect.equals(3, counter);
    }));
  });
}

testExtraMethods() {
  Events sentEvents = new Events()
    ..add(7)
    ..add(9)
    ..add(13)
    ..add(87)
    ..close();

  test("forEach", () {
    StreamController c = new StreamController();
    Events actualEvents = new Events();
    Future f = c.stream.forEach(actualEvents.add);
    f.then(expectAsync((_) {
      actualEvents.close();
      Expect.listEquals(sentEvents.events, actualEvents.events);
    }));
    sentEvents.replay(c);
  });

  test("forEachError", () {
    Events sentEvents = new Events()
      ..add(7)
      ..error("bad")
      ..add(87)
      ..close();
    StreamController c = new StreamController();
    Events actualEvents = new Events();
    Future f = c.stream.forEach(actualEvents.add);
    f.catchError(expectAsync((error) {
      Expect.equals("bad", error);
      Expect.listEquals((new Events()..add(7)).events, actualEvents.events);
    }));
    sentEvents.replay(c);
  });

  test("forEachError2", () {
    Events sentEvents = new Events()
      ..add(7)
      ..add(9)
      ..add(87)
      ..close();
    StreamController c = new StreamController();
    Events actualEvents = new Events();
    Future f = c.stream.forEach((x) {
      if (x == 9) throw "bad";
      actualEvents.add(x);
    });
    f.catchError(expectAsync((error) {
      Expect.equals("bad", error);
      Expect.listEquals((new Events()..add(7)).events, actualEvents.events);
    }));
    sentEvents.replay(c);
  });

  test("firstWhere", () {
    StreamController c = new StreamController();
    Future f = c.stream.firstWhere((x) => (x % 3) == 0);
    f.then(expectAsync((v) {
      Expect.equals(9, v);
    }));
    sentEvents.replay(c);
  });

  test("firstWhere 2", () {
    StreamController c = new StreamController();
    Future f = c.stream.firstWhere((x) => (x % 4) == 0);
    f.catchError(expectAsync((e) {}));
    sentEvents.replay(c);
  });

  test("firstWhere 3", () {
    StreamController c = new StreamController();
    Future f = c.stream.firstWhere((x) => (x % 4) == 0, orElse: () => 999);
    f.then(expectAsync((v) {
      Expect.equals(999, v);
    }));
    sentEvents.replay(c);
  });

  test("lastWhere", () {
    StreamController c = new StreamController();
    Future f = c.stream.lastWhere((x) => (x % 3) == 0);
    f.then(expectAsync((v) {
      Expect.equals(87, v);
    }));
    sentEvents.replay(c);
  });

  test("lastWhere 2", () {
    StreamController c = new StreamController();
    Future f = c.stream.lastWhere((x) => (x % 4) == 0);
    f.catchError(expectAsync((e) {}));
    sentEvents.replay(c);
  });

  test("lastWhere 3", () {
    StreamController c = new StreamController();
    Future f = c.stream.lastWhere((x) => (x % 4) == 0, orElse: () => 999);
    f.then(expectAsync((v) {
      Expect.equals(999, v);
    }));
    sentEvents.replay(c);
  });

  test("singleWhere", () {
    StreamController c = new StreamController();
    Future f = c.stream.singleWhere((x) => (x % 9) == 0);
    f.then(expectAsync((v) {
      Expect.equals(9, v);
    }));
    sentEvents.replay(c);
  });

  test("singleWhere 2", () {
    StreamController c = new StreamController();
    Future f = c.stream.singleWhere((x) => (x % 3) == 0); // Matches 9 and 87..
    f.catchError(expectAsync((error) {
      Expect.isTrue(error is StateError);
    }));
    sentEvents.replay(c);
  });

  test("first", () {
    StreamController c = new StreamController();
    Future f = c.stream.first;
    f.then(expectAsync((v) {
      Expect.equals(7, v);
    }));
    sentEvents.replay(c);
  });

  test("first empty", () {
    StreamController c = new StreamController();
    Future f = c.stream.first;
    f.catchError(expectAsync((error) {
      Expect.isTrue(error is StateError);
    }));
    Events emptyEvents = new Events()..close();
    emptyEvents.replay(c);
  });

  test("first error", () {
    StreamController c = new StreamController();
    Future f = c.stream.first;
    f.catchError(expectAsync((error) {
      Expect.equals("error", error);
    }));
    Events errorEvents = new Events()
      ..error("error")
      ..close();
    errorEvents.replay(c);
  });

  test("first error 2", () {
    StreamController c = new StreamController();
    Future f = c.stream.first;
    f.catchError(expectAsync((error) {
      Expect.equals("error", error);
    }));
    Events errorEvents = new Events()
      ..error("error")
      ..error("error2")
      ..close();
    errorEvents.replay(c);
  });

  test("last", () {
    StreamController c = new StreamController();
    Future f = c.stream.last;
    f.then(expectAsync((v) {
      Expect.equals(87, v);
    }));
    sentEvents.replay(c);
  });

  test("last empty", () {
    StreamController c = new StreamController();
    Future f = c.stream.last;
    f.catchError(expectAsync((error) {
      Expect.isTrue(error is StateError);
    }));
    Events emptyEvents = new Events()..close();
    emptyEvents.replay(c);
  });

  test("last error", () {
    StreamController c = new StreamController();
    Future f = c.stream.last;
    f.catchError(expectAsync((error) {
      Expect.equals("error", error);
    }));
    Events errorEvents = new Events()
      ..error("error")
      ..close();
    errorEvents.replay(c);
  });

  test("last error 2", () {
    StreamController c = new StreamController();
    Future f = c.stream.last;
    f.catchError(expectAsync((error) {
      Expect.equals("error", error);
    }));
    Events errorEvents = new Events()
      ..error("error")
      ..error("error2")
      ..close();
    errorEvents.replay(c);
  });

  test("elementAt", () {
    StreamController c = new StreamController();
    Future f = c.stream.elementAt(2);
    f.then(expectAsync((v) {
      Expect.equals(13, v);
    }));
    sentEvents.replay(c);
  });

  test("elementAt 2", () {
    StreamController c = new StreamController();
    Future f = c.stream.elementAt(20);
    f.catchError(expectAsync((error) {
      Expect.isTrue(error is RangeError);
    }));
    sentEvents.replay(c);
  });

  test("drain", () {
    StreamController c = new StreamController();
    Future f = c.stream.drain();
    f.then(expectAsync((v) {
      Expect.equals(null, v);
    }));
    sentEvents.replay(c);
  });

  test("drain error", () {
    StreamController c = new StreamController();
    Future f = c.stream.drain();
    f.catchError(expectAsync((error) {
      Expect.equals("error", error);
    }));
    Events errorEvents = new Events()
      ..error("error")
      ..error("error2")
      ..close();
    errorEvents.replay(c);
  });
}

testPause() {
  test("pause event-unpause", () {
    StreamProtocolTest test = new StreamProtocolTest();
    Completer completer = new Completer();
    test
      ..expectListen()
      ..expectData(42, () {
        test.pause(completer.future);
      })
      ..expectPause(() {
        completer.complete(null);
      })
      ..expectData(43)
      ..expectData(44)
      ..expectCancel()
      ..expectDone(test.terminate);
    test.listen();
    test.add(42);
    test.add(43);
    test.add(44);
    test.close();
  });

  test("pause twice event-unpause", () {
    StreamProtocolTest test = new StreamProtocolTest();
    Completer completer = new Completer();
    Completer completer2 = new Completer();
    test
      ..expectListen()
      ..expectData(42, () {
        test.pause(completer.future);
        test.pause(completer2.future);
      })
      ..expectPause(() {
        completer.future.then(completer2.complete);
        completer.complete(null);
      })
      ..expectData(43)
      ..expectData(44)
      ..expectCancel()
      ..expectDone(test.terminate);
    test
      ..listen()
      ..add(42)
      ..add(43)
      ..add(44)
      ..close();
  });

  test("pause twice direct-unpause", () {
    StreamProtocolTest test = new StreamProtocolTest();
    test
      ..expectListen()
      ..expectData(42, () {
        test.pause();
        test.pause();
      })
      ..expectPause(() {
        test.resume();
        test.resume();
      })
      ..expectData(43)
      ..expectData(44)
      ..expectCancel()
      ..expectDone(test.terminate);
    test
      ..listen()
      ..add(42)
      ..add(43)
      ..add(44)
      ..close();
  });

  test("pause twice direct-event-unpause", () {
    StreamProtocolTest test = new StreamProtocolTest();
    Completer completer = new Completer();
    test
      ..expectListen()
      ..expectData(42, () {
        test.pause();
        test.pause(completer.future);
        test.add(43);
        test.add(44);
        test.close();
      })
      ..expectPause(() {
        completer.future.then((v) => test.resume());
        completer.complete(null);
      })
      ..expectData(43)
      ..expectData(44)
      ..expectCancel()
      ..expectDone(test.terminate);
    test
      ..listen()
      ..add(42);
  });
}

class TestError {
  const TestError();
}

testRethrow() {
  TestError error = const TestError();

  testStream(name, streamValueTransform) {
    test("rethrow-$name-value", () {
      StreamController c = new StreamController();
      Stream s = streamValueTransform(c.stream, (v) {
        throw error;
      });
      s.listen((_) {
        Expect.fail("unexpected value");
      }, onError: expectAsync((e) {
        Expect.identical(error, e);
      }));
      c.add(null);
      c.close();
    });
  }

  testStreamError(name, streamErrorTransform) {
    test("rethrow-$name-error", () {
      StreamController c = new StreamController();
      Stream s = streamErrorTransform(c.stream, (e) {
        throw error;
      });
      s.listen((_) {
        Expect.fail("unexpected value");
      }, onError: expectAsync((e) {
        Expect.identical(error, e);
      }));
      c.addError("SOME ERROR");
      c.close();
    });
  }

  testFuture(name, streamValueTransform) {
    test("rethrow-$name-value", () {
      StreamController c = new StreamController();
      Future f = streamValueTransform(c.stream, (v) {
        throw error;
      });
      f.then((v) {
        Expect.fail("unreachable");
      }, onError: expectAsync((e) {
        Expect.identical(error, e);
      }));
      // Need two values to trigger compare for reduce.
      c.add(0);
      c.add(1);
      c.close();
    });
  }

  testStream("where", (s, act) => s.where(act));
  testStream("map", (s, act) => s.map(act));
  testStream("expand", (s, act) => s.expand(act));
  testStream("where", (s, act) => s.where(act));
  testStreamError("handleError", (s, act) => s.handleError(act));
  testStreamError("handleTest", (s, act) => s.handleError((v) {}, test: act));
  testFuture("forEach", (s, act) => s.forEach(act));
  testFuture("every", (s, act) => s.every(act));
  testFuture("any", (s, act) => s.any(act));
  testFuture("reduce", (s, act) => s.reduce((a, b) => act(b)));
  testFuture("fold", (s, act) => s.fold(0, (a, b) => act(b)));
  testFuture("drain", (s, act) => s.drain().then(act));
}

void testBroadcastController() {
  test("broadcast-controller-basic", () {
    StreamProtocolTest test = new StreamProtocolTest.broadcast();
    test
      ..expectListen()
      ..expectData(42)
      ..expectCancel()
      ..expectDone(test.terminate);
    test
      ..listen()
      ..add(42)
      ..close();
  });

  test("broadcast-controller-listen-twice", () {
    StreamProtocolTest test = new StreamProtocolTest.broadcast();
    test
      ..expectListen()
      ..expectData(42, () {
        test.listen();
        test.add(37);
        test.close();
      })
      // Order is not guaranteed between subscriptions if not sync.
      ..expectData(37)
      ..expectData(37)
      ..expectDone()
      ..expectCancel()
      ..expectDone(test.terminate);
    test.listen();
    test.add(42);
  });

  test("broadcast-controller-listen-twice-non-overlap", () {
    StreamProtocolTest test = new StreamProtocolTest.broadcast();
    test
      ..expectListen(() {
        test.add(42);
      })
      ..expectData(42, () {
        test.cancel();
      })
      ..expectCancel(() {
        test.listen();
      })
      ..expectListen(() {
        test.add(37);
      })
      ..expectData(37, () {
        test.close();
      })
      ..expectCancel()
      ..expectDone(test.terminate);
    test.listen();
  });

  test("broadcast-controller-individual-pause", () {
    StreamProtocolTest test = new StreamProtocolTest.broadcast();
    var sub1;
    test
      ..expectListen()
      ..expectData(42)
      ..expectData(42, () {
        sub1.pause();
      })
      ..expectData(43, () {
        sub1.cancel();
        test.listen();
        test.add(44);
        test.expectData(44);
        test.expectData(44, test.terminate);
      });
    sub1 = test.listen();
    test.listen();
    test.add(42);
    test.add(43);
  });

  test("broadcast-controller-add-in-callback", () {
    StreamProtocolTest test = new StreamProtocolTest.broadcast();
    test.expectListen();
    var sub = test.listen();
    test.add(42);
    sub.expectData(42, () {
      test.add(87);
      sub.cancel();
    });
    test.expectCancel(() {
      test.add(37);
      test.terminate();
    });
  });
}

void testAsBroadcast() {
  test("asBroadcast-not-canceled", () {
    StreamProtocolTest test = new StreamProtocolTest.asBroadcast();
    var sub;
    test
      ..expectListen()
      ..expectBroadcastListen((_) {
        test.add(42);
      })
      ..expectData(42, () {
        sub.cancel();
      })
      ..expectBroadcastCancel((_) {
        sub = test.listen();
      })
      ..expectBroadcastListen((_) {
        test.terminate();
      });
    sub = test.listen();
  });

  test("asBroadcast-canceled", () {
    StreamProtocolTest test = new StreamProtocolTest.asBroadcast();
    var sub;
    test
      ..expectListen()
      ..expectBroadcastListen((_) {
        test.add(42);
      })
      ..expectData(42, () {
        sub.cancel();
      })
      ..expectBroadcastCancel((originalSub) {
        originalSub.cancel();
      })
      ..expectCancel(test.terminate);
    sub = test.listen();
  });

  test("asBroadcast-pause-original", () {
    StreamProtocolTest test = new StreamProtocolTest.asBroadcast();
    var sub;
    test
      ..expectListen()
      ..expectBroadcastListen((_) {
        test.add(42);
        test.add(43);
      })
      ..expectData(42, () {
        sub.cancel();
      })
      ..expectBroadcastCancel((originalSub) {
        originalSub.pause(); // Pause before sending 43 from original sub.
      })
      ..expectPause(() {
        sub = test.listen();
      })
      ..expectBroadcastListen((originalSub) {
        originalSub.resume();
      })
      ..expectData(43)
      ..expectResume(() {
        test.close();
      })
      ..expectCancel()
      ..expectDone()
      ..expectBroadcastCancel((_) => test.terminate());
    sub = test.listen();
  });
}

void testSink(
    {required bool sync, required bool broadcast, required bool asBroadcast}) {
  String type =
      "${sync ? "S" : "A"}${broadcast ? "B" : "S"}${asBroadcast ? "aB" : ""}";
  test("$type-controller-sink", () {
    var done = expectAsync(() {});
    var c = broadcast
        ? new StreamController.broadcast(sync: sync)
        : new StreamController(sync: sync);
    var expected = new Events()
      ..add(42)
      ..error("error")
      ..add(1)
      ..add(2)
      ..add(3)
      ..add(4)
      ..add(5)
      ..add(43)
      ..close();
    var actual = new Events.capture(
        asBroadcast ? c.stream.asBroadcastStream() : c.stream);
    var sink = c.sink;
    sink.add(42);
    sink.addError("error");
    sink.addStream(new Stream.fromIterable([1, 2, 3, 4, 5])).then((_) {
      sink.add(43);
      return sink.close();
    }).then((_) {
      Expect.listEquals(expected.events, actual.events);
      done();
    });
  });

  test("$type-controller-sink-canceled", () {
    var done = expectAsync(() {});
    var c = broadcast
        ? new StreamController.broadcast(sync: sync)
        : new StreamController(sync: sync);
    var expected = new Events()
      ..add(42)
      ..error("error")
      ..add(1)
      ..add(2)
      ..add(3);
    var stream = asBroadcast ? c.stream.asBroadcastStream() : c.stream;
    var actual = new Events();
    var sub;
    // Cancel subscription after receiving "3" event.
    sub = stream.listen((v) {
      if (v == 3) sub.cancel();
      actual.add(v);
    }, onError: actual.error);
    var sink = c.sink;
    sink.add(42);
    sink.addError("error");
    sink.addStream(new Stream.fromIterable([1, 2, 3, 4, 5])).then((_) {
      Expect.listEquals(expected.events, actual.events);
      // Close controller as well. It has no listener. If it is a broadcast
      // stream, it will still be open, and we read the "done" future before
      // closing. A normal stream is already done when its listener cancels.
      Future doneFuture = sink.done;
      sink.close();
      return doneFuture;
    }).then((_) {
      // No change in events.
      Expect.listEquals(expected.events, actual.events);
      done();
    });
  });

  test("$type-controller-sink-paused", () {
    var done = expectAsync(() {});
    var c = broadcast
        ? new StreamController.broadcast(sync: sync)
        : new StreamController(sync: sync);
    var expected = new Events()
      ..add(42)
      ..error("error")
      ..add(1)
      ..add(2)
      ..add(3)
      ..add(4)
      ..add(5)
      ..add(43)
      ..close();
    var stream = asBroadcast ? c.stream.asBroadcastStream() : c.stream;
    var actual = new Events();
    var sub;
    var pauseIsDone = false;
    sub = stream.listen((v) {
      if (v == 3) {
        sub.pause(new Future.delayed(const Duration(milliseconds: 15), () {
          pauseIsDone = true;
        }));
      }
      actual.add(v);
    }, onError: actual.error, onDone: actual.close);
    var sink = c.sink;
    sink.add(42);
    sink.addError("error");
    sink.addStream(new Stream.fromIterable([1, 2, 3, 4, 5])).then((_) {
      sink.add(43);
      return sink.close();
    }).then((_) {
      if (asBroadcast || broadcast) {
        // The done-future of the sink completes when it passes
        // the done event to the asBroadcastStream controller, which is
        // before the final listener gets the event.
        // Wait for the done event to be *delivered* before testing the
        // events.
        actual.onDone(() {
          Expect.listEquals(expected.events, actual.events);
          done();
        });
      } else {
        Expect.listEquals(expected.events, actual.events);
        done();
      }
    });
  });

  test("$type-controller-addstream-error-stop", () {
    // Check that addStream defaults to ending after the first error.
    var done = expectAsync(() {});
    StreamController c = broadcast
        ? new StreamController.broadcast(sync: sync)
        : new StreamController(sync: sync);
    Stream stream = asBroadcast ? c.stream.asBroadcastStream() : c.stream;
    var actual = new Events.capture(stream);

    var source = new Events();
    source
      ..add(1)
      ..add(2)
      ..error("BAD")
      ..add(3)
      ..error("FAIL")
      ..close();

    var expected = new Events()
      ..add(1)
      ..add(2)
      ..error("BAD")
      ..close();
    StreamController sourceController = new StreamController();
    c.addStream(sourceController.stream, cancelOnError: true).then((_) {
      c.close().then((_) {
        Expect.listEquals(expected.events, actual.events);
        done();
      });
    });

    source.replay(sourceController);
  });

  test("$type-controller-addstream-error-forward", () {
    // Check that addStream with cancelOnError:false passes all data and errors
    // to the controller.
    var done = expectAsync(() {});
    StreamController c = broadcast
        ? new StreamController.broadcast(sync: sync)
        : new StreamController(sync: sync);
    Stream stream = asBroadcast ? c.stream.asBroadcastStream() : c.stream;
    var actual = new Events.capture(stream);

    var source = new Events();
    source
      ..add(1)
      ..add(2)
      ..addError("BAD")
      ..add(3)
      ..addError("FAIL")
      ..close();

    StreamController sourceController = new StreamController();
    c.addStream(sourceController.stream).then((_) {
      c.close().then((_) {
        Expect.listEquals(source.events, actual.events);
        done();
      });
    });

    source.replay(sourceController);
  });

  test("$type-controller-addstream-twice", () {
    // Using addStream twice on the same stream
    var done = expectAsync(() {});
    StreamController c = broadcast
        ? new StreamController.broadcast(sync: sync)
        : new StreamController(sync: sync);
    Stream stream = asBroadcast ? c.stream.asBroadcastStream() : c.stream;
    var actual = new Events.capture(stream);

    // Streams of five events, throws on 3.
    Stream s1 = new Stream.fromIterable([1, 2, 3, 4, 5])
        .map((x) => (x == 3 ? throw x : x));
    Stream s2 = new Stream.fromIterable([1, 2, 3, 4, 5])
        .map((x) => (x == 3 ? throw x : x));

    Events expected = new Events();
    expected
      ..add(1)
      ..add(2)
      ..error(3);
    expected
      ..add(1)
      ..add(2)
      ..error(3)
      ..add(4)
      ..add(5);
    expected..close();

    c.addStream(s1, cancelOnError: true).then((_) {
      c.addStream(s2, cancelOnError: false).then((_) {
        c.close().then((_) {
          Expect.listEquals(expected.events, actual.events);
          done();
        });
      });
    });
  });
}

main() {
  testController();
  testSingleController();
  testExtraMethods();
  testPause();
  testRethrow();
  testBroadcastController();
  testAsBroadcast();
  testSink(sync: true, broadcast: false, asBroadcast: false);
  testSink(sync: true, broadcast: false, asBroadcast: true);
  testSink(sync: true, broadcast: true, asBroadcast: false);
  testSink(sync: false, broadcast: false, asBroadcast: false);
  testSink(sync: false, broadcast: false, asBroadcast: true);
  testSink(sync: false, broadcast: true, asBroadcast: false);
}