免责声明:本答案仍在完善中。
基本前提:如果在流中存在“阻塞”,那么您的操作是错误的。
注意:没有状态可以离开可观察的lambda表达式。
步骤1:将整个数据集作为流
输入是员工的流。对于每个员工,您需要获取一个工资。让我们将其组成一个流。
/**
* @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
}
订阅后会发生什么:
- 从源中获取一个员工。
- 获取员工的工资。
- 输出一对员工和他们的工资。
这将重复进行,直到 employeesObservable
发出 onComplete
或出现错误时发出 onError
信号。
使用的操作符:
以下是如何将其连接到您的代码:
fun doStuff() {
val employeesObservable = employeeService.all()
val wageProvider = Function<Employee, Single<Int>> { employee ->
employeeService.calculateWage(employee.id).firstOrError()
}
val employeesAndWagesObservable =
getEmployeesAndWagesObservable(employeesObservable, wageProvider)
}
使用的运算符:
- first:从可观察对象中获取第一个项目并将其转换为单个流。
- timeout:如果您通过网络获取工资,则最好
.timeout
它。
下一步
选项1:结束
不要订阅,调用
val blockingIterable = employeesAndWagesObservable.blockingIterable()
blockingIterable.forEach { ... }
同步处理每个项目。坐下来,想想接下来的步骤,看演示文稿,读例子。
选项2:添加图层
.map
每个Pair<Employee, Int>
到一些抽象的PDF建筑块。
- 通过
Observable.fromCallable { ... }
将头部和底部打印机转换为Observables,它们也返回PDF构建块。
- 通过
Observable.concat(headerObs, employeeDataObs, footerObs)
以顺序方式合并所有这些内容。
.subscribe
此结果并开始将PDF构建块写入PDF编写器。
- 待办事项:
- 找出一种在订阅时懒惰地初始化PDF编写器的方法(不是在构建流之前),
- 在出现错误时删除输出,
- 在完成或出现错误时关闭输出流。
addHeaderToPDF
和addFooterToPDF
。calculateWage
应该返回一个Single
,它要么返回一个值,要么出现错误。 - Eugen PechaneccalculateWage
是一个Observable
,因为它会在员工数据发生变化时发出重新计算后的工资值。你能否给我展示一个正确的示例?我无法理清这个难题。 :( - szantogab