feat: add error handling and retry

android/build.gradle

@@ -40,10 +40,13 @@ android {
 }
 
 dependencies {
+    implementation project(':id')
     implementation project(':core')
     implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.2'
     implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.5.2'
 
+    coreLibraryDesugaring 'com.android.tools:desugar_jdk_libs:1.1.5'
+
     testImplementation 'io.mockk:mockk:1.10.6'
     testImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-test:1.3.2'
     androidTestImplementation 'androidx.test.ext:junit:1.1.3'

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

@@ -2,12 +2,18 @@ package com.amplitude.android
 
 import com.amplitude.core.Amplitude
 import com.amplitude.core.platform.plugins.AmplitudeDestination
+import com.amplitude.core.utilities.AnalyticsIdentityListener
+import com.amplitude.id.FileIdentityStorageProvider
+import com.amplitude.id.IdentityConfiguration
+import com.amplitude.id.IdentityContainer
 
 open class Amplitude(
     configuration: Configuration
 ): Amplitude(configuration) {
 
     override fun build() {
+        idContainer = IdentityContainer.getInstance(IdentityConfiguration(instanceName = configuration.instanceName, apiKey = configuration.apiKey, identityStorageProvider = FileIdentityStorageProvider()))
+        idContainer.identityManager.addIdentityListener(AnalyticsIdentityListener(store))
         add(AmplitudeDestination())
     }
 }

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

@@ -6,19 +6,22 @@ import com.amplitude.core.LoggerProvider
 import com.amplitude.core.StorageProvider
 import com.amplitude.android.utilities.AndroidLoggerProvider
 import com.amplitude.android.utilities.AndroidStorageProvider
+import com.amplitude.core.EventCallBack
 import com.amplitude.core.events.BaseEvent
+import com.amplitude.core.utilities.FileStorageProvider
 
 class Configuration(
     apiKey: String,
     context: Context,
     flushQueueSize: Int = FLUSH_QUEUE_SIZE,
     flushIntervalMillis: Int = FLUSH_INTERVAL_MILLIS,
+    instanceName: String = DEFAULT_INSTANCE,
     optOut: Boolean = false,
-    storageProvider: StorageProvider = AndroidStorageProvider(),
+    storageProvider: StorageProvider = FileStorageProvider(),
     loggerProvider: LoggerProvider = AndroidLoggerProvider(),
     minIdLength: Int? = null,
-    callback: ((BaseEvent) -> Unit)? = null,
+    callback: EventCallBack? = null,
     useAdvertisingIdForDeviceId: Boolean = false,
     useAppSetIdForDeviceId: Boolean = false,
     enableCoppaControl: Boolean = false
-) : Configuration(apiKey, flushQueueSize, flushIntervalMillis, optOut, storageProvider, loggerProvider, minIdLength, callback)
+) : Configuration(apiKey, flushQueueSize, flushIntervalMillis, instanceName, optOut, storageProvider, loggerProvider, minIdLength, callback)

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

@@ -4,19 +4,32 @@ import com.amplitude.core.Amplitude
 import com.amplitude.core.Storage
 import com.amplitude.core.StorageProvider
 import com.amplitude.core.events.BaseEvent
+import com.amplitude.core.utilities.FileStorageProvider
 
 class AndroidStorage(
     val amplitude: Amplitude
 ) : Storage {
-    override fun write(event: BaseEvent) {
+    override suspend fun writeEvent(event: BaseEvent) {
         TODO("Not yet implemented")
     }
 
-    override fun rollover() {
+    override suspend fun write(key: Storage.Constants, value: String) {
         TODO("Not yet implemented")
     }
 
-    override fun getEvents(): List<String> {
+    override suspend fun rollover() {
+        TODO("Not yet implemented")
+    }
+
+    override fun read(key: Storage.Constants): String? {
+        TODO("Not yet implemented")
+    }
+
+    override fun readEventsContent(): List<Any> {
+        TODO("Not yet implemented")
+    }
+
+    override fun getEventsString(content: Any): String {
         TODO("Not yet implemented")
     }
 }

core/build.gradle

@@ -16,6 +16,7 @@ repositories {
 
 dependencies {
     implementation "org.jetbrains.kotlin:kotlin-stdlib:$kotlin_version"
+    implementation project(':id')
     // MAIN DEPS
     compileOnly 'org.json:json:20211205'
     implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.2'

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

@@ -4,10 +4,15 @@ import com.amplitude.core.events.BaseEvent
 import com.amplitude.core.events.Identify
 import com.amplitude.core.events.Revenue
 import com.amplitude.core.events.RevenueEvent
+import com.amplitude.core.platform.ObservePlugin
 import com.amplitude.core.platform.Plugin
 import com.amplitude.core.platform.Timeline
 import com.amplitude.core.platform.plugins.AmplitudeDestination
 import com.amplitude.core.platform.plugins.ContextPlugin
+import com.amplitude.core.utilities.AnalyticsIdentityListener
+import com.amplitude.id.IMIdentityStorageProvider
+import com.amplitude.id.IdentityConfiguration
+import com.amplitude.id.IdentityContainer
 import kotlinx.coroutines.*
 import java.util.concurrent.Executors
 
@@ -17,12 +22,14 @@ open class Amplitude internal constructor(
     val amplitudeScope: CoroutineScope = CoroutineScope(SupervisorJob()),
     val amplitudeDispatcher: CoroutineDispatcher = Executors.newCachedThreadPool().asCoroutineDispatcher(),
     val networkIODispatcher: CoroutineDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher(),
-    val storageIODispatcher: CoroutineDispatcher = Executors.newFixedThreadPool(2).asCoroutineDispatcher()
+    val storageIODispatcher: CoroutineDispatcher = Executors.newFixedThreadPool(2).asCoroutineDispatcher(),
+    val retryDispatcher: CoroutineDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
 ){
 
     internal val timeline: Timeline
     val storage: Storage
     val logger: Logger
+    protected lateinit var idContainer: IdentityContainer
 
     init {
         require(configuration.isValid()) { "invalid configuration" }
@@ -38,6 +45,8 @@ open class Amplitude internal constructor(
     constructor(configuration: Configuration) : this(configuration, State())
 
     open fun build() {
+        idContainer = IdentityContainer.getInstance(IdentityConfiguration(instanceName = configuration.instanceName, apiKey = configuration.apiKey, identityStorageProvider = IMIdentityStorageProvider()))
+        idContainer.identityManager.addIdentityListener(AnalyticsIdentityListener(store))
         add(ContextPlugin())
         add(AmplitudeDestination())
 
@@ -60,7 +69,11 @@ open class Amplitude internal constructor(
     }
 
     fun identify(userId: String) {
+        this.idContainer.identityManager.editIdentity().setUserId(userId).commit()
+    }
 
+    fun setDeviceId(deviceId: String) {
+        this.idContainer.identityManager.editIdentity().setUserId(deviceId).commit()
     }
 
     fun groupIdentify(identify: Identify) {
@@ -101,12 +114,26 @@ open class Amplitude internal constructor(
     }
 
     fun add(plugin: Plugin) : Amplitude {
-        this.timeline.add(plugin)
+        when (plugin) {
+            is ObservePlugin -> {
+                this.store.add(plugin)
+            }
+            else -> {
+                this.timeline.add(plugin)
+            }
+        }
         return this
     }
 
     fun remove(plugin: Plugin): Amplitude {
-        this.timeline.remove(plugin)
+        when (plugin) {
+            is ObservePlugin -> {
+                this.store.remove(plugin)
+            }
+            else -> {
+                this.timeline.remove(plugin)
+            }
+        }
         return this
     }
 

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

@@ -4,20 +4,26 @@ import com.amplitude.core.events.BaseEvent
 import com.amplitude.core.utilities.ConsoleLoggerProvider
 import com.amplitude.core.utilities.InMemoryStorageProvider
 
+typealias EventCallBack = (BaseEvent, status: Int, message: String) -> Unit
+
 open class Configuration(
     val apiKey: String,
     val flushQueueSize: Int = FLUSH_QUEUE_SIZE,
     val flushIntervalMillis: Int = FLUSH_INTERVAL_MILLIS,
+    val instanceName: String = DEFAULT_INSTANCE,
     val optOut: Boolean = false,
     val storageProvider: StorageProvider = InMemoryStorageProvider(),
     val loggerProvider: LoggerProvider = ConsoleLoggerProvider(),
     val minIdLength: Int? = null,
-    val callback: ((BaseEvent) -> Unit)? = null
+    val callback: EventCallBack? = null,
+    val flushMaxRetries: Int = FLUSH_MAX_RETRIES
 ) {
 
     companion object {
         const val FLUSH_QUEUE_SIZE = 30
         const val FLUSH_INTERVAL_MILLIS = 30 * 1000 // 30s
+        const val FLUSH_MAX_RETRIES = 5
+        const val DEFAULT_INSTANCE = "\$default_instance"
     }
 
     fun isValid(): Boolean {

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

@@ -1,5 +1,31 @@
 package com.amplitude.core
 
+import com.amplitude.core.platform.ObservePlugin
+
 class State {
+    var userId: String? = null
+        set(value: String?) {
+            userId = value
+            plugins.forEach { plugin ->
+                plugin.onUserIdChanged(value)
+            }
+        }
+
+    var deviceId: String? = null
+        set(value: String?) {
+            deviceId = value
+            plugins.forEach { plugin ->
+                plugin.onDeviceIdChanged(value)
+            }
+        }
+
+    val plugins: MutableList<ObservePlugin> = mutableListOf()
+
+    fun add(plugin: ObservePlugin) = synchronized(plugins) {
+        plugins.add(plugin)
+    }
 
-}
+    fun remove(plugin: ObservePlugin) = synchronized(plugins) {
+        plugins.removeAll { it === plugin }
+    }
+}

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

@@ -3,11 +3,26 @@ package com.amplitude.core
 import com.amplitude.core.events.BaseEvent
 
 interface Storage {
-    fun write(event: BaseEvent)
 
-    fun rollover()
+    enum class Constants(val rawVal: String) {
+        LAST_EVENT_ID("last_event_id"),
+        PREVIOUS_SESSION_ID("previous_session_id"),
+        LAST_EVENT_TIME("last_event_time"),
+        OPT_OUT("opt_out"),
+        Events("events")
+    }
 
-    fun getEvents(): List<String>
+    suspend fun writeEvent(event: BaseEvent)
+
+    suspend fun write(key: Constants, value: String)
+
+    suspend fun rollover()
+
+    fun read(key: Constants): String?
+
+    fun readEventsContent(): List<Any>
+
+    fun getEventsString(content: Any) : String
 }
 
 interface StorageProvider {

core/src/main/java/com/amplitude/core/events/EventOptions.kt

@@ -1,12 +1,15 @@
 package com.amplitude.core.events
 
+import com.amplitude.core.EventCallBack
+import java.util.*
+
 open class EventOptions {
     var userId: String? = null
     var deviceId: String? = null
-    var timestamp: Long? = null
+    var timestamp: Long = System.currentTimeMillis()
     var eventId: Int? = null
     var sessionId: Long = -1
-    var insertId: String? = null
+    var insertId: String = UUID.randomUUID().toString()
     var locationLat: Double? = null
     var locationLng: Double? = null
     var appVersion: String? = null
@@ -37,5 +40,6 @@ open class EventOptions {
     var productId: String? = null
     var revenueType: String? = null
     var extra: Map<String, Any>? = null
-    var callback: ((BaseEvent) -> Unit)? = null
+    var callback: EventCallBack? = null
+    var attempts: Int = 0
 }