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

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

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

Flowable.toObservable介绍

[英]Converts the current Flowable into a non-backpressured Observable. Backpressure: Observables don't support backpressure thus the current Flowable is consumed in an unbounded manner (by requesting Long.MAX_VALUE). Scheduler: toObservable does not operate by default on a particular Scheduler.
[中]将可流动电流转换为非背压可观测电流。背压：可观测数据不支持背压，因此当前可流动数据以无限方式消耗（通过请求Long.MAX_值）。Scheduler:toObservable默认情况下不会在特定计划程序上运行。

代码示例

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

@Override
  public ObservableSource<Object> apply(Flowable<Object> f) throws Exception {
    return f.toObservable();
  }
});

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

@Override
  public Observable<Integer> apply(Flowable<Integer> v) {
    return v.toObservable();
  }
})

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

@Override
public Observable<Integer> apply(Flowable<Integer> upstream) {
  return upstream.toObservable();
}

代码示例来源：origin: requery/requery

/**
 * Converts the result stream to a {@link io.reactivex.Observable}. When the observable
 * terminates this result instance will be closed.
 *
 * @return observable stream of the results of this query.
 */
@CheckReturnValue
public Observable<E> observable() {
  return flowable().toObservable();
}

代码示例来源：origin: TeamNewPipe/NewPipe

@Override
public void startLoading(boolean forceLoad) {
  super.startLoading(forceLoad);
  resetFragment();
  subscriptionService.getSubscription().toObservable()
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(getSubscriptionObserver());
}

代码示例来源：origin: TeamNewPipe/NewPipe

@SuppressLint("CheckResult")
private void deleteChannel (ChannelInfoItem selectedItem) {
  subscriptionService.subscriptionTable()
      .getSubscription(selectedItem.getServiceId(), selectedItem.getUrl())
      .toObservable()
      .observeOn(Schedulers.io())
      .subscribe(getDeleteObserver());
  Toast.makeText(activity, getString(R.string.channel_unsubscribed), Toast.LENGTH_SHORT).show();
}

代码示例来源：origin: TeamNewPipe/NewPipe

@Override
public View onCreateView(@NonNull LayoutInflater inflater, ViewGroup container, Bundle savedInstanceState) {
  View v = inflater.inflate(R.layout.select_channel_fragment, container, false);
  recyclerView = v.findViewById(R.id.items_list);
  recyclerView.setLayoutManager(new LinearLayoutManager(getContext()));
  SelectChannelAdapter channelAdapter = new SelectChannelAdapter();
  recyclerView.setAdapter(channelAdapter);
  progressBar = v.findViewById(R.id.progressBar);
  emptyView = v.findViewById(R.id.empty_state_view);
  progressBar.setVisibility(View.VISIBLE);
  recyclerView.setVisibility(View.GONE);
  emptyView.setVisibility(View.GONE);
  SubscriptionService subscriptionService = SubscriptionService.getInstance(getContext());
  subscriptionService.getSubscription().toObservable()
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(getSubscriptionObserver());
  return v;
}

代码示例来源：origin: TeamNewPipe/NewPipe

public PlayQueueAdapter(final Context context, final PlayQueue playQueue) {
  if (playQueue.getBroadcastReceiver() == null) {
    throw new IllegalStateException("Play Queue has not been initialized.");
  }
  this.playQueueItemBuilder = new PlayQueueItemBuilder(context);
  this.playQueue = playQueue;
  playQueue.getBroadcastReceiver().toObservable().subscribe(getReactor());
}

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

@Test
public void fromPublisherDispose() {
  TestHelper.checkDisposed(Flowable.just(1).toObservable());
}

代码示例来源：origin: spring-projects/spring-framework

void registerAdapters(ReactiveAdapterRegistry registry) {
    registry.registerReactiveType(
        ReactiveTypeDescriptor.multiValue(io.reactivex.Flowable.class, io.reactivex.Flowable::empty),
        source -> (io.reactivex.Flowable<?>) source,
        Flowable::fromPublisher
    );
    registry.registerReactiveType(
        ReactiveTypeDescriptor.multiValue(io.reactivex.Observable.class, io.reactivex.Observable::empty),
        source -> ((io.reactivex.Observable<?>) source).toFlowable(BackpressureStrategy.BUFFER),
        source -> io.reactivex.Flowable.fromPublisher(source).toObservable()
    );
    registry.registerReactiveType(
        ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.Single.class),
        source -> ((io.reactivex.Single<?>) source).toFlowable(),
        source -> io.reactivex.Flowable.fromPublisher(source).toObservable().singleElement().toSingle()
    );
    registry.registerReactiveType(
        ReactiveTypeDescriptor.singleOptionalValue(io.reactivex.Maybe.class, io.reactivex.Maybe::empty),
        source -> ((io.reactivex.Maybe<?>) source).toFlowable(),
        source -> io.reactivex.Flowable.fromPublisher(source).toObservable().singleElement()
    );
    registry.registerReactiveType(
        ReactiveTypeDescriptor.noValue(io.reactivex.Completable.class, io.reactivex.Completable::complete),
        source -> ((io.reactivex.Completable) source).toFlowable(),
        source -> io.reactivex.Flowable.fromPublisher(source).toObservable().ignoreElements()
    );
  }
}

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

@Test
public void toObservableRange() {
  Flowable.range(1, 5).toObservable().test().assertResult(1, 2, 3, 4, 5);
}

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

@Test
public void toObservableEmpty() {
  Flowable.empty().toObservable().test().assertResult();
}

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

