Feature/real time flags (#28)

* Feature/real time flags (#5) * Compiles and test, need to add some tests then get it into a sample app * Should be testing but not getting the errors through Fuse so can't run into the logic * Probably gone as far as I can with 2.x fuel, let's try the 3.x * Move to Retrofit - seems to be going well so far, test runs * Tidying up, setTrait test not working * All the tests are passing so will finish the retrofit migration * Updated some of the logic and added setTraits * Checkpoint commit before trying generic converter * Generics working fine * All passing for flags and such with the new generic caching * Mostly swapped to Retrofit, now need to do the analytics * Analytics now over to retrofit * Add caching for the getFlags endpoint * Get rid of the last of Fuel * Another clear-out and all working fine on the tests * Now using Retrofit cache, remove the old stuff * Now just using HTTP caching * Delete the old caching logic * Finishing off, should be done for defaults and caching * Remove unneeded todo * Remove some more code * Still just playing around with it * Move cache configuration to its own data class * Tidy up the cache config and the tests * Update the comments * Now covers the caching tests * Tidy up some more of the tests * Some more tidying up * Default to caching disabled * Last few PR comments * Split the read and write timeout for HTTP * Initial basic implementation, let's try to get things hooked up to the server * Seems to be generally working * Checkpoint commit, seems to be generally working now just need to get the flags on update * Checkpoint commit before making the changes OK'd by Matthew to move the update clock into the event service * Ensure that the event source just reconnects if it loses the connection * Events and timers now all hooked-up and working in the manual integration test * Got the integration test working * Tidy everything up and move sensitive data to environment variables * Add a new test to cover the event stream going through a reconnect cycle * Added test for the live stream of flags, tidied up the imports and various thing, changed the logic a bit for when we need to do updates from events * Update FlagsmithClient/src/test/java/com/flagsmith/RealTimeUpdatesIntegrationTests.kt Co-authored-by: Matthew Elwell <mjelwell89@gmail.com> * Update environment variables in the github actions * Add some error checking on the environment variables so it's a bit more obvious what's going on if we don't configure properly * Noddy change to get the tests to run again * Try printing out unsuccessful responses in the integration tests * Push more more non-empty checks * Feature/real time flags (#5) (#21) * Feature/real time flags (#5) * Compiles and test, need to add some tests then get it into a sample app * Should be testing but not getting the errors through Fuse so can't run into the logic * Probably gone as far as I can with 2.x fuel, let's try the 3.x * Move to Retrofit - seems to be going well so far, test runs * Tidying up, setTrait test not working * All the tests are passing so will finish the retrofit migration * Updated some of the logic and added setTraits * Checkpoint commit before trying generic converter * Generics working fine * All passing for flags and such with the new generic caching * Mostly swapped to Retrofit, now need to do the analytics * Analytics now over to retrofit * Add caching for the getFlags endpoint * Get rid of the last of Fuel * Another clear-out and all working fine on the tests * Now using Retrofit cache, remove the old stuff * Now just using HTTP caching * Delete the old caching logic * Finishing off, should be done for defaults and caching * Remove unneeded todo * Remove some more code * Still just playing around with it * Move cache configuration to its own data class * Tidy up the cache config and the tests * Update the comments * Now covers the caching tests * Tidy up some more of the tests * Some more tidying up * Default to caching disabled * Last few PR comments * Split the read and write timeout for HTTP * Initial basic implementation, let's try to get things hooked up to the server * Seems to be generally working * Checkpoint commit, seems to be generally working now just need to get the flags on update * Checkpoint commit before making the changes OK'd by Matthew to move the update clock into the event service * Ensure that the event source just reconnects if it loses the connection * Events and timers now all hooked-up and working in the manual integration test * Got the integration test working * Tidy everything up and move sensitive data to environment variables * Add a new test to cover the event stream going through a reconnect cycle * Added test for the live stream of flags, tidied up the imports and various thing, changed the logic a bit for when we need to do updates from events * Update FlagsmithClient/src/test/java/com/flagsmith/RealTimeUpdatesIntegrationTests.kt Co-authored-by: Matthew Elwell <mjelwell89@gmail.com> * Update environment variables in the github actions * Add some error checking on the environment variables so it's a bit more obvious what's going on if we don't configure properly * Noddy change to get the tests to run again * Try printing out unsuccessful responses in the integration tests * Push more more non-empty checks --------- Co-authored-by: Matthew Elwell <mjelwell89@gmail.com> * Split the actions into push and pull-request versions * Remove the old script * Update the timeouts on the integration tests and make them a little more robust * Fix up some merge issues * Use the flow value directly in the unit tests and simplify the flow updates in the SDK so they're always passive * Quick update to see if we're getting an error back when pushing the flag * Bump to 4c runner * Move to Strings as the server prefers them. Increase the timeouts and make the check loops more CPU friendly * Updates to cover most of the review comments * Make some of the Flagsmith class external again, move the test-only portions of the API into the test folder * Suppress a case warning in the test code * Remove integration test env vars now that we know that they passed with them included * Exclude integration tests in the Kover run * Disable coverage on PR workflow * Remove some duplicated logic * Update construction of event source URL * Add some constants and rearrange a bit of code * Return the env key interceptor to its original position * Various updates to the integration tests to cover Matthew's comments * Tidied up the integration tests, also added a bit more to one of the timeouts as it was taking a while for the realtime updates to come though * Tidied some of the println() statements into the asserts, commented on the reasoning for some of the timeouts and delays --------- Co-authored-by: Gareth Reese <8297652+gazreese@users.noreply.github.com> Co-authored-by: Gareth Reese <gareth@foresightmobile.com>
Flagsmith · Nov 9, 2023 · 797c387 · 797c387
1 parent 9aab12d
commit 797c387
Show file tree

Hide file tree

Showing 15 changed files with 564 additions and 35 deletions.
diff --git a/.github/workflows/verify.yml → .github/workflows/verify-pull-request.yml b/.github/workflows/verify.yml → .github/workflows/verify-pull-request.yml
@@ -1,24 +1,19 @@
-name: Verify
+name: Verify pull request to main
 
 on:
-  push:
-    branches:
-      - "main"
   pull_request:
     branches:
       - "main"
 
 jobs:
   test:
     name: Run Unit Tests
-    runs-on: ubuntu-latest
+    runs-on: General-Purpose-4c-Runner
 
     steps:
       - uses: actions/checkout@v3
       - name: Preconfigure gradle
         uses: ./.github/actions/prepare-gradle
 
       - name: Run unit tests
-        run: ./gradlew check -x koverVerify
-      - name: Verify code coverage level
-        run: ./gradlew koverVerify
+        run: ./gradlew check -x koverVerify -P excludeIntegrationTests
diff --git a/.github/workflows/verify-push-main.yml b/.github/workflows/verify-push-main.yml
@@ -0,0 +1,27 @@
+name: Verify on push to main
+
+on:
+  push:
+    branches:
+      - "main"
+
+jobs:
+  test:
+    name: Run Unit Tests
+    runs-on: General-Purpose-4c-Runner
+
+    env:
+      INTEGRATION_TESTS_ENVIRONMENT_KEY: NTtWcerSBE5yj7a5optMSk
+      INTEGRATION_TESTS_FEATURE_NAME: integration-test-feature
+      INTEGRATION_TESTS_FEATURE_STATE_ID: 321715
+      INTEGRATION_TESTS_API_TOKEN: ${{ secrets.INTEGRATION_TESTS_API_TOKEN }}
+
+    steps:
+      - uses: actions/checkout@v3
+      - name: Preconfigure gradle
+        uses: ./.github/actions/prepare-gradle
+
+      - name: Run unit tests
+        run: ./gradlew check -x koverVerify
+      - name: Verify code coverage level
+        run: ./gradlew koverVerify
diff --git a/FlagsmithClient/build.gradle.kts b/FlagsmithClient/build.gradle.kts
@@ -68,11 +68,16 @@ android {
 
 dependencies {
     implementation("com.google.code.gson:gson:2.10")
+    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.6.4")
 
     // HTTP Client
     implementation("com.squareup.retrofit2:retrofit:2.9.0")
     implementation("com.squareup.retrofit2:converter-gson:2.9.0")
 
+    // Server Sent Events
+    implementation("com.squareup.okhttp3:okhttp-sse:4.11.0")
+    testImplementation("com.squareup.okhttp3:okhttp-sse:4.11.0")
+
     testImplementation("junit:junit:4.13.2")
     testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.6.4")
     testImplementation("org.mock-server:mockserver-netty-no-dependencies:5.14.0")
@@ -84,7 +89,8 @@ dependencies {
 kover {
     filters {
         classes {
-            excludes += listOf("${android.namespace}.BuildConfig")
+            excludes += listOf("${android.namespace}.BuildConfig", "com.flagsmith.test.*")
+
         }
     }
     verify {
@@ -101,6 +107,15 @@ kover {
 }
 
 tasks.withType(Test::class) {
+    // if the excludeIntegrationTests property is set
+    // then exclude tests with IntegrationTest in the name
+    // i.e. `gradle :FlagsmithClient:testDebugUnitTest --tests "com.flagsmith.*" -P excludeIntegrationTests`
+    if (project.hasProperty("excludeIntegrationTests")) {
+        exclude {
+            it.name.contains("IntegrationTest")
+        }
+    }
+
     testLogging {
         events(
             TestLogEvent.FAILED,

diff --git a/FlagsmithClient/src/main/java/com/flagsmith/Flagsmith.kt b/FlagsmithClient/src/main/java/com/flagsmith/Flagsmith.kt
@@ -1,10 +1,19 @@
 package com.flagsmith
 
 import android.content.Context
-import com.flagsmith.entities.*
+import android.util.Log
+import com.flagsmith.entities.Flag
+import com.flagsmith.entities.Identity
+import com.flagsmith.entities.IdentityFlagsAndTraits
+import com.flagsmith.entities.Trait
+import com.flagsmith.entities.TraitWithIdentity
 import com.flagsmith.internal.FlagsmithAnalytics
+import com.flagsmith.internal.FlagsmithEventService
+import com.flagsmith.internal.FlagsmithEventTimeTracker
 import com.flagsmith.internal.FlagsmithRetrofitService
 import com.flagsmith.internal.enqueueWithResult
+import kotlinx.coroutines.flow.MutableStateFlow
+import okhttp3.Cache
 
 /**
  * Flagsmith
@@ -21,27 +30,69 @@ import com.flagsmith.internal.enqueueWithResult
 class Flagsmith constructor(
     private val environmentKey: String,
     private val baseUrl: String = "https://edge.api.flagsmith.com/api/v1/",
+    private val eventSourceBaseUrl: String = "https://realtime.flagsmith.com/",
     private val context: Context? = null,
     private val enableAnalytics: Boolean = DEFAULT_ENABLE_ANALYTICS,
+    private val enableRealtimeUpdates: Boolean = false,
     private val analyticsFlushPeriod: Int = DEFAULT_ANALYTICS_FLUSH_PERIOD_SECONDS,
     private val cacheConfig: FlagsmithCacheConfig = FlagsmithCacheConfig(),
     private val defaultFlags: List<Flag> = emptyList(),
     private val requestTimeoutSeconds: Long = 4L,
     private val readTimeoutSeconds: Long = 6L,
-    private val writeTimeoutSeconds: Long = 6L
-) {
-    private val retrofit: FlagsmithRetrofitService = FlagsmithRetrofitService.create(
-        baseUrl = baseUrl, environmentKey = environmentKey, context = context, cacheConfig = cacheConfig,
-        requestTimeoutSeconds = requestTimeoutSeconds, readTimeoutSeconds = readTimeoutSeconds, writeTimeoutSeconds = writeTimeoutSeconds)
+    private val writeTimeoutSeconds: Long = 6L,
+    override var lastFlagFetchTime: Double = 0.0 // from FlagsmithEventTimeTracker
+) : FlagsmithEventTimeTracker {
+    private lateinit var retrofit: FlagsmithRetrofitService
+    private var cache: Cache? = null
+    private var lastUsedIdentity: String? = null
     private val analytics: FlagsmithAnalytics? =
         if (!enableAnalytics) null
         else if (context != null) FlagsmithAnalytics(context, retrofit, analyticsFlushPeriod)
         else throw IllegalArgumentException("Flagsmith requires a context to use the analytics feature")
 
+    private val eventService: FlagsmithEventService? =
+        if (!enableRealtimeUpdates) null
+        else FlagsmithEventService(eventSourceBaseUrl = eventSourceBaseUrl, environmentKey = environmentKey) { event ->
+            if (event.isSuccess) {
+                lastEventUpdate = event.getOrNull()?.updatedAt ?: lastEventUpdate
+
+                // Check whether this event is anything new
+                if (lastEventUpdate > lastFlagFetchTime) {
+                    // First evict the cache otherwise we'll be stuck with the old values
+                    cache?.evictAll()
+                    lastFlagFetchTime = lastEventUpdate
+
+                    // Now we can get the new values, which will automatically be emitted to the flagUpdateFlow
+                    getFeatureFlags(lastUsedIdentity) { res ->
+                        if (res.isFailure) {
+                            Log.e(
+                                "Flagsmith",
+                                "Error getting flags in SSE stream: ${res.exceptionOrNull()}"
+                            )
+                        } else {
+                            Log.i("Flagsmith", "Got flags due to SSE event: $event")
+                        }
+                    }
+                }
+            }
+        }
+
+    // The last time we got an event from the SSE stream or via the API
+    private var lastEventUpdate: Double = 0.0
+
+    /** Stream of flag updates from the SSE stream if enabled */
+    val flagUpdateFlow = MutableStateFlow<List<Flag>>(listOf())
+
     init {
         if (cacheConfig.enableCache && context == null) {
             throw IllegalArgumentException("Flagsmith requires a context to use the cache feature")
         }
+        val pair = FlagsmithRetrofitService.create<FlagsmithRetrofitService>(
+            baseUrl = baseUrl, environmentKey = environmentKey, context = context, cacheConfig = cacheConfig,
+            requestTimeoutSeconds = requestTimeoutSeconds, readTimeoutSeconds = readTimeoutSeconds,
+            writeTimeoutSeconds = writeTimeoutSeconds, timeTracker = this, klass = FlagsmithRetrofitService::class.java)
+        retrofit = pair.first
+        cache = pair.second
     }
 
     companion object {
@@ -50,12 +101,19 @@ class Flagsmith constructor(
     }
 
     fun getFeatureFlags(identity: String? = null, result: (Result<List<Flag>>) -> Unit) {
+        // Save the last used identity as we'll refresh with this if we get update events
+        lastUsedIdentity = identity
+
         if (identity != null) {
             retrofit.getIdentityFlagsAndTraits(identity).enqueueWithResult { res ->
+                flagUpdateFlow.tryEmit(res.getOrNull()?.flags ?: emptyList())
                 result(res.map { it.flags })
             }
         } else {
-            retrofit.getFlags().enqueueWithResult(defaults = defaultFlags, result = result)
+            retrofit.getFlags().enqueueWithResult(defaults = defaultFlags) { res ->
+                flagUpdateFlow.tryEmit(res.getOrNull() ?: emptyList())
+                result(res)
+            }
         }
     }
 
@@ -78,18 +136,19 @@ class Flagsmith constructor(
     fun getTrait(id: String, identity: String, result: (Result<Trait?>) -> Unit) =
         retrofit.getIdentityFlagsAndTraits(identity).enqueueWithResult { res ->
             result(res.map { value -> value.traits.find { it.key == id } })
-        }
+        }.also { lastUsedIdentity = identity }
 
     fun getTraits(identity: String, result: (Result<List<Trait>>) -> Unit) =
         retrofit.getIdentityFlagsAndTraits(identity).enqueueWithResult { res ->
             result(res.map { it.traits })
-        }
+        }.also { lastUsedIdentity = identity }
 
     fun setTrait(trait: Trait, identity: String, result: (Result<TraitWithIdentity>) -> Unit) =
         retrofit.postTraits(TraitWithIdentity(trait.key, trait.value, Identity(identity))).enqueueWithResult(result = result)
 
     fun getIdentity(identity: String, result: (Result<IdentityFlagsAndTraits>) -> Unit) =
         retrofit.getIdentityFlagsAndTraits(identity).enqueueWithResult(defaults = null, result = result)
+            .also { lastUsedIdentity = identity }
 
     private fun getFeatureFlag(
         featureId: String,
@@ -101,6 +160,5 @@ class Flagsmith constructor(
             analytics?.trackEvent(featureId)
             foundFlag
         })
-    }
-
+    }.also { lastUsedIdentity = identity }
 }
diff --git a/FlagsmithClient/src/main/java/com/flagsmith/entities/FlagEvent.kt b/FlagsmithClient/src/main/java/com/flagsmith/entities/FlagEvent.kt
@@ -0,0 +1,7 @@
+package com.flagsmith.entities
+
+import com.google.gson.annotations.SerializedName
+
+internal data class FlagEvent (
+    @SerializedName(value = "updated_at") val updatedAt: Double
+)
diff --git a/FlagsmithClient/src/main/java/com/flagsmith/entities/Trait.kt b/FlagsmithClient/src/main/java/com/flagsmith/entities/Trait.kt
@@ -2,7 +2,6 @@ package com.flagsmith.entities
 
 
 import com.google.gson.annotations.SerializedName
-import java.io.Reader
 
 data class Trait(
     val identifier: String? = null,

diff --git a/FlagsmithClient/src/main/java/com/flagsmith/internal/FlagsmithEventService.kt b/FlagsmithClient/src/main/java/com/flagsmith/internal/FlagsmithEventService.kt
@@ -0,0 +1,87 @@
+package com.flagsmith.internal
+
+import android.util.Log
+import com.flagsmith.entities.FlagEvent
+import com.google.gson.Gson
+import kotlinx.coroutines.flow.MutableStateFlow
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.Response
+import okhttp3.sse.EventSource
+import okhttp3.sse.EventSourceListener
+import okhttp3.sse.EventSources
+import java.util.concurrent.TimeUnit
+
+internal class FlagsmithEventService constructor(
+    private val eventSourceBaseUrl: String?,
+    private val environmentKey: String,
+    private val updates: (Result<FlagEvent>) -> Unit
+) {
+    private val sseClient = OkHttpClient.Builder()
+        .addInterceptor(FlagsmithRetrofitService.envKeyInterceptor(environmentKey))
+        .connectTimeout(6, TimeUnit.SECONDS)
+        .readTimeout(10, TimeUnit.MINUTES)
+        .writeTimeout(10, TimeUnit.MINUTES)
+        .build()
+
+    private val completeEventSourceUrl: String = eventSourceBaseUrl + "sse/environments/" +  environmentKey + "/stream"
+
+    private val sseRequest = Request.Builder()
+        .url(completeEventSourceUrl)
+        .header("Accept", "application/json")
+        .addHeader("Accept", "text/event-stream")
+        .build()
+
+    private var currentEventSource: EventSource? = null
+
+    var sseEventsFlow = MutableStateFlow(FlagEvent(updatedAt = 0.0))
+        private set
+
+    private val sseEventSourceListener = object : EventSourceListener() {
+        override fun onClosed(eventSource: EventSource) {
+            super.onClosed(eventSource)
+            Log.d(TAG, "onClosed: $eventSource")
+
+            // This isn't uncommon and is the nature of HTTP requests, so just reconnect
+            initEventSource()
+        }
+
+        override fun onEvent(eventSource: EventSource, id: String?, type: String?, data: String) {
+            super.onEvent(eventSource, id, type, data)
+            Log.d(TAG, "onEvent: $data")
+            if (type != null && type == "environment_updated" && data.isNotEmpty()) {
+                val flagEvent = Gson().fromJson(data, FlagEvent::class.java)
+                sseEventsFlow.tryEmit(flagEvent)
+                updates(Result.success(flagEvent))
+            }
+        }
+
+        override fun onFailure(eventSource: EventSource, t: Throwable?, response: Response?) {
+            super.onFailure(eventSource, t, response)
+            Log.d(TAG, "onFailure: ${t?.message}")
+            if (t != null)
+                updates(Result.failure(t))
+            else
+                updates(Result.failure(Throwable("Unknown error")))
+        }
+
+        override fun onOpen(eventSource: EventSource, response: Response) {
+            super.onOpen(eventSource, response)
+            Log.d(TAG, "onOpen: $eventSource")
+        }
+    }
+
+    init {
+        initEventSource()
+    }
+
+    private fun initEventSource() {
+        currentEventSource?.cancel()
+        currentEventSource = EventSources.createFactory(sseClient)
+            .newEventSource(request = sseRequest, listener = sseEventSourceListener)
+    }
+
+    companion object {
+        private const val TAG = "FlagsmithEventService"
+    }
+}
diff --git a/FlagsmithClient/src/main/java/com/flagsmith/internal/FlagsmithEventTimeTracker.kt b/FlagsmithClient/src/main/java/com/flagsmith/internal/FlagsmithEventTimeTracker.kt
@@ -0,0 +1,7 @@
+package com.flagsmith.internal
+
+/** Internal interface for objects that track the last time the flags were fetched from the server */
+interface FlagsmithEventTimeTracker {
+    /** The last time the flags were fetched from the server, as a Unix epoch */
+    var lastFlagFetchTime: Double
+}