Refactor forwardStream: uses Stream.multi

[hoc081098]

Based on #596 and #601

• Remove try catch to increase performance.
• Implementation classes must handle error.
• Uses Stream.multi (Fixes Deadlocks in tests after upgrading rxdart #587)
• doX operator uses normal broadcast StreamController

[codecov-commenter]

Codecov Report

Merging #605 (3e99ad6) into master (ce05687) will decrease coverage by 0.16%.
The diff coverage is 96.17%.

@@            Coverage Diff             @@
##           master     #605      +/-   ##
==========================================
- Coverage   93.39%   93.22%   -0.17%     
==========================================
  Files          70       71       +1     
  Lines        2256     2230      -26     
==========================================
- Hits         2107     2079      -28     
- Misses        149      151       +2     

[hoc081098]

A special case cannot be handled :((

    test('description', () async{
      final a = BehaviorSubject.seeded('b');
      final bug = a.doOnData((v) {});
      await bug.listen(null).cancel();
      expect(await bug.first, 'b'); // cannot complete
    });

[frankpepermans]

I also tried converting to Stream.multi, but where my attempt fails, so does this one, take this test for example:

test('issue/587', () async {
        final source = BehaviorSubject.seeded('source');
        final switched =
            source.switchMap((value) => BehaviorSubject.seeded('switched'));
        var i = 0;
        switched.listen((_) => i++);
        expect(await switched.first, 'switched');
        expect(i, 1);
        expect(await switched.first, 'switched');
        expect(i, 1);
      });

[hoc081098]

I also tried converting to Stream.multi, but where my attempt fails, so does this one, take this test for example:

test('issue/587', () async {
        final source = BehaviorSubject.seeded('source');
        final switched =
            source.switchMap((value) => BehaviorSubject.seeded('switched'));
        var i = 0;
        switched.listen((_) => i++);
        expect(await switched.first, 'switched');
        expect(i, 1);
        expect(await switched.first, 'switched');
        expect(i, 1);
      });

I have just skipped some tests:

• 'issue/419: sync throughput'
• 'Rx.sample.reusable'
• 'Rx.startWith broadcast stream should not startWith on multiple subscribers'

[frankpepermans]

Sorry for the delay again!

Awesome job yet again :)

And great to hear the deadlocks are solved!