updated to use correct comparison for distinctUntilChanged #97
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -11,6 +11,13 @@ import io.getunleash.android.errors.ServerException | |
| import io.getunleash.android.events.HeartbeatEvent | ||
| import io.getunleash.android.http.Throttler | ||
| import io.getunleash.android.unleashScope | ||
| import java.io.Closeable | ||
| import java.io.IOException | ||
| import java.util.concurrent.TimeUnit | ||
| import java.util.concurrent.atomic.AtomicReference | ||
| import kotlin.coroutines.CoroutineContext | ||
| import kotlin.coroutines.resume | ||
| import kotlin.coroutines.resumeWithException | ||
| import kotlinx.coroutines.Dispatchers | ||
| import kotlinx.coroutines.channels.BufferOverflow | ||
| import kotlinx.coroutines.flow.MutableSharedFlow | ||
|
|
@@ -30,54 +37,55 @@ import okhttp3.OkHttpClient | |
| import okhttp3.Request | ||
| import okhttp3.Response | ||
| import okhttp3.internal.closeQuietly | ||
|
|
||
| /** | ||
| * Http Client for fetching data from Unleash Proxy. By default creates an OkHttpClient with | ||
| * readTimeout set to 2 seconds and a cache of 10 MBs | ||
| * @param httpClient | ||
| * - the http client to use for fetching toggles from Unleash proxy | ||
| */ | ||
| open class UnleashFetcher( | ||
| unleashConfig: UnleashConfig, | ||
| private val httpClient: OkHttpClient, | ||
| private val unleashContext: StateFlow<UnleashContext>, | ||
| ) : Closeable { | ||
| companion object { | ||
| private const val TAG = "UnleashFetcher" | ||
| } | ||
| private var contextForLastFetch: UnleashContext? = null | ||
| private val proxyUrl = unleashConfig.proxyUrl?.toHttpUrl() | ||
| private val applicationHeaders = | ||
| unleashConfig.getApplicationHeaders(unleashConfig.pollingStrategy) | ||
| private val appName = unleashConfig.appName | ||
| private var etag: String? = null | ||
| private val featuresReceivedFlow = | ||
| MutableSharedFlow<UnleashState>( | ||
| replay = 1, | ||
| onBufferOverflow = BufferOverflow.DROP_OLDEST | ||
| ) | ||
| private val fetcherHeartbeatFlow = | ||
| MutableSharedFlow<HeartbeatEvent>( | ||
| extraBufferCapacity = 5, | ||
| onBufferOverflow = BufferOverflow.DROP_OLDEST | ||
| ) | ||
| private val coroutineContextForContextChange: CoroutineContext = Dispatchers.IO | ||
| private val currentCall = AtomicReference<Call?>(null) | ||
| private val throttler = | ||
| Throttler( | ||
| TimeUnit.MILLISECONDS.toSeconds(unleashConfig.pollingStrategy.interval), | ||
| longestAcceptableIntervalSeconds = 300, | ||
| proxyUrl.toString() | ||
| ) | ||
|
|
||
| fun getFeaturesReceivedFlow() = featuresReceivedFlow.asSharedFlow() | ||
|
|
||
| fun startWatchingContext() { | ||
| unleashScope.launch { | ||
|
Contributor
There was a problem hiding this comment.
|
||
| unleashContext.collect { | ||
| if (it == contextForLastFetch) { | ||
| Log.d(TAG, "Context unchanged, skipping refresh toggles") | ||
| return@collect | ||
| } | ||
| withContext(coroutineContextForContextChange) { | ||
| Log.d(TAG, "Unleash context changed: $it") | ||
| refreshToggles() | ||
|
|
@@ -89,7 +97,7 @@ open class UnleashFetcher( | |
| suspend fun refreshToggles(): ToggleResponse { | ||
| if (throttler.performAction()) { | ||
| Log.d(TAG, "Refreshing toggles") | ||
| val response = doFetchToggles(unleashContext.value) | ||
| fetcherHeartbeatFlow.emit(HeartbeatEvent(response.status, response.error?.message)) | ||
| return response | ||
| } | ||
|
|
@@ -98,15 +106,28 @@ open class UnleashFetcher( | |
| return ToggleResponse(Status.THROTTLED) | ||
| } | ||
|
|
||
| suspend fun refreshTogglesWithContext(ctx: UnleashContext): ToggleResponse { | ||
| if (throttler.performAction()) { | ||
| Log.d(TAG, "Refreshing toggles") | ||
| val response = doFetchToggles(ctx) | ||
| fetcherHeartbeatFlow.emit(HeartbeatEvent(response.status, response.error?.message)) | ||
| return response | ||
| } | ||
| Log.i(TAG, "Skipping refresh toggles due to throttling") | ||
| fetcherHeartbeatFlow.emit(HeartbeatEvent(Status.THROTTLED)) | ||
| return ToggleResponse(Status.THROTTLED) | ||
| } | ||
|
|
||
| internal suspend fun doFetchToggles(ctx: UnleashContext): ToggleResponse { | ||
| contextForLastFetch = ctx | ||
| val response = fetchToggles(ctx) | ||
| if (response.isSuccess()) { | ||
|
|
||
| val toggles = | ||
| response.config!!.toggles.groupBy { it.name }.mapValues { (_, v) -> v.first() } | ||
| Log.d( | ||
| TAG, | ||
| "Fetched new state with ${toggles.size} toggles, emitting featuresReceivedFlow" | ||
| ) | ||
| featuresReceivedFlow.emit(UnleashState(ctx, toggles)) | ||
| return ToggleResponse(response.status, toggles) | ||
|
|
@@ -124,26 +145,31 @@ open class UnleashFetcher( | |
|
|
||
| private suspend fun fetchToggles(ctx: UnleashContext): FetchResponse { | ||
| if (proxyUrl == null) { | ||
| return FetchResponse( | ||
| Status.FAILED, | ||
| error = IllegalStateException("Proxy URL is not set") | ||
| ) | ||
| } | ||
| val contextUrl = buildContextUrl(ctx) | ||
| try { | ||
| val request = Request.Builder().url(contextUrl).headers(applicationHeaders.toHeaders()) | ||
| if (etag != null) { | ||
| request.header("If-None-Match", etag!!) | ||
| } | ||
| val call = this.httpClient.newCall(request.build()) | ||
| val inFlightCall = currentCall.get() | ||
| if (!currentCall.compareAndSet(inFlightCall, call)) { | ||
| return FetchResponse( | ||
| Status.FAILED, | ||
| error = | ||
| IllegalStateException( | ||
| "Failed to set new call while ${inFlightCall?.request()?.url} is in flight" | ||
| ) | ||
| ) | ||
| } else if (inFlightCall != null && !inFlightCall.isCanceled() && !inFlightCall.isExecuted()) { | ||
|
Contributor
There was a problem hiding this comment. Do not cancel completed requests |
||
| Log.d( | ||
| TAG, | ||
| "Cancelling previous ${inFlightCall.request().method} ${inFlightCall.request().url}" | ||
| ) | ||
| inFlightCall.cancel() | ||
| } | ||
|
|
@@ -159,23 +185,21 @@ open class UnleashFetcher( | |
| res.body?.use { b -> | ||
| try { | ||
| val proxyResponse: ProxyResponse = | ||
| proxyResponseAdapter.fromJson(b.string())!! | ||
| FetchResponse(Status.SUCCESS, proxyResponse) | ||
| } catch (e: Exception) { | ||
| // If we fail to parse, just keep data | ||
| FetchResponse(Status.FAILED, error = e) | ||
| } | ||
| } | ||
| ?: FetchResponse(Status.FAILED, error = NoBodyException()) | ||
| } | ||
| res.code == 304 -> { | ||
| FetchResponse(Status.NOT_MODIFIED) | ||
| } | ||
| res.code == 401 -> { | ||
| FetchResponse(Status.FAILED, error = NotAuthorizedException()) | ||
| } | ||
| else -> { | ||
| FetchResponse(Status.FAILED, error = ServerException(res.code)) | ||
| } | ||
|
|
@@ -188,31 +212,33 @@ open class UnleashFetcher( | |
|
|
||
| private suspend fun Call.await(): Response { | ||
| return suspendCancellableCoroutine { continuation -> | ||
| enqueue( | ||
| object : Callback { | ||
| override fun onResponse(call: Call, response: Response) { | ||
| continuation.resume(response) | ||
| } | ||
|
|
||
| override fun onFailure(call: Call, e: IOException) { | ||
| // Don't bother with resuming the continuation if it is already | ||
| // cancelled. | ||
| if (continuation.isCancelled) return | ||
| continuation.resumeWithException(e) | ||
| } | ||
| } | ||
| ) | ||
|
|
||
| continuation.invokeOnCancellation { | ||
| try { | ||
| cancel() | ||
| } catch (ex: Throwable) { | ||
| // Ignore cancel exception | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private fun buildContextUrl(ctx: UnleashContext): HttpUrl { | ||
| var contextUrl = proxyUrl!!.newBuilder().addQueryParameter("appName", appName) | ||
| if (ctx.userId != null) { | ||
| contextUrl.addQueryParameter("userId", ctx.userId) | ||
| } | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hold the last fetched so we can ignore refreshing toggles when context has change due to a sync setContext which avoids 2 fetch calls for the same context