feat: implement connectivity observer

android/src/main/kotlin/com/rudderstack/sdk/kotlin/android/Analytics.kt

@@ -3,6 +3,7 @@ package com.rudderstack.sdk.kotlin.android
 import android.app.Activity
 import androidx.navigation.NavController
 import androidx.navigation.NavController.OnDestinationChangedListener
+import com.rudderstack.sdk.kotlin.android.connectivity.AndroidConnectivityObserverPlugin
 import com.rudderstack.sdk.kotlin.android.logger.AndroidLogger
 import com.rudderstack.sdk.kotlin.android.plugins.AndroidLifecyclePlugin
 import com.rudderstack.sdk.kotlin.android.plugins.AppInfoPlugin
@@ -203,6 +204,7 @@ class Analytics(
 
     private fun setup() {
         setLogger(logger = AndroidLogger())
+        add(AndroidConnectivityObserverPlugin(connectivityState))
         add(DeviceInfoPlugin())
         add(AppInfoPlugin())
         add(NetworkInfoPlugin())

android/src/main/kotlin/com/rudderstack/sdk/kotlin/android/connectivity/AndroidConnectivityObserverPlugin.kt

@@ -0,0 +1,118 @@
+package com.rudderstack.sdk.kotlin.android.connectivity
+
+import android.content.BroadcastReceiver
+import android.content.Context
+import android.content.Intent
+import android.content.IntentFilter
+import android.net.ConnectivityManager
+import android.net.Network
+import android.os.Build
+import androidx.annotation.VisibleForTesting
+import com.rudderstack.sdk.kotlin.android.utils.application
+import com.rudderstack.sdk.kotlin.android.utils.runBasedOnSDK
+import com.rudderstack.sdk.kotlin.core.Analytics
+import com.rudderstack.sdk.kotlin.core.internals.models.connectivity.ConnectivityState
+import com.rudderstack.sdk.kotlin.core.internals.plugins.Plugin
+import com.rudderstack.sdk.kotlin.core.internals.statemanagement.FlowState
+import com.rudderstack.sdk.kotlin.core.internals.utils.defaultExceptionHandler
+import com.rudderstack.sdk.kotlin.core.internals.utils.safelyExecute
+
+private const val MIN_SUPPORTED_VERSION = Build.VERSION_CODES.N
+
+/**
+ * Plugin to observe the network connectivity state of the Android device.
+ *
+ * It uses [ConnectivityManager] to observe the network connectivity changes for Android API level 24 and above.
+ * For lower API levels, it uses [BroadcastReceiver] to observe the network connectivity changes.
+ *
+ * In case of any exception while registering the connectivity observers, it sets the connection availability to `true`.
+ *
+ * @param connectivityState The state management for connectivity.
+ */
+@Suppress("MaximumLineLength")
+internal class AndroidConnectivityObserverPlugin(
+    private val connectivityState: FlowState<Boolean>
+) : Plugin {
+
+    override val pluginType: Plugin.PluginType = Plugin.PluginType.PreProcess
+    override lateinit var analytics: Analytics
+
+    private var connectivityManager: ConnectivityManager? = null
+    private var intentFilter: IntentFilter? = null
+
+    private val networkCallback by lazy { createNetworkCallback(connectivityState) }
+    private val broadcastReceiver by lazy { createBroadcastReceiver(connectivityState) }
+
+    override fun setup(analytics: Analytics) {
+        super.setup(analytics)
+
+        safelyExecute(
+            block = { registerConnectivityObserver() },
+            onException = { exception ->
+                defaultExceptionHandler(
+                    errorMsg = "Failed to register connectivity subscriber. Setting network availability to true. Exception:",
+                    exception = exception
+                )
+                connectivityState.dispatch(ConnectivityState.SetDefaultStateAction())
+            },
+        )
+    }
+
+    // Suppressing deprecation warning as we need to support lower API levels.
+    @Suppress("DEPRECATION")
+    @Throws(RuntimeException::class)
+    private fun registerConnectivityObserver() {
+        runBasedOnSDK(
+            minCompatibleVersion = MIN_SUPPORTED_VERSION,
+            onCompatibleVersion = {
+                connectivityManager = this.analytics.application.getSystemService(ConnectivityManager::class.java)
+                connectivityManager?.registerDefaultNetworkCallback(networkCallback)
+            },
+            onLegacyVersion = {
+                intentFilter = IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION)
+                this.analytics.application.registerReceiver(broadcastReceiver, intentFilter)
+            },
+        )
+    }
+
+    override fun teardown() {
+        runBasedOnSDK(
+            minCompatibleVersion = MIN_SUPPORTED_VERSION,
+            onCompatibleVersion = { this.connectivityManager?.unregisterNetworkCallback(networkCallback) },
+            onLegacyVersion = {
+                this.intentFilter?.let {
+                    this.analytics.application.unregisterReceiver(broadcastReceiver)
+                }
+            },
+        )
+    }
+}
+
+@VisibleForTesting
+internal fun createNetworkCallback(connectivityState: FlowState<Boolean>) = object : ConnectivityManager.NetworkCallback() {
+    override fun onAvailable(network: Network) {
+        super.onAvailable(network)
+        connectivityState.dispatch(ConnectivityState.EnableConnectivityAction())
+    }
+
+    override fun onLost(network: Network) {
+        super.onLost(network)
+        connectivityState.dispatch(ConnectivityState.DisableConnectivityAction())
+    }
+}
+
+@VisibleForTesting
+// Suppressing deprecation warning as we need to support lower API levels.
+@Suppress("DEPRECATION")
+internal fun createBroadcastReceiver(connectivityState: FlowState<Boolean>) = object : BroadcastReceiver() {
+    override fun onReceive(context: Context, intent: Intent?) {
+        (context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager).activeNetworkInfo?.let {
+            when (it.isConnected) {
+                true -> connectivityState.dispatch(ConnectivityState.EnableConnectivityAction())
+                false -> connectivityState.dispatch(ConnectivityState.DisableConnectivityAction())
+            }
+        } ?: run { // if activeNetworkInfo is null, it means the device is not connected to any network.
+            connectivityState.dispatch(ConnectivityState.DisableConnectivityAction())
+        }
+    }
+}

android/src/main/kotlin/com/rudderstack/sdk/kotlin/android/utils/AnalyticsUtils.kt

@@ -1,5 +1,7 @@
 package com.rudderstack.sdk.kotlin.android.utils
 
+import android.app.Application
+import com.rudderstack.sdk.kotlin.android.Configuration
 import com.rudderstack.sdk.kotlin.core.Analytics
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.Dispatchers
@@ -26,3 +28,14 @@ internal fun Analytics.runOnAnalyticsThread(block: suspend () -> Unit) = analyti
 internal fun AndroidAnalytics.runOnMainThread(block: suspend () -> Unit) = analyticsScope.launch(MAIN_DISPATCHER) {
     block()
 }
+
+/**
+ * Provides access to the [Application] instance associated with the Android Analytics object.
+ *
+ * This property retrieves the application instance from the [Configuration] of the Android Analytics object,
+ * ensuring that the correct application context is available for performing various SDK operations.
+ *
+ * @return The [Application] instance tied to the current [Analytics] configuration.
+ */
+internal val Analytics.application: Application
+    get() = (this.configuration as Configuration).application

android/src/main/kotlin/com/rudderstack/sdk/kotlin/android/utils/AndroidUtils.kt

@@ -0,0 +1,39 @@
+package com.rudderstack.sdk.kotlin.android.utils
+
+import android.os.Build
+import androidx.annotation.ChecksSdkIntAtLeast
+import com.rudderstack.sdk.kotlin.android.utils.AppSDKVersion.getVersionSDKInt
+
+/**
+ * Executes the provided lambda function based on the current SDK version.
+ *
+ * This function compares the current SDK version with a specified minimum compatible version.
+ * If the current SDK version is greater than or equal to the minimum version,
+ * it runs the `onCompatibleVersion` lambda. Otherwise, it runs the `onLegacyVersion` lambda.
+ *
+ * @param minCompatibleVersion The minimum SDK version required to execute `onCompatibleVersion`.
+ * @param onCompatibleVersion A lambda function to execute if the current SDK version is compatible.
+ * @param onLegacyVersion A lambda function to execute if the current SDK version is below the required minimum.
+ */
+@ChecksSdkIntAtLeast(parameter = 0, lambda = 1)
+internal inline fun runBasedOnSDK(minCompatibleVersion: Int, onCompatibleVersion: () -> Unit, onLegacyVersion: () -> Unit,) {
+    if (getVersionSDKInt() >= minCompatibleVersion) {
+        onCompatibleVersion()
+    } else {
+        onLegacyVersion()
+    }
+}
+
+/**
+ * A utility object to retrieve the current SDK version of the Android platform.
+ *
+ * This wrapper around `Build.VERSION.SDK_INT` allows for easier testing and mocking
+ * during unit tests. By encapsulating the SDK version retrieval in a separate function,
+ * it becomes possible to mock `getVersionSDKInt()` and simulate different SDK versions
+ * without relying on the actual device or emulator environment.
+ */
+internal object AppSDKVersion {
+    fun getVersionSDKInt(): Int {
+        return Build.VERSION.SDK_INT
+    }
+}