io.reactivex.Flowable.unsafeCreate()方法的使用及代码示例

x33g5p2x  于2022-01-19 转载在 其他  
字(9.1k)|赞(0)|评价(0)|浏览(157)

本文整理了Java中io.reactivex.Flowable.unsafeCreate()方法的一些代码示例，展示了Flowable.unsafeCreate()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Flowable.unsafeCreate()方法的具体详情如下：
包路径：io.reactivex.Flowable
类名称：Flowable
方法名：unsafeCreate

Flowable.unsafeCreate介绍

[英]Create a Flowable by wrapping a Publisher which has to be implemented according to the Reactive-Streams specification by handling backpressure and cancellation correctly; no safeguards are provided by the Flowable itself. Backpressure: This operator is a pass-through for backpressure and the behavior is determined by the provided Publisher implementation. Scheduler: unsafeCreate by default doesn't operate on any particular Scheduler.
[中]通过包装发布者创建一个可流动的，必须根据反应流规范通过正确处理背压和取消来实现；流动设备本身不提供任何保障措施。背压：此运算符是背压的传递，其行为由提供的发布者实现决定。计划程序：默认情况下，未完成创建不会在任何特定计划程序上运行。

代码示例

代码示例来源：origin: ReactiveX/RxJava

  1. @Override
  2.   public Flowable<Integer> apply(Disposable subscription) {
  3.     return Flowable.unsafeCreate(new Publisher<Integer>() {
  4.       @Override
  5.       public void subscribe(Subscriber<? super Integer> t1) {
  6.         throw new TestException();
  7.       }
  8.     });
  9.   }
  10. };

代码示例来源：origin: ReactiveX/RxJava

  1. @Override
  2.   public Flowable<Object> apply(Object opening) {
  3.     return Flowable.unsafeCreate(new Publisher<Object>() {
  4.       @Override
  5.       public void subscribe(Subscriber<? super Object> subscriber) {
  6.         subscriber.onSubscribe(new BooleanSubscription());
  7.         push(subscriber, new Object(), 100);
  8.         complete(subscriber, 101);
  9.       }
  10.     });
  11.   }
  12. };

代码示例来源：origin: ReactiveX/RxJava

  1. @Override
  2.   public Flowable<Object> apply(Object opening) {
  3.     return Flowable.unsafeCreate(new Publisher<Object>() {
  4.       @Override
  5.       public void subscribe(Subscriber<? super Object> subscriber) {
  6.         subscriber.onSubscribe(new BooleanSubscription());
  7.         push(subscriber, new Object(), 100);
  8.         complete(subscriber, 101);
  9.       }
  10.     });
  11.   }
  12. };

代码示例来源：origin: ReactiveX/RxJava

  1. @Override
  2.   public Flowable<String> apply(final String v) {
  3.     return Flowable.unsafeCreate(new Publisher<String>() {
  4.       @Override
  5.       public void subscribe(Subscriber<? super String> subscriber) {
  6.         subscriber.onSubscribe(new BooleanSubscription());
  7.         subscriber.onNext("value_after_map-" + v);
  8.         subscriber.onComplete();
  9.       }
  10.     }).subscribeOn(scheduler);
  11.   }
  12. });

代码示例来源：origin: ReactiveX/RxJava

  1. @Test(expected = NullPointerException.class)
  2. public void createNull() {
  3.   Flowable.unsafeCreate(null);
  4. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Override
  2.   public void subscribe(Subscriber<? super Flowable<String>> op) {
  3.     op.onSubscribe(new BooleanSubscription());
  4.     op.onNext(Flowable.unsafeCreate(f1));
  5.     op.onNext(Flowable.unsafeCreate(f2));
  6.     op.onError(new NullPointerException("throwing exception in parent"));
  7.   }
  8. });

