add okhttp client to the uploader

ConfidenceDemoApp/src/main/java/com/example/confidencedemoapp/MainVm.kt

@@ -8,7 +8,10 @@ import androidx.lifecycle.LiveData
 import androidx.lifecycle.MutableLiveData
 import androidx.lifecycle.viewModelScope
 import com.spotify.confidence.ConfidenceFeatureProvider
+import com.spotify.confidence.EventSender
+import com.spotify.confidence.EventsScope
 import com.spotify.confidence.InitialisationStrategy
+import com.spotify.confidence.eventSender
 import dev.openfeature.sdk.Client
 import dev.openfeature.sdk.EvaluationContext
 import dev.openfeature.sdk.FlagEvaluationDetails
@@ -31,6 +34,7 @@ class MainVm(app: Application) : AndroidViewModel(app) {
     private val _color: MutableLiveData<Color> = MutableLiveData(Color.Gray)
     val message: LiveData<String> = _message
     val color: LiveData<Color> = _color
+    private lateinit var eventSender: EventSender
 
     init {
         val start = System.currentTimeMillis()
@@ -45,14 +49,17 @@ class MainVm(app: Application) : AndroidViewModel(app) {
         client = OpenFeatureAPI.getClient()
         viewModelScope.launch {
             OpenFeatureAPI.setEvaluationContext(ctx)
-            OpenFeatureAPI.setProviderAndWait(
-                ConfidenceFeatureProvider.create(
-                    app.applicationContext,
-                    clientSecret,
-                    initialisationStrategy = strategy
-                ),
-                dispatcher = Dispatchers.IO
+            val provider = ConfidenceFeatureProvider.create(
+                app.applicationContext,
+                clientSecret,
+                initialisationStrategy = strategy
             )
+            OpenFeatureAPI.setProviderAndWait(provider, Dispatchers.IO)
+
+            eventSender = provider.eventSender(app.applicationContext)
+
+            eventSender.emit("eventDefinitions/navigate")
+
             Log.d(TAG, "client secret is $clientSecret")
             Log.d(TAG, "init took ${System.currentTimeMillis() - start} ms")
             refreshUi()
@@ -72,6 +79,7 @@ class MainVm(app: Application) : AndroidViewModel(app) {
         }.toComposeColor()
         _message.postValue(messageValue)
         _color.postValue(colorFlag)
+        eventSender.emit("eventDefinitions/navigate", mapOf("button" to "refresh_ui"))
     }
 
     fun updateContext() {

Provider/src/main/java/com/spotify/confidence/ConfidenceExtensions.kt

@@ -7,18 +7,38 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
 
+val confidenceFlushPolicy = object : FlushPolicy {
+    private var size = 0
+    override fun reset() {
+        size = 0
+    }
+
+    override fun hit(event: Event) {
+        size++
+    }
+
+    override fun shouldFlush(): Boolean {
+        return size > 4
+    }
+}
+
 fun ConfidenceFeatureProvider.eventSender(
     context: Context,
     dispatcher: CoroutineDispatcher = Dispatchers.IO
 ): EventSender = EventSenderImpl.create(
     clientSecret = this.clientSecret(),
     dispatcher = dispatcher,
+    flushPolicies = listOf(confidenceFlushPolicy),
     scope = EventsScope(
         fields = {
-            val evalContext = OpenFeatureAPI.getEvaluationContext()
-                ?.asMap()
-                ?.mapValues { Json.encodeToString(it.value) }
-            evalContext ?: mapOf()
+            OpenFeatureAPI.getEvaluationContext()?.let { evalContext ->
+                val map = mutableMapOf<String, String>()
+                map["targeting_key"] = evalContext.getTargetingKey()
+                evalContext.asMap().forEach {
+                    map[it.key] = Json.encodeToString(it.value)
+                }
+                map
+            } ?: mapOf()
         }
     ),
     context = context

Provider/src/main/java/com/spotify/confidence/EventSender.kt

@@ -5,7 +5,7 @@ import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.Dispatchers
 
 interface EventSender {
-    fun emit(definition: String, payload: Map<String, String>)
+    fun emit(definition: String, payload: Map<String, String> = mapOf())
     fun withScope(scope: EventsScope): EventSender
 }
 
@@ -24,7 +24,8 @@ class EventSenderImpl private constructor(
     private val scope: EventsScope = EventsScope()
 ) : EventSender {
     override fun emit(definition: String, payload: Map<String, String>) {
-        eventSenderEngine.emit(definition, payload + scope.fields())
+        val scope = scope.fields()
+        eventSenderEngine.emit(definition, payload + scope)
     }
 
     override fun withScope(scope: EventsScope): EventSender {

Provider/src/main/java/com/spotify/confidence/EventSenderEngine.kt

@@ -9,6 +9,7 @@ import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.launch
+import okhttp3.OkHttpClient
 
 internal class EventSenderEngine(
     private val eventStorage: EventStorage,
@@ -27,7 +28,8 @@ internal class EventSenderEngine(
             // do nothing
         }
     }
-    private lateinit var uploader: EventSenderUploader
+    private val uploader: EventSenderUploader =
+        EventSenderUploaderImpl(OkHttpClient(), dispatcher)
 
     init {
         coroutineScope.launch {

Provider/src/main/java/com/spotify/confidence/EventSenderUploader.kt

@@ -75,13 +75,14 @@ internal class EventSenderUploaderImpl(
             .post(eventsJson.encodeToString(events).toRequestBody())
             .build()
 
-        val statusCode = httpClient.newCall(httpRequest).await().code
+        val response = httpClient.newCall(httpRequest).await()
+        val statusCode = response.code
         /**
          * if server can't handle the batch, we should throw it away
          * except for rate limiting
          * here backend can be more specific
          */
-        (statusCode / 100) == 4 && statusCode != 429
+        !((statusCode / 100) == 4 && statusCode != 429)
     }
 
     companion object {

Provider/src/main/java/com/spotify/confidence/EventStorage.kt

@@ -2,9 +2,9 @@ package com.spotify.confidence
 
 import android.content.Context
 import kotlinx.coroutines.sync.Semaphore
-import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.encodeToString
 import java.io.File
+import java.io.OutputStream
 
 internal interface EventStorage {
     suspend fun rollover()
@@ -17,20 +17,22 @@ internal class EventStorageImpl(
     private val context: Context
 ) : EventStorage {
     private lateinit var currentFile: File
+    private var outputStream: OutputStream? = null
     private val semaphore: Semaphore = Semaphore(1)
 
     init {
-        resetCurrentFile(false)
+        resetCurrentFile()
     }
 
     override suspend fun rollover() = withLock {
         currentFile.renameTo(getFileWithName(currentFile.name + READY_TO_SENT_EXTENSION))
-        resetCurrentFile(true)
+        resetCurrentFile()
     }
 
     override suspend fun writeEvent(event: Event) = withLock {
         val delimiter = EVENT_WRITE_DELIMITER
-        currentFile.writeText(eventsJson.encodeToString(event) + delimiter)
+        val byteArray = (eventsJson.encodeToString(event) + delimiter).toByteArray()
+        outputStream?.write(byteArray)
     }
 
     override suspend fun batchReadyFiles(): List<File> {
@@ -46,27 +48,24 @@ internal class EventStorageImpl(
         return list
     }
 
-    override suspend fun eventsFor(file: File): List<Event> = file.readText()
-        .split(EVENT_WRITE_DELIMITER)
-        .map { eventsJson.decodeFromString(it) }
+    override suspend fun eventsFor(file: File): List<Event> {
+        val text = file.readText()
+        return text
+            .split(EVENT_WRITE_DELIMITER)
+            .filter { it.isNotEmpty() }
+            .map { eventsJson.decodeFromString(it) }
+    }
 
-    private fun maxIndex(): Int {
+    private fun latestWriteFile(): File? {
         val directory = context.getDir(DIRECTORY, Context.MODE_PRIVATE)
-        var maxIndex = 0
         for (file in directory.walk().iterator()) {
-            if (!file.name.endsWith(READY_TO_SENT_EXTENSION)) {
-                val index = indexForFile(file)
-                if (maxIndex < index) {
-                    maxIndex = index
+            if (!file.isDirectory) {
+                if (!file.name.endsWith(READY_TO_SENT_EXTENSION) && !file.isDirectory) {
+                    return file
                 }
             }
         }
-
-        return maxIndex
-    }
-
-    private fun indexForFile(file: File): Int {
-        return file.name.split("-")[1].toInt()
+        return null
     }
 
     private suspend fun withLock(body: () -> Unit) {
@@ -75,23 +74,20 @@ internal class EventStorageImpl(
         semaphore.release()
     }
 
-    private fun resetCurrentFile(newFile: Boolean) {
-        val maxIndex = maxIndex()
-        val index = if (newFile) {
-            maxIndex + 1
-        } else {
-            maxIndex
-        }
-        currentFile = getFileWithName(index.toString())
+    private fun resetCurrentFile() {
+        outputStream?.close()
+        currentFile = latestWriteFile()
+            ?: getFileWithName("events-${System.currentTimeMillis()}")
+        outputStream = currentFile.outputStream()
     }
     private fun getFileWithName(name: String): File {
         val directory = context.getDir(DIRECTORY, Context.MODE_PRIVATE)
-        return File(directory, "events-$name")
+        return File(directory, name)
     }
 
     companion object {
         const val DIRECTORY = "events"
-        const val READY_TO_SENT_EXTENSION = "ready"
+        const val READY_TO_SENT_EXTENSION = ".ready"
         const val EVENT_WRITE_DELIMITER = ",\n"
     }
 }