Example of Google Billing Library usage, using Kotlin coroutines

BillingManager.kt

package com.capybaradev.example.billing

import android.app.Activity
import android.content.Context
import com.android.billingclient.api.BillingClient
import com.android.billingclient.api.BillingClient.BillingResponseCode
import com.android.billingclient.api.BillingClientStateListener
import com.android.billingclient.api.BillingFlowParams
import com.android.billingclient.api.BillingResult
import com.android.billingclient.api.ConsumeParams
import com.android.billingclient.api.PendingPurchasesParams
import com.android.billingclient.api.ProductDetails
import com.android.billingclient.api.Purchase
import com.android.billingclient.api.Purchase.PurchaseState
import com.android.billingclient.api.QueryProductDetailsParams
import com.android.billingclient.api.QueryPurchasesParams
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.mapNotNull
import timber.log.Timber
import kotlin.coroutines.Continuation
import kotlin.coroutines.resume
import kotlin.coroutines.resumeWithException
import kotlin.coroutines.suspendCoroutine

class BillingManager(private val context: Context) {

    private var billingClient: BillingClient = buildBillingClient()
    private val purchaseFlow = MutableStateFlow<List<Purchase>>(emptyList())

    private val productList = listOf(
        QueryProductDetailsParams.Product.newBuilder()
            .setProductId(PRODUCT_ID_REMOVE_ADS)
            .setProductType(BillingClient.ProductType.INAPP)
            .build()
    )
    
    fun establishConnectionAsync(continuation: Continuation<Unit>? = null) {
        billingClient.startConnection(object : BillingClientStateListener {
            override fun onBillingSetupFinished(billingResult: BillingResult) {
                if (billingResult.responseCode == BillingResponseCode.OK) {
                    queryPurchasesAsync(continuation)
                }
            }

            override fun onBillingServiceDisconnected() {
                // NO-OP. Will be connected next time when needed
            }
        })
    }

    private fun buildBillingClient(): BillingClient {
        return BillingClient.newBuilder(context)
            .enablePendingPurchases(
                PendingPurchasesParams.newBuilder()
                    .enableOneTimeProducts()
                    .build()
            )
            .setListener { billingResult, purchases ->
                when (billingResult.responseCode) {
                    BillingResponseCode.OK -> {
                        purchaseFlow.tryEmit(purchases.orEmpty())
                    }

                    BillingResponseCode.USER_CANCELED -> {
                        Timber.e("User Cancelled")
                        Timber.d(billingResult.debugMessage)

                    }

                    else -> {
                        Timber.d(billingResult.debugMessage)
                    }
                }
            }.build()
    }

    private fun queryPurchasesAsync(continuation: Continuation<Unit>? = null) {
        val queryPurchasesParams = QueryPurchasesParams.newBuilder()
            .setProductType(BillingClient.ProductType.INAPP)
            .build()

        billingClient.queryPurchasesAsync(queryPurchasesParams) { billingResult, purchasesList ->
            if (billingResult.responseCode == BillingResponseCode.OK) {
                purchaseFlow.tryEmit(purchasesList)
            } else {
                Timber.d(billingResult.debugMessage)
            }
        }
        continuation?.resume(Unit)
    }

    fun endConnection() {
        billingClient.endConnection()
    }

    suspend fun launchBillingFlow(activity: Activity, params: BillingFlowParams): BillingResult {
        if (!billingClient.isReady) {
            billingClient = buildBillingClient()
            establishConnection()
        }

        return billingClient.launchBillingFlow(activity, params)
    }

    suspend fun loadAllProducts(): List<ProductDetails> {
        if (!billingClient.isReady) {
            billingClient = buildBillingClient()
            establishConnection()
        }

        return suspendCoroutine { continuation: Continuation<List<ProductDetails>> ->
            val params = QueryProductDetailsParams.newBuilder()
                .setProductList(productList)
                .build()

            billingClient.queryProductDetailsAsync(params) { billingResult, productDetailsList ->
                if (billingResult.responseCode == BillingResponseCode.OK && productDetailsList.isNotEmpty()) {
                    continuation.resume(productDetailsList)
                } else {
                    continuation.resumeWithException(BillingException("Initialisation error"))
                }
            }
        }
    }

    fun observeConsumablePurchase(productId: String): Flow<Purchase> {
        return purchaseFlow.mapNotNull { purchasesList ->
            purchasesList.find {
                it.products.contains(productId)
                        && it.purchaseState == PurchaseState.PURCHASED
            }
        }
    }

    suspend fun consumeProduct(purchase: Purchase) {
        if (!billingClient.isReady) {
            billingClient = buildBillingClient()
            establishConnection()
        }

        val consumeParams = ConsumeParams.newBuilder()
            .setPurchaseToken(purchase.purchaseToken)
            .build()

        suspendCoroutine {
            billingClient.consumeAsync(consumeParams) { billingResult, purchaseToken ->
                if (billingResult.responseCode == BillingResponseCode.OK) {
                    it.resume(purchase)
                }
            }
        }
    }

    private suspend fun establishConnection() = suspendCoroutine {
        establishConnectionAsync(it)
    }

    companion object {
        const val PRODUCT_ID_REMOVE_ADS = "remove_ads"
    }
}