代码示例来源：origin: ReactiveX/RxJava

  1. @Override
  2. public void subscribe(Subscriber<? super Flowable<String>> subscriber) {
  3.   subscriber.onSubscribe(new BooleanSubscription());
  4.   // simulate what would happen in an observable
  5.   subscriber.onNext(Flowable.unsafeCreate(w1));
  6.   subscriber.onNext(Flowable.unsafeCreate(w2));
  7.   subscriber.onComplete();
  8. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test
  2. public void mergeIterable() {
  3.   final Flowable<String> f1 = Flowable.unsafeCreate(new TestSynchronousFlowable());
  4.   final Flowable<String> f2 = Flowable.unsafeCreate(new TestSynchronousFlowable());
  5.   List<Flowable<String>> listOfFlowables = new ArrayList<Flowable<String>>();
  6.   listOfFlowables.add(f1);
  7.   listOfFlowables.add(f2);
  8.   Flowable<String> m = Flowable.mergeDelayError(listOfFlowables);
  9.   m.subscribe(stringSubscriber);
  10.   verify(stringSubscriber, never()).onError(any(Throwable.class));
  11.   verify(stringSubscriber, times(1)).onComplete();
  12.   verify(stringSubscriber, times(2)).onNext("hello");
  13. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test
  2. public void testMergeList() {
  3.   final Flowable<String> f1 = Flowable.unsafeCreate(new TestSynchronousFlowable());
  4.   final Flowable<String> f2 = Flowable.unsafeCreate(new TestSynchronousFlowable());
  5.   List<Flowable<String>> listOfFlowables = new ArrayList<Flowable<String>>();
  6.   listOfFlowables.add(f1);
  7.   listOfFlowables.add(f2);
  8.   Flowable<String> m = Flowable.merge(listOfFlowables);
  9.   m.subscribe(stringSubscriber);
  10.   verify(stringSubscriber, never()).onError(any(Throwable.class));
  11.   verify(stringSubscriber, times(1)).onComplete();
  12.   verify(stringSubscriber, times(2)).onNext("hello");
  13. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test
  2. public void testMergeArray() {
  3.   final Flowable<String> f1 = Flowable.unsafeCreate(new TestSynchronousFlowable());
  4.   final Flowable<String> f2 = Flowable.unsafeCreate(new TestSynchronousFlowable());
  5.   Flowable<String> m = Flowable.merge(f1, f2);
  6.   m.subscribe(stringSubscriber);
  7.   verify(stringSubscriber, never()).onError(any(Throwable.class));
  8.   verify(stringSubscriber, times(2)).onNext("hello");
  9.   verify(stringSubscriber, times(1)).onComplete();
  10. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test
  2. public void testMergeArray() {
  3.   final Flowable<String> f1 = Flowable.unsafeCreate(new TestSynchronousFlowable());
  4.   final Flowable<String> f2 = Flowable.unsafeCreate(new TestSynchronousFlowable());
  5.   Flowable<String> m = Flowable.mergeDelayError(f1, f2);
  6.   m.subscribe(stringSubscriber);
  7.   verify(stringSubscriber, never()).onError(any(Throwable.class));
  8.   verify(stringSubscriber, times(2)).onNext("hello");
  9.   verify(stringSubscriber, times(1)).onComplete();
  10. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test
  2. public void testOriginFails() {
  3.   Subscriber<String> subscriber = TestHelper.mockSubscriber();
  4.   Flowable<String> origin = Flowable.unsafeCreate(new FuncWithErrors(1));
  5.   origin.subscribe(subscriber);
  6.   InOrder inOrder = inOrder(subscriber);
  7.   inOrder.verify(subscriber, times(1)).onNext("beginningEveryTime");
  8.   inOrder.verify(subscriber, times(1)).onError(any(RuntimeException.class));
  9.   inOrder.verify(subscriber, never()).onNext("onSuccessOnly");
  10.   inOrder.verify(subscriber, never()).onComplete();
  11. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test
  2. public void testMergeList() {
  3.   final Flowable<String> f1 = Flowable.unsafeCreate(new TestSynchronousFlowable());
  4.   final Flowable<String> f2 = Flowable.unsafeCreate(new TestSynchronousFlowable());
  5.   List<Flowable<String>> listOfFlowables = new ArrayList<Flowable<String>>();
  6.   listOfFlowables.add(f1);
  7.   listOfFlowables.add(f2);
  8.   Flowable<String> m = Flowable.mergeDelayError(Flowable.fromIterable(listOfFlowables));
  9.   m.subscribe(stringSubscriber);
  10.   verify(stringSubscriber, never()).onError(any(Throwable.class));
  11.   verify(stringSubscriber, times(1)).onComplete();
  12.   verify(stringSubscriber, times(2)).onNext("hello");
  13. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test(expected = IllegalArgumentException.class)
  2. public void unsafeWithFlowable() {
  3.   Flowable.unsafeCreate(Flowable.just(1));
  4. }

代码示例来源：origin: ReactiveX/RxJava

  1. private static Flowable<String> infinite(final AtomicInteger produced) {
  2.   return Flowable.unsafeCreate(new Publisher<String>() {
  3.     @Override
  4.     public void subscribe(Subscriber<? super String> s) {
  5.       BooleanSubscription bs = new BooleanSubscription();
  6.       s.onSubscribe(bs);
  7.       while (!bs.isCancelled()) {
  8.         s.onNext("onNext");
  9.         produced.incrementAndGet();
  10.       }
  11.     }
  12.   }).subscribeOn(Schedulers.newThread());
  13. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test
  2. public void testMergeArrayWithThreading() {
  3.   final TestASynchronousFlowable f1 = new TestASynchronousFlowable();
  4.   final TestASynchronousFlowable f2 = new TestASynchronousFlowable();
  5.   Flowable<String> m = Flowable.merge(Flowable.unsafeCreate(f1), Flowable.unsafeCreate(f2));
  6.   TestSubscriber<String> ts = new TestSubscriber<String>(stringSubscriber);
  7.   m.subscribe(ts);
  8.   ts.awaitTerminalEvent();
  9.   ts.assertNoErrors();
  10.   verify(stringSubscriber, never()).onError(any(Throwable.class));
  11.   verify(stringSubscriber, times(2)).onNext("hello");
  12.   verify(stringSubscriber, times(1)).onComplete();
  13. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test
  2. public void testMultipleSubscribes() throws InterruptedException, ExecutionException {
  3.   final TestAsyncErrorObservable o = new TestAsyncErrorObservable("one", "two", null, "three");
  4.   Flowable<Notification<String>> m = Flowable.unsafeCreate(o).materialize();
  5.   assertEquals(3, m.toList().toFuture().get().size());
  6.   assertEquals(3, m.toList().toFuture().get().size());
  7. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test
  2. public void testSwitchShouldNotTriggerUnsubscribe() {
  3.   final BooleanSubscription bs = new BooleanSubscription();
  4.   Flowable.unsafeCreate(new Publisher<Long>() {
  5.     @Override
  6.     public void subscribe(final Subscriber<? super Long> subscriber) {
  7.       subscriber.onSubscribe(bs);
  8.       subscriber.onComplete();
  9.     }
  10.   }).switchIfEmpty(Flowable.<Long>never()).subscribe();
  11.   assertFalse(bs.isCancelled());
  12. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test
  2. @Ignore("Publisher.subscribe can't throw")
  3. public void testThrownErrorHandling() {
  4.   TestSubscriber<String> ts = new TestSubscriber<String>();
  5.   Flowable.unsafeCreate(new Publisher<String>() {
  6.     @Override
  7.     public void subscribe(Subscriber<? super String> s) {
  8.       throw new RuntimeException("fail");
  9.     }
  10.   }).subscribeOn(Schedulers.computation()).subscribe(ts);
  11.   ts.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS);
  12.   ts.assertTerminated();
  13. }

代码示例来源：origin: ReactiveX/RxJava

  1. @Test
  2. public void testOnError() {
  3.   TestSubscriber<String> ts = new TestSubscriber<String>();
  4.   Flowable.unsafeCreate(new Publisher<String>() {
  5.     @Override
  6.     public void subscribe(Subscriber<? super String> s) {
  7.       s.onSubscribe(new BooleanSubscription());
  8.       s.onError(new RuntimeException("fail"));
  9.     }
  10.   }).subscribeOn(Schedulers.computation()).subscribe(ts);
  11.   ts.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS);
  12.   ts.assertTerminated();
  13. }

内容来源于网络，如有侵权，请联系作者删除！

Flowable

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多

    Flowable类方法

    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com