storage changes for signals (#211)

Co-authored-by: Wenxi Zeng <[email protected]>

Author: wenxi-zeng

android/src/main/java/com/segment/analytics/kotlin/android/Storage.kt

@@ -25,7 +25,7 @@ class AndroidStorage(
 
     private val sharedPreferences: SharedPreferences =
         context.getSharedPreferences("analytics-android-$writeKey", Context.MODE_PRIVATE)
-    private val storageDirectory: File = context.getDir("segment-disk-queue", Context.MODE_PRIVATE)
+    override val storageDirectory: File = context.getDir("segment-disk-queue", Context.MODE_PRIVATE)
     internal val eventsFile =
         EventsFileManager(storageDirectory, writeKey, AndroidKVS(sharedPreferences))
 

core/src/main/java/com/segment/analytics/kotlin/core/Storage.kt

@@ -4,6 +4,7 @@ import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonObject
 import sovran.kotlin.Store
+import java.io.File
 
 /**
  * Storage interface that abstracts storage of
@@ -41,6 +42,8 @@ interface Storage {
         DeviceId("segment.device.id")
     }
 
+    val storageDirectory: File
+
     suspend fun subscribeToStore()
     suspend fun write(key: Constants, value: String)
     fun read(key: Constants): String?

core/src/main/java/com/segment/analytics/kotlin/core/platform/EventPipeline.kt

@@ -19,7 +19,7 @@ import kotlinx.serialization.json.jsonPrimitive
 import java.io.File
 import java.io.FileInputStream
 
-internal class EventPipeline(
+open class EventPipeline(
     private val analytics: Analytics,
     private val logTag: String,
     apiKey: String,
@@ -31,11 +31,15 @@ internal class EventPipeline(
 
     private var uploadChannel: Channel<String>
 
-    private val httpClient: HTTPClient = HTTPClient(apiKey, analytics.configuration.requestFactory)
+    protected open val httpClient: HTTPClient = HTTPClient(apiKey, analytics.configuration.requestFactory)
 
-    private val storage get() = analytics.storage
+    protected open val storage get() = analytics.storage
 
-    private val scope get() = analytics.analyticsScope
+    protected open val scope get() = analytics.analyticsScope
+
+    protected open val fileIODispatcher get() = analytics.fileIODispatcher
+
+    protected open val networkIODispatcher get() = analytics.networkIODispatcher
 
     var running: Boolean
         private set
@@ -87,7 +91,7 @@ internal class EventPipeline(
         unschedule()
     }
 
-    internal fun stringifyBaseEvent(payload: BaseEvent): String {
+    open fun stringifyBaseEvent(payload: BaseEvent): String {
         val finalPayload = EncodeDefaultsJson.encodeToJsonElement(payload)
             .jsonObject.filterNot { (k, v) ->
                 // filter out empty userId and traits values
@@ -98,7 +102,7 @@ internal class EventPipeline(
         return stringVal
     }
 
-    private fun write() = scope.launch(analytics.fileIODispatcher) {
+    private fun write() = scope.launch(fileIODispatcher) {
         for (event in writeChannel) {
             // write to storage
             val isPoison = (event.messageId == FLUSH_POISON)
@@ -122,10 +126,10 @@ internal class EventPipeline(
         }
     }
 
-    private fun upload() = scope.launch(analytics.networkIODispatcher) {
+    private fun upload() = scope.launch(networkIODispatcher) {
         uploadChannel.consumeEach {
             analytics.log("$logTag performing flush")
-            withContext(analytics.fileIODispatcher) {
+            withContext(fileIODispatcher) {
                 storage.rollover()
             }
 

core/src/main/java/com/segment/analytics/kotlin/core/utilities/StorageImpl.kt

@@ -22,7 +22,7 @@ class StorageImpl(
     private val ioDispatcher: CoroutineDispatcher
 ) : Subscriber, Storage {
 
-    private val storageDirectory = File("/tmp/analytics-kotlin/$writeKey")
+    override val storageDirectory = File("/tmp/analytics-kotlin/$writeKey")
     private val storageDirectoryEvents = File(storageDirectory, "events")
 
     internal val propertiesFile = PropertiesFile(storageDirectory, writeKey)