Revert "Run more db queries inside the IO pool, convert some RxJava references to Coroutines"

This reverts commit 18f02a85ac.

# Conflicts:
#	app/src/main/java/eu/kanade/tachiyomi/ui/setting/SettingsEhController.kt

app/src/main/java/exh/metadata/metadata/base/FlatMetadata.kt

@@ -2,8 +2,6 @@ package exh.metadata.metadata.base
 
 import com.pushtorefresh.storio.operations.PreparedOperation
 import eu.kanade.tachiyomi.data.database.DatabaseHelper
-import eu.kanade.tachiyomi.util.lang.runAsObservable
-import eu.kanade.tachiyomi.util.lang.withIOContext
 import exh.metadata.sql.models.SearchMetadata
 import exh.metadata.sql.models.SearchTag
 import exh.metadata.sql.models.SearchTitle
@@ -11,9 +9,6 @@ import exh.util.executeOnIO
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.async
 import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.runBlocking
 import kotlinx.serialization.InternalSerializationApi
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.serializer
@@ -37,35 +32,23 @@ data class FlatMetadata(
             }
 }
 
-interface PreparedSuspendOperation<T> : PreparedOperation<T> {
-    /**
-     * Creates a [Flow] that emits the result of of Operation
-     *
-     * Example:
-     *  override fun asFlow(): Flow<T> = flow { emit(operation()) }
-     *
-     */
-    fun asFlow(): Flow<T>
-
-    /**
-     * Executes operation asynchronously in the I/O thread pool.
-     */
-    suspend fun executeOnIO(): T
-}
-
-fun DatabaseHelper.getFlatMetadataForManga(mangaId: Long): PreparedSuspendOperation<FlatMetadata?> =
-    preparedOperationFromSuspend {
-        val meta = getSearchMetadataForManga(mangaId).executeOnIO()
+fun DatabaseHelper.getFlatMetadataForManga(mangaId: Long): PreparedOperation<FlatMetadata?> {
+    // We have to use fromCallable because StorIO messes up the thread scheduling if we use their rx functions
+    val single = Single.fromCallable {
+        val meta = getSearchMetadataForManga(mangaId).executeAsBlocking()
         if (meta != null) {
-            val tags = getSearchTagsForManga(mangaId).executeOnIO()
-            val titles = getSearchTitlesForManga(mangaId).executeOnIO()
+            val tags = getSearchTagsForManga(mangaId).executeAsBlocking()
+            val titles = getSearchTitlesForManga(mangaId).executeAsBlocking()
 
             FlatMetadata(meta, tags, titles)
         } else null
     }
 
-private fun <T> preparedOperationFromSuspend(operation: suspend () -> T): PreparedSuspendOperation<T> {
-    return object : PreparedSuspendOperation<T> {
+    return preparedOperationFromSingle(single)
+}
+
+private fun <T> preparedOperationFromSingle(single: Single<T>): PreparedOperation<T> {
+    return object : PreparedOperation<T> {
         /**
          * Creates [rx.Observable] that emits result of Operation.
          *
@@ -74,7 +57,7 @@ private fun <T> preparedOperationFromSuspend(operation: suspend () -> T): Prepar
          *
          * @return observable result of operation with only one [rx.Observer.onNext] call.
          */
-        override fun createObservable() = runAsObservable(operation)
+        override fun createObservable() = single.toObservable()
 
         /**
          * Executes operation synchronously in current thread.
@@ -83,11 +66,11 @@ private fun <T> preparedOperationFromSuspend(operation: suspend () -> T): Prepar
          * Notice: Blocking I/O operation should not be executed on the Main Thread,
          * it can cause ANR (Activity Not Responding dialog), block the UI and drop animations frames.
          * So please, execute blocking I/O operation only from background thread.
-         * See [androidx.annotation.WorkerThread].
+         * See [WorkerThread].
          *
          * @return nullable result of operation.
          */
-        override fun executeAsBlocking() = runBlocking { operation() }
+        override fun executeAsBlocking() = single.toBlocking().value()
 
         /**
          * Creates [rx.Observable] that emits result of Operation.
@@ -97,7 +80,7 @@ private fun <T> preparedOperationFromSuspend(operation: suspend () -> T): Prepar
          *
          * @return observable result of operation with only one [rx.Observer.onNext] call.
          */
-        override fun asRxObservable() = runAsObservable(operation)
+        override fun asRxObservable() = single.toObservable()
 
         /**
          * Creates [rx.Single] that emits result of Operation lazily when somebody subscribes to it.
@@ -106,21 +89,7 @@ private fun <T> preparedOperationFromSuspend(operation: suspend () -> T): Prepar
          *
          * @return single result of operation.
          */
-        override fun asRxSingle(): Single<T> = runAsObservable(operation).toSingle()
-
-        /**
-         * Creates a [Flow] that emits the result of of Operation
-         *
-         * Example:
-         *  override fun asFlow(): Flow<T> = flow { emit(operation()) }
-         *
-         */
-        override fun asFlow(): Flow<T> = flow { emit(operation()) }
-
-        /**
-         * Executes operation asynchronously in the I/O thread pool.
-         */
-        override suspend fun executeOnIO(): T = withIOContext { operation() }
+        override fun asRxSingle() = single
     }
 }