RxJava：将可观察到的拆分为两个列表

Question

使用：我必须更新存储在本地数据库中的当前订单。步骤如下：

1. 从后端下载数据(~800项）
2. 检查本地数据库是否已包含项。后端人员使用字符串作为主键。
3. 如果该项不在数据库中，请添加它。
4. 如果该项在数据库中，请更新它。

Observable<List<OrderDto>> getAllOrdersObservable = backendService.getAllOrders();

ordersDisposable = Observable.combineLatest(getStoredOrder(), getAllOrdersObservable, this::insertOrUpdateOrders)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .doOnComplete(() -> {
            Log.d(TAG, "Duration: " + (System.currentTimeMillis() - start) + " ms");
            showSyncButton();
            notifyListenersDataDownloaded();
        })
        .subscribe();


private Observable<List<Order>> getStoredOrder() {
    return Observable.defer(() -> {
        Log.d(TAG, "Started loading orders from database");
        Observable<List<Order>> just = Observable.just(orderDao.loadAll());
        Log.d(TAG, "Ended loading orders from database");
        return just;
    });
}

private List<Order> insertOrUpdateOrders(List<Order> orders, List<OrderDto> orderDtos) {
    List<Order> ordersToInsert = new LinkedList<>();
    List<Order> ordersToUpdate = new LinkedList<>();

    for (OrderDto orderDto : orderDtos) {
        Order order = getOrderByOrderNumber(orders, orderDto.getNumber());
        if (order != null) {
            dtoToEntityTransformer.updateFields(orderDto, order);
            ordersToUpdate.add(order);
        } else {
            order = dtoToEntityTransformer.transformToEntity(orderDto);
            ordersToInsert.add(order);
        }
    }

    orderDao.insertInTx(ordersToInsert);
    orderDao.updateInTx(ordersToUpdate);

    return orders;
}

问题

你有一个如何用反应的方式解决这个问题的想法吗？有没有一个运算符，允许把可观察到的东西分成两个列表。或者我应该使用一个全局变量（似乎不是一个好主意）来保存信息，哪些数据要插入，哪些要更新？

Answer 1

使用groupby，关键是它是否是更新/插入操作；您将得到一个包含密钥的GroupedObservable。使用该键执行适当的操作。