feat: offline support

android/src/main/java/com/amplitude/android/Amplitude.kt

@@ -7,6 +7,7 @@ import com.amplitude.android.plugins.AnalyticsConnectorIdentityPlugin
 import com.amplitude.android.plugins.AnalyticsConnectorPlugin
 import com.amplitude.android.plugins.AndroidContextPlugin
 import com.amplitude.android.plugins.AndroidLifecyclePlugin
+import com.amplitude.android.plugins.AndroidNetworkConnectivityCheckerPlugin
 import com.amplitude.core.Amplitude
 import com.amplitude.core.events.BaseEvent
 import com.amplitude.core.platform.plugins.AmplitudeDestination
@@ -56,6 +57,7 @@ open class Amplitude(
         }
         this.createIdentityContainer(identityConfiguration)
 
+        add(AndroidNetworkConnectivityCheckerPlugin())
         androidContextPlugin = AndroidContextPlugin()
         add(androidContextPlugin)
         add(GetAmpliExtrasPlugin())

android/src/main/java/com/amplitude/android/plugins/AndroidNetworkConnectivityCheckerPlugin.kt

@@ -0,0 +1,44 @@
+package com.amplitude.android.plugins
+
+import AndroidNetworkListener
+import com.amplitude.android.Configuration
+import com.amplitude.android.utilities.AndroidNetworkConnectivityChecker
+import com.amplitude.core.Amplitude
+import com.amplitude.core.events.BaseEvent
+import com.amplitude.core.platform.Plugin
+import kotlinx.coroutines.launch
+
+class AndroidNetworkConnectivityCheckerPlugin : Plugin {
+    override val type: Plugin.Type = Plugin.Type.Before
+    override lateinit var amplitude: Amplitude
+    private lateinit var networkConnectivityChecker: AndroidNetworkConnectivityChecker
+    private lateinit var networkListener: AndroidNetworkListener
+    private val networkChangeHandler =
+        object : AndroidNetworkListener.NetworkChangeCallback {
+            override fun onNetworkAvailable() {
+                println("AndroidNetworkListener, onNetworkAvailable")
+                amplitude.configuration.offline = false
+                amplitude.flush()
+            }
+
+            override fun onNetworkUnavailable() {
+                println("AndroidNetworkListener, onNetworkUnavailable")
+                amplitude.configuration.offline = true
+            }
+        }
+
+    override fun setup(amplitude: Amplitude) {
+        super.setup(amplitude)
+        networkConnectivityChecker = AndroidNetworkConnectivityChecker((amplitude.configuration as Configuration).context, amplitude.logger)
+        networkListener = AndroidNetworkListener((amplitude.configuration as Configuration).context)
+        networkListener.setNetworkChangeCallback(networkChangeHandler)
+        networkListener.startListening()
+    }
+
+    override fun execute(event: BaseEvent): BaseEvent? {
+        amplitude.amplitudeScope.launch(amplitude.storageIODispatcher) {
+            amplitude.configuration.offline = !networkConnectivityChecker.isConnected()
+        }
+        return super.execute(event)
+    }
+}

android/src/main/java/com/amplitude/android/utilities/AndroidNetworkConnectivityChecker.kt

@@ -0,0 +1,49 @@
+package com.amplitude.android.utilities
+
+import android.annotation.SuppressLint
+import android.content.Context
+import android.content.pm.PackageManager
+import android.net.ConnectivityManager
+import android.net.NetworkCapabilities
+import android.os.Build
+import com.amplitude.common.Logger
+
+class AndroidNetworkConnectivityChecker(private val context: Context, private val logger: Logger) {
+    companion object {
+        private const val ACCESS_NETWORK_STATE = "android.permission.ACCESS_NETWORK_STATE"
+    }
+
+    @SuppressLint("MissingPermission")
+    fun isConnected(): Boolean {
+        // Assume connection and proceed.
+        // Events will be treated like online
+        // regardless network connectivity
+        if (!hasPermission(context, ACCESS_NETWORK_STATE)) {
+            logger.warn(
+                @Suppress("ktlint:standard:max-line-length")
+                "No ACCESS_NETWORK_STATE permission, offline mode is not supported. To enable, add <uses-permission android:name=\"android.permission.ACCESS_NETWORK_STATE\" /> to your AndroidManifest.xml. Learn more at https://www.docs.developers.amplitude.com/data/sdks/android-kotlin/#offline-mode",
+            )
+            return true
+        }
+
+        val cm = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
+        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) {
+            val network = cm.activeNetwork ?: return false
+            val capabilities = cm.getNetworkCapabilities(network) ?: return false
+
+            return capabilities.hasTransport(NetworkCapabilities.TRANSPORT_WIFI) ||
+                capabilities.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR)
+        } else {
+            @SuppressLint("MissingPermission")
+            val networkInfo = cm.activeNetworkInfo
+            return networkInfo != null && networkInfo.isConnectedOrConnecting
+        }
+    }
+
+    private fun hasPermission(
+        context: Context,
+        permission: String,
+    ): Boolean {
+        return context.checkCallingOrSelfPermission(permission) == PackageManager.PERMISSION_GRANTED
+    }
+}

