• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Java ScalarSynchronousObservable类代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Java中rx.internal.util.ScalarSynchronousObservable的典型用法代码示例。如果您正苦于以下问题:Java ScalarSynchronousObservable类的具体用法?Java ScalarSynchronousObservable怎么用?Java ScalarSynchronousObservable使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。



ScalarSynchronousObservable类属于rx.internal.util包,在下文中一共展示了ScalarSynchronousObservable类的18个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Java代码示例。

示例1: handleScalarSynchronousObservable

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
private void handleScalarSynchronousObservable(ScalarSynchronousObservable<? extends T> t) {
    // fast-path for scalar, synchronous values such as Observable.from(int)
    /**
     * Without this optimization:
     * 
     * <pre> {@code
     * Benchmark                                          (size)   Mode   Samples        Score  Score error    Units
     * r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1         1  thrpt         5  2,418,452.409   130572.665    ops/s
     * r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1      1000  thrpt         5     5,690.456       94.958    ops/s
     * r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5          takes too long
     * 
     * With this optimization:
     * 
     * r.o.OperatorMergePerf.merge1SyncStreamOfN               1  thrpt         5  5,475,300.198   156741.334    ops/s
     * r.o.OperatorMergePerf.merge1SyncStreamOfN            1000  thrpt         5    68,932.278     1311.023    ops/s
     * r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5       64.405        0.611    ops/s
     * } </pre>
     * 
     */
    if (mergeProducer.requested == Long.MAX_VALUE) {
        handleScalarSynchronousObservableWithoutRequestLimits(t);
    } else {
        handleScalarSynchronousObservableWithRequestLimits(t);
    }
}
 
开发者ID:OpenNTF,项目名称:org.openntf.domino,代码行数:26,代码来源:OperatorMerge.java


示例2: handleScalarSynchronousObservableWithoutRequestLimits

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
private void handleScalarSynchronousObservableWithoutRequestLimits(ScalarSynchronousObservable<? extends T> t) {
    T value = t.get();
    if (getEmitLock()) {
        try {
            actual.onNext(value);
            return;
        } finally {
            if (releaseEmitLock()) {
                drainQueuesIfNeeded();
            }
            request(1);
        }
    } else {
        initScalarValueQueueIfNeeded();
        try {
            scalarValueQueue.onNext(value);
        } catch (MissingBackpressureException e) {
            onError(e);
        }
        return;
    }
}
 
开发者ID:OpenNTF,项目名称:org.openntf.domino,代码行数:23,代码来源:OperatorMerge.java


示例3: concatMapDelayError

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
@Experimental
public final <R> Observable<R> concatMapDelayError(Func1<? super T, ? extends Observable<? extends R>> func) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarFlatMap(func);
    }
    return create(new OnSubscribeConcatMap(this, func, 2, 2));
}
 
开发者ID:JackChan1999,项目名称:letv,代码行数:8,代码来源:Observable.java


示例4: flatMap

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
@Beta
public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func, int maxConcurrent) {
    if (getClass() == ScalarSynchronousObservable.class) {
        return ((ScalarSynchronousObservable) this).scalarFlatMap(func);
    }
    return merge(map(func), maxConcurrent);
}
 
开发者ID:JackChan1999,项目名称:letv,代码行数:8,代码来源:Observable.java


示例5: onNext

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public void onNext(Observable<? extends T> t) {
    if (t != null) {
        if (t instanceof ScalarSynchronousObservable) {
            tryEmit(((ScalarSynchronousObservable) t).get());
            return;
        }
        long j = this.uniqueId;
        this.uniqueId = 1 + j;
        InnerSubscriber<T> inner = new InnerSubscriber(this, j);
        addInner(inner);
        t.unsafeSubscribe(inner);
        emit();
    }
}
 
开发者ID:JackChan1999,项目名称:boohee_v5.6,代码行数:15,代码来源:OperatorMerge.java


示例6: testIsNotScalarSynchronousObservableAfterFlatMap

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
@Test
public void testIsNotScalarSynchronousObservableAfterFlatMap() {
    Observable<Integer> o = Observable.just(1).flatMap(new Func1<Integer, Observable<Integer>>() {

        @Override
        public Observable<Integer> call(Integer n) {
            return Observable.range(1, 100);
        }});
    Assert.assertFalse(o instanceof ScalarSynchronousObservable);
}
 
开发者ID:amsa-code,项目名称:risky,代码行数:11,代码来源:MergeTest.java


示例7: onNext

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
@Override
public void onNext(Observable<? extends T> t) {
    if (t instanceof ScalarSynchronousObservable) {
        handleScalarSynchronousObservable((ScalarSynchronousObservable)t);
    } else {
        if (t == null || isUnsubscribed()) {
            return;
        }
        synchronized (this) {
            // synchronized here because `wip` can be concurrently changed by children Observables
            wip++;
        }
        handleNewSource(t);
    }
}
 
开发者ID:OpenNTF,项目名称:org.openntf.domino,代码行数:16,代码来源:OperatorMerge.java


