feat: offline support

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

@@ -1,12 +1,14 @@
 package com.amplitude.android
 
+import AndroidNetworkListener
 import android.content.Context
 import com.amplitude.android.migration.ApiKeyStorageMigration
 import com.amplitude.android.migration.RemnantDataMigration
 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.utilities.AndroidNetworkConnectivityChecker
 import com.amplitude.core.Amplitude
 import com.amplitude.core.events.BaseEvent
 import com.amplitude.core.platform.plugins.AmplitudeDestination
@@ -16,11 +18,21 @@ import com.amplitude.id.IdentityConfiguration
 import kotlinx.coroutines.launch
 
 open class Amplitude(
-    configuration: Configuration
+    configuration: Configuration,
 ) : Amplitude(configuration) {
-
     internal var inForeground = false
     private lateinit var androidContextPlugin: AndroidContextPlugin
+    private var networkListener: AndroidNetworkListener
+    private val networkChangeHandler =
+        object : AndroidNetworkListener.NetworkChangeCallback {
+            override fun onNetworkAvailable() {
+                flush()
+            }
+
+            override fun onNetworkUnavailable() {
+                // Nothing to do so far
+            }
+        }
 
     val sessionId: Long
         get() {
@@ -29,6 +41,9 @@ open class Amplitude(
 
     init {
         registerShutdownHook()
+        networkListener = AndroidNetworkListener((this.configuration as Configuration).context)
+        networkListener.setNetworkChangeCallback(networkChangeHandler)
+        networkListener.startListening()
     }
 
     override fun createTimeline(): Timeline {
@@ -62,7 +77,7 @@ open class Amplitude(
         add(AndroidLifecyclePlugin())
         add(AnalyticsConnectorIdentityPlugin())
         add(AnalyticsConnectorPlugin())
-        add(AmplitudeDestination())
+        add(AmplitudeDestination(AndroidNetworkConnectivityChecker(this.configuration.context, this.logger)))
 
         (timeline as Timeline).start()
     }
@@ -113,11 +128,14 @@ open class Amplitude(
     }
 
     private fun registerShutdownHook() {
-        Runtime.getRuntime().addShutdownHook(object : Thread() {
-            override fun run() {
-                ([email protected] as Timeline).stop()
-            }
-        })
+        Runtime.getRuntime().addShutdownHook(
+            object : Thread() {
+                override fun run() {
+                    ([email protected] as Timeline).stop()
+                    ([email protected] as AndroidNetworkListener).stopListening()
+                }
+            },
+        )
     }
 
     companion object {

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

@@ -0,0 +1,47 @@
+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
+import com.amplitude.core.platform.NetworkConnectivityChecker
+
+class AndroidNetworkConnectivityChecker(private val context: Context, private val logger: Logger) : NetworkConnectivityChecker {
+    companion object {
+        private const val ACCESS_NETWORK_STATE = "android.permission.ACCESS_NETWORK_STATE"
+    }
+
+    @SuppressLint("MissingPermission")
+    override suspend fun isConnected(): Boolean {
+        // Assume connection and proceed.
+        // Events will be treated like online
+        // regardless network connectivity
+        if (!hasPermission(context, ACCESS_NETWORK_STATE)) {
+            logger.warn("No ACCESS_NETWORK_STATE permission, offline mode is not supported.")
+            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,15 +85,19 @@ class AndroidLifecyclePluginTest {
         every { anyConstructed<AndroidContextProvider>().mostRecentLocation } returns null
         every { anyConstructed<AndroidContextProvider>().appSetId } returns ""
 
-        configuration = Configuration(
-            apiKey = "api-key",
-            context = mockedContext,
-            storageProvider = InMemoryStorageProvider(),
-            loggerProvider = ConsoleLoggerProvider(),
-            identifyInterceptStorageProvider = InMemoryStorageProvider(),
-            identityStorageProvider = IMIdentityStorageProvider(),
-            trackingSessionEvents = false,
-        )
+        connectivityManager = mockk<ConnectivityManager>(relaxed = true)
+        every { mockedContext!!.getSystemService(Context.CONNECTIVITY_SERVICE) } returns connectivityManager
+
+        configuration =
+            Configuration(
+                apiKey = "api-key",
+                context = mockedContext,
+                storageProvider = InMemoryStorageProvider(),
+                loggerProvider = ConsoleLoggerProvider(),
+                identifyInterceptStorageProvider = InMemoryStorageProvider(),
+                identityStorageProvider = IMIdentityStorageProvider(),
+                trackingSessionEvents = false,
+            )
         amplitude = Amplitude(configuration)
     }
 

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

@@ -14,8 +14,13 @@ import kotlinx.coroutines.withContext
 import java.io.FileNotFoundException
 import java.util.concurrent.atomic.AtomicInteger
 
+interface NetworkConnectivityChecker {
+    suspend fun isConnected(): Boolean
+}
+
 class EventPipeline(
-    private val amplitude: Amplitude
+    private val amplitude: Amplitude,
+    private val networkConnectivityChecker: NetworkConnectivityChecker? = null
 ) {
 
     private val writeChannel: Channel<WriteQueueMessage>
@@ -99,6 +104,13 @@ class EventPipeline(
                 }
             }
 
+            // Skip flush when offline only if
+            // network connectivity is not null
+            // and network is not connected.
+            if (networkConnectivityChecker?.isConnected() == false) {
+                continue
+            }
+
             // if flush condition met, generate paths
             if (eventCount.incrementAndGet() >= getFlushCount() || triggerFlush) {
                 eventCount.set(0)

core/src/main/java/com/amplitude/core/platform/plugins/AmplitudeDestination.kt

@@ -7,10 +7,11 @@ import com.amplitude.core.events.IdentifyEvent
 import com.amplitude.core.events.RevenueEvent
 import com.amplitude.core.platform.DestinationPlugin
 import com.amplitude.core.platform.EventPipeline
+import com.amplitude.core.platform.NetworkConnectivityChecker
 import com.amplitude.core.platform.intercept.IdentifyInterceptor
 import kotlinx.coroutines.launch
 
-class AmplitudeDestination : DestinationPlugin() {
+class AmplitudeDestination(private val networkConnectivityChecker: NetworkConnectivityChecker? = null) : DestinationPlugin() {
     private lateinit var pipeline: EventPipeline
     private lateinit var identifyInterceptor: IdentifyInterceptor
 
@@ -66,7 +67,8 @@ class AmplitudeDestination : DestinationPlugin() {
 
         with(amplitude) {
             pipeline = EventPipeline(
-                amplitude
+                amplitude,
+                networkConnectivityChecker
             )
             pipeline.start()
             identifyInterceptor = IdentifyInterceptor(