本文整理了Java中io.reactivex.schedulers.Schedulers
类的一些代码示例,展示了Schedulers
类的具体用法。这些代码示例主要来源于Github
/Stackoverflow
/Maven
等平台,是从一些精选项目中提取出来的代码,具有较强的参考意义,能在一定程度帮忙到你。Schedulers
类的具体详情如下:
包路径:io.reactivex.schedulers.Schedulers
类名称:Schedulers
[英]Static factory methods for returning standard Scheduler instances.
The initial and runtime values of the various scheduler types can be overridden via the RxJavaPlugins.setInit(scheduler name)SchedulerHandler() and RxJavaPlugins.set(scheduler name)SchedulerHandler() respectively.
Supported system properties ( System.getProperty()):
代码示例来源:origin: ReactiveX/RxJava
@Override
public ObservableSource<Integer> apply(Integer i) throws Exception {
return i == 3 ? Observable.just(i) : Observable
.just(i)
.delay(1, TimeUnit.MILLISECONDS, Schedulers.io());
}
})
代码示例来源:origin: ReactiveX/RxJava
/**
* Shuts down the standard Schedulers.
* <p>The operation is idempotent and thread-safe.
*/
public static void shutdown() {
computation().shutdown();
io().shutdown();
newThread().shutdown();
single().shutdown();
trampoline().shutdown();
SchedulerPoolFactory.shutdown();
}
代码示例来源:origin: ReactiveX/RxJava
@Override
protected Scheduler getScheduler() {
return Schedulers.from(executor);
}
代码示例来源:origin: ReactiveX/RxJava
@Override
public Observable<Integer> apply(Integer t1) {
return composer(Observable.range(t1 * 10, 2), subscriptionCount, m)
.subscribeOn(Schedulers.computation());
}
}, new BiFunction<Integer, Integer, Integer>() {
代码示例来源:origin: ReactiveX/RxJava
@Override
protected Scheduler getScheduler() {
return Schedulers.newThread();
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void doubleObserveOn() {
Flowable.just(1).hide()
.observeOn(Schedulers.computation())
.observeOn(Schedulers.single())
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1);
}
代码示例来源:origin: ReactiveX/RxJava
public ObservableRefCount(ConnectableObservable<T> source) {
this(source, 1, 0L, TimeUnit.NANOSECONDS, Schedulers.trampoline());
}
代码示例来源:origin: ReactiveX/RxJava
@Override
public SingleSource<Integer> apply(Integer v) throws Exception {
return Single.just(v).subscribeOn(Schedulers.computation());
}
})
代码示例来源:origin: ReactiveX/RxJava
@Override
public Observable<String> apply(final GroupedObservable<Integer, Integer> group) {
return group.subscribeOn(Schedulers.newThread()).map(new Function<Integer, String>() {
@Override
public String apply(Integer t1) {
System.out.println("Received: " + t1 + " on group : " + group.getKey());
return "first groups: " + t1;
}
});
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void doubleObserveOnError() {
Flowable.error(new TestException())
.observeOn(Schedulers.computation())
.observeOn(Schedulers.single())
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertFailure(TestException.class);
}
代码示例来源:origin: ReactiveX/RxJava
public FlowableRefCount(ConnectableFlowable<T> source) {
this(source, 1, 0L, TimeUnit.NANOSECONDS, Schedulers.trampoline());
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void periodicDirectTaskRaceIO() throws Exception {
final Scheduler scheduler = Schedulers.io();
for (int i = 0; i < 100; i++) {
final Disposable d = scheduler.schedulePeriodicallyDirect(
Functions.EMPTY_RUNNABLE, 0, 0, TimeUnit.MILLISECONDS);
Thread.sleep(1);
d.dispose();
}
}
代码示例来源:origin: ReactiveX/RxJava
/**
* Starts the standard Schedulers.
* <p>The operation is idempotent and thread-safe.
*/
public static void start() {
computation().start();
io().start();
newThread().start();
single().start();
trampoline().start();
SchedulerPoolFactory.start();
}
代码示例来源:origin: ReactiveX/RxJava
@Override
public SingleSource<Integer> apply(Integer v) throws Exception {
return Single.just(v).subscribeOn(Schedulers.computation());
}
})
代码示例来源:origin: ReactiveX/RxJava
private static Observable<String> infinite(final AtomicInteger produced) {
return Observable.unsafeCreate(new ObservableSource<String>() {
@Override
public void subscribe(Observer<? super String> observer) {
Disposable bs = Disposables.empty();
observer.onSubscribe(bs);
while (!bs.isDisposed()) {
observer.onNext("onNext");
produced.incrementAndGet();
}
}
}).subscribeOn(Schedulers.newThread());
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void doubleObserveOnErrorConditional() {
Flowable.error(new TestException())
.observeOn(Schedulers.computation())
.distinct()
.observeOn(Schedulers.single())
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertFailure(TestException.class);
}
代码示例来源:origin: ReactiveX/RxJava
@Override
protected Scheduler getScheduler() {
return Schedulers.from(executor, true);
}
代码示例来源:origin: ReactiveX/RxJava
@Override
protected Scheduler getScheduler() {
return Schedulers.trampoline();
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void overrideIoScheduler() {
try {
RxJavaPlugins.setIoSchedulerHandler(replaceWithImmediate);
assertSame(ImmediateThinScheduler.INSTANCE, Schedulers.io());
} finally {
RxJavaPlugins.reset();
}
// make sure the reset worked
assertNotSame(ImmediateThinScheduler.INSTANCE, Schedulers.io());
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void moreThanMaxWorkers() {
final List<Worker> list = new ArrayList<Worker>();
SchedulerMultiWorkerSupport mws = (SchedulerMultiWorkerSupport)Schedulers.computation();
mws.createWorkers(max * 2, new WorkerCallback() {
@Override
public void onWorker(int i, Worker w) {
list.add(w);
}
});
assertEquals(max * 2, list.size());
}
内容来源于网络,如有侵权,请联系作者删除!