RxJava可观察到可完成，如何避免toBlocking（）

Question

我目前在Android和Kotlin上使用RxJava，但我有一个问题，如果不使用toBlocking（），我无法解决。

我在员工服务中有一个方法，它返回一个可观察的

fun all(): Observable<List<Employee>>

这一切都很好，因为每当员工发生变化时，这个可观察对象就会发出新的员工列表。但是我想从员工那里生成一个PDF文件，这显然不需要每次员工更改时都运行。另外，我想从PDF生成器方法返回一个可完成的对象。我想在PDF中添加一个标题，然后遍历员工并计算每个员工的工资，这也会返回一个可见值，这就是我现在使用Toblock的地方。我目前的做法是：

private fun generatePdf(outputStream: OutputStream): Completable {
    return employeeService.all().map { employees ->
        try {
                addHeaderToPDF()
                for (i in employees) {
                    val calculated = employeeService.calculateWage(i.id).toBlocking().first()
                    // Print calculated to PDF....
                }
                addFooterToPDF()
                return @map Completable.complete()
            }
            catch (e: Exception) {
                return @map Completable.error(e)
            }
        }.first().toCompletable()

有没有什么方法可以使用RxJava让这段代码更干净一点？

提前感谢！

Answer 1

我想到了这个：

    return employeeService.all().first()
            .doOnSubscribe { addHeaderToPDF() }
            .flatMapIterable { it }
            .flatMap { employeeService.calculateWage(it.id).first() }
            .doOnNext { printEmployeeWage(it) }
            .doOnCompleted { addFooterToPDF }
            .toCompletable()

这是应该这样做吗？：）

Answer 2

免责声明：此答案是正在进行的工作。

基本前提：如果流中有阻塞，那么就是做错了。

注意：任何状态都不能离开可观察的lambda。

输入是一个员工流。对于每个员工，你需要获得一份工资。让我们把它变成一条小溪。

/**
 * @param employeesObservable
 * Stream of employees we're interested in.
 * @param wageProvider
 * Transformation function which takes an employee and returns a [Single] of their wage.
 * @return
 * Observable stream spitting individual [Pair]s of employees and their wages.
 */
fun getEmployeesAndWagesObservable(
        employeesObservable: Observable<Employee>,
        wageProvider: Function<Employee, Single<Int>>
): Observable<Pair<Employee, Int>>? {
    val employeesAndWagesObservable: Observable<Pair<Employee, Int>>

    // Each Employee from the original stream will be converted
    // to a Single<Pair<Employee, Int>> via flatMapSingle operator.
    // Remember, we need a stream and Single is a stream.
    employeesAndWagesObservable = employeesObservable.flatMapSingle { employee ->
        // We need to get a source of wage value for current employee.
        // That source emits a single Int or errors.
        val wageForEmployeeSingle: Single<Int> = wageProvider.apply(employee)

        // Once the wage from said source is loaded...
        val employeeAndWageSingle: Single<Pair<Employee, Int> = wageForEmployeeSingle.map { wage ->
            // ... construct a Pair<Employee, Int>
            employee to wage
        }

        // This code is not executed now. It will be executed for each Employee
        // after the original Observable<Employee> starts spitting out items.
        // After subscribing to the resulting observable.
        return@flatMapSingle employeeAndWageSingle
    }

    return employeesAndWagesObservable
}

订阅时会发生什么：

1. 从源头带走一名员工。
2. 获取雇员的工资。
3. 吐出一对员工和他们的工资。

这一过程会重复，直到员工发出可维护的信号，或某个错误出现故障。

使用的运算符：

• FLAPMapSing：将实际值转换为具有某些转换值的新单一流。
• map：将实际值转换为其他实际值（没有嵌套流）。

Hee是如何将其连接到代码的：

fun doStuff() {
    val employeesObservable = employeeService.all()
    val wageProvider = Function<Employee, Single<Int>> { employee ->
        // Don't listen to changes. Take first wage and use that.
        employeeService.calculateWage(employee.id).firstOrError()
    }

    val employeesAndWagesObservable = 
            getEmployeesAndWagesObservable(employeesObservable, wageProvider)

    // Subscribe...
}

使用的运算符：

• 第一：从observable中获取第一项，并将其转换为单个流
• 超时：最好是。超时（timeout）如果您通过网络获取工资，则为工资

不要订阅，打电话

val blockingIterable = employeesAndWagesObservable.blockingIterable()
blockingIterable.forEach { ... }

并以同步方式处理每个项目。坐下来，找出下一步，观看演示文稿，阅读示例。

• <代码>。映射这些对中的每一对