本文整理了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

@Override
  public Flowable<Integer> apply(Disposable subscription) {
    return Flowable.unsafeCreate(new Publisher<Integer>() {
      @Override
      public void subscribe(Subscriber<? super Integer> t1) {
        throw new TestException();
      }
    });
  }
};

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

@Override
  public Flowable<Object> apply(Object opening) {
    return Flowable.unsafeCreate(new Publisher<Object>() {
      @Override
      public void subscribe(Subscriber<? super Object> subscriber) {
        subscriber.onSubscribe(new BooleanSubscription());
        push(subscriber, new Object(), 100);
        complete(subscriber, 101);
      }
    });
  }
};

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

@Override
  public Flowable<Object> apply(Object opening) {
    return Flowable.unsafeCreate(new Publisher<Object>() {
      @Override
      public void subscribe(Subscriber<? super Object> subscriber) {
        subscriber.onSubscribe(new BooleanSubscription());
        push(subscriber, new Object(), 100);
        complete(subscriber, 101);
      }
    });
  }
};

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

@Override
  public Flowable<String> apply(final String v) {
    return Flowable.unsafeCreate(new Publisher<String>() {
      @Override
      public void subscribe(Subscriber<? super String> subscriber) {
        subscriber.onSubscribe(new BooleanSubscription());
        subscriber.onNext("value_after_map-" + v);
        subscriber.onComplete();
      }
    }).subscribeOn(scheduler);
  }
});

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

private static Flowable<String> infinite(final AtomicInteger produced) {
  return Flowable.unsafeCreate(new Publisher<String>() {
    @Override
    public void subscribe(Subscriber<? super String> s) {
      BooleanSubscription bs = new BooleanSubscription();
      s.onSubscribe(bs);
      while (!bs.isCancelled()) {
        s.onNext("onNext");
        produced.incrementAndGet();
      }
    }
  }).subscribeOn(Schedulers.newThread());
}

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

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

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

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

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

@Test
public void testSwitchShouldNotTriggerUnsubscribe() {
  final BooleanSubscription bs = new BooleanSubscription();
  Flowable.unsafeCreate(new Publisher<Long>() {
    @Override
    public void subscribe(final Subscriber<? super Long> subscriber) {
      subscriber.onSubscribe(bs);
      subscriber.onComplete();
    }
  }).switchIfEmpty(Flowable.<Long>never()).subscribe();
  assertFalse(bs.isCancelled());
}

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

@Test
@Ignore("Publisher.subscribe can't throw")
public void testThrownErrorHandling() {
  TestSubscriber<String> ts = new TestSubscriber<String>();
  Flowable.unsafeCreate(new Publisher<String>() {
    @Override
    public void subscribe(Subscriber<? super String> s) {
      throw new RuntimeException("fail");
    }
  }).subscribeOn(Schedulers.computation()).subscribe(ts);
  ts.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS);
  ts.assertTerminated();
}

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

@Test
public void testOnError() {
  TestSubscriber<String> ts = new TestSubscriber<String>();
  Flowable.unsafeCreate(new Publisher<String>() {
    @Override
    public void subscribe(Subscriber<? super String> s) {
      s.onSubscribe(new BooleanSubscription());
      s.onError(new RuntimeException("fail"));
    }
  }).subscribeOn(Schedulers.computation()).subscribe(ts);
  ts.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS);
  ts.assertTerminated();
}