Use coroutines for Bangumi and Shikimori APIs

app/src/main/java/eu/kanade/tachiyomi/data/track/bangumi/Bangumi.kt

@@ -6,6 +6,7 @@ import eu.kanade.tachiyomi.R
 import eu.kanade.tachiyomi.data.database.models.Track
 import eu.kanade.tachiyomi.data.track.TrackService
 import eu.kanade.tachiyomi.data.track.model.TrackSearch
+import eu.kanade.tachiyomi.util.lang.runAsObservable
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
@@ -32,17 +33,17 @@ class Bangumi(private val context: Context, id: Int) : TrackService(id) {
     }
 
     override fun add(track: Track): Observable<Track> {
-        return api.addLibManga(track)
+        return runAsObservable({ api.addLibManga(track) })
     }
 
     override fun update(track: Track): Observable<Track> {
-        return api.updateLibManga(track)
+        return runAsObservable({ api.updateLibManga(track) })
     }
 
     override fun bind(track: Track): Observable<Track> {
-        return api.statusLibManga(track)
+        return runAsObservable({ api.statusLibManga(track) })
             .flatMap {
-                api.findLibManga(track).flatMap { remoteTrack ->
+                runAsObservable({ api.findLibManga(track) }).flatMap { remoteTrack ->
                     if (remoteTrack != null && it != null) {
                         track.copyPersonalFrom(remoteTrack)
                         track.library_id = remoteTrack.library_id
@@ -61,14 +62,14 @@ class Bangumi(private val context: Context, id: Int) : TrackService(id) {
     }
 
     override fun search(query: String): Observable<List<TrackSearch>> {
-        return api.search(query)
+        return runAsObservable({ api.search(query) })
     }
 
     override fun refresh(track: Track): Observable<Track> {
-        return api.statusLibManga(track)
+        return runAsObservable({ api.statusLibManga(track) })
             .flatMap {
                 track.copyPersonalFrom(it!!)
-                api.findLibManga(track)
+                runAsObservable({ api.findLibManga(track) })
                     .map { remoteTrack ->
                         if (remoteTrack != null) {
                             track.total_chapters = remoteTrack.total_chapters
@@ -103,7 +104,7 @@ class Bangumi(private val context: Context, id: Int) : TrackService(id) {
     override fun login(username: String, password: String) = login(password)
 
     fun login(code: String): Completable {
-        return api.accessToken(code).map { oauth: OAuth? ->
+        return runAsObservable({ api.accessToken(code) }).map { oauth: OAuth? ->
             interceptor.newAuth(oauth)
             if (oauth != null) {
                 saveCredentials(oauth.user_id.toString(), oauth.access_token)

app/src/main/java/eu/kanade/tachiyomi/data/track/bangumi/BangumiApi.kt

@@ -7,7 +7,7 @@ import eu.kanade.tachiyomi.data.track.TrackManager
 import eu.kanade.tachiyomi.data.track.model.TrackSearch
 import eu.kanade.tachiyomi.network.GET
 import eu.kanade.tachiyomi.network.POST
-import eu.kanade.tachiyomi.network.asObservableSuccess
+import eu.kanade.tachiyomi.network.await
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonObject
@@ -20,7 +20,6 @@ import okhttp3.CacheControl
 import okhttp3.FormBody
 import okhttp3.OkHttpClient
 import okhttp3.Request
-import rx.Observable
 import uy.kohesive.injekt.injectLazy
 import java.net.URLEncoder
 
@@ -30,59 +29,48 @@ class BangumiApi(private val client: OkHttpClient, interceptor: BangumiIntercept
 
     private val authClient = client.newBuilder().addInterceptor(interceptor).build()
 
-    fun addLibManga(track: Track): Observable<Track> {
+    suspend fun addLibManga(track: Track): Track {
         val body = FormBody.Builder()
             .add("rating", track.score.toInt().toString())
             .add("status", track.toBangumiStatus())
             .build()
-        return authClient.newCall(POST("$apiUrl/collection/${track.media_id}/update", body = body))
-            .asObservableSuccess()
-            .map {
-                track
-            }
+        authClient.newCall(POST("$apiUrl/collection/${track.media_id}/update", body = body)).await()
+        return track
     }
 
-    fun updateLibManga(track: Track): Observable<Track> {
+    suspend fun updateLibManga(track: Track): Track {
         // read status update
         val sbody = FormBody.Builder()
             .add("status", track.toBangumiStatus())
             .build()
-        return authClient.newCall(POST("$apiUrl/collection/${track.media_id}/update", body = sbody))
-            .asObservableSuccess()
-            .map {
-                track
-            }.flatMap {
-                // chapter update
-                val body = FormBody.Builder()
-                    .add("watched_eps", track.last_chapter_read.toString())
-                    .build()
-                authClient.newCall(POST("$apiUrl/subject/${track.media_id}/update/watched_eps", body = body))
-                    .asObservableSuccess()
-                    .map {
-                        track
-                    }
-            }
+        authClient.newCall(POST("$apiUrl/collection/${track.media_id}/update", body = sbody)).await()
+
+        // chapter update
+        val body = FormBody.Builder()
+            .add("watched_eps", track.last_chapter_read.toString())
+            .build()
+        authClient.newCall(POST("$apiUrl/subject/${track.media_id}/update/watched_eps", body = body)).await()
+
+        return track
     }
 
-    fun search(search: String): Observable<List<TrackSearch>> {
+    suspend fun search(search: String): List<TrackSearch> {
         val url = "$apiUrl/search/subject/${URLEncoder.encode(search, Charsets.UTF_8.name())}"
             .toUri()
             .buildUpon()
             .appendQueryParameter("max_results", "20")
             .build()
-        return authClient.newCall(GET(url.toString()))
-            .asObservableSuccess()
-            .map { netResponse ->
-                var responseBody = netResponse.body?.string().orEmpty()
-                if (responseBody.isEmpty()) {
-                    throw Exception("Null Response")
-                }
-                if (responseBody.contains("\"code\":404")) {
-                    responseBody = "{\"results\":0,\"list\":[]}"
-                }
-                val response = json.decodeFromString<JsonObject>(responseBody)["list"]?.jsonArray
-                response?.filter { it.jsonObject["type"]?.jsonPrimitive?.int == 1 }?.map { jsonToSearch(it.jsonObject) }
+        return authClient.newCall(GET(url.toString())).await().use {
+            var responseBody = it.body?.string().orEmpty()
+            if (responseBody.isEmpty()) {
+                throw Exception("Null Response")
+            }
+            if (responseBody.contains("\"code\":404")) {
+                responseBody = "{\"results\":0,\"list\":[]}"
             }
+            val response = json.decodeFromString<JsonObject>(responseBody)["list"]?.jsonArray
+            response?.filter { it.jsonObject["type"]?.jsonPrimitive?.int == 1 }?.map { jsonToSearch(it.jsonObject) }.orEmpty()
+        }
     }
 
     private fun jsonToSearch(obj: JsonObject): TrackSearch {
@@ -109,17 +97,15 @@ class BangumiApi(private val client: OkHttpClient, interceptor: BangumiIntercept
         }
     }
 
-    fun findLibManga(track: Track): Observable<Track?> {
-        return authClient.newCall(GET("$apiUrl/subject/${track.media_id}"))
-            .asObservableSuccess()
-            .map { netResponse ->
-                // get comic info
-                val responseBody = netResponse.body?.string().orEmpty()
-                jsonToTrack(json.decodeFromString(responseBody))
-            }
+    suspend fun findLibManga(track: Track): Track? {
+        return authClient.newCall(GET("$apiUrl/subject/${track.media_id}")).await().use {
+            // get comic info
+            val responseBody = it.body?.string().orEmpty()
+            jsonToTrack(json.decodeFromString(responseBody))
+        }
     }
 
-    fun statusLibManga(track: Track): Observable<Track?> {
+    suspend fun statusLibManga(track: Track): Track? {
         val urlUserRead = "$apiUrl/collection/${track.media_id}"
         val requestUserRead = Request.Builder()
             .url(urlUserRead)
@@ -127,30 +113,24 @@ class BangumiApi(private val client: OkHttpClient, interceptor: BangumiIntercept
             .get()
             .build()
 
-        // todo get user readed chapter here
-        return authClient.newCall(requestUserRead)
-            .asObservableSuccess()
-            .map { netResponse ->
-                val resp = netResponse.body?.string()
-                val coll = json.decodeFromString<Collection>(resp!!)
-                track.status = coll.status?.id!!
-                track.last_chapter_read = coll.ep_status!!
-                track
-            }
+        // TODO: get user readed chapter here
+        return authClient.newCall(requestUserRead).await().use {
+            val resp = it.body?.string()
+            val coll = json.decodeFromString<Collection>(resp!!)
+            track.status = coll.status?.id!!
+            track.last_chapter_read = coll.ep_status!!
+            track
+        }
     }
 
-    fun accessToken(code: String): Observable<OAuth> {
-        return client.newCall(accessTokenRequest(code))
-            .asObservableSuccess()
-            .map { netResponse ->
-                netResponse.use {
-                    val responseBody = it.body?.string().orEmpty()
-                    if (responseBody.isEmpty()) {
-                        throw Exception("Null Response")
-                    }
-                    json.decodeFromString<OAuth>(responseBody)
-                }
+    suspend fun accessToken(code: String): OAuth {
+        return client.newCall(accessTokenRequest(code)).await().use {
+            val responseBody = it.body?.string().orEmpty()
+            if (responseBody.isEmpty()) {
+                throw Exception("Null Response")
             }
+            json.decodeFromString<OAuth>(responseBody)
+        }
     }
 
     private fun accessTokenRequest(code: String) = POST(

app/src/main/java/eu/kanade/tachiyomi/data/track/shikimori/Shikimori.kt

@@ -6,6 +6,7 @@ import eu.kanade.tachiyomi.R
 import eu.kanade.tachiyomi.data.database.models.Track
 import eu.kanade.tachiyomi.data.track.TrackService
 import eu.kanade.tachiyomi.data.track.model.TrackSearch
+import eu.kanade.tachiyomi.util.lang.runAsObservable
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
@@ -44,15 +45,15 @@ class Shikimori(private val context: Context, id: Int) : TrackService(id) {
     }
 
     override fun add(track: Track): Observable<Track> {
-        return api.addLibManga(track, getUsername())
+        return runAsObservable({ api.addLibManga(track, getUsername()) })
     }
 
     override fun update(track: Track): Observable<Track> {
-        return api.updateLibManga(track, getUsername())
+        return runAsObservable({ api.updateLibManga(track, getUsername()) })
     }
 
     override fun bind(track: Track): Observable<Track> {
-        return api.findLibManga(track, getUsername())
+        return runAsObservable({ api.findLibManga(track, getUsername()) })
             .flatMap { remoteTrack ->
                 if (remoteTrack != null) {
                     track.copyPersonalFrom(remoteTrack)
@@ -68,11 +69,11 @@ class Shikimori(private val context: Context, id: Int) : TrackService(id) {
     }
 
     override fun search(query: String): Observable<List<TrackSearch>> {
-        return api.search(query)
+        return runAsObservable({ api.search(query) })
     }
 
     override fun refresh(track: Track): Observable<Track> {
-        return api.findLibManga(track, getUsername())
+        return runAsObservable({ api.findLibManga(track, getUsername()) })
             .map { remoteTrack ->
                 if (remoteTrack != null) {
                     track.copyPersonalFrom(remoteTrack)
@@ -107,7 +108,7 @@ class Shikimori(private val context: Context, id: Int) : TrackService(id) {
     override fun login(username: String, password: String) = login(password)
 
     fun login(code: String): Completable {
-        return api.accessToken(code).map { oauth: OAuth? ->
+        return runAsObservable({ api.accessToken(code) }).map { oauth: OAuth? ->
             interceptor.newAuth(oauth)
             if (oauth != null) {
                 val user = api.getCurrentUser()

app/src/main/java/eu/kanade/tachiyomi/data/track/shikimori/ShikimoriApi.kt

@@ -6,7 +6,7 @@ import eu.kanade.tachiyomi.data.track.TrackManager
 import eu.kanade.tachiyomi.data.track.model.TrackSearch
 import eu.kanade.tachiyomi.network.GET
 import eu.kanade.tachiyomi.network.POST
-import eu.kanade.tachiyomi.network.asObservableSuccess
+import eu.kanade.tachiyomi.network.await
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonArray
@@ -22,7 +22,6 @@ import okhttp3.FormBody
 import okhttp3.MediaType.Companion.toMediaType
 import okhttp3.OkHttpClient
 import okhttp3.RequestBody.Companion.toRequestBody
-import rx.Observable
 import uy.kohesive.injekt.injectLazy
 
 class ShikimoriApi(private val client: OkHttpClient, interceptor: ShikimoriInterceptor) {
@@ -32,7 +31,7 @@ class ShikimoriApi(private val client: OkHttpClient, interceptor: ShikimoriInter
     private val jsonMime = "application/json; charset=utf-8".toMediaType()
     private val authClient = client.newBuilder().addInterceptor(interceptor).build()
 
-    fun addLibManga(track: Track, user_id: String): Observable<Track> {
+    suspend fun addLibManga(track: Track, user_id: String): Track {
         val payload = buildJsonObject {
             putJsonObject("user_rate") {
                 put("user_id", user_id)
@@ -43,31 +42,26 @@ class ShikimoriApi(private val client: OkHttpClient, interceptor: ShikimoriInter
                 put("status", track.toShikimoriStatus())
             }
         }
-        return authClient.newCall(POST("$apiUrl/v2/user_rates", body = payload.toString().toRequestBody(jsonMime)))
-            .asObservableSuccess()
-            .map {
-                track
-            }
+        authClient.newCall(POST("$apiUrl/v2/user_rates", body = payload.toString().toRequestBody(jsonMime))).await()
+        return track
     }
 
-    fun updateLibManga(track: Track, user_id: String): Observable<Track> = addLibManga(track, user_id)
+    suspend fun updateLibManga(track: Track, user_id: String): Track = addLibManga(track, user_id)
 
-    fun search(search: String): Observable<List<TrackSearch>> {
+    suspend fun search(search: String): List<TrackSearch> {
         val url = "$apiUrl/mangas".toUri().buildUpon()
             .appendQueryParameter("order", "popularity")
             .appendQueryParameter("search", search)
             .appendQueryParameter("limit", "20")
             .build()
-        return authClient.newCall(GET(url.toString()))
-            .asObservableSuccess()
-            .map { netResponse ->
-                val responseBody = netResponse.body?.string().orEmpty()
-                if (responseBody.isEmpty()) {
-                    throw Exception("Null Response")
-                }
-                val response = json.decodeFromString<JsonArray>(responseBody)
-                response.map { jsonToSearch(it.jsonObject) }
+        return authClient.newCall(GET(url.toString())).await().use {
+            val responseBody = it.body?.string().orEmpty()
+            if (responseBody.isEmpty()) {
+                throw Exception("Null Response")
             }
+            val response = json.decodeFromString<JsonArray>(responseBody)
+            response.map { jsonToSearch(it.jsonObject) }
+        }
     }
 
     private fun jsonToSearch(obj: JsonObject): TrackSearch {
@@ -96,38 +90,34 @@ class ShikimoriApi(private val client: OkHttpClient, interceptor: ShikimoriInter
         }
     }
 
-    fun findLibManga(track: Track, user_id: String): Observable<Track?> {
+    suspend fun findLibManga(track: Track, user_id: String): Track? {
         val urlMangas = "$apiUrl/mangas".toUri().buildUpon()
             .appendPath(track.media_id.toString())
             .build()
-        return authClient.newCall(GET(urlMangas.toString()))
-            .asObservableSuccess()
-            .map { netResponse ->
-                val responseBody = netResponse.body?.string().orEmpty()
-                json.decodeFromString<JsonObject>(responseBody)
-            }.flatMap { mangas ->
-                val url = "$apiUrl/v2/user_rates".toUri().buildUpon()
-                    .appendQueryParameter("user_id", user_id)
-                    .appendQueryParameter("target_id", track.media_id.toString())
-                    .appendQueryParameter("target_type", "Manga")
-                    .build()
-                authClient.newCall(GET(url.toString()))
-                    .asObservableSuccess()
-                    .map { netResponse ->
-                        val responseBody = netResponse.body?.string().orEmpty()
-                        if (responseBody.isEmpty()) {
-                            throw Exception("Null Response")
-                        }
-                        val response = json.decodeFromString<JsonArray>(responseBody)
-                        if (response.size > 1) {
-                            throw Exception("Too much mangas in response")
-                        }
-                        val entry = response.map {
-                            jsonToTrack(it.jsonObject, mangas)
-                        }
-                        entry.firstOrNull()
-                    }
+        val mangas = authClient.newCall(GET(urlMangas.toString())).await().use {
+            val responseBody = it.body?.string().orEmpty()
+            json.decodeFromString<JsonObject>(responseBody)
+        }
+
+        val url = "$apiUrl/v2/user_rates".toUri().buildUpon()
+            .appendQueryParameter("user_id", user_id)
+            .appendQueryParameter("target_id", track.media_id.toString())
+            .appendQueryParameter("target_type", "Manga")
+            .build()
+        return authClient.newCall(GET(url.toString())).await().use {
+            val responseBody = it.body?.string().orEmpty()
+            if (responseBody.isEmpty()) {
+                throw Exception("Null Response")
             }
+            val response = json.decodeFromString<JsonArray>(responseBody)
+            if (response.size > 1) {
+                throw Exception("Too much mangas in response")
+            }
+            val entry = response.map {
+                jsonToTrack(it.jsonObject, mangas)
+            }
+            entry.firstOrNull()
+        }
     }
 
     fun getCurrentUser(): Int {
@@ -135,18 +125,14 @@ class ShikimoriApi(private val client: OkHttpClient, interceptor: ShikimoriInter
         return json.decodeFromString<JsonObject>(user)["id"]!!.jsonPrimitive.int
     }
 
-    fun accessToken(code: String): Observable<OAuth> {
-        return client.newCall(accessTokenRequest(code))
-            .asObservableSuccess()
-            .map { netResponse ->
-                netResponse.use {
-                    val responseBody = it.body?.string().orEmpty()
-                    if (responseBody.isEmpty()) {
-                        throw Exception("Null Response")
-                    }
-                    json.decodeFromString<OAuth>(responseBody)
-                }
+    suspend fun accessToken(code: String): OAuth {
+        return client.newCall(accessTokenRequest(code)).await().use {
+            val responseBody = it.body?.string().orEmpty()
+            if (responseBody.isEmpty()) {
+                throw Exception("Null Response")
             }
+            json.decodeFromString<OAuth>(responseBody)
+        }
     }
 
     private fun accessTokenRequest(code: String) = POST(