本文整理了Java中io.reactivex.Flowable.concatWith()
方法的一些代码示例,展示了Flowable.concatWith()
的具体用法。这些代码示例主要来源于Github
/Stackoverflow
/Maven
等平台,是从一些精选项目中提取出来的代码,具有较强的参考意义,能在一定程度帮忙到你。Flowable.concatWith()
方法的具体详情如下:
包路径:io.reactivex.Flowable
类名称:Flowable
方法名:concatWith
[英]Returns a Flowable that emits items from this Flowable and when it completes normally, the other CompletableSource is subscribed to and the returned Flowable emits its terminal events.
Backpressure: The operator does not interfere with backpressure between the current Flowable and the downstream consumer (i.e., acts as pass-through). When the operator switches to the Completable, backpressure is no longer present because Completable doesn't have items to apply backpressure to. Scheduler: concatWith does not operate by default on a particular Scheduler.
[中]返回一个从该流表中发出项的流表,当它正常完成时,订阅另一个CompletableSource,返回的流表发出其终端事件。
背压:操作员不会干扰可流动电流和下游耗电元件之间的背压(即,作为直通)。当操作员切换到Completable时,背压不再存在,因为Completable没有可应用背压的项目。调度程序:默认情况下,concatWith不会在特定调度程序上运行。
代码示例来源:origin: ReactiveX/RxJava
@Override
public Publisher<Integer> createPublisher(long elements) {
return
Flowable.range(1, Math.max(0, (int)elements - 1))
.concatWith(Single.just((int)elements))
;
}
}
代码示例来源:origin: ReactiveX/RxJava
@Override
public Flowable<String> apply(Resource resource) {
return Flowable.fromArray(resource.getTextFromWeb().split(" "))
.concatWith(Flowable.<String>error(new RuntimeException()));
}
};
代码示例来源:origin: ReactiveX/RxJava
@Test
public void skipLastTimedCustomSchedulerDelayError() {
Flowable.just(1).concatWith(Flowable.just(2).delay(500, TimeUnit.MILLISECONDS))
.skipLast(300, TimeUnit.MILLISECONDS, Schedulers.io(), true)
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normal() {
final TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
Flowable.range(1, 5)
.concatWith(Single.just(100))
.subscribe(ts);
ts.assertResult(1, 2, 3, 4, 5, 100);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normalNonDelayErrorOuter() {
Flowable.range(1, 10).concatWith(Flowable.<Integer>error(new TestException()))
.flatMapCompletable(new Function<Integer, CompletableSource>() {
@Override
public CompletableSource apply(Integer v) throws Exception {
return Completable.complete();
}
}, false, Integer.MAX_VALUE)
.test()
.assertFailure(TestException.class);
}
代码示例来源:origin: ReactiveX/RxJava
@SuppressWarnings("unchecked")
@Test
public void concatIterableDelayErrorWithError() {
Flowable.concatDelayError(
Arrays.asList(Flowable.just(1), Flowable.just(2),
Flowable.just(3).concatWith(Flowable.<Integer>error(new TestException())),
Flowable.just(4)))
.test()
.assertFailure(TestException.class, 1, 2, 3, 4);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void skipLastTimedDefaultScheduler() {
Flowable.just(1).concatWith(Flowable.just(2).delay(500, TimeUnit.MILLISECONDS))
.skipLast(300, TimeUnit.MILLISECONDS)
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void delayWithTimeDelayError() throws Exception {
Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
.delay(100, TimeUnit.MILLISECONDS, true)
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertFailure(TestException.class, 1);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void mergeDelayError3WithError() {
Flowable.mergeDelayError(
Flowable.just(1),
Flowable.just(2).concatWith(Flowable.<Integer>error(new TestException())),
Flowable.just(3)
)
.test()
.assertFailure(TestException.class, 1, 2, 3);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void delayError() {
TestSubscriber<Integer> ts = Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
.onBackpressureBuffer(true)
.test(0L)
.assertEmpty();
ts.request(1);
ts.assertFailure(TestException.class, 1);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void error() {
new FlowablePublishMulticast<Integer, Integer>(Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException())),
Functions.<Flowable<Integer>>identity(), 16, true)
.test()
.assertFailure(TestException.class, 1);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void cancelOther() {
CompletableSubject other = CompletableSubject.create();
TestSubscriber<Object> ts = Flowable.empty()
.concatWith(other)
.test();
assertTrue(other.hasObservers());
ts.cancel();
assertFalse(other.hasObservers());
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void scalarAndRangeBackpressured() {
TestSubscriber<Integer> ts = TestSubscriber.create(0);
Flowable.just(1).concatWith(Flowable.range(2, 3)).subscribe(ts);
ts.assertNoValues();
ts.request(5);
ts.assertValues(1, 2, 3, 4);
ts.assertComplete();
ts.assertNoErrors();
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void rangeAndEmptyBackpressured() {
TestSubscriber<Integer> ts = TestSubscriber.create(0);
Flowable.range(1, 2).concatWith(Flowable.<Integer>empty()).subscribe(ts);
ts.assertNoValues();
ts.request(5);
ts.assertValues(1, 2);
ts.assertComplete();
ts.assertNoErrors();
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void emptyAndScalarBackpressured() {
TestSubscriber<Integer> ts = TestSubscriber.create(0);
Flowable.<Integer>empty().concatWith(Flowable.just(1)).subscribe(ts);
ts.assertNoValues();
ts.request(5);
ts.assertValue(1);
ts.assertComplete();
ts.assertNoErrors();
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void scalarAndEmptyBackpressured() {
TestSubscriber<Integer> ts = TestSubscriber.create(0);
Flowable.just(1).concatWith(Flowable.<Integer>empty()).subscribe(ts);
ts.assertNoValues();
ts.request(5);
ts.assertValue(1);
ts.assertComplete();
ts.assertNoErrors();
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void concatObservableDelayErrorWithError() {
Flowable.concatDelayError(
Flowable.just(Flowable.just(1), Flowable.just(2),
Flowable.just(3).concatWith(Flowable.<Integer>error(new TestException())),
Flowable.just(4)))
.test()
.assertFailure(TestException.class, 1, 2, 3, 4);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void concatObservableDelayErrorTillEnd() {
Flowable.concatDelayError(
Flowable.just(Flowable.just(1), Flowable.just(2),
Flowable.just(3).concatWith(Flowable.<Integer>error(new TestException())),
Flowable.just(4)), 2, true)
.test()
.assertFailure(TestException.class, 1, 2, 3, 4);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void testInnerBackpressureWithAlignedBoundaries() {
TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
Flowable.range(0, Flowable.bufferSize() * 2)
.concatWith(Flowable.range(0, Flowable.bufferSize() * 2))
.observeOn(Schedulers.computation()) // observeOn has a backpressured RxRingBuffer
.subscribe(ts);
ts.awaitTerminalEvent();
ts.assertNoErrors();
assertEquals(Flowable.bufferSize() * 4, ts.valueCount());
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void observers() {
PublishProcessor<Integer> pp = PublishProcessor.create();
FlowableCache<Integer> cache = (FlowableCache<Integer>)Flowable.range(1, 5).concatWith(pp).cache();
assertFalse(cache.hasSubscribers());
assertEquals(0, cache.cachedEventCount());
TestSubscriber<Integer> ts = cache.test();
assertTrue(cache.hasSubscribers());
assertEquals(5, cache.cachedEventCount());
pp.onComplete();
ts.assertResult(1, 2, 3, 4, 5);
}
内容来源于网络,如有侵权,请联系作者删除!