Rxdart: switchMap doesn't cancel previous http request

[qiangjindong]

switchMap doesn't cancel previous http request. And when the streamcontroller close is called, the internal stream does not cancel. help me, thanks

class AssetManagementViewModel implements ViewModel {
  // Inputs
  void refresh() => _refresh.add(null);

  // Outputs
  late final Stream<List<Asset>> assets;

  final _refresh = StreamController<void>.broadcast();

  AssetManagementViewModel({required this.api}) {
    _getAssets() => api
        .getAssets(null)
        .map((event) => event.map((json) => Asset.fromJson(json)).toList());

    assets = _getAssets().asBroadcastStream().concatWith([
      _refresh.stream.switchMap((value) {
        return _getAssets(); // 2.don‘t cancel
      }).shareReplay(maxSize: 1)
    ]);
  }

  final AssetManagementAPI api;

  @override
  void dispose() {
    _refresh.close(); // 1.when call
  }
}

Here is the API stream generator，I can't find a better way

  Stream<List> getAssets(params) {
    print("will send getAssets request");
    final controller = StreamController<List>();
    rootBundle.loadString("lib/resource/assetData.json").then((value) {
      final list = jsonDecode(value);
      Future.delayed(Duration(seconds: 3)).then((value) {
        print("response getAssets request");
        controller.add(list);
        controller.close();
      });
    });
    controller.onCancel = () {
      print("cancel getAssets request");
      // dioToken.cancel ...
    };

    return controller.stream;
  }

[hoc081098]

Stream<List> getAssets(params) {
  return Rx.fromCallable(() => rootBundle.loadString("lib/resource/assetData.json"))
    .map((value) => jsonDecode(value) as List)
    .delay(const Duration(seconds: 3))
    .doOnListen(() => print('Listen'))
    .doOnCancel(() => print('Cancel'))
    .doOnData((d) => print('Data $d'));
}

and

class AssetManagementViewModel {
  ....

  final bag = CompositeSubscription():

  AssetManagementViewModel({required this.api}) {
    Stream<List<Asset>> _getAssets() => api
        .getAssets(null)
        .map((event) => event.map((json) => Asset.fromJson(json)).toList());

    assets = _refresh.stream
      .startWith(null)
      .switchMap((_) => _getAssets())
      .shareReplay(maxSize: 1)..listen(null).addTo(bag);
  }

  void dispose() {
    bag.dispose();
    ...
  }
}

Uses listen(null).addTo(bag) to keep the connection until ViewModel is disposed

[hoc081098]

And remember, Future cannot be cancelled -> When cancelling a Stream created from Future, we only mark the Stream as cancelled, and listeners will be removed from Stream, if source Future has been already executed, we cannot cancel that

[qiangjindong]

Stream<List> getAssets(params) {
  return Rx.fromCallable(() => rootBundle.loadString("lib/resource/assetData.json"))
    .map((value) => jsonDecode(value) as List)
    .delay(const Duration(seconds: 3))
    .doOnListen(() => print('Listen'))
    .doOnCancel(() => print('Cancel'))
    .doOnData((d) => print('Data $d'));
}

and

class AssetManagementViewModel {
  ....

  final bag = CompositeSubscription():

  AssetManagementViewModel({required this.api}) {
    Stream<List<Asset>> _getAssets() => api
        .getAssets(null)
        .map((event) => event.map((json) => Asset.fromJson(json)).toList());

    assets = _refresh.stream
      .startWith(null)
      .switchMap((_) => _getAssets())
      .shareReplay(maxSize: 1)..listen(null).addTo(bag);
  }

  void dispose() {
    bag.dispose();
    ...
  }
}

Uses listen(null).addTo(bag) to keep the connection until ViewModel is disposed

Thank you very much, it solved my headache. According to your suggestion, it works well now. Thanks again.