feat: switch storage for better thread safety

amplitude · Feb 22, 2024 · c516d7b · c516d7b
1 parent 111007f
commit c516d7b
Show file tree

Hide file tree

Showing 4 changed files with 57 additions and 22 deletions.
diff --git a/android/src/main/java/com/amplitude/android/utilities/AndroidStorage.kt b/android/src/main/java/com/amplitude/android/utilities/AndroidStorage.kt
@@ -35,7 +35,7 @@ class AndroidStorage(
         context.getSharedPreferences("${getPrefix()}-$storageKey", Context.MODE_PRIVATE)
     private val storageDirectory: File = context.getDir(getDir(), Context.MODE_PRIVATE)
     private val eventsFile =
-        EventsFileManager(storageDirectory, storageKey, AndroidKVS(sharedPreferences))
+        EventsFileManager(storageDirectory, storageKey, AndroidKVS(sharedPreferences), logger)
     private val eventCallbacksMap = mutableMapOf<String, EventCallBack>()
 
     override suspend fun writeEvent(event: BaseEvent) {

diff --git a/core/src/main/java/com/amplitude/core/utilities/EventsFileManager.kt b/core/src/main/java/com/amplitude/core/utilities/EventsFileManager.kt
@@ -1,10 +1,13 @@
 package com.amplitude.core.utilities
 
+import com.amplitude.common.Logger
 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 org.json.JSONException
+import org.json.JSONObject
 import java.io.BufferedReader
 import java.io.File
 import java.io.FileOutputStream
@@ -14,7 +17,8 @@ import java.util.concurrent.ConcurrentHashMap
 class EventsFileManager(
     private val directory: File,
     private val storageKey: String,
-    private val kvs: KeyValueStore
+    private val kvs: KeyValueStore,
+    private val logger: Logger
 ) {
     init {
         createDirectory(directory)
@@ -54,13 +58,7 @@ class EventsFileManager(
             }
         }
 
-        var contents = ""
-        if (file.length() == 0L) {
-            start(file)
-        } else if (file.length() > 1) {
-            contents += ","
-        }
-        contents += event
+        val contents = "${event}\n"
         writeToFile(contents.toByteArray(), file)
     }
 
@@ -122,8 +120,10 @@ class EventsFileManager(
         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)
+        writeEventsToFile(splitStrings.first, firstHalfFile)
+        writeEventsToFile(splitStrings.second, secondHalfFile)
+        firstHalfFile.renameTo(File(directory, firstHalfFile.nameWithoutExtension))
+        secondHalfFile.renameTo(File(directory, secondHalfFile.nameWithoutExtension))
         this.remove(filePath)
     }
 
@@ -135,7 +135,40 @@ class EventsFileManager(
         }
         filePathSet.add(filePath)
         File(filePath).bufferedReader().use<BufferedReader, String> {
-            return it.readText()
+            val content = it.readText()
+            val isCurrentVersion = content.endsWith("\n")
+            if (isCurrentVersion) {
+                // handle current version
+                val events = JSONArray();
+                content.split("\n").forEach {
+                    if (it.isNotEmpty()) {
+                        try {
+                            events.put(JSONObject(it))
+                        } catch (e: JSONException) {
+                            logger.error("Failed to parse event: $it")
+                        }
+                    }
+                }
+                return if (events.length() > 0) {
+                    events.toString()
+                } else {
+                    ""
+                }
+            } else {
+                // handle earlier versions
+                val normalizedContent = "[${content.trimStart('[').trimEnd(']', ',')}]"
+                if (normalizedContent.isEmpty()) {
+                    return ""
+                }
+                try {
+                    JSONArray(normalizedContent)
+                    return normalizedContent
+                } catch (e: JSONException) {
+                    logger.error("Failed to parse events: $normalizedContent, dropping file: $filePath")
+                    this.remove(filePath)
+                    return ""
+                }
+            }
         }
     }
 
@@ -148,9 +181,6 @@ class EventsFileManager(
             // if tmp file doesn't exist or empty then we don't need to do anything
             return
         }
-        // close events array and batch object
-        val contents = """]"""
-        writeToFile(contents.toByteArray(), file)
         file.renameTo(File(directory, file.nameWithoutExtension))
         incrementFileIndex()
         reset()
@@ -197,6 +227,11 @@ class EventsFileManager(
         file.renameTo(File(directory, file.nameWithoutExtension))
     }
 
+    private fun writeEventsToFile(events: List<JSONObject>, file: File) {
+        val contents = events.joinToString(separator = "\n", postfix = "\n") { it.toString()}
+        writeToFile(contents, file)
+    }
+
     private fun reset() {
         curFile.remove(storageKey)
     }

diff --git a/core/src/main/java/com/amplitude/core/utilities/FileStorage.kt b/core/src/main/java/com/amplitude/core/utilities/FileStorage.kt
@@ -28,7 +28,7 @@ class FileStorage(
     private val storageDirectoryEvents = File(storageDirectory, "events")
 
     private val propertiesFile = PropertiesFile(storageDirectory, storageKey, getPrefix(), null)
-    private val eventsFile = EventsFileManager(storageDirectoryEvents, storageKey, propertiesFile)
+    private val eventsFile = EventsFileManager(storageDirectoryEvents, storageKey, propertiesFile, logger)
     private val eventCallbacksMap = mutableMapOf<String, EventCallBack>()
 
     init {

diff --git a/core/src/main/java/com/amplitude/core/utilities/JSONUtil.kt b/core/src/main/java/com/amplitude/core/utilities/JSONUtil.kt
@@ -215,18 +215,18 @@ fun JSONArray.toEvents(): List<BaseEvent> {
     return events
 }
 
-internal fun JSONArray.split(): Pair<String, String> {
+internal fun JSONArray.split(): Pair<List<JSONObject>, List<JSONObject>> {
     val mid = this.length() / 2
-    val firstHalf = JSONArray()
-    val secondHalf = JSONArray()
+    val firstHalf = mutableListOf<JSONObject>()
+    val secondHalf = mutableListOf<JSONObject>()
     (0 until this.length()).forEach { index, ->
         if (index < mid) {
-            firstHalf.put(this.getJSONObject(index))
+            firstHalf.add(this.getJSONObject(index))
         } else {
-            secondHalf.put(this.getJSONObject(index))
+            secondHalf.add(this.getJSONObject(index))
         }
     }
-    return Pair(firstHalf.toString(), secondHalf.toString())
+    return Pair(firstHalf, secondHalf)
 }
 
 internal fun JSONObject.addValue(key: String, value: Any?) {