How to await billingClient.startConnection in Kotlin

I'm following the google billing integration instructions and have got stuck with how to await for the billing client connection result.

Whenever I need to query sku details or purchases I need to make sure that the billing client is initialized and connected. There are querySkuDetails and queryPurchasesAsync awaitable kotlin extension functions, but startConnection is based on listeners instead. Here are code samples from the docs.

private var billingClient = BillingClient.newBuilder(activity)
   .setListener(purchasesUpdatedListener)
   .enablePendingPurchases()
   .build()

billingClient.startConnection(object : BillingClientStateListener {
    override fun onBillingSetupFinished(billingResult: BillingResult) {
        if (billingResult.responseCode ==  BillingResponseCode.OK) {
            // The BillingClient is ready. You can query purchases here.
        }
    }
    override fun onBillingServiceDisconnected() {
        // Try to restart the connection on the next request to
        // Google Play by calling the startConnection() method.
    }
})

suspend fun querySkuDetails() {
    // prepare params
    // leverage querySkuDetails Kotlin extension function
    val skuDetailsResult = withContext(Dispatchers.IO) {
        billingClient.querySkuDetails(params.build())
    }
    // Process the result.
}

How to put all this together using suspend functions?

One way to create a suspending version of startConnection is the following:

/**
 * Returns immediately if this BillingClient is already connected, otherwise
 * initiates the connection and suspends until this client is connected.
 */
suspend fun BillingClient.ensureReady() {
    if (isReady) {
        return // fast path if already connected
    }
    return suspendCoroutine { cont ->
        startConnection(object : BillingClientStateListener {
            override fun onBillingSetupFinished(billingResult: BillingResult) {
                if (billingResult.responseCode == BillingResponseCode.OK) {
                    cont.resume(Unit)
                } else {
                    // you could also use a custom, more precise exception
                    cont.resumeWithException(RuntimeException("Billing setup failed: ${billingResult.debugMessage} (code ${billingResult.responseCode})"))
                }
            }

            override fun onBillingServiceDisconnected() {
                // no need to setup reconnection logic here, call ensureReady() 
                // before each purchase to reconnect as necessary
            }
        })
    }
}

This will fail if another coroutine already initiated a connection. If you want to deal with potential concurrent calls to this method, you can use a mutex to protect the connection part:

val billingConnectionMutex = Mutex()

/**
 * Returns immediately if this BillingClient is already connected, otherwise
 * initiates the connection and suspends until this client is connected.
 * If a connection is already in the process of being established, this
 * method just suspends until the billing client is ready.
 */
suspend fun BillingClient.ensureReady() {
    billingConnectionMutex.withLock {
        // fast path: avoid suspension if another coroutine already connected
        if (isReady) {
            return
        }
        connectOrThrow()
    }
}

private suspend fun BillingClient.connectOrThrow() = suspendCoroutine<Unit> { cont ->
    startConnection(object : BillingClientStateListener {
        override fun onBillingSetupFinished(billingResult: BillingResult) {
            if (billingResult.responseCode == BillingClient.BillingResponseCode.OK) {
                cont.resume(Unit)
            } else {
                cont.resumeWithException(RuntimeException("Billing setup failed: ${billingResult.debugMessage} (code ${billingResult.responseCode})"))
            }
        }

        override fun onBillingServiceDisconnected() {
            // no need to setup reconnection logic here, call ensureReady()
            // before each purchase to reconnect as necessary
        }
    })
}

Here, the release of the mutex corresponds to the end of connectOrThrow() for whichever other coroutine was holding it, so it's released as soon as the connection succeeds. If that other connection fails, this method will attempt the connection itself, and will succeed or fail on its own so the caller will be notified in case of error.

If you prefer to deal with result codes directly in if statements, you can return results instead of throwing:

private val billingConnectionMutex = Mutex()

private val resultAlreadyConnected = BillingResult.newBuilder()
    .setResponseCode(BillingClient.BillingResponseCode.OK)
    .setDebugMessage("Billing client is already connected")
    .build()

/**
 * Returns immediately if this BillingClient is already connected, otherwise
 * initiates the connection and suspends until this client is connected.
 * If a connection is already in the process of being established, this
 * method just suspends until the billing client is ready.
 */
suspend fun BillingClient.connect(): BillingResult = billingConnectionMutex.withLock {
    if (isReady) {
        // fast path: avoid suspension if already connected
        resultAlreadyConnected
    } else {
        unsafeConnect()
    }
}

private suspend fun BillingClient.unsafeConnect() = suspendCoroutine<BillingResult> { cont ->
    startConnection(object : BillingClientStateListener {
        override fun onBillingSetupFinished(billingResult: BillingResult) {
            cont.resume(billingResult)
        }
        override fun onBillingServiceDisconnected() {
            // no need to setup reconnection logic here, call ensureReady()
            // before each purchase to reconnect as necessary
        }
    })
}