Rxdart: BehaviorSubject bugged in 0.24.1 and 0.25.0-beta

[Mike278]

Probably the same as #477 but I'm opening a new issue because that one was not reopened.

These tests pass in rxdart 0.23.1:

test('rxdart #477/#500 - a', () async {
  final a = BehaviorSubject.seeded('a')
    .switchMap((_) => BehaviorSubject.seeded('a'))
    ..listen(print);
  await pumpEventQueue();
  expect(await a.first, 'a'); // this future does not complete in 0.24.1 - this is fixed in 0.25.0-beta
});

test('rxdart #477/#500 - b', () async {
  final b = BehaviorSubject.seeded('b')
    .map((_) => 'b')
    .switchMap((_) => BehaviorSubject.seeded('b'))
    ..listen(print);
  await pumpEventQueue();
  expect(await b.first, 'b'); // this future does not complete in 0.25.0-beta
});

[hoc081098]

https://github.com/ReactiveX/rxdart/issues/477#issuecomment-677954199

hey @Mike278
I think because map methods make the BehaviorSubject becomes a BroadcastStream, so we lose "replay latest value" behavior

Stream<S> map<S>(S convert(T event)) {
    return new _MapStream<T, S>(this, convert);
}

Temporary workaround: call shareValue after map

BehaviorSubject.seeded('seeded')
    .map((_) => 'map') 
    .shareValue()
    .switchMap((_) => BehaviorSubject.seeded('switchMap'))
      ..listen(print);

[Mike278]

I think because map methods make the BehaviorSubject becomes a BroadcastStream, so we lose "replay latest value" behavior

I don't think that's the case - the bug seems related to a listener being present:

final b = BehaviorSubject.seeded('b')
  .map((_) => 'b')
  .switchMap((_) => BehaviorSubject.seeded('b'));

// emits latest value as expected
expect(await b.first, 'b');
expect(await b.first, 'b');

// now add a listener
b.listen(print);
await pumpEventQueue();

// does not emit latest value
expect(await b.first, 'b');

[hoc081098]

@Mike278 This is how I understand it

final b = BehaviorSubject.seeded('b')                      [S1 - BehaviorSubject]
  .map((_) => 'b')                                         [S2 - _ForwardingStream]
  .switchMap((_) => BehaviorSubject.seeded('b'));          [S3 - _SyncBroadcastStreamController].stream

[onListen S3] -> [onListen S2] -> [onListen S1] -> receive latest value from BehaviorSubject
expect(await b.first, 'b');
[onCancel S3]

[onListen S3] -> [onListen S2] -> [onListen S1] -> receive latest value from BehaviorSubject
expect(await b.first, 'b');
[onCancel S3]

[onListen S3] -> [onListen S2] -> [onListen S1] -> receive latest value from BehaviorSubject
b.listen(print);
await pumpEventQueue();

Don't call onListen S3, because S3's Stream already has a listener. 
Simply listen to S3's Stream and S3's Stream doesn't replay latest value.
expect(await b.first, 'b');

[hoc081098]

• To preserve behavior of BehaviorSubject, we must use forwardStream everywhere, include built-in operators of Stream class (map, where, takeWhile, ...).
• Or introduce new extension methods with rx prefix (rxMap, rxWhere, rxTakeWhile, ...) to override default behavior of built-in operators:

extension RxMapExtension<T> on Stream<T> {
  Stream<R> rxMap<R>(R Function(T) converter) =>
      forwardStream(this, RxMapSink(converter));
}

class RxMapSink<T, R> implements ForwardingSink<T, R> {
  final R Function(T) _converter;

  RxMapSink(this._converter);

  @override
  void add(EventSink<R> sink, T data) => sink.add(_converter(data));

  @override
  void addError(EventSink<R> sink, Object error, [StackTrace st]) =>
      sink.addError(error, st);

  @override
  void close(EventSink<R> sink) => sink.close();

  @override
  FutureOr onCancel(EventSink<R> sink) {}

  @override
  void onListen(EventSink<R> sink) {}

  @override
  void onPause(EventSink<R> sink, [Future<dynamic> resumeSignal]) {}

  @override
  void onResume(EventSink<R> sink) {}
}

void main() {
  test('description', () async {
    final b = BehaviorSubject.seeded('b')
        .rxMap(_) => 'b')
        .switchMap((_) => BehaviorSubject.seeded('b'));

    // emits latest value as expected
    expect(await b.first, 'b');
    expect(await b.first, 'b');

    // now add a listener
    b.listen(print);
    await pumpEventQueue();

    // PASSED
    expect(await b.first, 'b');
  });
}

[frankpepermans]

I think we should override the built-in methods, but it still wouldn't be a bulletproof fix for Behavior/Replay Subject,
take map for example, it literally just returns a _MapStream internally, we have zero control at this point :/

I mean, anyone can write a Stream extension and just break the behavior of our Subjects this way, unfortunately I currently see no better option :(

[Mike278]

Is there something new in 0.24.1 that introduced that possibility, or has it always been possible? Curious what changed vs 0.23.1

[hoc081098]

• Before 0.24.1, all StreamTransformers in rxdart returns Single-Subscription Stream when input Stream is Single-Subscription or Broadcast Stream.
• Since 0.24.1, we changed StreamTransformer's behavior, a StreamTransform returns:
  
  
  
  • a Subject when input Stream is a Subject
  
  
  • a Single-Subscription Stream when input Stream is Single-Subscription.
  
  
  • a Broadcast Stream when input Stream is Broadcast Stream
  
  

_Sent from my Redmi 7A using FastHub_

[Mike278]

Sorry, it's not clear to me how those bullet points relate to this bug.

I guess my comment was more addressing "anyone can write a Stream extension and just break the behavior of our Subjects this way". I think the 0.24.0-dev.1 changelog answers my question:

as of this release, we've refactored the way Stream transformers are set up. Previous releases had some incorrect behavior when using certain operators
[...]
To properly fix this up, a new way of transforming Streams was introduced. Operators as of now use Stream.eventTransformed and we've refactored all operators to implement Sink instead.

Is this also the change that caused this bug? Doesn't seem like a worthy tradeoff from my perspective!

There were a few proposed solutions above, are any of them going to be implemented? This is still broken with 0.25.0-beta2.

[frankpepermans]

Mike,

Yes we want to fix this bug, ideally every bug that we discover or that is being reported.
We had to do the refactoring you mentioned to solve other issues that were equally breaking in their perspective.

Dart Streams were not built with behavior subjects in mind for example, so it's not trivial to implement some things, default Dart Stream ops such as map for example just plain return a MapStream under the hood.

The only hook we have there, is the onListen handler, so either we see what we can do there without causing other issues, or we override these default ops to suit rx.

But in case we do the latter, then anyone who writes Stream extensions that roughly do the same as map, will find similar issues you have.

Also, we do this in our free time, and we work on our own pace, as always we welcome contributions and pull requests.

[hoc081098]

Hey @Mike278, see pr #512