-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
10 changed files
with
551 additions
and
86 deletions.
There are no files selected for viewing
This file contains 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 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 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 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 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 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 |
|---|---|---|
| @@ -0,0 +1,134 @@ | ||
| package com.amplitude.experiment.flag | ||
|
|
||
| import com.amplitude.experiment.evaluation.EvaluationFlag | ||
| import com.amplitude.experiment.util.* | ||
| import com.amplitude.experiment.util.SdkStream | ||
| import kotlinx.serialization.decodeFromString | ||
| import okhttp3.OkHttpClient | ||
| import java.util.concurrent.CompletableFuture | ||
| import java.util.concurrent.ExecutionException | ||
| import java.util.concurrent.TimeUnit | ||
| import java.util.concurrent.TimeoutException | ||
| import java.util.concurrent.atomic.AtomicBoolean | ||
|
|
||
| internal open class FlagConfigStreamApiError(message: String?, cause: Throwable?): Exception(message, cause) { | ||
| constructor(message: String?) : this(message, null) | ||
| constructor(cause: Throwable?) : this(cause?.toString(), cause) | ||
| } | ||
| internal class FlagConfigStreamApiConnTimeoutError: FlagConfigStreamApiError("Initial connection timed out") | ||
| internal class FlagConfigStreamApiDataCorruptError: FlagConfigStreamApiError("Stream data corrupted") | ||
| internal class FlagConfigStreamApiStreamError(cause: Throwable?): FlagConfigStreamApiError("Stream error", cause) | ||
|
|
||
| private const val CONNECTION_TIMEOUT_MILLIS_DEFAULT = 2000L | ||
| private const val KEEP_ALIVE_TIMEOUT_MILLIS_DEFAULT = 17000L | ||
| private const val RECONN_INTERVAL_MILLIS_DEFAULT = 15 * 60 * 1000L | ||
| internal class FlagConfigStreamApi ( | ||
| deploymentKey: String, | ||
| serverUrl: String, | ||
| httpClient: OkHttpClient = OkHttpClient(), | ||
| connectionTimeoutMillis: Long = CONNECTION_TIMEOUT_MILLIS_DEFAULT, | ||
| keepaliveTimeoutMillis: Long = KEEP_ALIVE_TIMEOUT_MILLIS_DEFAULT, | ||
| reconnIntervalMillis: Long = RECONN_INTERVAL_MILLIS_DEFAULT | ||
| ) { | ||
| var onInitUpdate: ((List<EvaluationFlag>) -> Unit)? = null | ||
| var onUpdate: ((List<EvaluationFlag>) -> Unit)? = null | ||
| var onError: ((Exception?) -> Unit)? = null | ||
| private val stream: SdkStream = SdkStream( | ||
| "Api-Key $deploymentKey", | ||
| "$serverUrl/sdk/stream/v1/flags", | ||
| httpClient, | ||
| connectionTimeoutMillis, | ||
| keepaliveTimeoutMillis, | ||
| reconnIntervalMillis) | ||
|
|
||
| fun connect() { | ||
| val isInit = AtomicBoolean(true) | ||
| val connectTimeoutFuture = CompletableFuture<Unit>() | ||
| val updateTimeoutFuture = CompletableFuture<Unit>() | ||
| stream.onUpdate = { data -> | ||
| if (isInit.getAndSet(false)) { | ||
| // Stream is establishing. First data received. | ||
| // Resolve timeout. | ||
| connectTimeoutFuture.complete(Unit) | ||
|
|
||
| // Make sure valid data. | ||
| try { | ||
| val flags = getFlagsFromData(data) | ||
|
|
||
| try { | ||
| if (onInitUpdate != null) { | ||
| onInitUpdate?.let { it(flags) } | ||
| } else { | ||
| onUpdate?.let { it(flags) } | ||
| } | ||
| updateTimeoutFuture.complete(Unit) | ||
| } catch (e: Throwable) { | ||
| updateTimeoutFuture.completeExceptionally(e) | ||
| } | ||
| } catch (_: Throwable) { | ||
| updateTimeoutFuture.completeExceptionally(FlagConfigStreamApiDataCorruptError()) | ||
| } | ||
|
|
||
| } else { | ||
| // Stream has already established. | ||
| // Make sure valid data. | ||
| try { | ||
| val flags = getFlagsFromData(data) | ||
|
|
||
| try { | ||
| onUpdate?.let { it(flags) } | ||
| } catch (_: Throwable) { | ||
| // Don't care about application error. | ||
| } | ||
| } catch (_: Throwable) { | ||
| // Stream corrupted. Reconnect. | ||
| handleError(FlagConfigStreamApiDataCorruptError()) | ||
| } | ||
|
|
||
| } | ||
| } | ||
| stream.onError = { t -> | ||
| if (isInit.getAndSet(false)) { | ||
| connectTimeoutFuture.completeExceptionally(t) | ||
| updateTimeoutFuture.completeExceptionally(t) | ||
| } else { | ||
| handleError(FlagConfigStreamApiStreamError(t)) | ||
| } | ||
| } | ||
| stream.connect() | ||
|
|
||
| val t: Throwable | ||
| try { | ||
| connectTimeoutFuture.get(2000, TimeUnit.MILLISECONDS) | ||
| updateTimeoutFuture.get() | ||
| return | ||
| } catch (e: TimeoutException) { | ||
| // Timeouts should retry | ||
| t = FlagConfigStreamApiConnTimeoutError() | ||
| } catch (e: ExecutionException) { | ||
| val cause = e.cause | ||
| t = if (cause is StreamException) { | ||
| FlagConfigStreamApiStreamError(cause) | ||
| } else { | ||
| FlagConfigStreamApiError(e) | ||
| } | ||
| } catch (e: Throwable) { | ||
| t = FlagConfigStreamApiError(e) | ||
| } | ||
| close() | ||
| throw t | ||
| } | ||
|
|
||
| fun close() { | ||
| stream.cancel() | ||
| } | ||
|
|
||
| private fun getFlagsFromData(data: String): List<EvaluationFlag> { | ||
| return json.decodeFromString<List<EvaluationFlag>>(data) | ||
| } | ||
|
|
||
| private fun handleError(e: Exception?) { | ||
| close() | ||
| onError?.let { it(e) } | ||
| } | ||
| } |
Oops, something went wrong.