Repository: flutter
Issue: [in_app_purchase]InAppPurchase.instance.purchaseStream.listen increase call in every product consume.
Labels: p: in_app_purchase,package,team-ecosystem,P2,triaged-ecosystem

UPDATE : InAppPurchase.instance.purchaseStream.listen() onDone method is not calling after succesfull consume.
If you do not call  _subscription.cancel() after succesfull consume , it increase number of calls with same purchaseDetailsList object one by one...

Example:

```
Product B consumed  and money taken from user 
.listen((purchaseDetailsList) {
      ...
    }
called one time. 
Product B consumed again and money taken from user or process cancelled by user 
.listen((purchaseDetailsList) {
      ...
    }
```
called two time.

the loop goes this way


```
"final Stream<List<PurchaseDetails>> purchaseUpdated = InAppPurchase.instance.purchaseStream;
    _subscription = purchaseUpdated.listen((purchaseDetailsList) {
      _listenToPurchaseUpdated(purchaseDetailsList);
    }, onDone: () {
      _subscription.cancel();
    }, onError: (error) {
      debugPrint("_initializeIAP Error : $error");
    });"
```

hello, we have an app with comsumable products for android and ios. When a user starts to receive consumables in the app, if they start receiving the same consumables again without closing the app, the listen method starts making extra calls one by one. That's why we load multiple credits for a payment on the backend side. I am sharing the whole service below.
it always call listen method with the last purchased consumable
package version : 2.0.1

<details>
<summary> code </summary>

