How to use private class of com.github.kittinunf.fuel.coroutines package

...37 * @version 1.038 */39open class MessageHandler : CoroutineScope {40    override val coroutineContext: CoroutineContext = Job()41    private var enabled = false42    private var connectErrors = 043    private var reconnectCooldown = 0L44    private var sendErrors = 045    private var sendChannel: SendChannel<ApiMessage> = senderActor()46    private val messageStream = Channel<ApiMessage>(Channel.UNLIMITED)47    @UseExperimental(ExperimentalCoroutinesApi::class)48    var broadcast: BroadcastChannel<ApiMessage> = broadcast {49        while (true) {50            val msg = messageStream.receive()51            send(msg)52        }53    }54        private set55    private val keepOpenManager = FuelManager().apply {56        timeoutInMillisecond = 057        timeoutReadInMillisecond = 058    }59    var config: Config = Config()60    var logger = object : Logger {61        override fun info(message: String) = println("INFO: $message")62        override fun debug(message: String) = println("DEBUG: $message")63        override fun error(message: String) = println("ERROR: $message")64        override fun warn(message: String) = println("WARN: $message")65        override fun trace(message: String) = println("TRACE: $message")66    }67    suspend fun stop(message: String? = null) {68        if (message != null && config.announceDisconnect) {69            sendStatusUpdate(message)70        }71        enabled = false72        rcvJob?.cancel()73        rcvJob = null74    }75    private var rcvJob: Job? = null76    suspend fun start(message: String?, clear: Boolean) {77        logger.debug("starting connection")78        if (clear) {79            clear()80        }81        enabled = true82        rcvJob = messageBroadcast()83        if (message != null && config.announceConnect) {84            sendStatusUpdate(message)85        }86    }87    private suspend fun clear() {88        val url = "${config.url}/api/messages"89        val (request, response, result) = url.httpGet()90            .apply {91                if (config.token.isNotEmpty()) {92                    headers["Authorization"] = "Bearer ${config.token}"93                }94            }95            .awaitStringResponseResult()96        when (result) {97            is Result.Success -> {98                val messages: List<ApiMessage> = JSON.nonstrict.parse(ApiMessage.list, result.value)99                messages.forEach { msg ->100                    logger.trace("skipping $msg")101                }102                logger.debug("skipped ${messages.count()} messages")103            }104            is Result.Failure -> {105                logger.error("failed to clear messages")106                logger.error("url: $url")107                logger.error("cUrl: ${request.cUrlString()}")108                logger.error("response: $response")109                logger.error(result.error.exception.localizedMessage)110                result.error.exception.printStackTrace()111            }112        }113    }114    open suspend fun sendStatusUpdate(message: String) {115        transmit(ApiMessage(text = message))116    }117    open suspend fun transmit(msg: ApiMessage) {118//        if (streamConnection.isConnected || streamConnection.isConnecting) {119        if (msg.username.isEmpty())120            msg.username = config.systemUser121        if (msg.gateway.isEmpty()) {122            logger.error("missing gateway on message: $msg")123            return124        }125        logger.debug("Transmitting: $msg")126        sendChannel.send(msg)127//        }128    }129    @Deprecated("use coroutine api", level = DeprecationLevel.ERROR)130    fun checkConnection() {131    }132    @UseExperimental(ObsoleteCoroutinesApi::class)133    private fun CoroutineScope.senderActor() = actor<ApiMessage>(context = Dispatchers.IO) {134        consumeEach {135            try {136                logger.debug("sending $it")137                val url = "${config.url}/api/message"138                val (request, response, result) = url.httpPost()139                    .apply {140                        if (config.token.isNotEmpty()) {141                            headers["Authorization"] = "Bearer ${config.token}"142                        }143                    }144                    .jsonBody(it.encode())145                    .responseString()146                when (result) {147                    is Result.Success -> {148                        logger.debug("sent $it")149                        sendErrors = 0150                    }151                    is Result.Failure -> {152                        sendErrors++153                        logger.error("failed to deliver: $it")154                        logger.error("url: $url")155                        logger.error("cUrl: ${request.cUrlString()}")156                        logger.error("response: $response")157                        logger.error(result.error.exception.localizedMessage)158                        result.error.exception.printStackTrace()159//                    close()160                        throw result.error.exception161                    }162                }163            } catch (connectError: ConnectException) {164                connectError.printStackTrace()165                sendErrors++166            }167        }168    }169    private fun CoroutineScope.messageBroadcast() = launch(context = Dispatchers.IO + CoroutineName("msgBroadcaster")) {170        loop@ while (isActive) {171            logger.info("opening connection")172            val url = "${config.url}/api/stream"173            val (request, response, result) = keepOpenManager.request(Method.GET, url)174                .apply {175                    if (config.token.isNotEmpty()) {176                        headers["Authorization"] = "Bearer ${config.token}"177                    }178                }179                .responseObject(object : ResponseDeserializable<Unit> {180                    override fun deserialize(reader: Reader) =181                        runBlocking(Dispatchers.IO + CoroutineName("msgReceiver")) {182                            logger.info("connected successfully")183                            connectErrors = 0...

...18import kotlin.coroutines.resumeWithException19import kotlin.coroutines.suspendCoroutine20typealias JProxy = java.net.Proxy21object HttpUtils {22    private val proxyBlacklist = ConcurrentHashMap<Proxy, Long>()23    private val untrustedProxyList = mutableListOf<Proxy>()24    var currentProxy: Proxy? = null25    private fun setProxy(proxy: Proxy?) {26        proxy?.let {27            "Using proxy: $proxy".log()28            currentProxy = proxy29            currentProxy?.type = "current"30        }31        FuelManager.instance = FuelManager()32        FuelManager.instance.proxy = proxy?.let { JProxy(java.net.Proxy.Type.HTTP, InetSocketAddress(it.host, it.port)) }33        FuelManager.instance.timeoutInMillisecond = 500034        FuelManager.instance.timeoutReadInMillisecond = 500035    }36    private fun setProxies(list: List<Proxy>) {37        proxyBlacklist.clear()38        untrustedProxyList.clear()39        list.forEach {40            when (it.type) {41                "current" -> currentProxy = it42                "untrusted" -> untrustedProxyList.add(it)43                else -> proxyBlacklist[it] = it.added44            }45        }46    }47    private fun getBlackList(): List<Proxy> {48        return proxyBlacklist.map {49            it.key.added = it.value50            it.key51        }52    }53    fun storeProxies(dao: ProxyDao): Job {54        // slow but simple55        return GlobalScope.launch(Dispatchers.IO) {56            val blackList = HttpUtils.getBlackList().filter { it.added > System.currentTimeMillis() - 1000 * 60 * 60 * 24 }57            dao.deleteAll()58            dao.insertAll(blackList)59            dao.insertAll(untrustedProxyList)60            currentProxy?.let {61                it.type = "current"62                dao.insert(it)63            }64        }65    }66    fun loadProxies(dao: ProxyDao): Job {67        return GlobalScope.launch(Dispatchers.IO) {68            HttpUtils.setProxies(dao.getAll())69        }70    }71    suspend fun call4Json(method: HttpMethod, url: String, withProxy: Boolean = false): Json? {72        for (retries in 0..10) {73            try {74                return callHttp(method, url, withProxy)75            } catch (e: ProxyNotAvailableException) {76                if (currentProxy == null) {77                    return null78                }79                "Retrying with another proxy".log()80            } catch (e: Exception) {81                "Error connecting".logE(e)82                break83            }84        }85        return null86    }87    private suspend fun callHttp(method: HttpMethod = HttpMethod.GET, url: String, withProxy: Boolean = false): Json {88        setProxy(if (State.useProxy && withProxy) getProxy() else null)89        "Calling: $method $url".log()90        val caller = when (method) {91            HttpMethod.GET -> url.httpGet()92            HttpMethod.POST -> url.httpPost()93        }94        return suspendCoroutine { continuation ->95            caller.responseJson { _, resp, result ->96                try {97                    if (result is Result.Success && resp.headers["Content-Type"]?.firstOrNull()?.contains("application/json") == true) {98                        "Received result ${result.component1()?.content}".log()99                        continuation.resume(result.component1()!!)100                        return@responseJson101                    } else {102                        val currProxy = currentProxy103                        if (currProxy != null) {104                            "Blacklisting $currProxy".log()105                            currProxy.type = "blacklisted"106                            proxyBlacklist[currProxy] = System.currentTimeMillis()107                        }108                        result.component2().toString().logE()109                        throw ProxyNotAvailableException()110                    }111                } catch (e: Exception) {112                    continuation.resumeWithException(e)113                }114            }115        }116    }117    private suspend fun getProxy(): Proxy? {118        val currProxy = currentProxy119        if (currProxy != null && proxyBlacklist[currProxy] == null) {120            return currProxy121        }122        var fetched = false123        while (true) {124            if (untrustedProxyList.isEmpty()) {125                if (fetched) {126                    return null127                } // we have already fetched during this iteration, next fetch will bring the same results128                untrustedProxyList.addAll(fetchProxyList())129                fetched = true130            }131            val proxy = untrustedProxyList[0]132            val time = proxyBlacklist[proxy] ?: return proxy133            if (System.currentTimeMillis() - time > 1000 * 60 * 60 * 24) {134                proxyBlacklist.remove(proxy)135                return proxy136            }137            // it is still blacklisted, removing from a list138            untrustedProxyList.removeAt(0)139        }140    }141    private suspend fun fetchProxyList(): List<Proxy> {142        "Fetching proxy list".log()143        return suspendCoroutine { continuation ->144            try {145                Jsoup.connect("https://www.proxy" + "nova.com/proxy-server-list/country-ru/").get().run {146                    "Proxy list fetched".log()147                    val result = getElementById("tbl_p" + "roxy_list").select("tbody tr").map { row ->148                        if (!row.hasAttr("data-proxy-id")) {149                            return@map Proxy("", 0)150                        }151                        val columns = row.select("td")152                        val ip = columns[0].select("abbr").html().replace(Regex(".*?(\\d+\\.\\d+\\.\\d+\\.\\d+).*"), "$1")153                        val port = columns[1].select("a").text()154                        val speed = columns[3].select("small").text().split(" ")[0]155                        val type = columns[6].select("span").text()...

...23import kotlinx.serialization.decodeFromString24import kotlinx.serialization.encodeToString25import kotlinx.serialization.json.Json26class GatewayRepository {27    private val gatewayDataSource = GatewayDataSource()28    suspend fun retrieveAll(): Flow<LoadingResource<List<Gateway>>> {29        return flow {30            while (true) {31                try {32                    emit(LoadingResource.Loading())33                    delay(Constants.GATEWAY_LOADING)34                    emit(LoadingResource.Success(gatewayDataSource.retrieveAll()))35                } catch (ex: Exception) {36                    emit(LoadingResource.Error(ex, ex.message))37                }38                delay(Constants.REFRESH_GATEWAY_DELAY)39            }40        }41    }...

...13import timber.log.Timber14import java.time.LocalDate15@UnsafeSerializationApi16object RemoteRepo {17    private const val BASE_URL = "https://api.boerse-frankfurt.de/data"18    private const val PRICE_HISTORY_EP = "/price_history"19    private const val ISIN_PARAM_ID = "isin"20    private const val MIN_DATE_ID = "minDate"21    private const val MAX_DATE_ID = "maxDate"22    private val OFFSET_PARAM = "offset" to 023    private val LIMIT_PARAM = "limit" to 50524    private val MIC_PARAM = "mic" to "XETR"25    fun getHistory(26        isin: String,27        minDate: LocalDate,28        maxDate: LocalDate29    ): IO<Either<Throwable, History>> = IO {30        req(31            BASE_URL + PRICE_HISTORY_EP,32            listOf(33                OFFSET_PARAM,34                LIMIT_PARAM,35                MIC_PARAM,36                ISIN_PARAM_ID to isin,37                MIN_DATE_ID to minDate,38                MAX_DATE_ID to maxDate39            )40        )41    }42    private suspend inline fun <reified T : Any> req(43        url: String,44        params: Parameters? = null45    ): Either<Throwable, T> = Either.catch {46        val response = url.httpGet(params).awaitStringResult().catchable()47        response.deserialize<T>()48    }.flatten()49    private suspend inline fun <reified T : Any> String.deserialize(): Either<Throwable, T> =50        Either.catch {51            Json.decodeFromString(deserializer = T::class.serializer(), string = this)52        }.also {53            if (it.isLeft()) {54                Timber.e("Error deserializing to ${T::class.java}: $this")55            }56        }57    private fun com.github.kittinunf.result.Result<String, FuelError>.catchable(): String =58        fold(59            { it },60            {61                Timber.e("Error fetching url: $it")62                throw it63            }64        )65}...

...11import com.sar.shopaholism.domain.logger.Logger12import kotlinx.coroutines.CoroutineDispatcher13import kotlinx.coroutines.withContext14class WebApiClientImpl(15    private val fuelManager: FuelManager,16    private val logger: Logger,17    private var rateLimiter: RateLimiter? = null18) : WebApiClient {19    override fun setBasePath(basePath: String) {20        fuelManager.basePath = basePath21    }22    override fun setBaseParams(baseParams: Map<String, String>) {23        fuelManager.baseParams = baseParams.map { Pair<String, Any?>(it.key, it.value) }24    }25    fun setRateLimiter(rateLimiter: RateLimiter?) {26        this.rateLimiter = rateLimiter27    }28    private fun Request.toFuelRequest(): com.github.kittinunf.fuel.core.Request =29        fuelManager.request(30            method = Method.valueOf(method.value),31            path = path,32            parameters = parameters.map { Pair<String, Any?>(it.key, it.value) }33        )34    override suspend fun sendRequest(request: Request, dispatcher: CoroutineDispatcher): String =35        withContext(dispatcher) {36            if (rateLimiter?.isRateLimited() == true) {37                throw RateLimitedException("Rate limited")38            }39            try {40                return@withContext request.toFuelRequest().awaitString()41            } catch (e: Exception) {42                throw WebRequestUnsuccessfulException("Web request failed", e)...

...9import kotlinx.coroutines.runBlocking10import java.lang.Exception11//fun main() = threads()12 suspend fun main() = coroutines()13private fun threads() {14    repeat(10000) {15        val thread = Thread {16            callHello(it)17        }18        thread.start()19    }20}21private suspend fun coroutines() {22    coroutineScope {23        repeat(10000) {24            launch {25                callHelloCoroutine(it)26            }27        }28    }29}30private suspend fun callHelloCoroutine(i: Int) {31    val (request, response, result) = Fuel.get("http://localhost:8080/hello").awaitStringResponseResult()32    doSomething(result, i)33}34private fun callHello(i: Int) {35    val (request, response, result) = Fuel.get("http://localhost:8080/hello").responseString()36    doSomething(result, i)37}38private fun doSomething(result: Result<String, FuelError>, i: Int) {39    result.fold(40            { data -> println("$data - ${Thread.currentThread().name} - $i") },41            { error ->42                run {43                    println("An error of type ${error.exception} happened: ${error.message}")44                    throw Exception("Stop it")45                }46            }47    )48}...

...3import com.github.kittinunf.fuel.coroutines.awaitObjectResponseResult4import com.github.kittinunf.fuel.gson.gsonDeserializerOf5import com.github.kittinunf.fuel.httpGet6import kotlinx.coroutines.runBlocking7private const val workspaceId = "WORKSPACE_ID"8private const val apiKey = "API_KEY"9private const val defaultPwd = "api_token"10private const val user = "info@lunatech.com"11private const val reportingApiUrl = "https://toggl.com/reports/api/v2/details?user_agent=$user&workspace_id=$workspaceId"12private const val apiUrl = "https://toggl.com/api/v8/"13private const val workspaceClients = "workspaces/%s/clients"14fun fetchDetailsReport() {15    runBlocking {16        reportingApiUrl.httpGet()17            .authentication()18            .basic(apiKey, defaultPwd)19            .awaitObjectResponseResult(gsonDeserializerOf(DetailsReport::class.java)).third20            .fold(21                { data -> println(data) },22                { error -> println("An error of type ${error.exception} happened: ${error.message}") }23            )24    }25}...

...6import com.github.kittinunf.result.Result7import kotlinx.coroutines.delay8import kotlinx.serialization.ImplicitReflectionSerializer9import kotlinx.serialization.internal.ArrayListSerializer10private const val URL = "https://hoc081098.github.io/hoc081098.github.io/data.json"11@ImplicitReflectionSerializer12suspend fun getViewPagerItems(): Result<List<ViewPagerItem>, FuelError> {13    delay(2_000)14    return URL15        .httpGet()16        .awaitObjectResult(kotlinxDeserializerOf(ArrayListSerializer(ViewPagerItem.serializer())))17}...

1    import com.github.kittinunf.fuel.coroutines.awaitObjectResult2    import com.github.kittinunf.fuel.coroutines.awaitStringResult3    import com.github.kittinunf.fuel.coroutines.awaitByteArrayResult4    import com.github.kittinunf.fuel.coroutines.awaitResult5    import com.github.kittinunf.fuel.coroutines.awaitObjectResult6    import com.github.kittinunf.fuel.coroutines.awaitObjectResult7    import com.github.kittinunf.fuel.coroutines.awaitStringResult8    import com.github.kittinunf.fuel.coroutines.awaitByteArrayResult9    import com.github.kittinunf.fuel.coroutines.awaitResult10    import com.github.kittinunf.fuel.coroutines.awaitObjectResult11    import com.github.kittinunf.fuel.coroutines.awaitObjectResult12    import com.github.kittinunf.fuel.coroutines.awaitStringResult13    import com.github.kittinunf.fuel.coroutines.awaitByteArrayResult14    import com.github.kittinunf.fuel.coroutines.awaitResult15    import com.github.kittinunf.fuel.coroutines.awaitObjectResult16    import com.github.kittinunf.fuel.coroutines.awaitObjectResult17    import com.github.kittinunf.fuel.coroutines.awaitStringResult18    import com.github.kittinunf.fuel.coroutines.awaitByteArrayResult19    import com.github.kittinunf.fuel.coroutines.awaitResult20    import com.github.kittinunf.fuel.coroutines.awaitObjectResult21    import com.github.kittinunf.fuel.coroutines.awaitObjectResult22    import com.github.kittinunf.fuel.coroutines.awaitStringResult23    import com.github.kittinunf.fuel.coroutines.awaitByteArrayResult24    import com.github.kittinunf.fuel.coroutines.awaitResult25    import com.github.kittinunf.fuel.coroutines.awaitObjectResult26    import com

1@file:Suppress("unused")2import com.github.kittinunf.fuel.core.*3import kotlinx.coroutines.experimental.CommonPool4import kotlinx.coroutines.experimental.Deferred5import kotlinx.coroutines.experimental.async6suspend fun Request.awaitStringResponseResult(): ResponseResultOf<String> = async(CommonPool) {

1val deferred = GlobalScope.async(Dispatchers.IO) {2com.github.kittinunf.fuel.coroutines.awaitStringResponseResult(request)3}4val (request, response, result) = deferred.await()5}6}7}8@Throws(Exception::class)9fun main(args: Array<String>) {10val request = Request(Method.GET, url)11val result = request.awaitStringResponseResult()12println(result)13}14}15import com.github.kittinunf.fuel.core.Method16import com.github.kittinunf.fuel.core.Request17import com.github.kittinunf.fuel.coroutines.awaitStringResponseResult18import kotlin.coroutines.experimental.CoroutineContext19import kotlinx.coroutines.experimental.*20import kotlinx.coroutines.experimental.android.Main21import kotlinx.coroutines.experimental.runBlocking22class FuelCoroutines {23companion object {24suspend fun getResponseInCoroutine(request: Request, context: CoroutineContext = Main): String {25val deferred = GlobalScope.async(Dispatchers.IO) {26com.github.kittinunf.fuel.coroutines.awaitStringResponseResult(request)27}28val (request, response, result) = deferred.await()29}30}31}32@Throws(Exception::class)33fun main(args: Array<String>) {34val request = Request(Method.GET, url)35runBlocking {36val result = FuelCoroutines.getResponseInCoroutine(request)37println(result)38}39}40}41import com.github.kittinunf.fuel.core.Method42import com.github.kittinunf.fuel.core.Request43import com.github.kittinunf.fuel.coroutines.awaitStringResponseResult44import kotlin.coroutines.experimental.Coroutine