Feature/add flush policies (#122)

Add Flush Policies with concrete implementations of: CountBasedFlushPolicy and FrequencyFlushPolicy.

Author: didiergarcia

core/build.gradle

@@ -16,7 +16,7 @@ test {
 dependencies {
     // MAIN DEPS
     api 'com.segment:sovran-kotlin:1.2.1'
-    api "org.jetbrains.kotlinx:kotlinx-serialization-json:1.2.2"
+    api "org.jetbrains.kotlinx:kotlinx-serialization-json:1.4.1"
     implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.6.0'
 
     // TESTING

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

@@ -2,6 +2,7 @@ package com.segment.analytics.kotlin.core
 
 import com.segment.analytics.kotlin.core.Constants.DEFAULT_API_HOST
 import com.segment.analytics.kotlin.core.Constants.DEFAULT_CDN_HOST
+import com.segment.analytics.kotlin.core.platform.policies.FlushPolicy
 import com.segment.analytics.kotlin.core.utilities.ConcreteStorageProvider
 import kotlinx.coroutines.*
 import sovran.kotlin.Store
@@ -31,6 +32,7 @@ data class Configuration(
     var trackDeepLinks: Boolean = false,
     var flushAt: Int = 20,
     var flushInterval: Int = 30,
+    var flushPolicies: List<FlushPolicy> = emptyList<FlushPolicy>(),
     val defaultSettings: Settings = Settings(),
     var autoAddSegmentDestination: Boolean = true,
     var apiHost: String = DEFAULT_API_HOST,

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

@@ -1,17 +1,17 @@
 package com.segment.analytics.kotlin.core
 
+import kotlinx.serialization.DeserializationStrategy
 import kotlinx.serialization.KSerializer
 import kotlinx.serialization.SerialName
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.descriptors.PrimitiveKind
 import kotlinx.serialization.descriptors.PrimitiveSerialDescriptor
 import kotlinx.serialization.encoding.Decoder
 import kotlinx.serialization.encoding.Encoder
-import kotlinx.serialization.json.JsonArray
-import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.*
 import sovran.kotlin.Store
 import java.time.Instant
-import java.util.UUID
+import java.util.*
 
 typealias AnalyticsContext = JsonObject
 typealias Integrations = JsonObject
@@ -66,6 +66,8 @@ enum class EventType {
  * @see GroupEvent
  * @see AliasEvent
  */
+
+@Serializable(with = BaseEventSerializer::class)
 sealed class BaseEvent {
     // The type of event
     abstract var type: EventType
@@ -351,3 +353,16 @@ data class ScreenEvent(
         return result
     }
 }
+
+object BaseEventSerializer : JsonContentPolymorphicSerializer<BaseEvent>(BaseEvent::class) {
+    override fun selectDeserializer(element: JsonElement): DeserializationStrategy<out BaseEvent> {
+        return when (element.jsonObject["type"]?.jsonPrimitive?.content) {
+            "track" -> TrackEvent.serializer()
+            "screen" -> ScreenEvent.serializer()
+            "alias" -> AliasEvent.serializer()
+            "identify" -> IdentifyEvent.serializer()
+            "group" -> GroupEvent.serializer()
+            else -> throw Exception("Unknown Event: key 'type' not found or does not matches any event type")
+        }
+    }
+}

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

@@ -1,31 +1,36 @@
 package com.segment.analytics.kotlin.core.platform
 
 import com.segment.analytics.kotlin.core.*
-import com.segment.analytics.kotlin.core.platform.plugins.logger.*
-import kotlinx.coroutines.*
+import com.segment.analytics.kotlin.core.platform.plugins.logger.LogFilterKind
+import com.segment.analytics.kotlin.core.platform.plugins.logger.log
+import com.segment.analytics.kotlin.core.platform.plugins.logger.segmentLog
+import com.segment.analytics.kotlin.core.platform.policies.FlushPolicy
+import com.segment.analytics.kotlin.core.utilities.EncodeDefaultsJson
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import kotlinx.coroutines.channels.consumeEach
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.encodeToJsonElement
+import kotlinx.serialization.json.jsonObject
+import kotlinx.serialization.json.jsonPrimitive
 import java.io.File
 import java.io.FileInputStream
-import java.lang.Exception
-import java.util.concurrent.atomic.AtomicInteger
 
 internal class EventPipeline(
     private val analytics: Analytics,
     private val logTag: String,
     apiKey: String,
-    private val flushCount: Int = 20,
-    private val flushIntervalInMillis: Long = 30_000, // 30s
+    private val flushPolicies: List<FlushPolicy>,
     var apiHost: String = Constants.DEFAULT_API_HOST
 ) {
 
-    private val writeChannel: Channel<String>
+    private val writeChannel: Channel<BaseEvent>
 
     private val uploadChannel: Channel<String>
 
-    private val eventCount: AtomicInteger = AtomicInteger(0)
-
     private val httpClient: HTTPClient = HTTPClient(apiKey)
 
     private val storage get() = analytics.storage
@@ -37,7 +42,7 @@ internal class EventPipeline(
 
     companion object {
         internal const val FLUSH_POISON = "#!flush"
-
+        internal val FLUSH_EVENT = ScreenEvent(FLUSH_POISON, FLUSH_POISON, emptyJsonObject).apply { messageId = FLUSH_POISON }
         internal const val UPLOAD_SIG = "#!upload"
     }
 
@@ -50,12 +55,12 @@ internal class EventPipeline(
         registerShutdownHook()
     }
 
-    fun put(event: String) {
+    fun put(event: BaseEvent) {
         writeChannel.trySend(event)
     }
 
     fun flush() {
-        writeChannel.trySend(FLUSH_POISON)
+        writeChannel.trySend(FLUSH_EVENT)
     }
 
     fun start() {
@@ -68,32 +73,47 @@ internal class EventPipeline(
     fun stop() {
         uploadChannel.cancel()
         writeChannel.cancel()
+        unschedule()
         running = false
     }
 
+    fun stringifyBaseEvent(payload: BaseEvent): String {
+        val finalPayload = EncodeDefaultsJson.encodeToJsonElement(payload)
+            .jsonObject.filterNot { (k, v) ->
+                // filter out empty userId and traits values
+                (k == "userId" && v.jsonPrimitive.content.isBlank()) || (k == "traits" && v == emptyJsonObject)
+            }
+
+        val stringVal = Json.encodeToString(finalPayload)
+        return stringVal
+    }
+
     private fun write() = scope.launch(analytics.fileIODispatcher) {
         for (event in writeChannel) {
             // write to storage
-            val isPoison = (event == FLUSH_POISON)
+            val isPoison = (event.messageId == FLUSH_POISON)
             if (!isPoison) try {
-                storage.write(Storage.Constants.Events, event)
+                val stringVal = stringifyBaseEvent(event)
+                analytics.log("$logTag running $stringVal")
+                storage.write(Storage.Constants.Events, stringVal)
+
+                flushPolicies.forEach { flushPolicy -> flushPolicy.updateState(event) }
             }
             catch (e : Exception) {
                 Analytics.segmentLog("Error adding payload: $event", kind = LogFilterKind.ERROR)
             }
 
             // if flush condition met, generate paths
-            if (eventCount.incrementAndGet() >= flushCount || isPoison) {
-                eventCount.set(0)
+            if (isPoison || flushPolicies.any { it.shouldFlush() }) {
                 uploadChannel.trySend(UPLOAD_SIG)
+                flushPolicies.forEach { it.reset() }
             }
         }
     }
 
     private fun upload() = scope.launch(analytics.networkIODispatcher) {
         uploadChannel.consumeEach {
             analytics.log("$logTag performing flush")
-
             withContext(analytics.fileIODispatcher) {
                 storage.rollover()
             }
@@ -130,20 +150,16 @@ internal class EventPipeline(
         }
     }
 
-    private fun schedule() = scope.launch(analytics.fileIODispatcher) {
-        if (flushIntervalInMillis > 0) {
-            while (isActive && running) {
-                flush()
+    private fun schedule() {
+        flushPolicies.forEach { it.schedule(analytics) }
+    }
 
-                // use delay to do periodical task
-                // this is doable in coroutine, since delay only suspends, allowing thread to
-                // do other work and then come back. see:
-                // https://github.com/Kotlin/kotlinx.coroutines/issues/1632#issuecomment-545693827
-                delay(flushIntervalInMillis)
-            }
-        }
+    private fun unschedule() {
+        flushPolicies.forEach { it.unschedule() }
     }
 
+
+
     private fun handleUploadException(e: Exception, file: File): Boolean {
         var shouldCleanup = false
         if (e is HTTPException) {

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

@@ -1,19 +1,14 @@
 package com.segment.analytics.kotlin.core.platform.plugins
 
 import com.segment.analytics.kotlin.core.*
-import com.segment.analytics.kotlin.core.Constants.DEFAULT_API_HOST
 import com.segment.analytics.kotlin.core.platform.DestinationPlugin
 import com.segment.analytics.kotlin.core.platform.EventPipeline
 import com.segment.analytics.kotlin.core.platform.Plugin
 import com.segment.analytics.kotlin.core.platform.VersionedPlugin
-import com.segment.analytics.kotlin.core.platform.plugins.logger.log
-import com.segment.analytics.kotlin.core.utilities.EncodeDefaultsJson
+import com.segment.analytics.kotlin.core.platform.policies.CountBasedFlushPolicy
+import com.segment.analytics.kotlin.core.platform.policies.FlushPolicy
+import com.segment.analytics.kotlin.core.platform.policies.FrequencyFlushPolicy
 import kotlinx.serialization.Serializable
-import kotlinx.serialization.encodeToString
-import kotlinx.serialization.json.Json
-import kotlinx.serialization.json.encodeToJsonElement
-import kotlinx.serialization.json.jsonObject
-import kotlinx.serialization.json.jsonPrimitive
 
 @Serializable
 data class SegmentSettings(
@@ -28,10 +23,10 @@ data class SegmentSettings(
  * - We store events into a file with the batch api format (@link {https://segment.com/docs/connections/sources/catalog/libraries/server/http-api/#batch})
  * - We upload events on a dedicated thread using the batch api
  */
-class SegmentDestination : DestinationPlugin(), VersionedPlugin {
+class SegmentDestination: DestinationPlugin(), VersionedPlugin {
 
     private lateinit var pipeline: EventPipeline
-
+    var flushPolicies: List<FlushPolicy> = emptyList()
     override val key: String = "Segment.io"
 
     override fun track(payload: TrackEvent): BaseEvent {
@@ -59,23 +54,24 @@ class SegmentDestination : DestinationPlugin(), VersionedPlugin {
         return payload
     }
 
-    private inline fun <reified T : BaseEvent> enqueue(payload: T) {
-        // needs to be inline reified for encoding using Json
-        val finalPayload = EncodeDefaultsJson.encodeToJsonElement(payload)
-            .jsonObject.filterNot { (k, v) ->
-                // filter out empty userId and traits values
-                (k == "userId" && v.jsonPrimitive.content.isBlank()) || (k == "traits" && v == emptyJsonObject)
-            }
-
-        val stringVal = Json.encodeToString(finalPayload)
-        analytics.log("$key running $stringVal")
 
-        pipeline.put(stringVal)
+    private fun enqueue(payload: BaseEvent) {
+        pipeline.put(payload)
     }
 
     override fun setup(analytics: Analytics) {
         super.setup(analytics)
 
+        // convert flushAt and flushIntervals into FlushPolicies
+        flushPolicies = if (analytics.configuration.flushPolicies.isEmpty()) {
+            listOf(
+                CountBasedFlushPolicy(analytics.configuration.flushAt),
+                FrequencyFlushPolicy(analytics.configuration.flushInterval * 1000L)
+            )
+        } else {
+            analytics.configuration.flushPolicies
+        }
+
         // Add DestinationMetadata enrichment plugin
         add(DestinationMetadataPlugin())
 
@@ -84,8 +80,7 @@ class SegmentDestination : DestinationPlugin(), VersionedPlugin {
                 analytics,
                 key,
                 configuration.writeKey,
-                configuration.flushAt,
-                configuration.flushInterval * 1000L,
+                flushPolicies,
                 configuration.apiHost
             )
             pipeline.start()

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

@@ -0,0 +1,34 @@
+package com.segment.analytics.kotlin.core.platform.policies
+
+import com.segment.analytics.kotlin.core.BaseEvent
+
+/**
+ * A Count based Flush Policy that instructs the EventPipeline to flush at the
+ * given @param[flushAt]. The default value is 20. @param[flushAt] values should
+ * be >= 1 or they'll get the default value.
+ */
+class CountBasedFlushPolicy(var flushAt: Int = 20): FlushPolicy {
+
+
+    init {
+        // Make sure to only take valid counts or fallback to our default.
+        flushAt = when {
+            flushAt >= 1 -> flushAt
+            else -> 20
+        }
+    }
+
+    private var count: Int = 0
+
+    override fun shouldFlush(): Boolean {
+        return count >= flushAt
+    }
+
+    override fun updateState(event: BaseEvent) {
+        count++
+    }
+
+    override fun reset() {
+        count = 0
+    }
+}

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

@@ -0,0 +1,34 @@
+package com.segment.analytics.kotlin.core.platform.policies
+
+import com.segment.analytics.kotlin.core.Analytics
+import com.segment.analytics.kotlin.core.BaseEvent
+import kotlinx.coroutines.CoroutineScope
+
+interface FlushPolicy {
+
+    /**
+     * Called when the policy becomes active. We should start any periodic flushing
+     * we want here.
+     */
+    fun schedule(analytics: Analytics) = Unit
+
+    /**
+     * Called when policy should stop running any scheduled flushes
+     */
+    fun unschedule() = Unit
+
+    /**
+     * Called to check whether or not the events should be flushed.
+     */
+    fun shouldFlush(): Boolean
+
+    /**
+     * Called as events are added to the timeline and JSON Stringified.
+     */
+    fun updateState(event: BaseEvent) = Unit
+
+    /**
+     * Called after the events are flushed.
+     */
+    fun reset() = Unit
+}