android/src/main/java/com/amplitude/android/utilities/AndroidNetworkListener.kt

@@ -0,0 +1,100 @@
+import android.annotation.SuppressLint
+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.net.NetworkCapabilities
+import android.net.NetworkRequest
+import android.os.Build
+import java.lang.IllegalArgumentException
+
+class AndroidNetworkListener(private val context: Context) {
+    private var networkCallback: NetworkChangeCallback? = null
+    private var networkCallbackForLowerApiLevels: BroadcastReceiver? = null
+    private var networkCallbackForHigherApiLevels: ConnectivityManager.NetworkCallback? = null
+
+    interface NetworkChangeCallback {
+        fun onNetworkAvailable()
+
+        fun onNetworkUnavailable()
+    }
+
+    fun setNetworkChangeCallback(callback: NetworkChangeCallback) {
+        this.networkCallback = callback
+    }
+
+    fun startListening() {
+        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.LOLLIPOP) {
+            setupNetworkCallback()
+        } else {
+            setupBroadcastReceiver()
+        }
+    }
+
+    @SuppressLint("NewApi", "MissingPermission")
+    // startListening() checks API level
+    // ACCESS_NETWORK_STATE permission should be added manually by users to enable this feature
+    private fun setupNetworkCallback() {
+        val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
+        val networkRequest =
+            NetworkRequest.Builder()
+                .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
+                .build()
+
+        networkCallbackForHigherApiLevels =
+            object : ConnectivityManager.NetworkCallback() {
+                override fun onAvailable(network: Network) {
+                    networkCallback?.onNetworkAvailable()
+                }
+
+                override fun onLost(network: Network) {
+                    networkCallback?.onNetworkUnavailable()
+                }
+            }
+
+        connectivityManager.registerNetworkCallback(networkRequest, networkCallbackForHigherApiLevels!!)
+    }
+
+    private fun setupBroadcastReceiver() {
+        networkCallbackForLowerApiLevels =
+            object : BroadcastReceiver() {
+                @SuppressLint("MissingPermission")
+                override fun onReceive(
+                    context: Context,
+                    intent: Intent,
+                ) {
+                    if (ConnectivityManager.CONNECTIVITY_ACTION == intent.action) {
+                        val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
+                        val activeNetwork = connectivityManager.activeNetworkInfo
+                        val isConnected = activeNetwork?.isConnectedOrConnecting == true
+
+                        if (isConnected) {
+                            networkCallback?.onNetworkAvailable()
+                        } else {
+                            networkCallback?.onNetworkUnavailable()
+                        }
+                    }
+                }
+            }
+
+        val filter = IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION)
+        context.registerReceiver(networkCallbackForLowerApiLevels, filter)
+    }
+
+    fun stopListening() {
+        try {
+            if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.LOLLIPOP) {
+                val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
+                networkCallbackForHigherApiLevels?.let { connectivityManager.unregisterNetworkCallback(it) }
+            } else {
+                networkCallbackForLowerApiLevels?.let { context.unregisterReceiver(it) }
+            }
+        } catch (e: IllegalArgumentException) {
+            // callback was already unregistered.
+        } catch (e: IllegalStateException) {
+            // shutdown process is in progress and certain operations are not allowed.
+        }
+    }
+}

android/src/test/java/com/amplitude/android/AmplitudeRobolectricTests.kt

@@ -2,6 +2,7 @@ package com.amplitude.android
 
 import android.app.Application
 import android.content.Context
+import android.net.ConnectivityManager
 import com.amplitude.core.events.BaseEvent
 import com.amplitude.core.utilities.ConsoleLoggerProvider
 import com.amplitude.id.IMIdentityStorageProvider
