android - 如何在RxAndroid 2.0中正确过滤/计数？

给定返回可观察票据列表的存储库：

Observable<List<Bill>> getBills();

我只想在一张或多张账单未付的情况下显示一个视图。我正在尝试以下代码：

repository.getBills()
    .flatMapIterable(bills -> bills)
    .filter(bill -> !bill.isPaid())
    .count()
    .map(count -> count > 0)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(overdue -> {
        if (!overdue) return;
        mView.showWarning();
    });

但是onSuccess和onError都没有被调用。
我知道存储库至少包含一个过期项，因为以下代码打印未付款的账单：

repository.getBills()
    .subscribeOn(Schedulers.io())
    .flatMapIterable(bills -> bills)
    .filter(bill -> !bill.isPaid())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        bill -> Timber.d(bill.toString()),
        e -> Timber.e(e.getMessage(), e),
        () -> Timber.d("Completed")
    );

最佳答案

TL；博士：
如果它从未完成，count将不起作用。如果要检查未付项目，可以使用any运算符、takeUntil或takeWhile。跳到这个答案的第三项。
完整答案：
有三个可能的问题：
它可能发生在showWarning()内部。我运行了下面的代码
打印日期：

findViewById(R.id.doSomething).setOnClickListener(v -> {
    clearWarning();
    getBills()
        .subscribeOn(Schedulers.io())
        .flatMapIterable(bills -> bills)
        .filter(bill -> !bill.isPaid())
        .count()
        .map(count -> count > 0)
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            due -> {
              if (!due) return;
              showWarning();
            }
        );
});

使用以下getBills()：

private Observable<List<Bill>> getBills() {
    Bill sampleBill = new Bill();
    List<Bill> bills = new ArrayList<>(1);
    bills.add(sampleBill);
    return Observable.just(bills);
}

Bill是一个仅在false中返回isPaid()的伪类：

class Bill {
    public boolean isPaid() {
        return false;
    }
}

我有一个TextView用于showWarning()和clearWarning()的
正确打印“到期”
另一个选项是getBills()内部的问题。是不是
源代码成功完成（我的意思是，它调用onComplete()）？您可以手动调用它或使用Single，但随后需要在toObservable()之前调用flatMapIterable()。
根据documentation：
如果源Observable因错误而终止，count将通过
此错误通知不会首先发出项目。如果
可观测源根本不终止，计数也不会发出
不终止的项目。
如果你不能在getBills()中改变观测值，但只需要
当有未付账单时，您可以使用takeWhile，
takeUntil或any：

findViewById(R.id.doSomething).setOnClickListener(v -> {
  clearWarning();
  getBills()
      .flatMapIterable(bills -> bills)
      .takeUntil(bill -> !bill.isPaid())
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(
          bill -> {
            Log.d("POTATO", "Number: " + bill.getNumber() + " Paid: " + bill.isPaid());
          },
          e -> Log.e("POTATO", "Error"),
          () -> {
            Log.d("POTATO", "Complete");
            showWarning();
          }
      );
});   }

在本例中，我将getBills()更改为永不完成：

 private Observable<List<Bill>> getBills() {
    List<Bill> bills = new ArrayList<>();
    bills.add(new Bill(1, true));
    bills.add(new Bill(2, true));
    bills.add(new Bill(3, false));
    bills.add(new Bill(4, false));
    return Observable.create(
        emitter -> emitter.onNext(bills)
    );
  }

现在Bill类是
跟随：

class Bill {

  private final int number;
  private boolean isPaid;

  Bill(int number, boolean isPaid) {
    this.number = number;
    this.isPaid = isPaid;
  }

  int getNumber() {
    return number;
  }

  boolean isPaid() {
    return isPaid;
  }
}

Log打印
数字：1支付：真
数字：2付费：真
编号：3已付：假
完成
然后调用showWarning()。当然，takeWhile的返回值应该与takeUntil相反。当有未付款项时，两者都将接受已付款项并停止，但takeWhile甚至不会发出未付款项（Number: 3: Paid: false不会出现在日志中，但它将在2之后立即完成）。any将得到任何符合条件的项目，这对您来说可能就足够了。请注意，这是一个完全不同的解决方案。如果源只发出付费项目，它将永远不会完成。但无论如何你应该在某处有一个unsubscribe。