- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我正在尝试使用 rx Java 中的背压在我的 Android 应用程序中创建无限滚动。我希望它仅调用外部服务请求的次数(在调用 request(1)
之后)。但使用 flatmap 之后,每个 subscribe
都会加载 16 个页面。
在我的代码下方,具有预期结果。几乎每个测试都因第一个请求而失败(n=16)
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import rx.Observable;
import rx.observers.TestSubscriber;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import static java.util.Collections.emptyList;
import static org.mockito.Mockito.*;
import static rx.internal.util.UtilityFunctions.identity;
public class ServiceObservablesTest {
public static <T> Observable<List<T>> create(DataProvider<T> dataProvider) {
Observable<Observable<List<T>>> metaObservalble = Observable.create(subscriber -> {
AtomicInteger pageNumber = new AtomicInteger();
subscriber.setProducer(n -> {
// at subscribe rxJava makes request for 16 elements - probably because of flatMap
// after first request with 16 elements everything seems to work fine even if i ignore the 'n' param
Observable<List<T>> page = dataProvider.requestPage(pageNumber.getAndIncrement());
subscriber.onNext(page);
});
});
return metaObservalble.flatMap(identity()).takeWhile(page -> !page.isEmpty());
}
public interface DataProvider<T> {
Observable<List<T>> requestPage(int page);
}
private DataProvider provider;
@Before
public void setUp() throws Exception {
provider = Mockito.mock(DataProvider.class);
List<Object> list = Arrays.asList(new Object());
when(provider.requestPage(anyInt())).thenReturn(Observable.just(list));
}
@Test
public void shouldRequestOnlyFirstPageOnSubscribe() {
//given
TestSubscriber<List<Object>> subscriber = new TestSubscriber<>(1);
Observable<List<Object>> flightsObservable = create(provider);
//when
flightsObservable.subscribe(subscriber);
//then
subscriber.assertValueCount(1);
subscriber.assertNotCompleted();
verify(provider, times(1)).requestPage(0);
verify(provider, never()).requestPage(1);
}
@Test
public void shouldRequestNumberOfPagesSpecified() {
//given
int requested_pages = 5;
TestSubscriber<List<Object>> subscriber = new TestSubscriber<>(0);
Observable<List<Object>> flightsObservable = create(provider);
//when
flightsObservable.subscribe(subscriber);
subscriber.requestMore(requested_pages);
//then
subscriber.assertValueCount(requested_pages);
subscriber.assertNotCompleted();
for (int i = 0; i < requested_pages; i++) {
verify(provider, times(1)).requestPage(i);
}
verify(provider, never()).requestPage(requested_pages);
}
@Test
public void shouldCompleteAfterRetrievingEmptyResult() {
//given
int emptyPage = 2;
when(provider.requestPage(emptyPage)).thenReturn(Observable.just(emptyList()));
TestSubscriber<List<Object>> subscriber = new TestSubscriber<>(100);
Observable<List<Object>> flightsObservable = create(provider);
//when
flightsObservable.subscribe(subscriber);
//then
subscriber.assertValueCount(emptyPage);
subscriber.assertCompleted();
verify(provider, times(1)).requestPage(0); //requested at subscribe
for (int i = 1; i <= emptyPage; i++) {
verify(provider, times(1)).requestPage(i);
}
verify(provider, never()).requestPage(emptyPage + 1);
}
@Test
public void shouldRequestNextPageWhenRequestedMore() {
//given
TestSubscriber<List<Object>> subscriber = new TestSubscriber<>(1);
Observable<List<Object>> flightsObservable = create(provider);
//when
flightsObservable.subscribe(subscriber);
subscriber.requestMore(1);
//then
subscriber.assertValueCount(2);
verify(provider, times(1)).requestPage(0);
verify(provider, times(1)).requestPage(1);
verify(provider, never()).requestPage(2);
//when
subscriber.requestMore(1);
//then
subscriber.assertValueCount(3);
subscriber.assertNotCompleted();
verify(provider, times(1)).requestPage(0);
verify(provider, times(1)).requestPage(1);
verify(provider, times(1)).requestPage(2);
verify(provider, never()).requestPage(3);
}
@Test
public void shouldWorkWithMultipleSubscribers() {
//given
TestSubscriber<List<Object>> subscriber1 = new TestSubscriber<>(1);
TestSubscriber<List<Object>> subscriber2 = new TestSubscriber<>(1);
Observable<List<Object>> flightsObservable = create(provider);
//when
flightsObservable.subscribe(subscriber1);
flightsObservable.subscribe(subscriber2);
//then
subscriber1.assertValueCount(1);
subscriber2.assertValueCount(1);
verify(provider, times(2)).requestPage(0);
verify(provider, never()).requestPage(1);
//when
subscriber1.requestMore(1);
//then
subscriber1.assertValueCount(2);
subscriber2.assertValueCount(1);
verify(provider, times(2)).requestPage(0);
verify(provider, times(1)).requestPage(1);
verify(provider, never()).requestPage(2);
//when
subscriber2.requestMore(1);
//then
subscriber1.assertValueCount(2);
subscriber2.assertValueCount(2);
verify(provider, times(2)).requestPage(0);
verify(provider, times(2)).requestPage(1);
verify(provider, never()).requestPage(2);
}
}
最佳答案
背压旨在协商并发消费者生产者行为,并允许程序作者设置策略来解决当数据生成速率超过数据消耗速率时要做什么。
也就是说,您会看到组合可观察量的运算符,例如 merge
将为您提供与您所需的数据量不相符的请求量。合并时,外部 observable(Observables 中的 Observable)将始终收到 RxAndroid 上 16(RxJava 中 128)的请求。然后,当它接收到 List 的内部 Observables 时,每个内部 Observables 将收到一个基于下游订阅者请求量的请求。如果您尝试写 Observable<Observable<T>>
你将被迫写一个 OnSubscribe<Observable<List<T>>>
内部管理合并行为的函数,使其成为 Observable<List<T>>
而不是Observable<Observable<List<T>>
。编写此代码将强制您订阅数据提供者返回的可观察对象以解包并 onNext List<T>
。
我建议您将屏幕 y 位置映射到页尾事件,然后使用扫描将其转换为单调递增的数字,然后将该数字 concatMap 到对 DataProvider.requestPage()
的调用中。 .
screenYPositions
.map(this::isUninitializedOrNearEndOfPage)
.scan(1, (event, pageNumber) -> pageNumber + 1 )
.concatMap(dataProvider::requestPage)
.subscribe(testSubscriber);
关于rx-java - RxJava 背压和对生产者的调用次数,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/31967251/
我用 Cocoapods 创建了一个简单的项目并安装了 RxSwift 和 RxCocoa。 我写了这段代码: import UIKit import RxSwift class ViewContro
我对 Rx 真的很陌生,只是想知道 subscribeOn 的顺序如何影响 Observable //This will not print anything Observable.just("wha
我有一个 PublishSubject 每 X 秒发出一个信号,我想只考虑 Y 秒后发出的第一个项目。 例子 observable A 每秒发出一次“滴答声” observable B 应该每 5 秒
我有以下流。 Observable.just(Unit) // execute immediately .mergeWith(tryAgainRelay) // execute again w
我可以看到 RX 适用于 Android 和 UI 事件处理。我很难看到 RX 在后端提供了什么好处。 RX Java 是为后端处理而设计的,还是这个概念太过分了? 最佳答案 实际上,RxJava 最
我有一个关于什么是更好的 RxJava 模式以保持可观察状态的问题。 为简单起见,假设我们有一个 StateManager需要跟踪系统中某个状态(假设它是一个简单的 bool 标志)并以可观察的方式公
我遇到了以下问题。我有一个从具有 isMember 属性的服务器请求的基本对象。如果该属性为 true,我想向端点 1、2 和 3 发出请求。 如果该属性为 false,我只想向端点 1 发出请求。
我正在使用 RxJava,但也许能够翻译另一个实现的答案。 我有一个 Observable 发出一系列项目,我想将它们分成 10 组。如下所示: observable .buffer(10)
有没有办法计算 RxAndroid 中流中已处理的元素数量? 我有类似的事情: Observable.fromArray(new String[]{"these", "are", "my", "val
如果我的客户端断开连接(错误)超过 10 秒,我会尝试显示一个弹出窗口。但是,当重新连接时,我也会关闭弹出窗口(true)。如果为真,我需要立即关闭弹出窗口。 我认为我需要做的是根据值(假)进行去抖,
收到 onNext() 后如何自动退订? 现在我使用这个代码: rxObservable .compose(bindToLifecycle()) // unsubscribe automaticall
我在 Ettus x310 上有一个简单的 C++ 测试程序,以前可以用,但现在不行了。我试图简单地设置单个 USRP 的两个 channel 的两个中心频率。当我尝试在第二个 channel 上设置
我有一个返回 Observable 的网络调用,我有另一个网络调用,它不是依赖于第一个 Observable 的 rx,我需要以某种方式转换这一切都与 Rx 相关。 Observable respon
我正在尝试根据 Rxjava 中的某些条件创建组列表。 以下是我的回复: { "dates":[ { "date":18, "value":
RX 6800 XT是最近很强的一款显卡,让很多的用户都很喜爱,详细还有很多没有入手的用户对他不是很了解吧,下面就带来了RX 6800 XT评测、跑分、价格、参数、图片,快来一起看看吧。 RX
我有一个重复元素列表,比如: Observable source = Observable.just("A", "B", "A", "C", "C", "A"); 我想按它们的值以及它们出现的次数对它
如果我将 RX-main 包添加到 WPF 应用程序,我会安装以下依赖包: 接收接口(interface) 接收核心 Rx-Linq Rx 平台服务 主要接收 添加 Rx-main 后,可以安装两个与
com.myapp.test.debug E/MessageQueue-JNI:rx.exceptions.OnErrorNotImplementedException 在 rx.Observable
这个问题在这里已经有了答案: RxJava operator that emits if different from last (1 个回答) 关闭 5 年前。 我有一个 bool 型 Obser
首先,我在两个 shell 中同时运行 vnstat -tr -i eth0 和 iftop。奇怪的是,vnstat 中的 rx 比 iftop 中的峰值 rx 大得多。我该如何解决这个问题? vns
我是一名优秀的程序员,十分优秀!