如何在 WorkManager 中处理 Rx？

Question

我实现了AlarmManager，以便在用户向任务添加截止日期时发送通知。但是，当用户关闭设备时，所有警报都会丢失。现在我正在更新BroadcastReceiver 以接收android.intent.action.BOOT_COMPLETED 并重新安排为每个任务设置的所有警报。

我的第一次尝试是在BroadcastReceiver 中获取所有到期日期高于当前时间的任务的Rx Single，然后重新安排所有警报。问题是一旦BroadcastReceiver 没有生命周期，我就无法处置Observable。另外，it seems that this is not a good approach。

在我的研究中，IntentService 是这个案例的一个很好的解决方案，但我正在进入新的 WorkManager 库，OneTimeWorkRequest 看起来是一个很好且简单的解决方案。

Worker 正在被调用并正确执行，但我无法释放 Observable，因为从未调用过 onStopped 方法。

这是基于this snippet的实现：

class TaskAlarmWorker(context: Context, params: WorkerParameters) :
    Worker(context, params), KoinComponent {

    private val daoRepository: DaoRepository by inject()

    private val compositeDisposable = CompositeDisposable()

    override fun doWork(): Result {
        Timber.d("doWork")

        val result = LinkedBlockingQueue<Result>()

        val disposable =
            daoRepository.getTaskDao().getAllTasks().applySchedulers().subscribe(
            { result.put(Result.SUCCESS) },
            { result.put(Result.FAILURE) }
        )

        compositeDisposable.add(disposable)

        return try {
            result.take()
        } catch (e: InterruptedException) {
            Result.RETRY
        }
    }

    override fun onStopped(cancelled: Boolean) {
        Timber.d("onStopped")
        compositeDisposable.clear()
    }
}

• WorkManager 是否适合这种情况？
• 是否可以正确处理Observable？

Answer 1

• 是的WorkManager 是一个很好的解决方案（甚至可能是最好的解决方案）

• 您应该使用RxWorker 而不是Worker。这是一个例子：

  1. 实现它。将androidx.work:work-rxjava2:$work_version 添加到您的build.gradle 文件作为依赖项。

  2. 从RxWorker 类扩展您的类，然后覆盖createWork() 函数。

class TaskAlarmWorker(context: Context, params: WorkerParameters) :
    RxWorker(context, params), KoinComponent {

    private val daoRepository: DaoRepository by inject()  

    override fun createWork(): Single<Result> {
        Timber.d("doRxWork")

     return daoRepository.getTaskDao().getAllTasks()
                .doOnSuccess { /* process result somehow */ }
                .map { Result.success() }
                .onErrorReturn { Result.failure() }          

    }

}

关于 RxWorker 的重要说明：

• createWork() 方法在主线程上调用，但返回的 single 在后台线程上订阅。
• 您无需担心销毁 Observer，因为 RxWorker 会在工作停止时自动销毁它。
• 返回值为 Result.failure() 的 Single 和返回错误的 Single 都会导致 worker 进入失败状态。
• 您可以覆盖onStopped 函数以执行更多操作。

阅读更多：

• How to use WorkManager with RxJava
• Stackoverflow answer

Answer 2

你可以用onStoped()方法清除它然后compositeDisposable.dispose();

然后拨打super.onStoped()