Rxdart: BehaviorSubject + switchMap does not emit when expected after 24.1 upgrade

[Mike278]

Tried again to upgrade rxdart from 23.1 to 24.1, but some stream-powered widgets in our app stopped loading.

I've reduced it down to this case:

test("rxdart 23.1 -> 24.1", () async {
  final a = BehaviorSubject.seeded('a');
  final bug = a.switchMap((_) => BehaviorSubject.seeded('b'));
  bug.listen((b) => print('data! $b'));
  print(await bug.first);
  print(await bug.first); // hangs here with rxdart 24.1, passes with rxdart 23.1
});

[hoc081098]

Hi @Mike278, IMO

• Because inside BehaviorSubject, we use StartWithStreamTransformer to replay latest value.
  
  
  
  • RxDart 23.1: StartWithStreamTransformer adds value synchronously
  
  
  • RxDart 24.1: StartWithStreamTransformer adds value based on sync param in BehaviorSubject.seeded(value, { bool sync: false }). By default,sync` is false, so event added asynchronously.
  
  
• Fix: make BehaviorSubject is sync eg:
  ```dart
  BehaviorSubject.seeded(seedValue, sync: true)

test​(​"rxdart 23.1 -> 24.1"​, () ​async​ {
  ​final​ a ​=​ ​BehaviorSubject​.​seeded​(​'a', sync: true​);
  ​final​ bug ​=​ a.​switchMap​((_) ​=>​ ​BehaviorSubject​.​seeded​(​'b', sync: true​));
  bug.​listen​((b) ​=>​ ​print​(​'data! $​b​'​));
  ​print​(​await​ bug.first);
  ​print​(​await​ bug.first); ​// hangs here with rxdart 24.1, passes with rxdart 23.1​
});
```

[Mike278]

Thanks for the reply!

Adding sync: true just makes it hang one line earlier at the first ​print​(​await​ bug.first); - did something different happen when you ran it?

I'm not sure I understand how StreamController's sync parameter would have helped here - the problem is not that an event is delivered at an unexpected time, it's that the future returned by the second call to bug.first does not complete at all.

The bug seems to be triggered by a specific sequence of listeners. For example 24.1 has the same behavior as 23.1 if you comment out bug.​listen​((b) ​=>​ ​print​(​'data! $​b​'​)); or swap it with the line below:

BehaviorSubject<String> a;
Stream<String> bug;
setUp(() {
  a = BehaviorSubject.seeded('a');
  bug = a.switchMap((_) => BehaviorSubject.seeded('b'));
});

test("v1", () async {
  bug.listen((b) => print('data! $b'));
  print(await bug.first);
  print(await bug.first); // hangs here with rxdart 24.1, passes with rxdart 23.1
});

test("v2", () async {
  // bug.listen((b) => print('data! $b'));
  print(await bug.first);
  print(await bug.first); // works
});

test("v3", () async {
  print(await bug.first);
  bug.listen((b) => print('data! $b'));
  print(await bug.first); // works
});

I also discovered a possibly related bug in #478

[benoitskipr]

having the same issue, does not feel like normal behaviorsubject ... behavior

[Mike278]

Any movement on this bug? It prevents us from upgrading rxdart.

[frankpepermans]

I'm looking into this, but it's not an easy fix, plus the current covid crisis & regular workload don't make it easy to commit a lot of time to rxdart at the moment :/

[Mike278]

Thanks @frankpepermans, appreciate the update :)

[frankpepermans]

pre-release is up: https://pub.dev/packages/rxdart/versions/0.25.0-beta

[Mike278]

I gave it a try and while the repro test now passes, I'm still seeing the issue with actual widgets in our app. I'll try to get a minimal repro tomorrow.

Edit: I clicked around for a bit longer and found that cancelling the listener throws a state error with 0.25.0-beta:

test("rxdart 23.1 -> ^0.25.0-beta", () async {
  final a = BehaviorSubject.seeded('a');
  final bug = a.switchMap((_) => BehaviorSubject.seeded('b'));
  await bug.listen((b) => print('data! $b')).cancel();
  print(await bug.first);  // Bad state: No element
});

// stacktrace:
dart:async                                                          _StreamImpl.listen
package:rxdart/src/streams/defer.dart 37:18                         DeferStream.listen
dart:async                                                          Stream.first

Bad state: No element

[Mike278]

Finally had a chance to look at this again and was able to reduce it to this case:

test('rxdart ^0.25.0-beta', () async {
  final bug = BehaviorSubject.seeded('seeded')
    .map((_) => 'map') // passes if either this or below is commented out
    .switchMap((_) => BehaviorSubject.seeded('switchMap'))
    ..listen(print); // passes if either this or above is commented out
  await pumpEventQueue();
  expect(await bug.first, 'switchMap'); // future never completes
});

with:

# pubspec.yaml:
dependency_overrides:
  rxdart:
    git:
      url: https://github.com/ReactiveX/rxdart.git
      ref: master

# pubspec.lock
rxdart:
  dependency: "direct main"
  description:
    path: "."
    ref: master
    resolved-ref: "4dcefadf6f72cc5061f7b9346e93179cfd2515d6"
    url: "https://github.com/ReactiveX/rxdart.git"
  source: git
  version: "0.25.0-beta"

[hoc081098]

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);