@Test
public void toObservableJust() {
  Flowable.just(1).toObservable().test().assertResult(1);
}

代码示例来源：origin: TeamNewPipe/NewPipe

final Flowable<List<SearchHistoryEntry>> flowable = historyRecordManager
    .getRelatedSearches(query, 3, 25);
final Observable<List<SuggestionItem>> local = flowable.toObservable()
    .map(searchHistoryEntries -> {
      List<SuggestionItem> result = new ArrayList<>();

代码示例来源：origin: TeamNewPipe/NewPipe

private void monitorSubscription(final ChannelInfo info) {
  final Consumer<Throwable> onError = (Throwable throwable) -> {
      animateView(headerSubscribeButton, false, 100);
      showSnackBarError(throwable, UserAction.SUBSCRIPTION,
          NewPipe.getNameOfService(currentInfo.getServiceId()),
          "Get subscription status",
          0);
  };
  final Observable<List<SubscriptionEntity>> observable = subscriptionService.subscriptionTable()
      .getSubscription(info.getServiceId(), info.getUrl())
      .toObservable();
  disposables.add(observable
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(getSubscribeUpdateMonitor(info), onError));
  disposables.add(observable
      // Some updates are very rapid (when calling the updateSubscription(info), for example)
      // so only update the UI for the latest emission ("sync" the subscribe button's state)
      .debounce(100, TimeUnit.MILLISECONDS)
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe((List<SubscriptionEntity> subscriptionEntities) ->
          updateSubscribeButton(!subscriptionEntities.isEmpty())
          , onError));
}

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

@Test
public void toObservableError() {
  Flowable.error(new TestException()).toObservable().test().assertFailure(TestException.class);
}

代码示例来源：origin: org.springframework/spring-core

void registerAdapters(ReactiveAdapterRegistry registry) {
    registry.registerReactiveType(
        ReactiveTypeDescriptor.multiValue(io.reactivex.Flowable.class, io.reactivex.Flowable::empty),
        source -> (io.reactivex.Flowable<?>) source,
        Flowable::fromPublisher
    );
    registry.registerReactiveType(
        ReactiveTypeDescriptor.multiValue(io.reactivex.Observable.class, io.reactivex.Observable::empty),
        source -> ((io.reactivex.Observable<?>) source).toFlowable(BackpressureStrategy.BUFFER),
        source -> io.reactivex.Flowable.fromPublisher(source).toObservable()
    );
    registry.registerReactiveType(
        ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.Single.class),
        source -> ((io.reactivex.Single<?>) source).toFlowable(),
        source -> io.reactivex.Flowable.fromPublisher(source).toObservable().singleElement().toSingle()
    );
    registry.registerReactiveType(
        ReactiveTypeDescriptor.singleOptionalValue(io.reactivex.Maybe.class, io.reactivex.Maybe::empty),
        source -> ((io.reactivex.Maybe<?>) source).toFlowable(),
        source -> io.reactivex.Flowable.fromPublisher(source).toObservable().singleElement()
    );
    registry.registerReactiveType(
        ReactiveTypeDescriptor.noValue(io.reactivex.Completable.class, io.reactivex.Completable::complete),
        source -> ((io.reactivex.Completable) source).toFlowable(),
        source -> io.reactivex.Flowable.fromPublisher(source).toObservable().ignoreElements()
    );
  }
}

代码示例来源：origin: pwittchen/ReactiveNetwork

@Override public Observable<Connectivity> observeNetworkConnectivity(final Context context) {
 final String service = Context.CONNECTIVITY_SERVICE;
 final ConnectivityManager manager = (ConnectivityManager) context.getSystemService(service);
 networkCallback = createNetworkCallback(context);
 registerIdleReceiver(context);
 final NetworkRequest request =
   new NetworkRequest.Builder().addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
     .addCapability(NetworkCapabilities.NET_CAPABILITY_NOT_RESTRICTED)
     .build();
 manager.registerNetworkCallback(request, networkCallback);
 return connectivitySubject.toFlowable(BackpressureStrategy.LATEST).doOnCancel(new Action() {
  @Override public void run() {
   tryToUnregisterCallback(manager);
   tryToUnregisterReceiver(context);
  }
 }).startWith(Connectivity.create(context)).distinctUntilChanged().toObservable();
}

代码示例来源：origin: florent37/RxGps

@RequiresPermission(anyOf = {Manifest.permission.ACCESS_COARSE_LOCATION, Manifest.permission.ACCESS_FINE_LOCATION})
public Observable<Location> updates(@NonNull LocationRequest locationRequest, @NonNull Looper looper, long timeoutTime, @NonNull TimeUnit timeoutUnit) {
  return updatesInternal(locationRequest, looper, timeoutTime, timeoutUnit, BackpressureStrategy.MISSING).toObservable();
}

代码示例来源：origin: io.vertx/vertx-rx-java2

private Observable<String> inTransaction(Exception e) throws Exception {
  return client.rxGetConnection().flatMapObservable(conn -> {
   return rxInsertExtraFolks(conn)
    .andThen(uniqueNames(conn).toObservable())
    .compose(upstream -> e == null ? upstream : upstream.concatWith(Observable.error(e)))
    .compose(SQLClientHelper.txObservableTransformer(conn))
    .concatWith(rxAssertAutoCommit(conn).toObservable())
    .doFinally(conn::close);
  });
 }
}

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

Flowable

相关文章

    查看更多

    热门标签

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

    最新文章

    更多

    Flowable类方法

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