Add onDoneResume extension method for Stream

[jarekb123]

Stream.onDoneResume intercepts done event and switches to new stream.

Example

// Part of UploadBloc
  @override
  Stream<UploadState> mapEventToState(UploadEvent event) async* {
    yield* _repository
        .upload(event.file)
        .map((progress) => UploadState(
              progress: progress.floor(),
              isFinished: false,
              hasError: false,
            ))
        .onDoneResume(() => _onFinished())
        .onErrorResume((_) => _onError());
  }

  Stream<UploadState> _onFinished() async* {
    yield state.copyWith.call(isFinished: true);
  }

  Stream<UploadState> _onError() async* {
    yield state.copyWith.call(hasError: true);
  }

[codecov-io]

Codecov Report

Merging #440 into master will decrease coverage by 0.07%. The diff coverage is n/a.

@@            Coverage Diff             @@
##           master     #440      +/-   ##
==========================================
- Coverage   92.61%   92.54%   -0.08%     
==========================================
  Files          63       63              
  Lines        2073     2080       +7     
==========================================
+ Hits         1920     1925       +5     
- Misses        153      155       +2     

[frankpepermans]

I like it 👍

I did find an issue with our concatWith while looking at this, it is available as a Stream operator, but will always yield a single subscription Stream.

For example, below will throw currently, I'll fix that up asap:

final s = Stream.fromIterable([1, 2, 3])
      .asBroadcastStream()
      .concatWith(Stream.fromIterable([4, 5, 6]).asBroadcastStream());
  s.listen(null);
  s.listen(null);

[jarekb123]

@frankpepermans Fixed in #441 PR :)

[brianegan]

Hey there -- thanks for contributing!

Overall, I feel pretty strongly that we should not add this to the library. The Stream API is already quite large and hard for folks to learn, and I don't think adding more operators that just aliases of other operators will help that situation.

The current concatWith operator fulfills this job just fine, requires less typing, and doesn't increase the API surface area.

// Part of UploadBloc
  @override
  Stream<UploadState> mapEventToState(UploadEvent event) async* {
    yield* _repository
        .upload(event.file)
        .map((progress) => UploadState(
              progress: progress.floor(),
              isFinished: false,
              hasError: false,
            ))
        .concatWith([_onFinished()])
        .onErrorResume((_) => _onError());
  }

[frankpepermans]

@brianegan I do like onDoneResume as a name better than concatWith, if we do keep concatWith, maybe we oughta drop passing a List in favour of a Stream. (But concatWith is more Rx-y)

You can always concatWith again using chaining. (same for zipWith, mergeWith, ...)?

[brianegan]

I guess my argument is: concatWith is a standard Rx operator found in every implementation, onDoneResume is not a standard Rx operator. I'm not sure it makes sense for us to introduce the same functionality with two names, or remove concatWith in favor of a non-standard name.

We could have it take a single Stream instead of a list of Streams.

I don't feel too strongly about this one, but it would remove functionality for people who might be using it.

[narcodico]

What do you guys feel about making concatWith take a single Stream and add a secondary operator concatWithMany which would take multiple Streams. This would somewhat adhere to naming conventions found in other Rx implementations like Rx.NET having operators such as selectMany. Alternatively those operators could be named concat and concatMany, a bit less verbose.

[frankpepermans]

I would prefer it to take just a single Stream indeed, same with zipWith and mergeWith.

We aren't 1.0 yet, so I feel we still can.