RxConnectionCheckehelp observe network state on android with RxJava Subjects

RxConnectionChecker.kt

import android.annotation.SuppressLint
import android.net.ConnectivityManager
import android.net.Network
import android.net.NetworkCapabilities
import android.net.NetworkRequest
import android.os.Build
import io.reactivex.rxjava3.core.Observable
import io.reactivex.rxjava3.subjects.BehaviorSubject
import timber.log.Timber
import java.util.concurrent.atomic.AtomicInteger

class ConnectionCheckerImpl(
    private val connectivityManager: ConnectivityManager
) : ConnectionChecker {

    override val connectionState: BehaviorSubject<Boolean> = BehaviorSubject.create()
    private val connectionsCount = AtomicInteger(0)

    init {
        registerCallBack()
    }

    @SuppressLint("MissingPermission")
    override fun isNetworkAvailable(): Boolean {
        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) {
            val capabilities: NetworkCapabilities? =
                connectivityManager.getNetworkCapabilities(connectivityManager.activeNetwork)
            if (capabilities != null) {
                return when {
                    capabilities.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR) -> true
                    capabilities.hasTransport(NetworkCapabilities.TRANSPORT_WIFI) -> true
                    capabilities.hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET) -> true
                    else -> false
                }
            }
        } else {
            try {
                val activeNetworkInfo = connectivityManager.activeNetworkInfo
                if (activeNetworkInfo != null && activeNetworkInfo.isConnected) {
                    return true
                }
            } catch (e: Exception) {
                Timber.w(e)
            }
        }
        notifyObservers()
        return false
    }

    private fun notifyObservers() {
        if (connectionState.hasObservers()) {
            connectionState.onNext(connectionsCount.get() > 0)
        }
    }

    @SuppressLint("MissingPermission")
    private fun registerCallBack() {

        val request = NetworkRequest.Builder()
            .addTransportType(NetworkCapabilities.TRANSPORT_CELLULAR)
            .addTransportType(NetworkCapabilities.TRANSPORT_WIFI)
            .addTransportType(NetworkCapabilities.TRANSPORT_ETHERNET)
            .build()

        val callback = object : ConnectivityManager.NetworkCallback() {
            override fun onAvailable(network: Network) {
                super.onAvailable(network)
                connectionsCount.incrementAndGet()
                notifyObservers()
            }

            override fun onLost(network: Network) {
                super.onLost(network)
                connectionsCount.decrementAndGet()
                notifyObservers()
            }
        }

        connectivityManager.registerNetworkCallback(request, callback)
    }
}

/**
 * @property connectionState - subscribe to monitor
 * network state in real time
 * @property isNetworkAvailable - single check for network state
 **/
interface ConnectionChecker {
    val connectionState: Observable<Boolean>
    fun isNetworkAvailable(): Boolean
}