Как я могу убедиться, что RXJava Observable излучает только и точно 1 элемент

Question

У меня есть сценарий использования, который я должен проверить, что RXJava Observable испускает только 1 элемент (не больше и не меньше), я знаю, что это асинхронный мир, и Observable должен определить период ожидания, чтобы закрыть потоки к тому времени мы сможем реализовать наблюдаемые испущенные 0 или 1 или более элементов, проблема, с которой я столкнулся, связана со значениями производительности и тайм-аутов, потому что это может повлиять на мое обслуживание клиентов.Вот мой текущий код, и я хочу его улучшить, избегая блокирующих операторов в новом коде:

import rx.Observable;
...
Observable<Items> oItems = service.getItems(requestParams);
List resultData = new ArrayList();
oItems .toBlocking().subscribe((arg) -> {
    resultData.add(arg.getId());
}
if (resultData.size() == 1) {
    return Observable.just(resultData.get(0));
}
return Observable.empty();

Спасибо за ваши мысли и идеи.

Answer 1

используйте .scan() так будет:

Observable<Boolean> isOneItemObservable = oItems
    .scan(0, (amount, o) -> amount + 1)
    .map( amount -> amount == 1)