Remove Rx from EventHub and TimelineCases (#3446)

* remove Rx from EventHub and TimelineCases

* fix tests

* fix AccountViewModel.unblockDomain

* remove debug logging

app/src/main/java/com/keylesspalace/tusky/MainActivity.kt

@@ -43,12 +43,10 @@ import androidx.core.content.ContextCompat
 import androidx.core.content.pm.ShortcutManagerCompat
 import androidx.core.view.GravityCompat
 import androidx.core.view.MenuProvider
-import androidx.lifecycle.Lifecycle
 import androidx.lifecycle.lifecycleScope
 import androidx.preference.PreferenceManager
 import androidx.viewpager2.widget.MarginPageTransformer
 import at.connyduck.calladapter.networkresult.fold
-import autodispose2.androidx.lifecycle.autoDispose
 import com.bumptech.glide.Glide
 import com.bumptech.glide.RequestManager
 import com.bumptech.glide.load.resource.bitmap.RoundedCorners
@@ -61,7 +59,6 @@ import com.google.android.material.tabs.TabLayout.OnTabSelectedListener
 import com.google.android.material.tabs.TabLayoutMediator
 import com.keylesspalace.tusky.appstore.AnnouncementReadEvent
 import com.keylesspalace.tusky.appstore.CacheUpdater
-import com.keylesspalace.tusky.appstore.Event
 import com.keylesspalace.tusky.appstore.EventHub
 import com.keylesspalace.tusky.appstore.MainTabsChangedEvent
 import com.keylesspalace.tusky.appstore.ProfileEditedEvent
@@ -133,7 +130,6 @@ import com.mikepenz.materialdrawer.widget.AccountHeaderView
 import dagger.android.DispatchingAndroidInjector
 import dagger.android.HasAndroidInjector
 import de.c1710.filemojicompat_ui.helpers.EMOJI_PREFERENCE
-import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import io.reactivex.rxjava3.schedulers.Schedulers
 import kotlinx.coroutines.launch
 import javax.inject.Inject
@@ -287,10 +283,8 @@ class MainActivity : BottomSheetActivity(), ActionButtonActivity, HasAndroidInje
 
         setupTabs(showNotificationTab)
 
-        eventHub.events
-            .observeOn(AndroidSchedulers.mainThread())
-            .autoDispose(this, Lifecycle.Event.ON_DESTROY)
-            .subscribe { event: Event? ->
+        lifecycleScope.launch {
+            eventHub.events.collect { event ->
                 when (event) {
                     is ProfileEditedEvent -> onFetchUserInfoSuccess(event.newProfileData)
                     is MainTabsChangedEvent -> {
@@ -308,6 +302,7 @@ class MainActivity : BottomSheetActivity(), ActionButtonActivity, HasAndroidInje
                     }
                 }
             }
+        }
 
         Schedulers.io().scheduleDirect {
             // Flush old media that was cached for sharing

app/src/main/java/com/keylesspalace/tusky/TabPreferenceActivity.kt

@@ -348,7 +348,9 @@ class TabPreferenceActivity : BaseActivity(), Injectable, ItemInteractionListene
     override fun onPause() {
         super.onPause()
         if (tabsChanged) {
-            eventHub.dispatch(MainTabsChangedEvent(currentTabs))
+            lifecycleScope.launch {
+                eventHub.dispatch(MainTabsChangedEvent(currentTabs))
+            }
         }
     }
 

app/src/main/java/com/keylesspalace/tusky/appstore/CacheUpdater.kt

@@ -8,7 +8,6 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.rx3.asFlow
 import javax.inject.Inject
 
 class CacheUpdater @Inject constructor(
@@ -24,7 +23,7 @@ class CacheUpdater @Inject constructor(
         val timelineDao = appDatabase.timelineDao()
 
         scope.launch {
-            eventHub.events.asFlow().collect { event ->
+            eventHub.events.collect { event ->
                 val accountId = accountManager.activeAccount?.id ?: return@collect
                 when (event) {
                     is FavoriteEvent ->

app/src/main/java/com/keylesspalace/tusky/appstore/Events.kt

@@ -5,21 +5,21 @@ import com.keylesspalace.tusky.entity.Account
 import com.keylesspalace.tusky.entity.Poll
 import com.keylesspalace.tusky.entity.Status
 
-data class FavoriteEvent(val statusId: String, val favourite: Boolean) : Dispatchable
-data class ReblogEvent(val statusId: String, val reblog: Boolean) : Dispatchable
-data class BookmarkEvent(val statusId: String, val bookmark: Boolean) : Dispatchable
-data class MuteConversationEvent(val statusId: String, val mute: Boolean) : Dispatchable
-data class UnfollowEvent(val accountId: String) : Dispatchable
-data class BlockEvent(val accountId: String) : Dispatchable
-data class MuteEvent(val accountId: String) : Dispatchable
-data class StatusDeletedEvent(val statusId: String) : Dispatchable
-data class StatusComposedEvent(val status: Status) : Dispatchable
-data class StatusScheduledEvent(val status: Status) : Dispatchable
-data class StatusEditedEvent(val originalId: String, val status: Status) : Dispatchable
-data class ProfileEditedEvent(val newProfileData: Account) : Dispatchable
-data class PreferenceChangedEvent(val preferenceKey: String) : Dispatchable
-data class MainTabsChangedEvent(val newTabs: List<TabData>) : Dispatchable
-data class PollVoteEvent(val statusId: String, val poll: Poll) : Dispatchable
-data class DomainMuteEvent(val instance: String) : Dispatchable
-data class AnnouncementReadEvent(val announcementId: String) : Dispatchable
-data class PinEvent(val statusId: String, val pinned: Boolean) : Dispatchable
+data class FavoriteEvent(val statusId: String, val favourite: Boolean) : Event
+data class ReblogEvent(val statusId: String, val reblog: Boolean) : Event
+data class BookmarkEvent(val statusId: String, val bookmark: Boolean) : Event
+data class MuteConversationEvent(val statusId: String, val mute: Boolean) : Event
+data class UnfollowEvent(val accountId: String) : Event
+data class BlockEvent(val accountId: String) : Event
+data class MuteEvent(val accountId: String) : Event
+data class StatusDeletedEvent(val statusId: String) : Event
+data class StatusComposedEvent(val status: Status) : Event
+data class StatusScheduledEvent(val status: Status) : Event
+data class StatusEditedEvent(val originalId: String, val status: Status) : Event
+data class ProfileEditedEvent(val newProfileData: Account) : Event
+data class PreferenceChangedEvent(val preferenceKey: String) : Event
+data class MainTabsChangedEvent(val newTabs: List<TabData>) : Event
+data class PollVoteEvent(val statusId: String, val poll: Poll) : Event
+data class DomainMuteEvent(val instance: String) : Event
+data class AnnouncementReadEvent(val announcementId: String) : Event
+data class PinEvent(val statusId: String, val pinned: Boolean) : Event

app/src/main/java/com/keylesspalace/tusky/appstore/EventsHub.kt

@@ -1,20 +1,19 @@
 package com.keylesspalace.tusky.appstore
 
-import io.reactivex.rxjava3.core.Observable
-import io.reactivex.rxjava3.subjects.PublishSubject
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
 import javax.inject.Inject
 import javax.inject.Singleton
 
 interface Event
-interface Dispatchable : Event
 
 @Singleton
 class EventHub @Inject constructor() {
 
-    private val eventsSubject = PublishSubject.create<Event>()
-    val events: Observable<Event> = eventsSubject
+    private val sharedEventFlow: MutableSharedFlow<Event> = MutableSharedFlow()
+    val events: Flow<Event> = sharedEventFlow
 
-    fun dispatch(event: Dispatchable) {
-        eventsSubject.onNext(event)
+    suspend fun dispatch(event: Event) {
+        sharedEventFlow.emit(event)
     }
 }

app/src/main/java/com/keylesspalace/tusky/components/account/AccountViewModel.kt

@@ -3,6 +3,7 @@ package com.keylesspalace.tusky.components.account
 import android.util.Log
 import androidx.lifecycle.MutableLiveData
 import androidx.lifecycle.viewModelScope
+import at.connyduck.calladapter.networkresult.fold
 import com.keylesspalace.tusky.appstore.BlockEvent
 import com.keylesspalace.tusky.appstore.DomainMuteEvent
 import com.keylesspalace.tusky.appstore.EventHub
@@ -21,9 +22,6 @@ import com.keylesspalace.tusky.util.Success
 import io.reactivex.rxjava3.core.Single
 import io.reactivex.rxjava3.disposables.Disposable
 import kotlinx.coroutines.launch
-import retrofit2.Call
-import retrofit2.Callback
-import retrofit2.Response
 import java.util.concurrent.TimeUnit
 import javax.inject.Inject
 
@@ -47,12 +45,13 @@ class AccountViewModel @Inject constructor(
     private var noteDisposable: Disposable? = null
 
     init {
-        eventHub.events
-            .subscribe { event ->
+        viewModelScope.launch {
+            eventHub.events.collect { event ->
                 if (event is ProfileEditedEvent && event.newProfileData.id == accountData.value?.data?.id) {
                     accountData.postValue(Success(event.newProfileData))
                 }
-            }.autoDispose()
+            }
+        }
     }
 
     private fun obtainAccount(reload: Boolean = false) {
@@ -133,42 +132,30 @@ class AccountViewModel @Inject constructor(
     }
 
     fun blockDomain(instance: String) {
-        mastodonApi.blockDomain(instance).enqueue(object : Callback<Any> {
-            override fun onResponse(call: Call<Any>, response: Response<Any>) {
-                if (response.isSuccessful) {
-                    eventHub.dispatch(DomainMuteEvent(instance))
-                    val relation = relationshipData.value?.data
-                    if (relation != null) {
-                        relationshipData.postValue(Success(relation.copy(blockingDomain = true)))
-                    }
-                } else {
-                    Log.e(TAG, "Error muting %s".format(instance))
+        viewModelScope.launch {
+            mastodonApi.blockDomain(instance).fold({
+                eventHub.dispatch(DomainMuteEvent(instance))
+                val relation = relationshipData.value?.data
+                if (relation != null) {
+                    relationshipData.postValue(Success(relation.copy(blockingDomain = true)))
                 }
-            }
-
-            override fun onFailure(call: Call<Any>, t: Throwable) {
-                Log.e(TAG, "Error muting %s".format(instance), t)
-            }
-        })
+            }, { e ->
+                Log.e(TAG, "Error muting $instance", e)
+            })
+        }
     }
 
     fun unblockDomain(instance: String) {
-        mastodonApi.unblockDomain(instance).enqueue(object : Callback<Any> {
-            override fun onResponse(call: Call<Any>, response: Response<Any>) {
-                if (response.isSuccessful) {
-                    val relation = relationshipData.value?.data
-                    if (relation != null) {
-                        relationshipData.postValue(Success(relation.copy(blockingDomain = false)))
-                    }
-                } else {
-                    Log.e(TAG, "Error unmuting %s".format(instance))
+        viewModelScope.launch {
+            mastodonApi.unblockDomain(instance).fold({
+                val relation = relationshipData.value?.data
+                if (relation != null) {
+                    relationshipData.postValue(Success(relation.copy(blockingDomain = false)))
                 }
-            }
-
-            override fun onFailure(call: Call<Any>, t: Throwable) {
-                Log.e(TAG, "Error unmuting %s".format(instance), t)
-            }
-        })
+            }, { e ->
+                Log.e(TAG, "Error unmuting $instance", e)
+            })
+        }
     }
 
     fun changeShowReblogsState() {

app/src/main/java/com/keylesspalace/tusky/components/conversation/ConversationsFragment.kt

@@ -36,7 +36,6 @@ import androidx.recyclerview.widget.LinearLayoutManager
 import androidx.recyclerview.widget.RecyclerView
 import androidx.recyclerview.widget.SimpleItemAnimator
 import at.connyduck.sparkbutton.helpers.Utils
-import autodispose2.androidx.lifecycle.autoDispose
 import com.google.android.material.color.MaterialColors
 import com.keylesspalace.tusky.R
 import com.keylesspalace.tusky.StatusListActivity
@@ -62,7 +61,6 @@ import com.mikepenz.iconics.IconicsDrawable
 import com.mikepenz.iconics.typeface.library.googlematerial.GoogleMaterial
 import com.mikepenz.iconics.utils.colorInt
 import com.mikepenz.iconics.utils.sizeDp
-import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.launch
@@ -205,14 +203,13 @@ class ConversationsFragment :
             }
         }
 
-        eventHub.events
-            .observeOn(AndroidSchedulers.mainThread())
-            .autoDispose(this, Lifecycle.Event.ON_DESTROY)
-            .subscribe { event ->
+        lifecycleScope.launch {
+            eventHub.events.collect { event ->
                 if (event is PreferenceChangedEvent) {
                     onPreferenceChanged(event.preferenceKey)
                 }
             }
+        }
     }
 
     override fun onCreateMenu(menu: Menu, menuInflater: MenuInflater) {

app/src/main/java/com/keylesspalace/tusky/components/conversation/ConversationsViewModel.kt

@@ -23,6 +23,7 @@ import androidx.paging.Pager
 import androidx.paging.PagingConfig
 import androidx.paging.cachedIn
 import androidx.paging.map
+import at.connyduck.calladapter.networkresult.fold
 import com.keylesspalace.tusky.db.AccountManager
 import com.keylesspalace.tusky.db.AppDatabase
 import com.keylesspalace.tusky.network.MastodonApi
@@ -30,7 +31,6 @@ import com.keylesspalace.tusky.usecase.TimelineCases
 import com.keylesspalace.tusky.util.EmptyPagingSource
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.rx3.await
 import javax.inject.Inject
 
 class ConversationsViewModel @Inject constructor(
@@ -61,51 +61,47 @@ class ConversationsViewModel @Inject constructor(
 
     fun favourite(favourite: Boolean, conversation: ConversationViewData) {
         viewModelScope.launch {
-            try {
-                timelineCases.favourite(conversation.lastStatus.id, favourite).await()
-
+            timelineCases.favourite(conversation.lastStatus.id, favourite).fold({
                 val newConversation = conversation.toEntity(
                     accountId = accountManager.activeAccount!!.id,
                     favourited = favourite
                 )
 
                 saveConversationToDb(newConversation)
-            } catch (e: Exception) {
+            }, { e ->
                 Log.w(TAG, "failed to favourite status", e)
-            }
+            })
         }
     }
 
     fun bookmark(bookmark: Boolean, conversation: ConversationViewData) {
         viewModelScope.launch {
-            try {
-                timelineCases.bookmark(conversation.lastStatus.id, bookmark).await()
-
+            timelineCases.bookmark(conversation.lastStatus.id, bookmark).fold({
                 val newConversation = conversation.toEntity(
                     accountId = accountManager.activeAccount!!.id,
                     bookmarked = bookmark
                 )
 
                 saveConversationToDb(newConversation)
-            } catch (e: Exception) {
+            }, { e ->
                 Log.w(TAG, "failed to bookmark status", e)
-            }
+            })
         }
     }
 
     fun voteInPoll(choices: List<Int>, conversation: ConversationViewData) {
         viewModelScope.launch {
-            try {
-                val poll = timelineCases.voteInPoll(conversation.lastStatus.id, conversation.lastStatus.status.poll?.id!!, choices).await()
-                val newConversation = conversation.toEntity(
-                    accountId = accountManager.activeAccount!!.id,
-                    poll = poll
-                )
+            timelineCases.voteInPoll(conversation.lastStatus.id, conversation.lastStatus.status.poll?.id!!, choices)
+                .fold({ poll ->
+                    val newConversation = conversation.toEntity(
+                        accountId = accountManager.activeAccount!!.id,
+                        poll = poll
+                    )
 
-                saveConversationToDb(newConversation)
-            } catch (e: Exception) {
-                Log.w(TAG, "failed to vote in poll", e)
-            }
+                    saveConversationToDb(newConversation)
+                }, { e ->
+                    Log.w(TAG, "failed to vote in poll", e)
+                })
         }
     }
 
@@ -160,7 +156,7 @@ class ConversationsViewModel @Inject constructor(
                 timelineCases.muteConversation(
                     conversation.lastStatus.id,
                     !(conversation.lastStatus.status.muted ?: false)
-                ).await()
+                )
 
                 val newConversation = conversation.toEntity(
                     accountId = accountManager.activeAccount!!.id,

app/src/main/java/com/keylesspalace/tusky/components/instancemute/fragment/InstanceListFragment.kt

@@ -5,9 +5,11 @@ import android.util.Log
 import android.view.View
 import androidx.fragment.app.Fragment
 import androidx.lifecycle.Lifecycle
+import androidx.lifecycle.lifecycleScope
 import androidx.recyclerview.widget.DividerItemDecoration
 import androidx.recyclerview.widget.LinearLayoutManager
 import androidx.recyclerview.widget.RecyclerView
+import at.connyduck.calladapter.networkresult.fold
 import autodispose2.androidx.lifecycle.AndroidLifecycleScopeProvider.from
 import autodispose2.autoDispose
 import com.google.android.material.snackbar.Snackbar
@@ -23,9 +25,7 @@ import com.keylesspalace.tusky.util.show
 import com.keylesspalace.tusky.util.viewBinding
 import com.keylesspalace.tusky.view.EndlessOnScrollListener
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
-import retrofit2.Call
-import retrofit2.Callback
-import retrofit2.Response
+import kotlinx.coroutines.launch
 import java.io.IOException
 import javax.inject.Inject
 
@@ -64,39 +64,25 @@ class InstanceListFragment : Fragment(R.layout.fragment_instance_list), Injectab
     }
 
     override fun mute(mute: Boolean, instance: String, position: Int) {
-        if (mute) {
-            api.blockDomain(instance).enqueue(object : Callback<Any> {
-                override fun onFailure(call: Call<Any>, t: Throwable) {
-                    Log.e(TAG, "Error muting domain $instance")
-                }
-
-                override fun onResponse(call: Call<Any>, response: Response<Any>) {
-                    if (response.isSuccessful) {
-                        adapter.addItem(instance)
-                    } else {
-                        Log.e(TAG, "Error muting domain $instance")
-                    }
-                }
-            })
-        } else {
-            api.unblockDomain(instance).enqueue(object : Callback<Any> {
-                override fun onFailure(call: Call<Any>, t: Throwable) {
-                    Log.e(TAG, "Error unmuting domain $instance")
-                }
-
-                override fun onResponse(call: Call<Any>, response: Response<Any>) {
-                    if (response.isSuccessful) {
-                        adapter.removeItem(position)
-                        Snackbar.make(binding.recyclerView, getString(R.string.confirmation_domain_unmuted, instance), Snackbar.LENGTH_LONG)
-                            .setAction(R.string.action_undo) {
-                                mute(true, instance, position)
-                            }
-                            .show()
-                    } else {
-                        Log.e(TAG, "Error unmuting domain $instance")
-                    }
-                }
-            })
+        viewLifecycleOwner.lifecycleScope.launch {
+            if (mute) {
+                api.blockDomain(instance).fold({
+                    adapter.addItem(instance)
+                }, { e ->
+                    Log.e(TAG, "Error muting domain $instance", e)
+                })
+            } else {
+                api.unblockDomain(instance).fold({
+                    adapter.removeItem(position)
+                    Snackbar.make(binding.recyclerView, getString(R.string.confirmation_domain_unmuted, instance), Snackbar.LENGTH_LONG)
+                        .setAction(R.string.action_undo) {
+                            mute(true, instance, position)
+                        }
+                        .show()
+                }, { e ->
+                    Log.e(TAG, "Error unmuting domain $instance", e)
+                })
+            }
         }
     }
 

app/src/main/java/com/keylesspalace/tusky/components/notifications/NotificationsViewModel.kt

@@ -62,7 +62,6 @@ import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.rx3.asFlow
 import kotlinx.coroutines.rx3.await
 import retrofit2.HttpException
 import javax.inject.Inject
@@ -357,7 +356,7 @@ class NotificationsViewModel @Inject constructor(
         )
 
         viewModelScope.launch {
-            eventHub.events.asFlow()
+            eventHub.events
                 .filterIsInstance<PreferenceChangedEvent>()
                 .filter { StatusDisplayOptions.prefKeys.contains(it.preferenceKey) }
                 .map {
@@ -420,23 +419,23 @@ class NotificationsViewModel @Inject constructor(
                                 timelineCases.bookmark(
                                     action.statusViewData.actionableId,
                                     action.state
-                                ).await()
+                                )
                             is StatusAction.Favourite ->
                                 timelineCases.favourite(
                                     action.statusViewData.actionableId,
                                     action.state
-                                ).await()
+                                )
                             is StatusAction.Reblog ->
                                 timelineCases.reblog(
                                     action.statusViewData.actionableId,
                                     action.state
-                                ).await()
+                                )
                             is StatusAction.VoteInPoll ->
                                 timelineCases.voteInPoll(
                                     action.statusViewData.actionableId,
                                     action.poll.id,
                                     action.choices
-                                ).await()
+                                )
                         }
                         uiSuccess.emit(StatusActionSuccess.from(action))
                     } catch (e: Exception) {
@@ -447,7 +446,7 @@ class NotificationsViewModel @Inject constructor(
 
         // Handle events that should refresh the list
         viewModelScope.launch {
-            eventHub.events.asFlow().collectLatest {
+            eventHub.events.collectLatest {
                 when (it) {
                     is BlockEvent -> uiSuccess.emit(UiSuccess.Block)
                     is MuteEvent -> uiSuccess.emit(UiSuccess.Mute)
@@ -504,7 +503,7 @@ class NotificationsViewModel @Inject constructor(
      * @return Flow of relevant preferences that change the UI
      */
     // TODO: Preferences should be in a repository
-    private fun getUiPrefs() = eventHub.events.asFlow()
+    private fun getUiPrefs() = eventHub.events
         .filterIsInstance<PreferenceChangedEvent>()
         .filter { UiPrefs.prefKeys.contains(it.preferenceKey) }
         .map { toPrefs() }

app/src/main/java/com/keylesspalace/tusky/components/preference/AccountPreferencesFragment.kt

@@ -21,6 +21,7 @@ import android.os.Build
 import android.os.Bundle
 import android.util.Log
 import androidx.annotation.DrawableRes
+import androidx.lifecycle.lifecycleScope
 import androidx.preference.PreferenceFragmentCompat
 import com.google.android.material.color.MaterialColors
 import com.google.android.material.snackbar.Snackbar
@@ -57,6 +58,7 @@ import com.mikepenz.iconics.IconicsDrawable
 import com.mikepenz.iconics.typeface.library.googlematerial.GoogleMaterial
 import com.mikepenz.iconics.utils.colorInt
 import com.mikepenz.iconics.utils.sizeRes
+import kotlinx.coroutines.launch
 import retrofit2.Call
 import retrofit2.Callback
 import retrofit2.Response
@@ -198,7 +200,6 @@ class AccountPreferencesFragment : PreferenceFragmentCompat(), Injectable {
                     setOnPreferenceChangeListener { _, newValue ->
                         setIcon(getIconForVisibility(Status.Visibility.byString(newValue as String)))
                         syncWithServer(visibility = newValue)
-                        eventHub.dispatch(PreferenceChangedEvent(key))
                         true
                     }
                 }
@@ -221,7 +222,6 @@ class AccountPreferencesFragment : PreferenceFragmentCompat(), Injectable {
 
                     setOnPreferenceChangeListener { _, newValue ->
                         syncWithServer(language = (newValue as String))
-                        eventHub.dispatch(PreferenceChangedEvent(key))
                         true
                     }
                 }
@@ -237,7 +237,6 @@ class AccountPreferencesFragment : PreferenceFragmentCompat(), Injectable {
                     setOnPreferenceChangeListener { _, newValue ->
                         setIcon(getIconForSensitivity(newValue as Boolean))
                         syncWithServer(sensitive = newValue)
-                        eventHub.dispatch(PreferenceChangedEvent(key))
                         true
                     }
                 }
@@ -246,7 +245,7 @@ class AccountPreferencesFragment : PreferenceFragmentCompat(), Injectable {
             preferenceCategory(R.string.pref_title_timelines) {
                 // TODO having no activeAccount in this fragment does not really make sense, enforce it?
                 //   All other locations here make it optional, however.
-                val accountPreferenceHandler = AccountPreferenceHandler(accountManager.activeAccount!!, accountManager, eventHub)
+                val accountPreferenceHandler = AccountPreferenceHandler(accountManager.activeAccount!!, accountManager, ::dispatchEvent)
 
                 switchPreference {
                     key = PrefKeys.MEDIA_PREVIEW_ENABLED
@@ -354,6 +353,12 @@ class AccountPreferencesFragment : PreferenceFragmentCompat(), Injectable {
         activity?.overridePendingTransition(R.anim.slide_from_right, R.anim.slide_to_left)
     }
 
+    private fun dispatchEvent(event: PreferenceChangedEvent) {
+        lifecycleScope.launch {
+            eventHub.dispatch(event)
+        }
+    }
+
     companion object {
         fun newInstance() = AccountPreferencesFragment()
     }

app/src/main/java/com/keylesspalace/tusky/components/preference/PreferencesActivity.kt

@@ -23,6 +23,7 @@ import android.util.Log
 import androidx.activity.OnBackPressedCallback
 import androidx.fragment.app.Fragment
 import androidx.fragment.app.commit
+import androidx.lifecycle.lifecycleScope
 import androidx.preference.Preference
 import androidx.preference.PreferenceFragmentCompat
 import androidx.preference.PreferenceManager
@@ -38,6 +39,7 @@ import com.keylesspalace.tusky.util.getNonNullString
 import com.keylesspalace.tusky.util.setAppNightMode
 import dagger.android.DispatchingAndroidInjector
 import dagger.android.HasAndroidInjector
+import kotlinx.coroutines.launch
 import javax.inject.Inject
 
 class PreferencesActivity :
@@ -155,8 +157,9 @@ class PreferencesActivity :
                 restartActivitiesOnBackPressedCallback.isEnabled = true
             }
         }
-
-        eventHub.dispatch(PreferenceChangedEvent(key))
+        lifecycleScope.launch {
+            eventHub.dispatch(PreferenceChangedEvent(key))
+        }
     }
 
     private fun restartCurrentActivity() {

app/src/main/java/com/keylesspalace/tusky/components/scheduled/ScheduledStatusActivity.kt

@@ -28,7 +28,6 @@ import androidx.lifecycle.lifecycleScope
 import androidx.paging.LoadState
 import androidx.recyclerview.widget.DividerItemDecoration
 import androidx.recyclerview.widget.LinearLayoutManager
-import autodispose2.androidx.lifecycle.autoDispose
 import com.google.android.material.color.MaterialColors
 import com.keylesspalace.tusky.BaseActivity
 import com.keylesspalace.tusky.R
@@ -46,7 +45,6 @@ import com.mikepenz.iconics.IconicsDrawable
 import com.mikepenz.iconics.typeface.library.googlematerial.GoogleMaterial
 import com.mikepenz.iconics.utils.colorInt
 import com.mikepenz.iconics.utils.sizeDp
-import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.launch
 import javax.inject.Inject
@@ -119,14 +117,13 @@ class ScheduledStatusActivity :
             }
         }
 
-        eventHub.events
-            .observeOn(AndroidSchedulers.mainThread())
-            .autoDispose(this)
-            .subscribe { event ->
+        lifecycleScope.launch {
+            eventHub.events.collect { event ->
                 if (event is StatusScheduledEvent) {
                     adapter.refresh()
                 }
             }
+        }
     }
 
     override fun onCreateMenu(menu: Menu, menuInflater: MenuInflater) {

app/src/main/java/com/keylesspalace/tusky/components/search/SearchViewModel.kt

@@ -16,11 +16,14 @@
 package com.keylesspalace.tusky.components.search
 
 import android.util.Log
+import androidx.lifecycle.ViewModel
 import androidx.lifecycle.viewModelScope
 import androidx.paging.Pager
 import androidx.paging.PagingConfig
 import androidx.paging.cachedIn
 import at.connyduck.calladapter.networkresult.NetworkResult
+import at.connyduck.calladapter.networkresult.fold
+import at.connyduck.calladapter.networkresult.onFailure
 import com.keylesspalace.tusky.components.search.adapter.SearchPagingSourceFactory
 import com.keylesspalace.tusky.db.AccountEntity
 import com.keylesspalace.tusky.db.AccountManager
@@ -28,10 +31,8 @@ import com.keylesspalace.tusky.entity.DeletedStatus
 import com.keylesspalace.tusky.entity.Status
 import com.keylesspalace.tusky.network.MastodonApi
 import com.keylesspalace.tusky.usecase.TimelineCases
-import com.keylesspalace.tusky.util.RxAwareViewModel
 import com.keylesspalace.tusky.util.toViewData
 import com.keylesspalace.tusky.viewdata.StatusViewData
-import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.async
 import kotlinx.coroutines.launch
@@ -41,7 +42,7 @@ class SearchViewModel @Inject constructor(
     mastodonApi: MastodonApi,
     private val timelineCases: TimelineCases,
     private val accountManager: AccountManager
-) : RxAwareViewModel() {
+) : ViewModel() {
 
     var currentQuery: String = ""
 
@@ -115,22 +116,18 @@ class SearchViewModel @Inject constructor(
     }
 
     fun reblog(statusViewData: StatusViewData.Concrete, reblog: Boolean) {
-        timelineCases.reblog(statusViewData.id, reblog)
-            .observeOn(AndroidSchedulers.mainThread())
-            .subscribe(
-                { setRebloggedForStatus(statusViewData, reblog) },
-                { t -> Log.d(TAG, "Failed to reblog status ${statusViewData.id}", t) }
-            )
-            .autoDispose()
-    }
-
-    private fun setRebloggedForStatus(statusViewData: StatusViewData.Concrete, reblog: Boolean) {
-        updateStatus(
-            statusViewData.status.copy(
-                reblogged = reblog,
-                reblog = statusViewData.status.reblog?.copy(reblogged = reblog)
-            )
-        )
+        viewModelScope.launch {
+            timelineCases.reblog(statusViewData.id, reblog).fold({
+                updateStatus(
+                    statusViewData.status.copy(
+                        reblogged = reblog,
+                        reblog = statusViewData.status.reblog?.copy(reblogged = reblog)
+                    )
+                )
+            }, { t ->
+                Log.d(TAG, "Failed to reblog status ${statusViewData.id}", t)
+            })
+        }
     }
 
     fun contentHiddenChange(statusViewData: StatusViewData.Concrete, isShowing: Boolean) {
@@ -144,27 +141,24 @@ class SearchViewModel @Inject constructor(
     fun voteInPoll(statusViewData: StatusViewData.Concrete, choices: MutableList<Int>) {
         val votedPoll = statusViewData.status.actionableStatus.poll!!.votedCopy(choices)
         updateStatus(statusViewData.status.copy(poll = votedPoll))
-        timelineCases.voteInPoll(statusViewData.id, votedPoll.id, choices)
-            .observeOn(AndroidSchedulers.mainThread())
-            .doOnError { t -> Log.d(TAG, "Failed to vote in poll: ${statusViewData.id}", t) }
-            .subscribe()
-            .autoDispose()
+        viewModelScope.launch {
+            timelineCases.voteInPoll(statusViewData.id, votedPoll.id, choices)
+                .onFailure { t -> Log.d(TAG, "Failed to vote in poll: ${statusViewData.id}", t) }
+        }
     }
 
     fun favorite(statusViewData: StatusViewData.Concrete, isFavorited: Boolean) {
         updateStatus(statusViewData.status.copy(favourited = isFavorited))
-        timelineCases.favourite(statusViewData.id, isFavorited)
-            .onErrorReturnItem(statusViewData.status)
-            .subscribe()
-            .autoDispose()
+        viewModelScope.launch {
+            timelineCases.favourite(statusViewData.id, isFavorited)
+        }
     }
 
     fun bookmark(statusViewData: StatusViewData.Concrete, isBookmarked: Boolean) {
         updateStatus(statusViewData.status.copy(bookmarked = isBookmarked))
-        timelineCases.bookmark(statusViewData.id, isBookmarked)
-            .onErrorReturnItem(statusViewData.status)
-            .subscribe()
-            .autoDispose()
+        viewModelScope.launch {
+            timelineCases.bookmark(statusViewData.id, isBookmarked)
+        }
     }
 
     fun muteAccount(accountId: String, notifications: Boolean, duration: Int?) {
@@ -174,7 +168,9 @@ class SearchViewModel @Inject constructor(
     }
 
     fun pinAccount(status: Status, isPin: Boolean) {
-        timelineCases.pin(status.id, isPin)
+        viewModelScope.launch {
+            timelineCases.pin(status.id, isPin)
+        }
     }
 
     fun blockAccount(accountId: String) {
@@ -191,10 +187,9 @@ class SearchViewModel @Inject constructor(
 
     fun muteConversation(statusViewData: StatusViewData.Concrete, mute: Boolean) {
         updateStatus(statusViewData.status.copy(muted = mute))
-        timelineCases.muteConversation(statusViewData.id, mute)
-            .onErrorReturnItem(statusViewData.status)
-            .subscribe()
-            .autoDispose()
+        viewModelScope.launch {
+            timelineCases.muteConversation(statusViewData.id, mute)
+        }
     }
 
     private fun updateStatusViewData(newStatusViewData: StatusViewData.Concrete) {

app/src/main/java/com/keylesspalace/tusky/components/timeline/TimelineFragment.kt

@@ -77,6 +77,7 @@ import com.mikepenz.iconics.utils.colorInt
 import com.mikepenz.iconics.utils.sizeDp
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import io.reactivex.rxjava3.core.Observable
+import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.launch
 import java.io.IOException
@@ -299,10 +300,8 @@ class TimelineFragment :
             })
         }
 
-        eventHub.events
-            .observeOn(AndroidSchedulers.mainThread())
-            .autoDispose(this, Lifecycle.Event.ON_DESTROY)
-            .subscribe { event ->
+        viewLifecycleOwner.lifecycleScope.launch {
+            eventHub.events.collect { event ->
                 when (event) {
                     is PreferenceChangedEvent -> {
                         onPreferenceChanged(event.preferenceKey)
@@ -316,6 +315,7 @@ class TimelineFragment :
                     }
                 }
             }
+        }
     }
 
     override fun onCreateMenu(menu: Menu, menuInflater: MenuInflater) {

app/src/main/java/com/keylesspalace/tusky/components/timeline/viewmodel/TimelineViewModel.kt

@@ -22,6 +22,7 @@ import androidx.lifecycle.viewModelScope
 import androidx.paging.PagingData
 import at.connyduck.calladapter.networkresult.fold
 import at.connyduck.calladapter.networkresult.getOrElse
+import at.connyduck.calladapter.networkresult.getOrThrow
 import com.keylesspalace.tusky.appstore.BlockEvent
 import com.keylesspalace.tusky.appstore.BookmarkEvent
 import com.keylesspalace.tusky.appstore.DomainMuteEvent
@@ -49,8 +50,6 @@ import com.keylesspalace.tusky.viewdata.StatusViewData
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.rx3.asFlow
-import kotlinx.coroutines.rx3.await
 import retrofit2.HttpException
 
 abstract class TimelineViewModel(
@@ -101,7 +100,6 @@ abstract class TimelineViewModel(
 
         viewModelScope.launch {
             eventHub.events
-                .asFlow()
                 .collect { event -> handleEvent(event) }
         }
 
@@ -110,7 +108,7 @@ abstract class TimelineViewModel(
 
     fun reblog(reblog: Boolean, status: StatusViewData.Concrete): Job = viewModelScope.launch {
         try {
-            timelineCases.reblog(status.actionableId, reblog).await()
+            timelineCases.reblog(status.actionableId, reblog).getOrThrow()
         } catch (t: Exception) {
             ifExpected(t) {
                 Log.d(TAG, "Failed to reblog status " + status.actionableId, t)
@@ -120,7 +118,7 @@ abstract class TimelineViewModel(
 
     fun favorite(favorite: Boolean, status: StatusViewData.Concrete): Job = viewModelScope.launch {
         try {
-            timelineCases.favourite(status.actionableId, favorite).await()
+            timelineCases.favourite(status.actionableId, favorite).getOrThrow()
         } catch (t: Exception) {
             ifExpected(t) {
                 Log.d(TAG, "Failed to favourite status " + status.actionableId, t)
@@ -130,7 +128,7 @@ abstract class TimelineViewModel(
 
     fun bookmark(bookmark: Boolean, status: StatusViewData.Concrete): Job = viewModelScope.launch {
         try {
-            timelineCases.bookmark(status.actionableId, bookmark).await()
+            timelineCases.bookmark(status.actionableId, bookmark).getOrThrow()
         } catch (t: Exception) {
             ifExpected(t) {
                 Log.d(TAG, "Failed to bookmark status " + status.actionableId, t)
@@ -148,7 +146,7 @@ abstract class TimelineViewModel(
         updatePoll(votedPoll, status)
 
         try {
-            timelineCases.voteInPoll(status.actionableId, poll.id, choices).await()
+            timelineCases.voteInPoll(status.actionableId, poll.id, choices).getOrThrow()
         } catch (t: Exception) {
             ifExpected(t) {
                 Log.d(TAG, "Failed to vote in poll: " + status.actionableId, t)

app/src/main/java/com/keylesspalace/tusky/components/trending/viewmodel/TrendingViewModel.kt

@@ -28,7 +28,6 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.filterIsInstance
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.rx3.asFlow
 import okio.IOException
 import javax.inject.Inject
 
@@ -55,7 +54,7 @@ class TrendingViewModel @Inject constructor(
         // or deleted. Unfortunately, there's nothing in the event to determine if it's a filter
         // that was modified, so refresh on every preference change.
         viewModelScope.launch {
-            eventHub.events.asFlow()
+            eventHub.events
                 .filterIsInstance<PreferenceChangedEvent>()
                 .collect {
                     invalidate()

app/src/main/java/com/keylesspalace/tusky/components/viewthread/ViewThreadViewModel.kt

@@ -20,6 +20,7 @@ import androidx.lifecycle.ViewModel
 import androidx.lifecycle.viewModelScope
 import at.connyduck.calladapter.networkresult.fold
 import at.connyduck.calladapter.networkresult.getOrElse
+import at.connyduck.calladapter.networkresult.getOrThrow
 import com.google.gson.Gson
 import com.keylesspalace.tusky.appstore.BlockEvent
 import com.keylesspalace.tusky.appstore.BookmarkEvent
@@ -50,8 +51,6 @@ import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.rx3.asFlow
-import kotlinx.coroutines.rx3.await
 import retrofit2.HttpException
 import javax.inject.Inject
 
@@ -85,7 +84,6 @@ class ViewThreadViewModel @Inject constructor(
 
         viewModelScope.launch {
             eventHub.events
-                .asFlow()
                 .collect { event ->
                     when (event) {
                         is FavoriteEvent -> handleFavEvent(event)
@@ -195,7 +193,7 @@ class ViewThreadViewModel @Inject constructor(
 
     fun reblog(reblog: Boolean, status: StatusViewData.Concrete): Job = viewModelScope.launch {
         try {
-            timelineCases.reblog(status.actionableId, reblog).await()
+            timelineCases.reblog(status.actionableId, reblog).getOrThrow()
         } catch (t: Exception) {
             ifExpected(t) {
                 Log.d(TAG, "Failed to reblog status " + status.actionableId, t)
@@ -205,7 +203,7 @@ class ViewThreadViewModel @Inject constructor(
 
     fun favorite(favorite: Boolean, status: StatusViewData.Concrete): Job = viewModelScope.launch {
         try {
-            timelineCases.favourite(status.actionableId, favorite).await()
+            timelineCases.favourite(status.actionableId, favorite).getOrThrow()
         } catch (t: Exception) {
             ifExpected(t) {
                 Log.d(TAG, "Failed to favourite status " + status.actionableId, t)
@@ -215,7 +213,7 @@ class ViewThreadViewModel @Inject constructor(
 
     fun bookmark(bookmark: Boolean, status: StatusViewData.Concrete): Job = viewModelScope.launch {
         try {
-            timelineCases.bookmark(status.actionableId, bookmark).await()
+            timelineCases.bookmark(status.actionableId, bookmark).getOrThrow()
         } catch (t: Exception) {
             ifExpected(t) {
                 Log.d(TAG, "Failed to bookmark status " + status.actionableId, t)
@@ -235,7 +233,7 @@ class ViewThreadViewModel @Inject constructor(
         }
 
         try {
-            timelineCases.voteInPoll(status.actionableId, poll.id, choices).await()
+            timelineCases.voteInPoll(status.actionableId, poll.id, choices).getOrThrow()
         } catch (t: Exception) {
             ifExpected(t) {
                 Log.d(TAG, "Failed to vote in poll: " + status.actionableId, t)

app/src/main/java/com/keylesspalace/tusky/fragment/SFragment.kt

@@ -33,11 +33,9 @@ import androidx.appcompat.app.AlertDialog
 import androidx.appcompat.widget.PopupMenu
 import androidx.core.app.ActivityOptionsCompat
 import androidx.fragment.app.Fragment
-import androidx.lifecycle.Lifecycle
 import androidx.lifecycle.lifecycleScope
 import at.connyduck.calladapter.networkresult.fold
-import autodispose2.AutoDispose
-import autodispose2.androidx.lifecycle.AndroidLifecycleScopeProvider
+import at.connyduck.calladapter.networkresult.onFailure
 import com.google.android.material.snackbar.Snackbar
 import com.keylesspalace.tusky.BaseActivity
 import com.keylesspalace.tusky.BottomSheetActivity
@@ -61,7 +59,6 @@ import com.keylesspalace.tusky.util.openLink
 import com.keylesspalace.tusky.util.parseAsMastodonHtml
 import com.keylesspalace.tusky.view.showMuteAccountDialog
 import com.keylesspalace.tusky.viewdata.AttachmentViewData
-import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import kotlinx.coroutines.launch
 import javax.inject.Inject
 
@@ -276,30 +273,19 @@ abstract class SFragment : Fragment(), Injectable {
                     return@setOnMenuItemClickListener true
                 }
                 R.id.pin -> {
-                    timelineCases.pin(status.id, !status.isPinned())
-                        .observeOn(AndroidSchedulers.mainThread())
-                        .doOnError { e: Throwable ->
-                            val message = e.message ?: getString(if (status.isPinned()) R.string.failed_to_unpin else R.string.failed_to_pin)
+                    lifecycleScope.launch {
+                        timelineCases.pin(status.id, !status.isPinned()).onFailure { e: Throwable ->
+                            val message = e.message
+                                ?: getString(if (status.isPinned()) R.string.failed_to_unpin else R.string.failed_to_pin)
                             Snackbar.make(requireView(), message, Snackbar.LENGTH_LONG).show()
                         }
-                        .to(
-                            AutoDispose.autoDisposable(
-                                AndroidLifecycleScopeProvider.from(this, Lifecycle.Event.ON_DESTROY)
-                            )
-                        )
-                        .subscribe()
+                    }
                     return@setOnMenuItemClickListener true
                 }
                 R.id.status_mute_conversation -> {
-                    timelineCases.muteConversation(status.id, status.muted != true)
-                        .onErrorReturnItem(status)
-                        .observeOn(AndroidSchedulers.mainThread())
-                        .to(
-                            AutoDispose.autoDisposable(
-                                AndroidLifecycleScopeProvider.from(this, Lifecycle.Event.ON_DESTROY)
-                            )
-                        )
-                        .subscribe()
+                    lifecycleScope.launch {
+                        timelineCases.muteConversation(status.id, status.muted != true)
+                    }
                     return@setOnMenuItemClickListener true
                 }
             }

app/src/main/java/com/keylesspalace/tusky/network/MastodonApi.kt

@@ -235,54 +235,54 @@ interface MastodonApi {
     ): NetworkResult<DeletedStatus>
 
     @POST("api/v1/statuses/{id}/reblog")
-    fun reblogStatus(
+    suspend fun reblogStatus(
         @Path("id") statusId: String
-    ): Single<Status>
+    ): NetworkResult<Status>
 
     @POST("api/v1/statuses/{id}/unreblog")
-    fun unreblogStatus(
+    suspend fun unreblogStatus(
         @Path("id") statusId: String
-    ): Single<Status>
+    ): NetworkResult<Status>
 
     @POST("api/v1/statuses/{id}/favourite")
-    fun favouriteStatus(
+    suspend fun favouriteStatus(
         @Path("id") statusId: String
-    ): Single<Status>
+    ): NetworkResult<Status>
 
     @POST("api/v1/statuses/{id}/unfavourite")
-    fun unfavouriteStatus(
+    suspend fun unfavouriteStatus(
         @Path("id") statusId: String
-    ): Single<Status>
+    ): NetworkResult<Status>
 
     @POST("api/v1/statuses/{id}/bookmark")
-    fun bookmarkStatus(
+    suspend fun bookmarkStatus(
         @Path("id") statusId: String
-    ): Single<Status>
+    ): NetworkResult<Status>
 
     @POST("api/v1/statuses/{id}/unbookmark")
-    fun unbookmarkStatus(
+    suspend fun unbookmarkStatus(
         @Path("id") statusId: String
-    ): Single<Status>
+    ): NetworkResult<Status>
 
     @POST("api/v1/statuses/{id}/pin")
-    fun pinStatus(
+    suspend fun pinStatus(
         @Path("id") statusId: String
-    ): Single<Status>
+    ): NetworkResult<Status>
 
     @POST("api/v1/statuses/{id}/unpin")
-    fun unpinStatus(
+    suspend fun unpinStatus(
         @Path("id") statusId: String
-    ): Single<Status>
+    ): NetworkResult<Status>
 
     @POST("api/v1/statuses/{id}/mute")
-    fun muteConversation(
+    suspend fun muteConversation(
         @Path("id") statusId: String
-    ): Single<Status>
+    ): NetworkResult<Status>
 
     @POST("api/v1/statuses/{id}/unmute")
-    fun unmuteConversation(
+    suspend fun unmuteConversation(
         @Path("id") statusId: String
-    ): Single<Status>
+    ): NetworkResult<Status>
 
     @GET("api/v1/scheduled_statuses")
     fun scheduledStatuses(
@@ -450,14 +450,14 @@ interface MastodonApi {
 
     @FormUrlEncoded
     @POST("api/v1/domain_blocks")
-    fun blockDomain(
+    suspend fun blockDomain(
         @Field("domain") domain: String
-    ): Call<Any>
+    ): NetworkResult<Unit>
 
     @FormUrlEncoded
     // @DELETE doesn't support fields
     @HTTP(method = "DELETE", path = "api/v1/domain_blocks", hasBody = true)
-    fun unblockDomain(@Field("domain") domain: String): Call<Any>
+    suspend fun unblockDomain(@Field("domain") domain: String): NetworkResult<Unit>
 
     @GET("api/v1/favourites")
     suspend fun favourites(
@@ -648,10 +648,10 @@ interface MastodonApi {
 
     @FormUrlEncoded
     @POST("api/v1/polls/{id}/votes")
-    fun voteInPoll(
+    suspend fun voteInPoll(
         @Path("id") id: String,
         @Field("choices[]") choices: List<Int>
-    ): Single<Poll>
+    ): NetworkResult<Poll>
 
     @GET("api/v1/announcements")
     suspend fun listAnnouncements(

app/src/main/java/com/keylesspalace/tusky/settings/AccountPreferenceHandler.kt

@@ -1,7 +1,6 @@
 package com.keylesspalace.tusky.settings
 
 import androidx.preference.PreferenceDataStore
-import com.keylesspalace.tusky.appstore.EventHub
 import com.keylesspalace.tusky.appstore.PreferenceChangedEvent
 import com.keylesspalace.tusky.db.AccountEntity
 import com.keylesspalace.tusky.db.AccountManager
@@ -9,7 +8,7 @@ import com.keylesspalace.tusky.db.AccountManager
 class AccountPreferenceHandler(
     private val account: AccountEntity,
     private val accountManager: AccountManager,
-    private val eventHub: EventHub
+    private val dispatchEvent: (PreferenceChangedEvent) -> Unit
 ) : PreferenceDataStore() {
 
     override fun getBoolean(key: String, defValue: Boolean): Boolean {
@@ -30,6 +29,6 @@ class AccountPreferenceHandler(
 
         accountManager.saveAccount(account)
 
-        eventHub.dispatch(PreferenceChangedEvent(key))
+        dispatchEvent(PreferenceChangedEvent(key))
     }
 }

app/src/main/java/com/keylesspalace/tusky/usecase/TimelineCases.kt

@@ -17,6 +17,7 @@ package com.keylesspalace.tusky.usecase
 
 import android.util.Log
 import at.connyduck.calladapter.networkresult.NetworkResult
+import at.connyduck.calladapter.networkresult.fold
 import at.connyduck.calladapter.networkresult.onFailure
 import at.connyduck.calladapter.networkresult.onSuccess
 import com.keylesspalace.tusky.appstore.BlockEvent
@@ -36,7 +37,6 @@ import com.keylesspalace.tusky.entity.Status
 import com.keylesspalace.tusky.network.MastodonApi
 import com.keylesspalace.tusky.util.getServerErrorMessage
 import io.reactivex.rxjava3.core.Single
-import io.reactivex.rxjava3.disposables.CompositeDisposable
 import javax.inject.Inject
 
 /**
@@ -48,52 +48,42 @@ class TimelineCases @Inject constructor(
     private val eventHub: EventHub
 ) {
 
-    /**
-     * Unused yet but can be use for cancellation later. It's always a good idea to save
-     * Disposables.
-     */
-    private val cancelDisposable = CompositeDisposable()
-
-    fun reblog(statusId: String, reblog: Boolean): Single<Status> {
-        val call = if (reblog) {
+    suspend fun reblog(statusId: String, reblog: Boolean): NetworkResult<Status> {
+        return if (reblog) {
             mastodonApi.reblogStatus(statusId)
         } else {
             mastodonApi.unreblogStatus(statusId)
-        }
-        return call.doAfterSuccess {
+        }.onSuccess {
             eventHub.dispatch(ReblogEvent(statusId, reblog))
         }
     }
 
-    fun favourite(statusId: String, favourite: Boolean): Single<Status> {
-        val call = if (favourite) {
+    suspend fun favourite(statusId: String, favourite: Boolean): NetworkResult<Status> {
+        return if (favourite) {
             mastodonApi.favouriteStatus(statusId)
         } else {
             mastodonApi.unfavouriteStatus(statusId)
-        }
-        return call.doAfterSuccess {
+        }.onSuccess {
             eventHub.dispatch(FavoriteEvent(statusId, favourite))
         }
     }
 
-    fun bookmark(statusId: String, bookmark: Boolean): Single<Status> {
-        val call = if (bookmark) {
+    suspend fun bookmark(statusId: String, bookmark: Boolean): NetworkResult<Status> {
+        return if (bookmark) {
             mastodonApi.bookmarkStatus(statusId)
         } else {
             mastodonApi.unbookmarkStatus(statusId)
-        }
-        return call.doAfterSuccess {
+        }.onSuccess {
             eventHub.dispatch(BookmarkEvent(statusId, bookmark))
         }
     }
 
-    fun muteConversation(statusId: String, mute: Boolean): Single<Status> {
-        val call = if (mute) {
+    suspend fun muteConversation(statusId: String, mute: Boolean): NetworkResult<Status> {
+        return if (mute) {
             mastodonApi.muteConversation(statusId)
         } else {
             mastodonApi.unmuteConversation(statusId)
-        }
-        return call.doAfterSuccess {
+        }.onSuccess {
             eventHub.dispatch(MuteConversationEvent(statusId, mute))
         }
     }
@@ -122,25 +112,27 @@ class TimelineCases @Inject constructor(
             .onFailure { Log.w(TAG, "Failed to delete status", it) }
     }
 
-    fun pin(statusId: String, pin: Boolean): Single<Status> {
-        // Replace with extension method if we use RxKotlin
-        return (if (pin) mastodonApi.pinStatus(statusId) else mastodonApi.unpinStatus(statusId))
-            .doOnError { e ->
-                Log.w(TAG, "Failed to change pin state", e)
-            }
-            .onErrorResumeNext(::convertError)
-            .doAfterSuccess {
-                eventHub.dispatch(PinEvent(statusId, pin))
-            }
+    suspend fun pin(statusId: String, pin: Boolean): NetworkResult<Status> {
+        return if (pin) {
+            mastodonApi.pinStatus(statusId)
+        } else {
+            mastodonApi.unpinStatus(statusId)
+        }.fold({ status ->
+            eventHub.dispatch(PinEvent(statusId, pin))
+            NetworkResult.success(status)
+        }, { e ->
+            Log.w(TAG, "Failed to change pin state", e)
+            NetworkResult.failure(TimelineError(e.getServerErrorMessage()))
+        })
     }
 
-    fun voteInPoll(statusId: String, pollId: String, choices: List<Int>): Single<Poll> {
+    suspend fun voteInPoll(statusId: String, pollId: String, choices: List<Int>): NetworkResult<Poll> {
         if (choices.isEmpty()) {
-            return Single.error(IllegalStateException())
+            return NetworkResult.failure(IllegalStateException())
         }
 
-        return mastodonApi.voteInPoll(pollId, choices).doAfterSuccess {
-            eventHub.dispatch(PollVoteEvent(statusId, it))
+        return mastodonApi.voteInPoll(pollId, choices).onSuccess { poll ->
+            eventHub.dispatch(PollVoteEvent(statusId, poll))
         }
     }
 
@@ -152,10 +144,6 @@ class TimelineCases @Inject constructor(
         return mastodonApi.rejectFollowRequest(accountId)
     }
 
-    private fun <T : Any> convertError(e: Throwable): Single<T> {
-        return Single.error(TimelineError(e.getServerErrorMessage()))
-    }
-
     companion object {
         private const val TAG = "TimelineCases"
     }

app/src/test/java/com/keylesspalace/tusky/components/notifications/NotificationsViewModelTestStatusAction.kt

@@ -18,10 +18,10 @@
 package com.keylesspalace.tusky.components.notifications
 
 import app.cash.turbine.test
+import at.connyduck.calladapter.networkresult.NetworkResult
 import com.google.common.truth.Truth.assertThat
 import com.keylesspalace.tusky.FilterV1Test.Companion.mockStatus
 import com.keylesspalace.tusky.viewdata.StatusViewData
-import io.reactivex.rxjava3.core.Single
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runTest
 import org.junit.Test
@@ -73,7 +73,7 @@ class NotificationsViewModelTestStatusAction : NotificationsViewModelTestBase()
     @Test
     fun `bookmark succeeds && emits UiSuccess`() = runTest {
         // Given
-        timelineCases.stub { onBlocking { bookmark(any(), any()) } doReturn Single.just(status) }
+        timelineCases.stub { onBlocking { bookmark(any(), any()) } doReturn NetworkResult.success(status) }
 
         viewModel.uiSuccess.test {
             // When
@@ -111,7 +111,7 @@ class NotificationsViewModelTestStatusAction : NotificationsViewModelTestBase()
     fun `favourite succeeds && emits UiSuccess`() = runTest {
         // Given
         timelineCases.stub {
-            onBlocking { favourite(any(), any()) } doReturn Single.just(status)
+            onBlocking { favourite(any(), any()) } doReturn NetworkResult.success(status)
         }
 
         viewModel.uiSuccess.test {
@@ -149,7 +149,7 @@ class NotificationsViewModelTestStatusAction : NotificationsViewModelTestBase()
     @Test
     fun `reblog succeeds && emits UiSuccess`() = runTest {
         // Given
-        timelineCases.stub { onBlocking { reblog(any(), any()) } doReturn Single.just(status) }
+        timelineCases.stub { onBlocking { reblog(any(), any()) } doReturn NetworkResult.success(status) }
 
         viewModel.uiSuccess.test {
             // When
@@ -187,7 +187,7 @@ class NotificationsViewModelTestStatusAction : NotificationsViewModelTestBase()
     fun `voteinpoll succeeds && emits UiSuccess`() = runTest {
         // Given
         timelineCases.stub {
-            onBlocking { voteInPoll(any(), any(), any()) } doReturn Single.just(status.poll!!)
+            onBlocking { voteInPoll(any(), any(), any()) } doReturn NetworkResult.success(status.poll!!)
         }
 
         viewModel.uiSuccess.test {

app/src/test/java/com/keylesspalace/tusky/components/viewthread/ViewThreadViewModelTest.kt

@@ -221,9 +221,9 @@ class ViewThreadViewModelTest {
 
         viewModel.loadThread(threadId)
 
-        eventHub.dispatch(FavoriteEvent(statusId = "1", false))
-
         runBlocking {
+            eventHub.dispatch(FavoriteEvent(statusId = "1", false))
+
             assertEquals(
                 ThreadUiState.Success(
                     statusViewData = listOf(
@@ -245,9 +245,9 @@ class ViewThreadViewModelTest {
 
         viewModel.loadThread(threadId)
 
-        eventHub.dispatch(ReblogEvent(statusId = "2", true))
-
         runBlocking {
+            eventHub.dispatch(ReblogEvent(statusId = "2", true))
+
             assertEquals(
                 ThreadUiState.Success(
                     statusViewData = listOf(
@@ -269,9 +269,9 @@ class ViewThreadViewModelTest {
 
         viewModel.loadThread(threadId)
 
-        eventHub.dispatch(BookmarkEvent(statusId = "3", false))
-
         runBlocking {
+            eventHub.dispatch(BookmarkEvent(statusId = "3", false))
+
             assertEquals(
                 ThreadUiState.Success(
                     statusViewData = listOf(

app/src/test/java/com/keylesspalace/tusky/usecase/TimelineCasesTest.kt

@@ -1,12 +1,15 @@
 package com.keylesspalace.tusky.usecase
 
 import androidx.test.ext.junit.runners.AndroidJUnit4
+import app.cash.turbine.test
+import at.connyduck.calladapter.networkresult.NetworkResult
 import com.keylesspalace.tusky.appstore.EventHub
 import com.keylesspalace.tusky.appstore.PinEvent
 import com.keylesspalace.tusky.entity.Status
 import com.keylesspalace.tusky.network.MastodonApi
-import io.reactivex.rxjava3.core.Single
+import kotlinx.coroutines.runBlocking
 import okhttp3.ResponseBody.Companion.toResponseBody
+import org.junit.Assert.assertEquals
 import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
@@ -16,7 +19,7 @@ import org.mockito.kotlin.stub
 import org.robolectric.annotation.Config
 import retrofit2.HttpException
 import retrofit2.Response
-import java.util.Date
+import java.util.*
 
 @Config(sdk = [28])
 @RunWith(AndroidJUnit4::class)
@@ -38,21 +41,21 @@ class TimelineCasesTest {
     @Test
     fun `pin success emits PinEvent`() {
         api.stub {
-            onBlocking { pinStatus(statusId) } doReturn Single.just(mockStatus(pinned = true))
+            onBlocking { pinStatus(statusId) } doReturn NetworkResult.success(mockStatus(pinned = true))
         }
 
-        val events = eventHub.events.test()
-        timelineCases.pin(statusId, true)
-            .test()
-            .assertComplete()
-
-        events.assertValue(PinEvent(statusId, true))
+        runBlocking {
+            eventHub.events.test {
+                timelineCases.pin(statusId, true)
+                assertEquals(PinEvent(statusId, true), awaitItem())
+            }
+        }
     }
 
     @Test
     fun `pin failure with server error throws TimelineError with server message`() {
         api.stub {
-            onBlocking { pinStatus(statusId) } doReturn Single.error(
+            onBlocking { pinStatus(statusId) } doReturn NetworkResult.failure(
                 HttpException(
                     Response.error<Status>(
                         422,
@@ -61,9 +64,12 @@ class TimelineCasesTest {
                 )
             )
         }
-        timelineCases.pin(statusId, true)
-            .test()
-            .assertError { it.message == "Validation Failed: You have already pinned the maximum number of toots" }
+        runBlocking {
+            assertEquals(
+                "Validation Failed: You have already pinned the maximum number of toots",
+                timelineCases.pin(statusId, true).exceptionOrNull()?.message
+            )
+        }
     }
 
     private fun mockStatus(pinned: Boolean = false): Status {