Migrate extension list fetch to coroutine

app/src/main/java/eu/kanade/tachiyomi/data/updater/UpdaterService.kt

@@ -17,9 +17,7 @@ import uy.kohesive.injekt.injectLazy
 import java.io.File
 
 class UpdaterService : IntentService(UpdaterService::class.java.name) {
-    /**
+
-     * Network helper
-     */
     private val network: NetworkHelper by injectLazy()
 
     /**

app/src/main/java/eu/kanade/tachiyomi/extension/ExtensionManager.kt

@@ -15,8 +15,6 @@ import eu.kanade.tachiyomi.source.SourceManager
 import eu.kanade.tachiyomi.util.lang.launchNow
 import kotlinx.coroutines.async
 import rx.Observable
-import rx.android.schedulers.AndroidSchedulers
-import rx.schedulers.Schedulers
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
 
@@ -146,11 +144,13 @@ class ExtensionManager(
      * Finds the available extensions in the [api] and updates [availableExtensions].
      */
     fun findAvailableExtensions() {
-        api.findExtensions()
+        launchNow {
-                .onErrorReturn { emptyList() }
+            availableExtensions = try {
-                .subscribeOn(Schedulers.io())
+                api.findExtensions()
-                .observeOn(AndroidSchedulers.mainThread())
+            } catch (e: Exception) {
-                .subscribe { availableExtensions = it }
+                emptyList()
+            }
+        }
     }
 
     /**

app/src/main/java/eu/kanade/tachiyomi/extension/api/ExtensionGithubApi.kt

@@ -9,26 +9,20 @@ import com.google.gson.JsonArray
 import eu.kanade.tachiyomi.extension.model.Extension
 import eu.kanade.tachiyomi.network.GET
 import eu.kanade.tachiyomi.network.NetworkHelper
-import eu.kanade.tachiyomi.network.asObservableSuccess
+import eu.kanade.tachiyomi.network.await
 import okhttp3.Response
-import rx.Observable
 import uy.kohesive.injekt.injectLazy
 
 internal class ExtensionGithubApi {
 
     private val network: NetworkHelper by injectLazy()
 
-    private val client get() = network.client
-
     private val gson: Gson by injectLazy()
 
-    private val repoUrl = "https://raw.githubusercontent.com/inorichi/tachiyomi-extensions/repo"
+    suspend fun findExtensions(): List<Extension.Available> {
+        val call = GET("$REPO_URL/index.json")
 
-    fun findExtensions(): Observable<List<Extension.Available>> {
+        return parseResponse(network.client.newCall(call).await())
-        val call = GET("$repoUrl/index.json")
-
-        return client.newCall(call).asObservableSuccess()
-                .map(::parseResponse)
     }
 
     private fun parseResponse(response: Response): List<Extension.Available> {
@@ -43,13 +37,17 @@ internal class ExtensionGithubApi {
             val versionName = element["version"].string
             val versionCode = element["code"].int
             val lang = element["lang"].string
-            val icon = "$repoUrl/icon/${apkName.replace(".apk", ".png")}"
+            val icon = "$REPO_URL/icon/${apkName.replace(".apk", ".png")}"
 
             Extension.Available(name, pkgName, versionName, versionCode, lang, apkName, icon)
         }
     }
 
     fun getApkUrl(extension: Extension.Available): String {
-        return "$repoUrl/apk/${extension.apkName}"
+        return "$REPO_URL/apk/${extension.apkName}"
+    }
+
+    companion object {
+        private const val REPO_URL = "https://raw.githubusercontent.com/inorichi/tachiyomi-extensions/repo"
     }
 }

app/src/main/java/eu/kanade/tachiyomi/network/OkHttpExtensions.kt

@@ -1,13 +1,14 @@
 package eu.kanade.tachiyomi.network
 
-import okhttp3.Call
+import kotlinx.coroutines.suspendCancellableCoroutine
-import okhttp3.OkHttpClient
+import okhttp3.*
-import okhttp3.Request
-import okhttp3.Response
 import rx.Observable
 import rx.Producer
 import rx.Subscription
+import java.io.IOException
 import java.util.concurrent.atomic.AtomicBoolean
+import kotlin.coroutines.resume
+import kotlin.coroutines.resumeWithException
 
 fun Call.asObservable(): Observable<Response> {
     return Observable.unsafeCreate { subscriber ->
@@ -46,6 +47,31 @@ fun Call.asObservable(): Observable<Response> {
     }
 }
 
+// Based on https://github.com/gildor/kotlin-coroutines-okhttp
+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
+            }
+        }
+    }
+}
+
 fun Call.asObservableSuccess(): Observable<Response> {
     return asObservable().doOnNext { response ->
         if (!response.isSuccessful) {