反应式编程-使用Rx更新金额

Question

我有两个Subject（plusOne[itemId]，minusOne[itemId]），它们发出的项目ID的数量应该更改一（加或减）。

用户可以用相同的项目id发送多个信号。最终，我想订阅一个会发出Pair[itemId,amountToSet] 的Observable

UI看起来像是一个项目行列表，每个行都包含"+"one_answers"-"按钮，它们触发相应的onNext（itemId）。

我曾尝试过通过对itemid进行分组并减少序列来解决问题，但问题是reduce只有在调用onCompleted时才会生效，而我的受试者不应该完成他们的排放。

以下是无法使用reduce:的代码

plusItem.asObservable().groupBy(id -> id).subscribe(new Action1<GroupedObservable<Long, Long>>() {
    @Override
    public void call(GroupedObservable<Long, Long> groupedObservable) {
        System.out.println("Composed a group with key: " + groupedObservable.getKey());
        groupedObservable
                .map(id -> 1)
                .startWith(0)
                .reduce((integer, integer2) -> integer + integer2)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer sum) {
                        System.out.println(Integer.toString(sum));
                    }
                });
    }
});

解决方案

private Observable<Pair<Long,Integer>> getUpdateObservable(
    Observable clickSource,
        Observable<Map<Long,Integer>> initValues,
        Func1<Long,Integer> groupMapFunc
) {
    return clickSource
            .asObservable()
            .groupBy(id -> id)
            .flatMap(new Func1<GroupedObservable<Long, Long>, Observable<Pair<Long, Integer>>>() {
                @Override
                public Observable<Pair<Long, Integer>> call(GroupedObservable<Long, Long> groupedObservable) {
                    return groupedObservable
                            .map(groupMapFunc)
                            .scan((integer, integer2) -> integer + integer2)
                            .withLatestFrom(initValues, new Func2<Integer, Map<Long, Integer>, Integer>() {
                                @Override
                                public Integer call(Integer integer, Map<Long, Integer> basket) {
                                    return basket.get(groupedObservable.getKey()) + integer;
                                }
                            })
                            .map(new Func1<Integer, Pair<Long, Integer>>() {
                                @Override
                                public Pair<Long, Integer> call(Integer integer) {
                                    return new Pair<>(groupedObservable.getKey(), integer);
                                }
                            });
                }
            });
}

你能提出什么建议吗？

谢谢！

Answer 1

您可以使用scan返回中间值

plusItem
  .asObservable()
  .groupBy(id -> id)
  .subscribe(new Action1<GroupedObservable<Long, Long>>() {
    @Override
    public void call(GroupedObservable<Long, Long> groupedObservable) {
        System.out.println("Composed a group with key: " + groupedObservable.getKey());
        groupedObservable
                .map(id -> 1)
                .scan(0, (integer, integer2) -> integer + integer2)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer sum) {
                        System.out.println(Integer.toString(sum));
                    }
                });
    }
});