示例8: handleScalarSynchronousObservableWithRequestLimits

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
private void handleScalarSynchronousObservableWithRequestLimits(ScalarSynchronousObservable<? extends T> t) {
    if (getEmitLock()) {
        boolean emitted = false;
        try {
            long r = mergeProducer.requested;
            if (r > 0) {
                emitted = true;
                actual.onNext(t.get());
                MergeProducer.REQUESTED.decrementAndGet(mergeProducer);
                // we handle this Observable without ever incrementing the wip or touching other machinery so just return here
                return;
            }
        } finally {
            if (releaseEmitLock()) {
                drainQueuesIfNeeded();
            }
            if (emitted) {
                request(1);
            }
        }
    }

    // if we didn't return above we need to enqueue
    // enqueue the values for later delivery
    initScalarValueQueueIfNeeded();
    try {
        scalarValueQueue.onNext(t.get());
    } catch (MissingBackpressureException e) {
        onError(e);
    }
}
 
开发者ID:OpenNTF,项目名称:org.openntf.domino,代码行数:32,代码来源:OperatorMerge.java


示例9: just

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public static <T> Observable<T> just(T value) {
    return ScalarSynchronousObservable.create(value);
}
 
开发者ID:JackChan1999,项目名称:letv,代码行数:4,代码来源:Observable.java


示例10: merge

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public static <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source) {
    if (source.getClass() == ScalarSynchronousObservable.class) {
        return ((ScalarSynchronousObservable) source).scalarFlatMap(UtilityFunctions.identity());
    }
    return source.lift(OperatorMerge.instance(false));
}
 
开发者ID:JackChan1999,项目名称:letv,代码行数:7,代码来源:Observable.java


示例11: concatMap

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public final <R> Observable<R> concatMap(Func1<? super T, ? extends Observable<? extends R>> func) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarFlatMap(func);
    }
    return create(new OnSubscribeConcatMap(this, func, 2, 0));
}
 
开发者ID:JackChan1999,项目名称:letv,代码行数:7,代码来源:Observable.java


示例12: observeOn

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarScheduleOn(scheduler);
    }
    return lift(new OperatorObserveOn(scheduler, delayError, bufferSize));
}
 
开发者ID:JackChan1999,项目名称:letv,代码行数:7,代码来源:Observable.java


示例13: subscribeOn

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public final Observable<T> subscribeOn(Scheduler scheduler) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarScheduleOn(scheduler);
    }
    return create(new OperatorSubscribeOn(this, scheduler));
}
 
开发者ID:JackChan1999,项目名称:letv,代码行数:7,代码来源:Observable.java


示例14: just

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public static final <T> Observable<T> just(T value) {
    return ScalarSynchronousObservable.create(value);
}
 
开发者ID:JackChan1999,项目名称:boohee_v5.6,代码行数:4,代码来源:Observable.java


示例15: merge

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public static final <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source) {
    if (source.getClass() == ScalarSynchronousObservable.class) {
        return ((ScalarSynchronousObservable) source).scalarFlatMap(UtilityFunctions.identity());
    }
    return source.lift(OperatorMerge.instance(false));
}
 
开发者ID:JackChan1999,项目名称:boohee_v5.6,代码行数:7,代码来源:Observable.java


示例16: observeOn

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public final Observable<T> observeOn(Scheduler scheduler) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarScheduleOn(scheduler);
    }
    return lift(new OperatorObserveOn(scheduler));
}
 
开发者ID:JackChan1999,项目名称:boohee_v5.6,代码行数:7,代码来源:Observable.java


示例17: subscribeOn

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public final Observable<T> subscribeOn(Scheduler scheduler) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarScheduleOn(scheduler);
    }
    return nest().lift(new OperatorSubscribeOn(scheduler));
}
 
开发者ID:JackChan1999,项目名称:boohee_v5.6,代码行数:7,代码来源:Observable.java


示例18: just

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
/**
 * Returns an Observable that emits a single item and then completes.
 * <p>
 * <img width="640" height="310" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/just.png" alt="">
 * <p>
 * To convert any object into an Observable that emits that object, pass that object into the {@code just}
 * method.
 * <p>
 * This is similar to the {@link #from(java.lang.Object[])} method, except that {@code from} will convert
 * an {@link Iterable} object into an Observable that emits each of the items in the Iterable, one at a
 * time, while the {@code just} method converts an Iterable into an Observable that emits the entire
 * Iterable as a single item.
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code just} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 * 
 * @param value
 *            the item to emit
 * @param <T>
 *            the type of that item
 * @return an Observable that emits {@code value} as a single item and then completes
 * @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#just">RxJava wiki: just</a>
 */
public final static <T> Observable<T> just(final T value) {
    return ScalarSynchronousObservable.create(value);
}
 
开发者ID:davidmoten,项目名称:rxjava-parallel,代码行数:28,代码来源:Observable.java



注:本文中的rx.internal.util.ScalarSynchronousObservable类示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Java LanguageSpecific类代码示例发布时间:2022-05-22
下一篇:
Java ConsoleColorProvider类代码示例发布时间:2022-05-22
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap