feat: add error handling and retry

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

@@ -3,9 +3,9 @@ package com.amplitude.android
 import android.content.Context
 import com.amplitude.android.utilities.AndroidLoggerProvider
 import com.amplitude.core.Configuration
+import com.amplitude.core.EventCallBack
 import com.amplitude.core.LoggerProvider
 import com.amplitude.core.StorageProvider
-import com.amplitude.core.events.BaseEvent
 import com.amplitude.core.utilities.FileStorageProvider
 
 class Configuration(
@@ -18,7 +18,7 @@ class Configuration(
     storageProvider: StorageProvider = FileStorageProvider(),
     loggerProvider: LoggerProvider = AndroidLoggerProvider(),
     minIdLength: Int? = null,
-    callback: ((BaseEvent) -> Unit)? = null,
+    callback: EventCallBack? = null,
     val useAdvertisingIdForDeviceId: Boolean = false,
     val useAppSetIdForDeviceId: Boolean = false,
     val newDeviceIdPerInstall: Boolean = false,

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

@@ -1,9 +1,14 @@
 package com.amplitude.android.utilities
 
 import com.amplitude.core.Amplitude
+import com.amplitude.core.Configuration
 import com.amplitude.core.Storage
 import com.amplitude.core.StorageProvider
 import com.amplitude.core.events.BaseEvent
+import com.amplitude.core.platform.EventPipeline
+import com.amplitude.core.utilities.ResponseHandler
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineScope
 
 class AndroidStorage(
     val amplitude: Amplitude
@@ -24,7 +29,23 @@ class AndroidStorage(
         TODO("Not yet implemented")
     }
 
-    override fun getEvents(): List<String> {
+    override fun readEventsContent(): List<Any> {
+        TODO("Not yet implemented")
+    }
+
+    override fun getEventsString(content: Any): String {
+        TODO("Not yet implemented")
+    }
+
+    override fun getResponseHandler(
+        storage: Storage,
+        eventPipeline: EventPipeline,
+        configuration: Configuration,
+        scope: CoroutineScope,
+        dispatcher: CoroutineDispatcher,
+        events: Any,
+        eventsString: String
+    ): ResponseHandler {
         TODO("Not yet implemented")
     }
 }

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

@@ -31,9 +31,9 @@ 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

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

@@ -4,6 +4,8 @@ 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,
@@ -13,12 +15,14 @@ open class Configuration(
     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"
     }
 

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

@@ -1,6 +1,10 @@
 package com.amplitude.core
 
 import com.amplitude.core.events.BaseEvent
+import com.amplitude.core.platform.EventPipeline
+import com.amplitude.core.utilities.ResponseHandler
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineScope
 
 interface Storage {
 
@@ -20,7 +24,11 @@ interface Storage {
 
     fun read(key: Constants): String?
 
-    fun getEvents(): List<String>
+    fun readEventsContent(): List<Any>
+
+    fun getEventsString(content: Any): String
+
+    fun getResponseHandler(storage: Storage, eventPipeline: EventPipeline, configuration: Configuration, scope: CoroutineScope, dispatcher: CoroutineDispatcher, events: Any, eventsString: String): ResponseHandler
 }
 
 interface StorageProvider {

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

@@ -1,5 +1,7 @@
 package com.amplitude.core.events
 
+import com.amplitude.core.EventCallBack
+
 open class EventOptions {
     var userId: String? = null
     var deviceId: String? = null
@@ -37,5 +39,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
+    internal var attempts: Int = 0
 }

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

@@ -13,7 +13,7 @@ import kotlinx.coroutines.withContext
 import java.lang.Exception
 import java.util.concurrent.atomic.AtomicInteger
 
-internal class EventPipeline(
+class EventPipeline(
     private val amplitude: Amplitude
 ) {
 
@@ -35,6 +35,8 @@ internal class EventPipeline(
     var scheduled: Boolean
         private set
 
+    var flushSizeDivider: AtomicInteger = AtomicInteger(1)
+
     companion object {
         internal const val UPLOAD_SIG = "#!upload"
     }
@@ -50,6 +52,7 @@ internal class EventPipeline(
     }
 
     fun put(event: BaseEvent) {
+        event.attempts += 1
         writeChannel.trySend(WriteQueueMessage(WriteQueueMessageType.EVENT, event))
     }
 
@@ -76,7 +79,9 @@ internal class EventPipeline(
             if (!triggerFlush && message.event != null) try {
                 storage.writeEvent(message.event)
             } catch (e: Exception) {
-                e.printStackTrace()
+                e.message?.let {
+                    amplitude.logger.error(it)
+                }
             }
 
             // if flush condition met, generate paths
@@ -96,27 +101,32 @@ internal class EventPipeline(
                 storage.rollover()
             }
 
-            val eventsData = storage.getEvents()
+            val eventsData = storage.readEventsContent()
             for (events in eventsData) {
-                if (events.isEmpty()) continue
+                val eventsString = storage.getEventsString(events)
+                if (eventsString.isEmpty()) continue
 
                 try {
                     val connection = httpClient.upload()
                     connection.outputStream?.let {
-                        connection.setEvents(events)
+                        connection.setEvents(eventsString)
                         // Upload the payloads.
                         connection.close()
                     }
+                    val responseHandler = storage.getResponseHandler(storage, this@EventPipeline, amplitude.configuration, scope, amplitude.retryDispatcher, events, eventsString)
+                    responseHandler?.handle(connection.response)
                 } catch (e: Exception) {
-                    e.printStackTrace()
+                    e.message?.let {
+                        amplitude.logger.error(it)
+                    }
                 }
-                // @TODO: handle failures and retry
             }
         }
     }
 
     private fun getFlushCount(): Int {
-        return amplitude.configuration.flushQueueSize
+        val count = amplitude.configuration.flushQueueSize / flushSizeDivider.get()
+        return count.takeUnless { it == 0 } ?: 1
     }
 
     private fun getFlushIntervalInMillis(): Long {

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

@@ -4,6 +4,7 @@ import com.amplitude.core.Amplitude
 import com.amplitude.core.Constants
 import com.amplitude.core.events.BaseEvent
 import com.amplitude.core.platform.Plugin
+import java.util.UUID
 
 class ContextPlugin : Plugin {
     override val type: Plugin.Type = Plugin.Type.Before
@@ -14,6 +15,12 @@ class ContextPlugin : Plugin {
     }
 
     private fun applyContextData(event: BaseEvent) {
+        event.timestamp ?: run {
+            event.timestamp = System.currentTimeMillis()
+        }
+        event.insertId ?: run {
+            event.insertId = UUID.randomUUID().toString()
+        }
         event.library = Constants.SDK_LIBRARY + "/" + Constants.SDK_VERSION
         event.userId = amplitude.store.userId
         event.deviceId = amplitude.store.deviceId

core/src/main/java/com/amplitude/core/utilities/EventsFileManager.kt

@@ -4,6 +4,7 @@ import com.amplitude.id.utilities.KeyValueStore
 import com.amplitude.id.utilities.createDirectory
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import org.json.JSONArray
 import java.io.File
 import java.io.FileOutputStream
 
@@ -101,6 +102,24 @@ class EventsFileManager(
         finish()
     }
 
+    /**
+     * Split one file to two smaller file
+     * This is used to handle payload too large error response
+     */
+    fun splitFile(filePath: String, events: JSONArray) {
+        val originalFile = File(filePath)
+        if (!originalFile.exists()) {
+            return
+        }
+        val fileName = originalFile.name
+        val firstHalfFile = File(directory, "$fileName-1.tmp")
+        val secondHalfFile = File(directory, "$fileName-2.tmp")
+        val splitStrings = events.split()
+        writeToFile(splitStrings.first, firstHalfFile)
+        writeToFile(splitStrings.second, secondHalfFile)
+        this.remove(filePath)
+    }
+
     private fun finish() {
         val file = currentFile()
         if (!file.exists()) {
@@ -135,6 +154,17 @@ class EventsFileManager(
         }
     }
 
+    private fun writeToFile(content: String, file: File) {
+        file.createNewFile()
+        val fileOS = FileOutputStream(file, true)
+        fileOS.run {
+            write(content.toByteArray())
+            flush()
+        }
+        file.renameTo(File(directory, file.nameWithoutExtension))
+        fileOS.close()
+    }
+
     private fun reset() {
         os = null
         curFile = null

core/src/main/java/com/amplitude/core/utilities/FileResponseHandler.kt

@@ -0,0 +1,96 @@
+package com.amplitude.core.utilities
+
+import com.amplitude.core.Configuration
+import com.amplitude.core.events.BaseEvent
+import com.amplitude.core.platform.EventPipeline
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.launch
+import org.json.JSONArray
+
+internal class FileResponseHandler(
+    private val storage: FileStorage,
+    private val eventPipeline: EventPipeline,
+    private val configuration: Configuration,
+    private val scope: CoroutineScope,
+    private val dispatcher: CoroutineDispatcher,
+    private val eventFilePath: String,
+    private val eventsString: String
+) : ResponseHandler {
+
+    override fun handleSuccessResponse(successResponse: SuccessResponse) {
+        val events = JSONArray(eventsString).toEvents()
+        triggerEventsCallback(events, HttpStatus.SUCCESS.code, "Event sent success.")
+        scope.launch(dispatcher) {
+            storage.removeFile(eventFilePath)
+        }
+    }
+
+    override fun handleBadRequestResponse(badRequestResponse: BadRequestResponse) {
+        val events = JSONArray(eventsString).toEvents()
+        if (events.size == 1) {
+            triggerEventsCallback(events, HttpStatus.BAD_REQUEST.code, badRequestResponse.error)
+            storage.removeFile(eventFilePath)
+            return
+        }
+        val droppedIndices = badRequestResponse.getEventIndicesToDrop()
+        val eventsToDrop = mutableListOf<BaseEvent>()
+        val eventsToRetry = mutableListOf<BaseEvent>()
+        events.forEachIndexed { index, event ->
+            if (droppedIndices.contains(index) || badRequestResponse.isEventSilenced(event)) {
+                eventsToDrop.add(event)
+            } else {
+                eventsToRetry.add(event)
+            }
+        }
+        triggerEventsCallback(eventsToDrop, HttpStatus.BAD_REQUEST.code, badRequestResponse.error)
+        eventsToRetry.forEach {
+            eventPipeline.put(it)
+        }
+        scope.launch(dispatcher) {
+            storage.removeFile(eventFilePath)
+        }
+    }
+
+    override fun handlePayloadTooLargeResponse(payloadTooLargeResponse: PayloadTooLargeResponse) {
+        val rawEvents = JSONArray(eventsString)
+        if (rawEvents.length() == 1) {
+            val events = rawEvents.toEvents()
+            triggerEventsCallback(events, HttpStatus.PAYLOAD_TOO_LARGE.code, payloadTooLargeResponse.error)
+            scope.launch(dispatcher) {
+                storage.removeFile(eventFilePath)
+            }
+            return
+        }
+        // split file into two
+        scope.launch(dispatcher) {
+            storage.splitEventFile(eventFilePath, rawEvents)
+        }
+    }
+
+    override fun handleTooManyRequestsResponse(tooManyRequestsResponse: TooManyRequestsResponse) {
+        // wait for next time to pick it up
+    }
+
+    override fun handleTimeoutResponse(timeoutResponse: TimeoutResponse) {
+        // wait for next time to try again
+    }
+
+    override fun handleFailedResponse(failedResponse: FailedResponse) {
+        // wait for next time to try again
+    }
+
+    private fun triggerEventsCallback(events: List<BaseEvent>, status: Int, message: String) {
+        events.forEach { event ->
+            configuration.callback?.let {
+                it(event, status, message)
+            }
+            event.insertId?.let { insertId ->
+                storage.getEventCallback(insertId)?.let {
+                    it(event, status, message)
+                    storage.removeEventCallback(insertId)
+                }
+            }
+        }
+    }
+}