```

// ignore_for_file: prefer_final_fields, depend_on_referenced_packages, directives_ordering, null_closures, unnecessary_null_comparison
import 'dart:async';
import 'dart:io';
import 'package:falsepeti/constants/api_urls.dart';
import 'package:falsepeti/models/credit.dart';
import 'package:falsepeti/providers/api_mode_provider.dart';
import 'package:falsepeti/services/shared_prefs_service.dart';
import 'package:http/http.dart' as http;
import 'package:falsepeti/providers/credits_provider.dart';
import 'package:falsepeti/providers/user_provider.dart';
import 'package:flutter/material.dart';
import 'package:in_app_purchase_android/in_app_purchase_android.dart';
import 'package:in_app_purchase/in_app_purchase.dart';
import 'package:provider/provider.dart';

class PurchaseModel {
  final dynamic pckg;
  final ProductDetails productDetail;

  PurchaseModel({this.pckg, required this.productDetail});
}

class PurchaseService {
  bool pendingPurchases = false;
  late BuildContext context;
  late InAppPurchase _connection;
  late StreamSubscription<List<PurchaseDetails>> _subscription;
  late List<ProductDetails> _playProducts = [];
  late List<Credit> _apiPackages = [];
  late List<PurchaseModel> _products = [];
  int completed = 0;
  void dispose() {
    _subscription.cancel();
  }

  PurchaseService({required this.context});
  void init() {
    _connection = InAppPurchase.instance;
    _initializeIAP();
    _proccessPastPurchases();
    debugPrint("Puschase Service initialized!");
  }

  Future<void> _initializeIAP() async {
    final Stream<List<PurchaseDetails>> purchaseUpdated = InAppPurchase.instance.purchaseStream;
    _subscription = purchaseUpdated.listen((purchaseDetailsList) {
      _listenToPurchaseUpdated(purchaseDetailsList);
    }, onDone: () {
      _subscription.cancel();
    }, onError: (error) {
      debugPrint("_initializeIAP Error : $error");
    });
  }

//BIND PLAY PRODUCTS AND API PACKAGES
  Future<void> _mapProducts() async {
    _playProducts.sort((pda, pdb) {
      final pa = _apiPackages.firstWhere((package) => package.sku == pda.id, orElse: null);
      final pb = _apiPackages.firstWhere((package) => package.sku == pdb.id, orElse: null);

      if (pa == null || pb == null) {
        return 0;
      }
      // ignore: cast_nullable_to_non_nullable
      return (pa.orderNumber as int).compareTo(pb.orderNumber as int);
    });

    _products.clear();
    final UserProvider userProvider = Provider.of<UserProvider>(context, listen: false);
    for (final ProductDetails pd in _playProducts) {
      final pckg = _apiPackages.firstWhere((package) => package.sku == pd.id);
      if (pckg == null) continue;
      if (userProvider.user.usedSpecialPackageForMobile == 1) {
        if (pd.id == "special_for_mobile") continue;
        if (pd.id == "ios_special_for_mobile") continue;
      }
      _products.add(PurchaseModel(pckg: pckg, productDetail: pd));
    }
  }

  //FETCH API PACKAGES
  Future<void> _getApiProducts() async {
    final CreditsProvider creditProvider = Provider.of<CreditsProvider>(context, listen: false);
    _apiPackages.clear();

    for (final package in creditProvider.credits) {
      _apiPackages.add(package);
    }
  }

  Future<List<PurchaseModel>> getProducts() async {
    await _getApiProducts();
    await getPlayProducts();
    _mapProducts();
    return _products;
  }

  //FETCH PLAY PRODUCTS
  Future<void> getPlayProducts() async {
    final bool isAvailable = await _connection.isAvailable();
    if (!isAvailable) return;
    _playProducts.clear();
    final ProductDetailsResponse pdr = await _connection.queryProductDetails(_apiPackages.map<String>((credit) => credit.sku ?? "").toSet());
    _playProducts.addAll(pdr.productDetails);
  }

  Future<bool> buyProduct({required PurchaseModel purchaseModel}) async {
    return _connection.buyConsumable(
      purchaseParam: PurchaseParam(
        productDetails: purchaseModel.productDetail,
      ),
    );
  }

  Future<void> _proccessPastPurchases() async {
    debugPrint("===== Purchase Service _proccessPastPurchases ===========");
    await _connection.restorePurchases();
    final Stream<List<PurchaseDetails>> purchaseUpdated = InAppPurchase.instance.purchaseStream;
    purchaseUpdated.listen((purchaseDetailsList) {
      _processPurchaseDetailsList(purchaseDetailsList, past: true);
    });
  }

  void _listenToPurchaseUpdated(List<PurchaseDetails> purchaseDetailsList) {
    _processPurchaseDetailsList(purchaseDetailsList);
  }

  Future<void> _processPurchaseDetailsList(List<PurchaseDetails> purchaseDetailsList, {bool past = false}) async {
    // ignore: unused_local_variable
    PurchaseDetails pending;

    if (purchaseDetailsList.isEmpty) {
      debugPrint("_processPurchaseDetailsList() Purchase List Empty ");
    } else {
      pending = purchaseDetailsList.first;
      for (final PurchaseDetails purchaseDetails in purchaseDetailsList) {
        if (purchaseDetails.status == PurchaseStatus.pending) {
          pending = purchaseDetails;
        }
        await _processPurchase(purchaseDetails);
      }
    }
  }

  Future<void> _processPurchase(PurchaseDetails purchaseDetails) async {
    switch (purchaseDetails.status) {
      case PurchaseStatus.purchased:
        await _processPurchasedPurchase(purchaseDetails);
        break;
      case PurchaseStatus.pending:
        await _processPendingPurchase(purchaseDetails);
        break;
      case PurchaseStatus.restored:
        await _processPendingPurchase(purchaseDetails);
        break;
      case PurchaseStatus.error:
        await _processErrorPurchase(purchaseDetails);
        break;
      case PurchaseStatus.canceled:
        await _processErrorPurchase(purchaseDetails);
        break;
    }
  }

  Future<void> _processPendingPurchase(PurchaseDetails purchaseDetails) async {
    if (Platform.isAndroid) {
      // Acquire the InAppPurchaseAndroidPlatformAddition instance.
      final InAppPurchaseAndroidPlatformAddition androidAddition = _connection.getPlatformAddition<InAppPurchaseAndroidPlatformAddition>();
      // Consume an Android purchase.
      await ApiModeProvider().getCurrentApiMode();
      final String _url = ApiUrls.playStorePending;
      final String _token = await SharedPrefsService().getToken() ?? "";
      final Map<String, dynamic> playData = {
        "product_id": purchaseDetails.productID,
        "purchase_token": purchaseDetails.verificationData.serverVerificationData,
        "order_id": purchaseDetails.purchaseID,
        "important": "1" // this is for apiclient if there is no token it will use oldtoken
      };
      await http.post(Uri.parse(_url),
          body: playData,
          headers: {"Authorization": "Bearer $_token", "Accept": "application/json", "Content-Type": "application/x-www-form-urlencoded"}).catchError((e) {
        debugPrint("_processPendingPurchase error : $e");
      });
      await androidAddition.consumePurchase(purchaseDetails).catchError((e) {
        debugPrint("_processPendingPurchase comsumePurchase Error : $e");
      });
    }

    if (Platform.isIOS) {}

    return;
  }

  Future<void> _processPurchasedPurchase(PurchaseDetails purchaseDetails) async {
    if (Platform.isAndroid) {
      final InAppPurchaseAndroidPlatformAddition androidAddition = _connection.getPlatformAddition<InAppPurchaseAndroidPlatformAddition>();
// Consume an Android purchase.
      await ApiModeProvider().getCurrentApiMode();
      final String _url = ApiUrls.playStoreInPayment;
      final String _token = await SharedPrefsService().getToken() ?? "";
      final Map<String, dynamic> playPurchasedData = {
        "product_id": purchaseDetails.productID,
        "purchase_token": purchaseDetails.verificationData.serverVerificationData,
        "important": "1" // this is for apiclient if there is no token it will use oldtoken
      };
      debugPrint("Android backend load request");
      await http.post(Uri.parse(_url),
          body: playPurchasedData,
          headers: {"Authorization": "Bearer $_token", "Accept": "application/json", "Content-Type": "application/x-www-form-urlencoded"}).catchError((e) {
        debugPrint("_processPurchasedPurchase error : $e");
      });

      await androidAddition.consumePurchase(purchaseDetails).catchError((e) {
        debugPrint("_processPurchasedPurchase comsumePurchase Error : $e");
      }).then((value) => {completed = 1});
    }
    if (Platform.isIOS) {
      await ApiModeProvider().getCurrentApiMode();
      final String _url = ApiUrls.appStoreInPayment;
      final String _token = await SharedPrefsService().getToken() ?? "";
      final Map<String, dynamic> appStorePurchasedData = {
        "product_id": purchaseDetails.productID,
        "purchase_token": purchaseDetails.verificationData.serverVerificationData,
        "important": "1" // this is for apiclient if there is no token it will use oldtoken
      };
      await http.post(Uri.parse(_url),
          body: appStorePurchasedData,
          headers: {"Authorization": "Bearer $_token", "Accept": "application/json", "Content-Type": "application/x-www-form-urlencoded"}).catchError((e) {
        debugPrint("_processPurchasedPurchase error : $e");
      });

      await _connection.completePurchase(purchaseDetails).catchError((e) {
        debugPrint("_processPurchasedPurchase completePurchase Error : $e");
      }).then((value) => {completed = 1});
    }
  }

  Future<void> _processErrorPurchase(PurchaseDetails purchaseDetails) async {
    if (Platform.isAndroid) {
      completed = 2;
    }

    if (Platform.isIOS) {
      // in in error status will be complete too.
      await _connection.completePurchase(purchaseDetails);
      completed = 2;
    }
  }
}

```


</details>

<details>
```

Doctor summary (to see all details, run flutter doctor -v):
[√] Flutter (Channel stable, 2.8.1, on Microsoft Windows [Version 10.0.19044.1526], locale tr-TR)
[√] Android toolchain - develop for Android devices (Android SDK version 33.0.0-rc1)
[√] Chrome - develop for the web
[√] Android Studio (version 2021.1)
[!] Android Studio (version 4.1)
    X Unable to determine bundled Java version.
[√] VS Code (version 1.64.2)
[√] Connected device (3 available)

```

</details>