@@ -23,15 +24,17 @@ import kotlin.io.path.absolutePathString
 class AmplitudeRobolectricTests {
     private lateinit var amplitude: Amplitude
     private var context: Context? = null
+    private lateinit var connectivityManager: ConnectivityManager
 
     var tempDir = TempDirectory()
 
     @ExperimentalCoroutinesApi
     @Before
     fun setup() {
         context = mockk<Application>(relaxed = true)
+        connectivityManager = mockk<ConnectivityManager>(relaxed = true)
         every { context!!.getDir(any(), any()) } returns File(tempDir.create("data").absolutePathString())
-
+        every { context!!.getSystemService(Context.CONNECTIVITY_SERVICE) } returns connectivityManager
         amplitude = Amplitude(createConfiguration())
     }
 

android/src/test/java/com/amplitude/android/AmplitudeSessionTest.kt

@@ -1,6 +1,8 @@
 package com.amplitude.android
 
 import android.app.Application
+import android.content.Context
+import android.net.ConnectivityManager
 import com.amplitude.android.plugins.AndroidLifecyclePlugin
 import com.amplitude.common.android.AndroidContextProvider
 import com.amplitude.core.Storage
@@ -64,6 +66,8 @@ class AmplitudeSessionTest {
 
     private fun createConfiguration(storageProvider: StorageProvider? = null): Configuration {
         val context = mockk<Application>(relaxed = true)
+        var connectivityManager = mockk<ConnectivityManager>(relaxed = true)
+        every { context!!.getSystemService(Context.CONNECTIVITY_SERVICE) } returns connectivityManager
 
         return Configuration(
             apiKey = "api-key",

android/src/test/java/com/amplitude/android/AmplitudeTest.kt

@@ -2,6 +2,7 @@ package com.amplitude.android
 
 import android.app.Application
 import android.content.Context
+import android.net.ConnectivityManager
 import com.amplitude.analytics.connector.AnalyticsConnector
 import com.amplitude.analytics.connector.Identity
 import com.amplitude.android.plugins.AndroidLifecyclePlugin
@@ -38,10 +39,14 @@ open class StubPlugin : EventPlugin {
 class AmplitudeTest {
     private var context: Context? = null
     private var amplitude: Amplitude? = null
+    private lateinit var connectivityManager: ConnectivityManager
 
     @BeforeEach
     fun setUp() {
         context = mockk<Application>(relaxed = true)
+        connectivityManager = mockk<ConnectivityManager>(relaxed = true)
+        every { context!!.getSystemService(Context.CONNECTIVITY_SERVICE) } returns connectivityManager
+
         mockkStatic(AndroidLifecyclePlugin::class)
 
         mockkConstructor(AndroidContextProvider::class)

android/src/test/java/com/amplitude/android/plugins/AndroidLifecyclePluginTest.kt

@@ -2,10 +2,12 @@ package com.amplitude.android.plugins
 
 import android.app.Activity
 import android.app.Application
+import android.content.Context
 import android.content.Intent
 import android.content.pm.ActivityInfo
 import android.content.pm.PackageInfo
 import android.content.pm.PackageManager
+import android.net.ConnectivityManager
 import android.net.Uri
 import android.os.Bundle
 import com.amplitude.android.Amplitude
@@ -44,6 +46,7 @@ class AndroidLifecyclePluginTest {
 
     private val mockedContext = mockk<Application>(relaxed = true)
     private var mockedPackageManager: PackageManager
+    private lateinit var connectivityManager: ConnectivityManager
 
     init {
         val packageInfo = PackageInfo()
@@ -82,6 +85,9 @@ class AndroidLifecyclePluginTest {
         every { anyConstructed<AndroidContextProvider>().mostRecentLocation } returns null
         every { anyConstructed<AndroidContextProvider>().appSetId } returns ""
 
+        connectivityManager = mockk<ConnectivityManager>(relaxed = true)
+        every { mockedContext!!.getSystemService(Context.CONNECTIVITY_SERVICE) } returns connectivityManager
+
         configuration = Configuration(
             apiKey = "api-key",
             context = mockedContext,

core/src/main/java/com/amplitude/core/Configuration.kt

@@ -32,6 +32,8 @@ open class Configuration @JvmOverloads constructor(
     open var identityStorageProvider: IdentityStorageProvider = IMIdentityStorageProvider(),
 ) {
 
+    var offline: Boolean = false
+
     companion object {
         const val FLUSH_QUEUE_SIZE = 30
         const val FLUSH_INTERVAL_MILLIS = 30 * 1000 // 30s

core/src/main/java/com/amplitude/core/platform/EventPipeline.kt

@@ -99,6 +99,11 @@ class EventPipeline(
                 }
             }
 
+            // Skip flush when offline
+            if (amplitude.configuration.offline) {
+                continue
+            }
+
             // if flush condition met, generate paths
             if (eventCount.incrementAndGet() >= getFlushCount() || triggerFlush) {
                 eventCount.set(0)