Observe rx observables in IO

horizontalsystems · Nov 16, 2023 · a9a273f · a9a273f
1 parent 2bd239f
commit a9a273f
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 0 deletions.
diff --git a/...rc/main/kotlin/io/horizontalsystems/bitcoincore/apisync/blockchair/BlockchairApiSyncer.kt b/...rc/main/kotlin/io/horizontalsystems/bitcoincore/apisync/blockchair/BlockchairApiSyncer.kt
@@ -38,6 +38,7 @@ class BlockchairApiSyncer(
     override fun sync() {
         scanSingle()
             .subscribeOn(Schedulers.io())
+            .observeOn(Schedulers.io())
             .subscribe({}, {
                 handleError(it)
             }).let {
@@ -48,6 +49,7 @@ class BlockchairApiSyncer(
     override fun syncLastBlock() {
         syncLastBlockSingle()
             .subscribeOn(Schedulers.io())
+            .observeOn(Schedulers.io())
             .subscribe({}, {
                 handleError(it)
             }).let {

diff --git a/bitcoincore/src/main/kotlin/io/horizontalsystems/bitcoincore/apisync/legacy/ApiSyncer.kt b/bitcoincore/src/main/kotlin/io/horizontalsystems/bitcoincore/apisync/legacy/ApiSyncer.kt
@@ -34,6 +34,7 @@ class ApiSyncer(
     override fun sync() {
         val disposable = blockHashDiscovery.discoverBlockHashes()
             .subscribeOn(Schedulers.io())
+            .observeOn(Schedulers.io())
             .subscribe(
                 { (publicKeys, blockHashes) ->
                     val sortedUniqueBlockHashes = blockHashes.distinctBy { it.height }.sortedBy { it.height }